Sleepy Events Listener (for Sleepy v5)

sleepy-v5-listener.py

'''
Sleepy Events Listener by wyf9 (for Sleepy v5)
Dependencies:
    $ uv add requests sseclient-py deepdiff rich
Or run it directly:
    $ uv run --with httpx-sse --with deepdiff --with rich sleepy-v5-listener.py http://sleepy.host # REPLACE HOST
'''

import asyncio
import json
from datetime import datetime
from sys import argv

import httpx
from deepdiff import DeepDiff
from httpx_sse import aconnect_sse
from rich.console import Console

try:
    HOST = argv[1]
except Exception:
    HOST = input('Input Host: ')
if not HOST.startswith(('http://', 'https://')):
    print('Invaild host!')
    exit(1)
HOST = HOST.rstrip('/')

console = Console()
old_event = None


async def main():
    delay = 1

    async with httpx.AsyncClient(timeout=None) as client:
        while True:
            try:
                async with aconnect_sse(
                    client,
                    "GET",
                    f"{HOST}/api/status/events",
                ) as event_source:

                    delay = 1

                    async for event in event_source.aiter_sse():
                        if event.event != "update":
                            continue

                        data = json.loads(event.data)

                        print(
                            f"\n========== [{datetime.now().isoformat()}] =========="
                        )
                        print(f"========== Event #{event.id or '?'} ==========")

                        global old_event

                        if old_event is None:
                            console.print_json(data=data)
                        else:
                            diff = DeepDiff(
                                old_event,
                                data,
                                ignore_order=True,
                            )

                            console.print_json(data=diff.to_dict())

                        old_event = data

            except KeyboardInterrupt:
                break

            except Exception as e:
                print(f"Disconnected: {e}")
                print(f"Retry in {delay}s...")
                await asyncio.sleep(delay)
                delay = min(delay * 2, 30)


asyncio.run(main())