events.py

# events.py - In-memory event logging for real-time dashboard
import collections
import threading
import time
from typing import Dict, List, Any

_LOG = collections.deque(maxlen=50)
_LOCK = threading.Lock()

def push(kind: str, **kv) -> None:
    """Add an event to the log with timestamp."""
    with _LOCK:
        _LOG.appendleft({
            "t": time.strftime("%H:%M:%S"), 
            "kind": kind, 
            **kv
        })

def dump() -> List[Dict[str, Any]]:
    """Get all logged events (most recent first)."""
    with _LOCK:
        return list(_LOG)

def clear() -> None:
    """Clear all logged events."""
    with _LOCK:
        _LOG.clear()