Collect stats for Home Assistant state_changed events

ha-event-state_changed-stats.py

import asyncio
import websockets
import json
from datetime import datetime, timedelta

url = 'ws://127.0.0.1:8123/api/websocket'
access_token = ''

buckets = set()
message_count = {}
message_size = {}

async def main():
    async with websockets.connect(url) as websocket:

        await websocket.send(json.dumps({'type': 'auth', 'access_token': access_token}))
        await websocket.send(json.dumps({'id': 1, 'type': 'subscribe_events', 'event_type': 'state_changed' }))

        end_time = datetime.now() + timedelta(hours=1)

        while datetime.now() < end_time:
            try:
                message = await asyncio.wait_for(websocket.recv(), timeout=1.0)
            except asyncio.TimeoutError:
                continue

            data = json.loads(message)
            if data.get('id') == 1:
                ts = datetime.now().replace(microsecond=0)
                if ts not in buckets:
                    message_count[ts] = 0
                    message_size[ts] = 0
                message_count[ts] += 1
                message_size[ts] += len(message)
                buckets.add(ts)

    for ts in sorted(buckets):
        print(f"{ts}\t{message_count[ts]}\t{message_size[ts]}")
    print(f"Average{' '*12}\t{round(sum(message_count.values()) / len(message_count.values()),1)}\t{round(sum(message_size.values()) / len(message_size.values()),1)}")
    print(f"Total{' '*14}\t{sum(message_count.values())}\t{sum(message_size.values())}")

asyncio.run(main())