"""A-side OmniDaemon that owns local control/video OmniSocket sessions.""" from __future__ import annotations import argparse import copy import json import os from pathlib import Path import queue import signal import socketserver import sys import threading import time from dataclasses import dataclass from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler from typing import Any import yaml from . import DEFAULT_CONFIG_PATH, DEFAULT_SOCKET_PATH, VERSION from .control_codec import ANALOG_EVENT_CODES, EVENT_NAME_TO_ID, make_control_packet def utc_iso_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def load_omnisocket_api(): from omnisocket import CONTROL_DEFAULTS, MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS return CONTROL_DEFAULTS, MSG_TYPE_BINARY, Session, VIDEO_DEFAULTS def _merge_kcp_defaults(defaults: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: merged = dict(defaults) for key in ("nodelay", "interval_ms", "resend", "nc", "sndwnd", "rcvwnd", "mtu"): if key in override: merged[key] = int(override[key]) return merged def _load_config(config_path: str | None) -> dict[str, Any]: path = Path(os.getenv("OMNIDAEMON_CONFIG", config_path or str(DEFAULT_CONFIG_PATH))) raw: dict[str, Any] = {} if path.exists(): with path.open("r", encoding="utf-8") as file: raw = yaml.safe_load(file) or {} control_defaults, _msg_type_binary, _session_cls, video_defaults = load_omnisocket_api() transport = dict(raw.get("transport", {})) control = dict(raw.get("control_sender", {})) video = dict(raw.get("video_receiver", {})) daemon_cfg = dict(raw.get("daemon", {})) policy = dict(raw.get("policy", {})) def _profile(name: str, fps: int, max_frame_kb: int) -> dict[str, int]: section = dict(policy.get(name, {})) return { "fps": int(section.get("fps", fps)), "max_frame_bytes": int(section.get("max_frame_kb", max_frame_kb)) * 1024, } return { "config_path": str(path), "transport": { "server_addr": str(transport.get("server_addr", "")), "relay_via": str(transport.get("relay_via", "")), "bind_ip": str(transport.get("bind_ip", "")), "bind_device": str(transport.get("bind_device", "")), }, "control_sender": { "peer_id": str(control.get("peer_id", "peer-a-ctrl")), "target_peer": str(control.get("target_peer", "peer-b-ctrl")), "stats_interval_ms": int(control.get("stats_interval_ms", 100)), "kcp": _merge_kcp_defaults(control_defaults, control), }, "video_receiver": { "peer_id": str(video.get("peer_id", "peer-a-video")), "buffer_bytes": int(video.get("buffer_bytes", 1024 * 1024)), "stats_interval_ms": int(video.get("stats_interval_ms", 100)), "kcp": _merge_kcp_defaults(video_defaults, video), }, "daemon": { "socket_path": os.getenv( "OMNIDAEMON_SOCKET", str(daemon_cfg.get("socket_path", DEFAULT_SOCKET_PATH)), ), "reconnect_delay_ms": int(daemon_cfg.get("reconnect_delay_ms", 2000)), "telemetry_interval_ms": int(daemon_cfg.get("telemetry_interval_ms", 100)), "analog_send_hz": int(daemon_cfg.get("analog_send_hz", 100)), "frame_stale_ms": int(daemon_cfg.get("frame_stale_ms", 500)), }, "policy": { "health_window_ms": int(policy.get("health_window_ms", 2000)), "green_srtt_ms": int(policy.get("green_srtt_ms", 35)), "yellow_srtt_ms": int(policy.get("yellow_srtt_ms", 60)), "retrans_red_threshold": int(policy.get("retrans_red_threshold", 10)), "profile_green": _profile("profile_green", 15, 60), "profile_yellow": _profile("profile_yellow", 10, 40), "profile_red": _profile("profile_red", 5, 20), }, } def _default_stats() -> dict[str, int]: return { "send_calls": 0, "send_bytes": 0, "send_errors": 0, "recv_calls": 0, "recv_bytes": 0, "recv_timeouts": 0, "recv_errors": 0, "connected": 0, } def _default_kcp_metrics() -> dict[str, Any]: return { "connected": 0, "has_conv": 0, "conv": 0, "local_addr": "", "remote_addr": "", "rto_ms": 0, "srtt_ms": 0, "srttvar_ms": 0, "bytes_sent": 0, "bytes_received": 0, "in_pkts": 0, "out_pkts": 0, "in_segs": 0, "out_segs": 0, "retrans_segs": 0, "fast_retrans_segs": 0, "early_retrans_segs": 0, "lost_segs": 0, "repeat_segs": 0, "in_errs": 0, "kcp_in_errs": 0, "ring_buffer_snd_queue": 0, "ring_buffer_rcv_queue": 0, "ring_buffer_snd_buffer": 0, } @dataclass(slots=True) class QueuedControlEvent: seq_id: int source: str event_code: str drive_value: float client_time_ms: int class ControlSessionManager: def __init__(self, config: dict[str, Any]) -> None: control_defaults, _msg_type_binary, session_cls, _video_defaults = load_omnisocket_api() transport = config["transport"] control_cfg = config["control_sender"] daemon_cfg = config["daemon"] self._session_cls = session_cls self._connect_kwargs = { "server_addr": transport["server_addr"], "relay_via": transport["relay_via"], "bind_ip": transport["bind_ip"], "bind_device": transport["bind_device"], "peer_id": control_cfg["peer_id"], "stats_interval_ms": control_cfg["stats_interval_ms"], **_merge_kcp_defaults(control_defaults, control_cfg["kcp"]), } self._target_peer = control_cfg["target_peer"] self._analog_interval = 1.0 / max(1, daemon_cfg["analog_send_hz"]) self._reconnect_delay_s = max(0.25, daemon_cfg["reconnect_delay_ms"] / 1000.0) self._lock = threading.Lock() self._stop_event = threading.Event() self._wake_event = threading.Event() self._discrete_queue: queue.Queue[QueuedControlEvent] = queue.Queue() self._pending_analog: dict[str, QueuedControlEvent] = {} self._session = None self._connected = False self._seq_id = 0 self._last_seq_id: int | None = None self._last_error = "" self._last_connect_attempt = 0.0 self._next_analog_flush = 0.0 self._thread = threading.Thread( target=self._run, name="omni-a-side-control", daemon=True, ) def start(self) -> None: self._thread.start() def stop(self) -> None: self._stop_event.set() self._wake_event.set() if self._thread.is_alive(): self._thread.join(timeout=2.0) self._disconnect("control daemon stopped", drop_pending=True) def send_event( self, *, source: str, event_code: str, drive_value: float, client_time_ms: int | None, ) -> dict[str, Any]: if event_code not in EVENT_NAME_TO_ID: return { "accepted": False, "assigned_seq_id": None, "control_connected": self.is_connected(), "queue_depth": self.queue_depth(), "error": f"unknown event_code: {event_code}", } if not self.is_connected(): return { "accepted": False, "assigned_seq_id": None, "control_connected": False, "queue_depth": self.queue_depth(), "error": self.last_error() or "control session is not connected", } with self._lock: seq_id = self._seq_id self._seq_id += 1 event = QueuedControlEvent( seq_id=seq_id, source=source, event_code=event_code, drive_value=float(drive_value), client_time_ms=int(client_time_ms or time.time() * 1000), ) if event_code in ANALOG_EVENT_CODES: self._pending_analog[event_code] = event else: self._discrete_queue.put_nowait(event) self._wake_event.set() return { "accepted": True, "assigned_seq_id": seq_id, "control_connected": True, "queue_depth": self.queue_depth(), "error": "", } def is_connected(self) -> bool: with self._lock: return self._connected and self._session is not None def queue_depth(self) -> int: with self._lock: return self._discrete_queue.qsize() + len(self._pending_analog) def last_error(self) -> str: with self._lock: return self._last_error def snapshot(self) -> dict[str, Any]: with self._lock: session = self._session connected = self._connected and session is not None last_error = self._last_error last_seq_id = self._last_seq_id queue_depth = self._discrete_queue.qsize() + len(self._pending_analog) stats = _default_stats() metrics = _default_kcp_metrics() if session is not None: try: stats = dict(session.stats()) metrics = dict(session.kcp_metrics()) connected = bool(stats.get("connected") and metrics.get("connected")) except Exception as error: # pragma: no cover - runtime integration connected = False last_error = str(error) return { "connected": connected, "queue_depth": queue_depth, "last_seq_id": last_seq_id, "last_error": last_error, "peer_id": self._connect_kwargs["peer_id"], "target_peer": self._target_peer, "server_addr": self._connect_kwargs["server_addr"], "relay_via": self._connect_kwargs["relay_via"], "stats": stats, "kcp_metrics": metrics, } def _run(self) -> None: analog_order = ("set_surge", "set_sway", "set_spin") while not self._stop_event.is_set(): if not self.is_connected(): now = time.monotonic() if now - self._last_connect_attempt >= self._reconnect_delay_s: self._last_connect_attempt = now self._connect() self._wake_event.wait(0.2) self._wake_event.clear() continue sent_any = False while not self._stop_event.is_set(): try: event = self._discrete_queue.get_nowait() except queue.Empty: break if not self._send_packet(event): break sent_any = True now = time.monotonic() if now >= self._next_analog_flush: analog_batch: list[QueuedControlEvent] = [] with self._lock: for event_code in analog_order: event = self._pending_analog.pop(event_code, None) if event is not None: analog_batch.append(event) if analog_batch: for event in analog_batch: if not self._send_packet(event): break sent_any = True self._next_analog_flush = now + self._analog_interval if not sent_any: wait_s = max(0.02, self._next_analog_flush - time.monotonic()) self._wake_event.wait(wait_s) self._wake_event.clear() def _connect(self) -> None: session = self._session_cls() try: session.connect(**self._connect_kwargs) except Exception as error: # pragma: no cover - runtime integration with self._lock: self._connected = False self._last_error = str(error) try: session.close() except Exception: pass return with self._lock: self._session = session self._connected = True self._last_error = "" self._next_analog_flush = time.monotonic() def _disconnect(self, error_message: str, *, drop_pending: bool) -> None: with self._lock: session = self._session self._session = None self._connected = False if error_message: self._last_error = error_message if drop_pending: self._pending_analog.clear() while not self._discrete_queue.empty(): try: self._discrete_queue.get_nowait() except queue.Empty: break if session is not None: try: session.close() except Exception: pass def _send_packet(self, event: QueuedControlEvent) -> bool: with self._lock: session = self._session if session is None: return False packet = make_control_packet(event.seq_id, event.event_code, event.drive_value) try: session.send(to=self._target_peer, data=packet.encode()) except Exception as error: # pragma: no cover - runtime integration self._disconnect(str(error), drop_pending=True) return False with self._lock: self._last_seq_id = event.seq_id self._last_error = "" return True class VideoSessionManager: def __init__(self, config: dict[str, Any]) -> None: _control_defaults, msg_type_binary, session_cls, video_defaults = load_omnisocket_api() transport = config["transport"] video_cfg = config["video_receiver"] daemon_cfg = config["daemon"] self._msg_type_binary = msg_type_binary self._session_cls = session_cls self._connect_kwargs = { "server_addr": transport["server_addr"], "relay_via": transport["relay_via"], "bind_ip": transport["bind_ip"], "bind_device": transport["bind_device"], "peer_id": video_cfg["peer_id"], "stats_interval_ms": video_cfg["stats_interval_ms"], **_merge_kcp_defaults(video_defaults, video_cfg["kcp"]), } self._buffer_bytes = video_cfg["buffer_bytes"] self._frame_stale_s = max(0.1, daemon_cfg["frame_stale_ms"] / 1000.0) self._reconnect_delay_s = max(0.25, daemon_cfg["reconnect_delay_ms"] / 1000.0) self._config_path = config["config_path"] self._lock = threading.Lock() self._stop_event = threading.Event() self._session = None self._latest_frame: bytes | None = None self._latest_received_at = 0.0 self._latest_sequence: int | None = None self._frames_received = 0 self._last_error = "" self._last_connect_attempt = 0.0 self._thread = threading.Thread( target=self._run, name="omni-a-side-video", daemon=True, ) def start(self) -> None: self._thread.start() def stop(self) -> None: self._stop_event.set() if self._thread.is_alive(): self._thread.join(timeout=2.0) self._disconnect("video daemon stopped") def get_latest_frame(self) -> bytes | None: with self._lock: if self._latest_frame is None: return None if time.time() - self._latest_received_at > self._frame_stale_s: return None return self._latest_frame def snapshot(self) -> dict[str, Any]: with self._lock: session = self._session latest_received_at = self._latest_received_at latest_sequence = self._latest_sequence frames_received = self._frames_received last_error = self._last_error has_recent_frame = ( self._latest_frame is not None and time.time() - self._latest_received_at <= self._frame_stale_s ) stats = _default_stats() metrics = _default_kcp_metrics() connected = session is not None if session is not None: try: stats = dict(session.stats()) metrics = dict(session.kcp_metrics()) connected = bool(stats.get("connected") and metrics.get("connected")) except Exception as error: # pragma: no cover - runtime integration connected = False last_error = str(error) return { "connected": connected, "has_recent_frame": has_recent_frame, "latest_received_at": latest_received_at, "latest_sequence": latest_sequence, "frames_received": frames_received, "last_error": last_error, "buffer_bytes": self._buffer_bytes, "peer_id": self._connect_kwargs["peer_id"], "server_addr": self._connect_kwargs["server_addr"], "relay_via": self._connect_kwargs["relay_via"], "config_path": self._config_path, "stats": stats, "kcp_metrics": metrics, } def _run(self) -> None: while not self._stop_event.is_set(): if not self._is_connected(): now = time.monotonic() if now - self._last_connect_attempt >= self._reconnect_delay_s: self._last_connect_attempt = now self._connect() time.sleep(0.2) continue with self._lock: session = self._session if session is None: continue buffer = bytearray(self._buffer_bytes) try: while not self._stop_event.is_set(): meta = session.recv_into(buffer, timeout_ms=200) if meta is None: continue if meta.get("msg_type") != self._msg_type_binary: continue frame = bytes(buffer[: int(meta["body_len"])]) jpeg_frame = self._extract_jpeg_frame(frame) if jpeg_frame is None: with self._lock: self._last_error = "received non-JPEG binary frame" continue with self._lock: self._latest_frame = jpeg_frame self._latest_received_at = time.time() self._latest_sequence = self._extract_sequence(frame) self._frames_received += 1 self._last_error = "" except Exception as error: # pragma: no cover - runtime integration self._disconnect(str(error)) time.sleep(0.2) def _connect(self) -> None: session = self._session_cls() try: session.connect(**self._connect_kwargs) except Exception as error: # pragma: no cover - runtime integration with self._lock: self._session = None self._last_error = str(error) try: session.close() except Exception: pass return with self._lock: self._session = session self._last_error = "" def _disconnect(self, error_message: str) -> None: with self._lock: session = self._session self._session = None self._last_error = error_message if session is not None: try: session.close() except Exception: pass def _is_connected(self) -> bool: with self._lock: return self._session is not None @staticmethod def _extract_jpeg_frame(frame: bytes) -> bytes | None: if frame.startswith(b"\xff\xd8"): return frame if len(frame) > 8 and frame[8:10] == b"\xff\xd8": return frame[8:] return None @staticmethod def _extract_sequence(frame: bytes) -> int | None: if len(frame) >= 8 and not frame.startswith(b"\xff\xd8"): return int.from_bytes(frame[:8], "big") return None class PolicyEngine: def __init__(self, policy_cfg: dict[str, Any]) -> None: self._policy_cfg = policy_cfg def classify( self, *, control_connected: bool, avg_srtt_ms: float, retrans_delta: int, lost_delta: int, ) -> tuple[str, dict[str, int]]: if not control_connected: return "red", dict(self._policy_cfg["profile_red"]) if ( avg_srtt_ms >= self._policy_cfg["yellow_srtt_ms"] or lost_delta > 0 or retrans_delta >= self._policy_cfg["retrans_red_threshold"] ): return "red", dict(self._policy_cfg["profile_red"]) if avg_srtt_ms >= self._policy_cfg["green_srtt_ms"] or retrans_delta > 0: return "yellow", dict(self._policy_cfg["profile_yellow"]) return "green", dict(self._policy_cfg["profile_green"]) class TelemetrySampler: def __init__( self, config: dict[str, Any], control_manager: ControlSessionManager, video_manager: VideoSessionManager, ) -> None: self._config = config self._control_manager = control_manager self._video_manager = video_manager self._policy_engine = PolicyEngine(config["policy"]) self._interval_s = max(0.1, config["daemon"]["telemetry_interval_ms"] / 1000.0) self._window_s = max(0.5, config["policy"]["health_window_ms"] / 1000.0) self._lock = threading.Lock() self._stop_event = threading.Event() self._thread = threading.Thread( target=self._run, name="omni-a-side-telemetry", daemon=True, ) self._state = self._build_initial_state() self._control_history: list[dict[str, Any]] = [] self._last_totals: dict[str, Any] | None = None self._started_at = utc_iso_now() def start(self) -> None: self._thread.start() def stop(self) -> None: self._stop_event.set() if self._thread.is_alive(): self._thread.join(timeout=2.0) def snapshot(self) -> dict[str, Any]: with self._lock: return copy.deepcopy(self._state) def _run(self) -> None: while not self._stop_event.is_set(): self._sample_once() self._stop_event.wait(self._interval_s) def _sample_once(self) -> None: now = time.time() control = self._control_manager.snapshot() video = self._video_manager.snapshot() control_metrics = control["kcp_metrics"] video_metrics = video["kcp_metrics"] self._control_history.append( { "ts": now, "connected": bool(control["connected"]), "srtt_ms": float(control_metrics.get("srtt_ms", 0.0) or 0.0), "retrans_segs": int(control_metrics.get("retrans_segs", 0) or 0), "lost_segs": int(control_metrics.get("lost_segs", 0) or 0), } ) self._control_history = [ item for item in self._control_history if now - item["ts"] <= self._window_s ] avg_srtt_ms = 0.0 jitter_ms = 0.0 retrans_delta_window = 0 lost_delta_window = 0 srtt_values = [item["srtt_ms"] for item in self._control_history if item["connected"]] if srtt_values: avg_srtt_ms = sum(srtt_values) / len(srtt_values) if len(srtt_values) >= 2: jitter_samples = [ abs(srtt_values[index] - srtt_values[index - 1]) for index in range(1, len(srtt_values)) ] jitter_ms = sum(jitter_samples) / len(jitter_samples) if len(self._control_history) >= 2: first = self._control_history[0] last = self._control_history[-1] retrans_delta_window = max(0, last["retrans_segs"] - first["retrans_segs"]) lost_delta_window = max(0, last["lost_segs"] - first["lost_segs"]) health_band, profile = self._policy_engine.classify( control_connected=bool(control["connected"]), avg_srtt_ms=avg_srtt_ms, retrans_delta=retrans_delta_window, lost_delta=lost_delta_window, ) total_bytes_sent = int(control_metrics.get("bytes_sent", 0)) + int(video_metrics.get("bytes_sent", 0)) total_bytes_received = int(control_metrics.get("bytes_received", 0)) + int(video_metrics.get("bytes_received", 0)) total_out_segs = int(control_metrics.get("out_segs", 0)) + int(video_metrics.get("out_segs", 0)) total_retrans = int(control_metrics.get("retrans_segs", 0)) + int(video_metrics.get("retrans_segs", 0)) tx_kbps = 0 rx_kbps = 0 retrans_pct = 0.0 if self._last_totals is not None: dt = max(0.001, now - self._last_totals["ts"]) sent_delta = max(0, total_bytes_sent - self._last_totals["bytes_sent"]) recv_delta = max(0, total_bytes_received - self._last_totals["bytes_received"]) out_seg_delta = max(0, total_out_segs - self._last_totals["out_segs"]) retrans_delta = max(0, total_retrans - self._last_totals["retrans"]) tx_kbps = int((sent_delta * 8.0) / dt / 1000.0) rx_kbps = int((recv_delta * 8.0) / dt / 1000.0) if out_seg_delta > 0: retrans_pct = round((retrans_delta / out_seg_delta) * 100.0, 2) self._last_totals = { "ts": now, "bytes_sent": total_bytes_sent, "bytes_received": total_bytes_received, "out_segs": total_out_segs, "retrans": total_retrans, } network = { "peer_status": "online" if control["connected"] or video["connected"] else "offline", "latency_ms": round(avg_srtt_ms or float(control_metrics.get("srtt_ms", 0) or 0.0), 1), "jitter_ms": round(jitter_ms, 1), # Keep packet_loss_pct as a compatibility alias for existing clients. "retrans_pct": round(retrans_pct, 2), "packet_loss_pct": round(retrans_pct, 2), "tx_kbps": tx_kbps, "rx_kbps": rx_kbps, "signal_dbm": None, "transport": "OmniSocket / daemon", "source_mode": "daemon-live", "updated_at": utc_iso_now(), } video_status = { "available": bool(video["has_recent_frame"]), "source_mode": "omnisocket-jpeg-live" if video["has_recent_frame"] else "omnisocket-waiting", "frame_count": int(video["frames_received"]), "fps": int(profile["fps"]), "frame_dir": "omni-daemon://latest-frame", "source_detail": ( f"peer stream active, frames={video['frames_received']}" if video["has_recent_frame"] else (video["last_error"] or "waiting for latest JPEG frame from daemon") ), "receiver": { "backend_ready": True, "mode": "daemon", "connected": bool(video["connected"]), "has_recent_frame": bool(video["has_recent_frame"]), "frames_received": int(video["frames_received"]), "latest_sequence": video["latest_sequence"], "last_error": video["last_error"], "config_path": video["config_path"], "server_addr": video["server_addr"], "relay_via": video["relay_via"], "peer_id": video["peer_id"], "buffer_bytes": int(video["buffer_bytes"]), "stats": video["stats"], "kcp_metrics": video_metrics, }, } state = { "network": network, "video": video_status, "control": { "connected": bool(control["connected"]), "queue_depth": int(control["queue_depth"]), "last_seq_id": control["last_seq_id"], "last_error": control["last_error"], "peer_id": control["peer_id"], "target_peer": control["target_peer"], "server_addr": control["server_addr"], "relay_via": control["relay_via"], "stats": control["stats"], "kcp_metrics": control_metrics, }, "policy": { "health_band": health_band, "recommended_video_profile": profile, }, "daemon": { "started_at": self._started_at, "socket_path": self._config["daemon"]["socket_path"], "version": VERSION, }, } with self._lock: self._state = state @staticmethod def _build_initial_state() -> dict[str, Any]: return { "network": { "peer_status": "offline", "latency_ms": 0.0, "jitter_ms": 0.0, "retrans_pct": 0.0, "packet_loss_pct": 0.0, "tx_kbps": 0, "rx_kbps": 0, "signal_dbm": None, "transport": "OmniSocket / daemon", "source_mode": "daemon-starting", "updated_at": utc_iso_now(), }, "video": { "available": False, "source_mode": "omnisocket-waiting", "frame_count": 0, "fps": 15, "frame_dir": "omni-daemon://latest-frame", "source_detail": "daemon is starting", "receiver": { "backend_ready": True, "mode": "daemon", "connected": False, "has_recent_frame": False, "frames_received": 0, "latest_sequence": None, "last_error": "", "config_path": "", "server_addr": "", "relay_via": "", "peer_id": "", "buffer_bytes": 0, }, }, "control": { "connected": False, "queue_depth": 0, "last_seq_id": None, "last_error": "", "peer_id": "", "target_peer": "", "server_addr": "", "relay_via": "", "stats": _default_stats(), "kcp_metrics": _default_kcp_metrics(), }, "policy": { "health_band": "red", "recommended_video_profile": { "fps": 5, "max_frame_bytes": 20 * 1024, }, }, "daemon": { "started_at": utc_iso_now(), "socket_path": DEFAULT_SOCKET_PATH, "version": VERSION, }, } class ThreadingUnixHTTPServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): daemon_threads = True class OmniDaemonHTTPHandler(BaseHTTPRequestHandler): server_version = "OmniDaemonHTTP/1.0" protocol_version = "HTTP/1.1" def do_GET(self) -> None: # pragma: no cover - exercised by runtime integration app: ASideOmniDaemon = self.server.app # type: ignore[attr-defined] if self.path == "/v1/health": state = app.get_state() control = state["control"] video = state["video"]["receiver"] policy = state["policy"] status = "ok" if policy["health_band"] == "green" else "degraded" if not control["connected"] and not video["connected"]: status = "unavailable" self._send_json( HTTPStatus.OK, { "status": status, "health_band": policy["health_band"], "control_connected": control["connected"], "video_connected": video["connected"], "updated_at": state["network"]["updated_at"], }, ) return if self.path == "/v1/state": self._send_json(HTTPStatus.OK, app.get_state()) return if self.path == "/v1/control/status": self._send_json(HTTPStatus.OK, app.get_state()["control"]) return if self.path == "/v1/video/frame": frame = app.get_latest_frame() if frame is None: self._send_json( HTTPStatus.SERVICE_UNAVAILABLE, {"error": "no fresh JPEG frame available"}, ) return self._send_bytes(HTTPStatus.OK, frame, "image/jpeg") return self._send_json(HTTPStatus.NOT_FOUND, {"error": f"unknown path: {self.path}"}) def do_POST(self) -> None: # pragma: no cover - exercised by runtime integration app: ASideOmniDaemon = self.server.app # type: ignore[attr-defined] if self.path != "/v1/control/event": self._send_json(HTTPStatus.NOT_FOUND, {"error": f"unknown path: {self.path}"}) return try: payload = self._read_json() except ValueError as error: self._send_json(HTTPStatus.BAD_REQUEST, {"error": str(error)}) return result = app.send_control_event( source=str(payload.get("source", "unknown")), event_code=str(payload.get("event_code", "")), drive_value=float(payload.get("drive_value", 1.0)), client_time_ms=payload.get("client_time_ms"), ) status = HTTPStatus.OK if result.get("accepted") else HTTPStatus.SERVICE_UNAVAILABLE self._send_json(status, result) def log_message(self, format: str, *args: Any) -> None: # noqa: A003 return def _read_json(self) -> dict[str, Any]: raw_length = self.headers.get("Content-Length") if raw_length is None: raise ValueError("missing Content-Length") try: length = int(raw_length) except ValueError as error: raise ValueError("invalid Content-Length") from error raw = self.rfile.read(length) try: payload = json.loads(raw.decode("utf-8")) except json.JSONDecodeError as error: raise ValueError(f"invalid JSON body: {error}") from error if not isinstance(payload, dict): raise ValueError("request body must be a JSON object") return payload def _send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None: raw = json.dumps(payload).encode("utf-8") self._send_bytes(status, raw, "application/json; charset=utf-8") def _send_bytes(self, status: HTTPStatus, payload: bytes, content_type: str) -> None: self.send_response(status.value) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(payload))) self.send_header("Cache-Control", "no-store") self.send_header("Connection", "keep-alive") self.end_headers() self.wfile.write(payload) class ASideOmniDaemon: def __init__(self, config_path: str | None = None) -> None: self._config = _load_config(config_path) self._control_manager = ControlSessionManager(self._config) self._video_manager = VideoSessionManager(self._config) self._telemetry = TelemetrySampler( self._config, self._control_manager, self._video_manager, ) self._server: ThreadingUnixHTTPServer | None = None @property def socket_path(self) -> str: return self._config["daemon"]["socket_path"] def start(self) -> None: self._control_manager.start() self._video_manager.start() self._telemetry.start() self._server = self._build_server() def stop(self) -> None: if self._server is not None: self._server.shutdown() self._server.server_close() self._server = None try: os.unlink(self.socket_path) except FileNotFoundError: pass self._telemetry.stop() self._video_manager.stop() self._control_manager.stop() def serve_forever(self) -> None: self.start() assert self._server is not None self._server.serve_forever() def get_state(self) -> dict[str, Any]: return self._telemetry.snapshot() def get_latest_frame(self) -> bytes | None: return self._video_manager.get_latest_frame() def send_control_event( self, *, source: str, event_code: str, drive_value: float, client_time_ms: int | None, ) -> dict[str, Any]: return self._control_manager.send_event( source=source, event_code=event_code, drive_value=drive_value, client_time_ms=client_time_ms, ) def _build_server(self) -> ThreadingUnixHTTPServer: socket_path = self.socket_path socket_dir = os.path.dirname(socket_path) if socket_dir: os.makedirs(socket_dir, exist_ok=True) try: os.unlink(socket_path) except FileNotFoundError: pass server = ThreadingUnixHTTPServer(socket_path, OmniDaemonHTTPHandler) server.app = self # type: ignore[attr-defined] return server def main(argv: list[str] | None = None) -> None: parser = argparse.ArgumentParser(description="Run the A-side OmniDaemon") parser.add_argument("--config", dest="config_path", help="Path to daemon YAML config") args = parser.parse_args(argv) app = ASideOmniDaemon(config_path=args.config_path) print( ( "A-side OmniDaemon starting " f"(config={app._config['config_path']}, " f"socket={app._config['daemon']['socket_path']})" ), file=sys.stderr, flush=True, ) def _handle_signal(_signum: int, _frame: Any) -> None: app.stop() raise SystemExit(0) signal.signal(signal.SIGINT, _handle_signal) signal.signal(signal.SIGTERM, _handle_signal) try: app.start() print( ( "A-side OmniDaemon ready " f"(state: curl --unix-socket {app.socket_path} http://localhost/v1/state)" ), file=sys.stderr, flush=True, ) assert app._server is not None app._server.serve_forever() except KeyboardInterrupt: pass finally: app.stop() if __name__ == "__main__": main()