65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
"""Route history and live subscribers for the dashboard.
|
|
|
|
The inference path is synchronous, so new records are handed to asyncio queues
|
|
with call_soon_threadsafe.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
|
|
class LiveHub:
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._subs: list[tuple[asyncio.AbstractEventLoop, asyncio.Queue]] = []
|
|
|
|
def subscribe(self, loop: asyncio.AbstractEventLoop, queue: asyncio.Queue) -> None:
|
|
with self._lock:
|
|
self._subs.append((loop, queue))
|
|
|
|
def unsubscribe(self, queue: asyncio.Queue) -> None:
|
|
with self._lock:
|
|
self._subs = [(loop, item) for loop, item in self._subs if item is not queue]
|
|
|
|
def publish(self, record: dict) -> None:
|
|
with self._lock:
|
|
subs = list(self._subs)
|
|
for loop, queue in subs:
|
|
loop.call_soon_threadsafe(_offer, queue, record)
|
|
|
|
|
|
def _offer(queue: asyncio.Queue, record: dict) -> None:
|
|
if queue.full():
|
|
try:
|
|
queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
pass
|
|
try:
|
|
queue.put_nowait(record)
|
|
except asyncio.QueueFull:
|
|
pass
|
|
|
|
|
|
def read_routes(path: Path, limit: int = 300) -> list[dict]:
|
|
if limit <= 0 or not path.is_file():
|
|
return []
|
|
records: list[dict] = []
|
|
with path.open(encoding="utf-8") as handle:
|
|
for line in handle:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
item = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(item, dict):
|
|
records.append(item)
|
|
if len(records) > limit:
|
|
return records[-limit:]
|
|
return records
|