mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-07 09:23:06 +02:00
Add CompanionAPIEndpoints integration in APIEndpoints class
- Introduced CompanionAPIEndpoints to handle routes under /api/companion/*. - Enhanced the APIEndpoints class to create a nested companion object for improved modularity and organization of companion-related API functionality.
This commit is contained in:
agessaman
@@ -11,6 +11,7 @@ from repeater.config import update_global_flood_policy
|
||||
from .cad_calibration_engine import CADCalibrationEngine
|
||||
from .auth.middleware import require_auth
|
||||
from .auth_endpoints import AuthAPIEndpoints
|
||||
from .companion_endpoints import CompanionAPIEndpoints
|
||||
from pymc_core.protocol import CryptoUtils
|
||||
|
||||
logger = logging.getLogger("HTTPServer")
|
||||
@@ -146,6 +147,9 @@ class APIEndpoints:
|
||||
# Create nested auth object for /api/auth/* routes
|
||||
self.auth = AuthAPIEndpoints()
|
||||
|
||||
# Create nested companion object for /api/companion/* routes
|
||||
self.companion = CompanionAPIEndpoints(daemon_instance, event_loop, self.config)
|
||||
|
||||
def _is_cors_enabled(self):
|
||||
return self.config.get("web", {}).get("cors_enabled", False)
|
||||
|
||||
|
||||
@@ -0,0 +1,576 @@
|
||||
"""
|
||||
Companion Bridge REST API and SSE event stream endpoints.
|
||||
|
||||
Mounted as a nested CherryPy object at /api/companion/ via APIEndpoints.
|
||||
Provides browser-accessible REST endpoints that proxy into the CompanionBridge
|
||||
async methods, plus a Server-Sent Events stream for real-time push callbacks.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import queue
|
||||
import time
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
import cherrypy
|
||||
from .auth.middleware import require_auth
|
||||
|
||||
logger = logging.getLogger("CompanionAPI")
|
||||
|
||||
|
||||
class CompanionAPIEndpoints:
|
||||
"""REST + SSE endpoints for a companion bridge.
|
||||
|
||||
CherryPy auto-mounts this at ``/api/companion/`` when assigned as
|
||||
``APIEndpoints.companion``. All async bridge calls are dispatched
|
||||
to the daemon's event loop via ``asyncio.run_coroutine_threadsafe``.
|
||||
"""
|
||||
|
||||
def __init__(self, daemon_instance=None, event_loop=None, config=None):
|
||||
self.daemon_instance = daemon_instance
|
||||
self.event_loop = event_loop
|
||||
self.config = config or {}
|
||||
|
||||
# SSE clients: each gets a thread-safe queue
|
||||
self._sse_clients: list[queue.Queue] = []
|
||||
self._sse_lock = threading.Lock()
|
||||
|
||||
# Flag: have we registered push callbacks yet?
|
||||
self._callbacks_registered = False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _get_bridge(self, name: Optional[str] = None, companion_hash: Optional[int] = None):
|
||||
"""Return the companion bridge, or raise 503/404 if unavailable.
|
||||
|
||||
Resolution order (mirrors room-server pattern):
|
||||
1. *name* — look up via identity_manager by registered name.
|
||||
2. *companion_hash* — direct lookup in ``companion_bridges`` dict.
|
||||
3. Neither — return the first (and typically only) bridge.
|
||||
"""
|
||||
if not self.daemon_instance:
|
||||
raise cherrypy.HTTPError(503, "Daemon not initialized")
|
||||
bridges = getattr(self.daemon_instance, "companion_bridges", {})
|
||||
if not bridges:
|
||||
raise cherrypy.HTTPError(503, "No companion bridges configured")
|
||||
|
||||
# --- resolve by name via identity_manager (same pattern as room servers) ---
|
||||
if name is not None:
|
||||
identity_manager = getattr(self.daemon_instance, "identity_manager", None)
|
||||
if identity_manager:
|
||||
for reg_name, identity, _cfg in identity_manager.get_identities_by_type("companion"):
|
||||
if reg_name == name:
|
||||
hash_byte = identity.get_public_key()[0]
|
||||
bridge = bridges.get(hash_byte)
|
||||
if bridge:
|
||||
return bridge
|
||||
raise cherrypy.HTTPError(404, f"Companion '{name}' not found")
|
||||
|
||||
# --- resolve by hash (fallback) ---
|
||||
if companion_hash is not None:
|
||||
bridge = bridges.get(companion_hash)
|
||||
if not bridge:
|
||||
raise cherrypy.HTTPError(404, f"Companion 0x{companion_hash:02X} not found")
|
||||
return bridge
|
||||
|
||||
# --- default: first bridge ---
|
||||
return next(iter(bridges.values()))
|
||||
|
||||
def _resolve_bridge_params(self, params) -> dict:
|
||||
"""Extract optional companion name/hash from request params.
|
||||
|
||||
Returns kwargs suitable for ``_get_bridge(**result)``.
|
||||
Follows the room-server convention: ``companion_name`` is the
|
||||
primary selector, ``companion_hash`` is the fallback.
|
||||
"""
|
||||
name = params.get("companion_name")
|
||||
raw_hash = params.get("companion_hash")
|
||||
result: dict = {}
|
||||
if name is not None:
|
||||
result["name"] = str(name)
|
||||
elif raw_hash is not None:
|
||||
try:
|
||||
result["companion_hash"] = int(str(raw_hash), 0)
|
||||
except (ValueError, TypeError):
|
||||
raise cherrypy.HTTPError(400, "Invalid companion_hash")
|
||||
return result
|
||||
|
||||
def _run_async(self, coro, timeout: float = 30.0):
|
||||
"""Run an async coroutine on the daemon event loop and return result."""
|
||||
if self.event_loop is None:
|
||||
raise cherrypy.HTTPError(503, "Event loop not available")
|
||||
future = asyncio.run_coroutine_threadsafe(coro, self.event_loop)
|
||||
return future.result(timeout=timeout)
|
||||
|
||||
@staticmethod
|
||||
def _success(data, **kwargs):
|
||||
result = {"success": True, "data": data}
|
||||
result.update(kwargs)
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def _error(msg):
|
||||
return {"success": False, "error": str(msg)}
|
||||
|
||||
def _require_post(self):
|
||||
if cherrypy.request.method != "POST":
|
||||
cherrypy.response.headers["Allow"] = "POST"
|
||||
raise cherrypy.HTTPError(405, "Method not allowed. Use POST.")
|
||||
|
||||
def _get_json_body(self) -> dict:
|
||||
"""Read and parse the JSON request body."""
|
||||
try:
|
||||
raw = cherrypy.request.body.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
except (json.JSONDecodeError, ValueError) as exc:
|
||||
raise cherrypy.HTTPError(400, f"Invalid JSON body: {exc}")
|
||||
|
||||
def _pub_key_from_hex(self, hex_str: str) -> bytes:
|
||||
"""Decode a hex public key, raising 400 on error."""
|
||||
try:
|
||||
key = bytes.fromhex(hex_str)
|
||||
if len(key) != 32:
|
||||
raise ValueError("Expected 32-byte key")
|
||||
return key
|
||||
except (ValueError, TypeError) as exc:
|
||||
raise cherrypy.HTTPError(400, f"Invalid public key: {exc}")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# SSE push-event plumbing
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _ensure_callbacks(self):
|
||||
"""Register push callbacks on the bridge (once)."""
|
||||
if self._callbacks_registered:
|
||||
return
|
||||
try:
|
||||
bridge = self._get_bridge()
|
||||
except cherrypy.HTTPError:
|
||||
return # bridge not yet available
|
||||
|
||||
def _make_cb(event_name):
|
||||
"""Create a callback that serialises event data for SSE clients."""
|
||||
def _cb(*args, **kwargs):
|
||||
payload = self._serialise_event(event_name, args, kwargs)
|
||||
self._broadcast_sse(payload)
|
||||
return _cb
|
||||
|
||||
callback_names = [
|
||||
"message_received",
|
||||
"channel_message_received",
|
||||
"advert_received",
|
||||
"contact_path_updated",
|
||||
"send_confirmed",
|
||||
"login_result",
|
||||
]
|
||||
for name in callback_names:
|
||||
register_fn = getattr(bridge, f"on_{name}", None)
|
||||
if register_fn:
|
||||
register_fn(_make_cb(name))
|
||||
|
||||
self._callbacks_registered = True
|
||||
|
||||
@staticmethod
|
||||
def _serialise_event(event_name: str, args: tuple, kwargs: dict) -> dict:
|
||||
"""Convert callback arguments to a JSON-safe dict."""
|
||||
data: dict = {"event": event_name, "timestamp": int(time.time())}
|
||||
for i, arg in enumerate(args):
|
||||
data[f"arg{i}"] = _to_json_safe(arg)
|
||||
for k, v in kwargs.items():
|
||||
data[k] = _to_json_safe(v)
|
||||
return data
|
||||
|
||||
def _broadcast_sse(self, payload: dict):
|
||||
"""Put *payload* into every active SSE client queue."""
|
||||
with self._sse_lock:
|
||||
dead = []
|
||||
for q in self._sse_clients:
|
||||
try:
|
||||
q.put_nowait(payload)
|
||||
except queue.Full:
|
||||
dead.append(q)
|
||||
for q in dead:
|
||||
self._sse_clients.remove(q)
|
||||
|
||||
# ==================================================================
|
||||
# REST Endpoints
|
||||
# ==================================================================
|
||||
|
||||
# ----- Index / listing -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def index(self, **kwargs):
|
||||
"""GET /api/companion/ — list configured companions."""
|
||||
bridges = getattr(self.daemon_instance, "companion_bridges", {})
|
||||
identity_manager = getattr(self.daemon_instance, "identity_manager", None)
|
||||
|
||||
# Build name lookup from identity_manager (same pattern as room servers)
|
||||
name_by_hash: dict[int, str] = {}
|
||||
if identity_manager:
|
||||
for reg_name, identity, _cfg in identity_manager.get_identities_by_type("companion"):
|
||||
name_by_hash[identity.get_public_key()[0]] = reg_name
|
||||
|
||||
items = []
|
||||
for h, b in bridges.items():
|
||||
items.append({
|
||||
"companion_name": name_by_hash.get(h, ""),
|
||||
"companion_hash": f"0x{h:02X}",
|
||||
"node_name": b.prefs.node_name,
|
||||
"public_key": b.get_public_key().hex(),
|
||||
"is_running": b.is_running,
|
||||
"contacts_count": b.contacts.get_count(),
|
||||
"channels_count": b.channels.get_count(),
|
||||
})
|
||||
return self._success(items)
|
||||
|
||||
# ----- Identity -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def self_info(self, **kwargs):
|
||||
"""GET /api/companion/self_info — node identity and preferences."""
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(kwargs))
|
||||
prefs = bridge.get_self_info()
|
||||
return self._success({
|
||||
"public_key": bridge.get_public_key().hex(),
|
||||
"node_name": prefs.node_name,
|
||||
"adv_type": prefs.adv_type,
|
||||
"tx_power_dbm": prefs.tx_power_dbm,
|
||||
"frequency_hz": prefs.frequency_hz,
|
||||
"bandwidth_hz": prefs.bandwidth_hz,
|
||||
"spreading_factor": prefs.spreading_factor,
|
||||
"coding_rate": prefs.coding_rate,
|
||||
"latitude": prefs.latitude,
|
||||
"longitude": prefs.longitude,
|
||||
})
|
||||
|
||||
# ----- Contacts -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def contacts(self, **kwargs):
|
||||
"""GET /api/companion/contacts — list all contacts."""
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(kwargs))
|
||||
since = int(kwargs.get("since", 0))
|
||||
contacts = bridge.get_contacts(since=since)
|
||||
items = []
|
||||
for c in contacts:
|
||||
items.append({
|
||||
"public_key": c.public_key.hex() if isinstance(c.public_key, bytes) else c.public_key,
|
||||
"name": c.name,
|
||||
"adv_type": c.adv_type,
|
||||
"flags": c.flags,
|
||||
"out_path_len": c.out_path_len,
|
||||
"last_advert_timestamp": c.last_advert_timestamp,
|
||||
"lastmod": c.lastmod,
|
||||
"gps_lat": c.gps_lat,
|
||||
"gps_lon": c.gps_lon,
|
||||
})
|
||||
return self._success(items)
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def contact(self, **kwargs):
|
||||
"""GET /api/companion/contact?pub_key=<hex> — get single contact."""
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(kwargs))
|
||||
pk_hex = kwargs.get("pub_key")
|
||||
if not pk_hex:
|
||||
raise cherrypy.HTTPError(400, "pub_key required")
|
||||
pub_key = self._pub_key_from_hex(pk_hex)
|
||||
c = bridge.get_contact_by_key(pub_key)
|
||||
if not c:
|
||||
raise cherrypy.HTTPError(404, "Contact not found")
|
||||
return self._success({
|
||||
"public_key": c.public_key.hex() if isinstance(c.public_key, bytes) else c.public_key,
|
||||
"name": c.name,
|
||||
"adv_type": c.adv_type,
|
||||
"flags": c.flags,
|
||||
"out_path_len": c.out_path_len,
|
||||
"out_path": c.out_path.hex() if isinstance(c.out_path, bytes) else "",
|
||||
"last_advert_timestamp": c.last_advert_timestamp,
|
||||
"lastmod": c.lastmod,
|
||||
"gps_lat": c.gps_lat,
|
||||
"gps_lon": c.gps_lon,
|
||||
})
|
||||
|
||||
# ----- Channels -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def channels(self, **kwargs):
|
||||
"""GET /api/companion/channels — list configured channels."""
|
||||
try:
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(kwargs))
|
||||
items = []
|
||||
for idx in range(bridge.channels.max_channels):
|
||||
ch = bridge.channels.get(idx)
|
||||
if ch:
|
||||
items.append({
|
||||
"index": idx,
|
||||
"name": ch.name,
|
||||
# Don't expose the PSK secret over REST
|
||||
})
|
||||
return self._success(items)
|
||||
except cherrypy.HTTPError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(f"channels endpoint error: {exc}", exc_info=True)
|
||||
return self._error(str(exc))
|
||||
|
||||
# ----- Statistics -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def stats(self, **kwargs):
|
||||
"""GET /api/companion/stats?type=packets — local companion stats."""
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(kwargs))
|
||||
stats_type_map = {"core": 0, "radio": 1, "packets": 2}
|
||||
stype = stats_type_map.get(kwargs.get("type", "packets"), 2)
|
||||
return self._success(bridge.get_stats(stype))
|
||||
|
||||
# ----- Messaging -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def send_text(self, **kwargs):
|
||||
"""POST /api/companion/send_text {pub_key, text, txt_type?, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
pub_key = self._pub_key_from_hex(body.get("pub_key", ""))
|
||||
text = body.get("text", "")
|
||||
if not text:
|
||||
raise cherrypy.HTTPError(400, "text required")
|
||||
txt_type = int(body.get("txt_type", 0))
|
||||
result = self._run_async(
|
||||
bridge.send_text_message(pub_key, text, txt_type=txt_type)
|
||||
)
|
||||
return self._success({
|
||||
"sent": result.success,
|
||||
"is_flood": result.is_flood,
|
||||
"expected_ack": result.expected_ack,
|
||||
})
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def send_channel_message(self, **kwargs):
|
||||
"""POST /api/companion/send_channel_message {channel_idx, text, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
channel_idx = int(body.get("channel_idx", 0))
|
||||
text = body.get("text", "")
|
||||
if not text:
|
||||
raise cherrypy.HTTPError(400, "text required")
|
||||
success = self._run_async(bridge.send_channel_message(channel_idx, text))
|
||||
return self._success({"sent": success})
|
||||
|
||||
# ----- Login -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def login(self, **kwargs):
|
||||
"""POST /api/companion/login {pub_key, password?, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
pub_key = self._pub_key_from_hex(body.get("pub_key", ""))
|
||||
password = body.get("password", "")
|
||||
result = self._run_async(
|
||||
bridge.send_login(pub_key, password), timeout=15.0
|
||||
)
|
||||
return self._success(_to_json_safe(result))
|
||||
|
||||
# ----- Status / Telemetry Requests -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def request_status(self, **kwargs):
|
||||
"""POST /api/companion/request_status {pub_key, timeout?, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
pub_key = self._pub_key_from_hex(body.get("pub_key", ""))
|
||||
timeout = float(body.get("timeout", 15.0))
|
||||
result = self._run_async(
|
||||
bridge.send_status_request(pub_key, timeout=timeout),
|
||||
timeout=timeout + 5.0,
|
||||
)
|
||||
return self._success(_to_json_safe(result))
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def request_telemetry(self, **kwargs):
|
||||
"""POST /api/companion/request_telemetry {pub_key, want_base?, want_location?, want_environment?, timeout?, companion_name?}"""
|
||||
self._require_post()
|
||||
try:
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
pub_key = self._pub_key_from_hex(body.get("pub_key", ""))
|
||||
timeout = float(body.get("timeout", 10.0))
|
||||
result = self._run_async(
|
||||
bridge.send_telemetry_request(
|
||||
pub_key,
|
||||
want_base=bool(body.get("want_base", True)),
|
||||
want_location=bool(body.get("want_location", True)),
|
||||
want_environment=bool(body.get("want_environment", True)),
|
||||
timeout=timeout,
|
||||
),
|
||||
timeout=timeout + 5.0,
|
||||
)
|
||||
# Ensure all values are JSON-serialisable (telemetry may contain bytes)
|
||||
return self._success(_to_json_safe(result))
|
||||
except cherrypy.HTTPError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(f"request_telemetry endpoint error: {exc}", exc_info=True)
|
||||
return self._error(str(exc))
|
||||
|
||||
# ----- Repeater Commands -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def send_command(self, **kwargs):
|
||||
"""POST /api/companion/send_command {pub_key, command, parameters?, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
pub_key = self._pub_key_from_hex(body.get("pub_key", ""))
|
||||
command = body.get("command", "")
|
||||
if not command:
|
||||
raise cherrypy.HTTPError(400, "command required")
|
||||
parameters = body.get("parameters")
|
||||
result = self._run_async(
|
||||
bridge.send_repeater_command(pub_key, command, parameters),
|
||||
timeout=20.0,
|
||||
)
|
||||
return self._success(_to_json_safe(result))
|
||||
|
||||
# ----- Path / Routing -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def reset_path(self, **kwargs):
|
||||
"""POST /api/companion/reset_path {pub_key, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
pub_key = self._pub_key_from_hex(body.get("pub_key", ""))
|
||||
ok = bridge.reset_path(pub_key)
|
||||
return self._success({"reset": ok})
|
||||
|
||||
# ----- Device Configuration -----
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def set_advert_name(self, **kwargs):
|
||||
"""POST /api/companion/set_advert_name {advert_name, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
name = body.get("advert_name", body.get("name", ""))
|
||||
if not name:
|
||||
raise cherrypy.HTTPError(400, "name required")
|
||||
bridge.set_advert_name(name)
|
||||
return self._success({"name": bridge.prefs.node_name})
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@require_auth
|
||||
def set_advert_location(self, **kwargs):
|
||||
"""POST /api/companion/set_advert_location {latitude, longitude, companion_name?}"""
|
||||
self._require_post()
|
||||
body = self._get_json_body()
|
||||
bridge = self._get_bridge(**self._resolve_bridge_params(body))
|
||||
lat = float(body.get("latitude", 0.0))
|
||||
lon = float(body.get("longitude", 0.0))
|
||||
bridge.set_advert_latlon(lat, lon)
|
||||
return self._success({"latitude": lat, "longitude": lon})
|
||||
|
||||
# ==================================================================
|
||||
# SSE Event Stream
|
||||
# ==================================================================
|
||||
|
||||
@cherrypy.expose
|
||||
def events(self, **kwargs):
|
||||
"""GET /api/companion/events — Server-Sent Events stream for push callbacks.
|
||||
|
||||
Connect with ``EventSource('/api/companion/events?token=JWT')``.
|
||||
Auth is handled by the CherryPy tool-level require_auth (supports
|
||||
query-param JWT tokens needed by the browser EventSource API).
|
||||
"""
|
||||
self._ensure_callbacks()
|
||||
|
||||
cherrypy.response.headers["Content-Type"] = "text/event-stream"
|
||||
cherrypy.response.headers["Cache-Control"] = "no-cache"
|
||||
cherrypy.response.headers["Connection"] = "keep-alive"
|
||||
cherrypy.response.headers["X-Accel-Buffering"] = "no"
|
||||
|
||||
client_queue: queue.Queue = queue.Queue(maxsize=256)
|
||||
with self._sse_lock:
|
||||
self._sse_clients.append(client_queue)
|
||||
|
||||
def generate():
|
||||
try:
|
||||
yield f"data: {json.dumps({'event': 'connected', 'timestamp': int(time.time())})}\n\n"
|
||||
|
||||
while True:
|
||||
try:
|
||||
item = client_queue.get(timeout=15.0)
|
||||
yield f"data: {json.dumps(item)}\n\n"
|
||||
except queue.Empty:
|
||||
# Keep-alive comment
|
||||
yield f"data: {json.dumps({'event': 'keepalive', 'timestamp': int(time.time())})}\n\n"
|
||||
except GeneratorExit:
|
||||
pass
|
||||
except Exception as exc:
|
||||
logger.debug(f"SSE stream ended: {exc}")
|
||||
finally:
|
||||
with self._sse_lock:
|
||||
if client_queue in self._sse_clients:
|
||||
self._sse_clients.remove(client_queue)
|
||||
|
||||
return generate()
|
||||
|
||||
events._cp_config = {"response.stream": True}
|
||||
|
||||
|
||||
# ======================================================================
|
||||
# Utility: make arbitrary objects JSON-serialisable for SSE events
|
||||
# ======================================================================
|
||||
|
||||
def _to_json_safe(obj):
|
||||
"""Convert common companion objects to JSON-safe dicts/values."""
|
||||
if obj is None or isinstance(obj, (bool, int, float, str)):
|
||||
return obj
|
||||
if isinstance(obj, bytes):
|
||||
return obj.hex()
|
||||
if isinstance(obj, bytearray):
|
||||
return bytes(obj).hex()
|
||||
if isinstance(obj, dict):
|
||||
return {k: _to_json_safe(v) for k, v in obj.items()}
|
||||
if isinstance(obj, (list, tuple)):
|
||||
return [_to_json_safe(v) for v in obj]
|
||||
# Dataclass / namedtuple with __dict__
|
||||
if hasattr(obj, "__dict__"):
|
||||
return {k: _to_json_safe(v) for k, v in obj.__dict__.items() if not k.startswith("_")}
|
||||
return str(obj)
|
||||
Reference in New Issue
Block a user