198 lines
7.6 KiB
Python
198 lines
7.6 KiB
Python
"""Connection supervisor.
|
|
|
|
Owns the adapter's lifecycle so the web app never has to care whether the robot
|
|
is up. One background task:
|
|
|
|
disconnected -> connecting -> connected -> (health poll) -> disconnected -> ...
|
|
|
|
Connection failures are normal, not exceptional: they are published as status
|
|
events and retried with capped exponential backoff. The HTTP server keeps serving
|
|
the dashboard throughout, which is the whole point of the offline-first design.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from typing import Any, Dict, Optional
|
|
|
|
from ..config.settings import Settings
|
|
from ..core.events import EventBus, EventType
|
|
from .base import RobotAdapter, RobotError, RobotState
|
|
from .factory import create_robot
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RobotManager:
|
|
def __init__(self, settings: Settings, bus: EventBus) -> None:
|
|
self._settings = settings
|
|
self._bus = bus
|
|
self._adapter: RobotAdapter = create_robot(settings)
|
|
self._state = RobotState.DISCONNECTED
|
|
self._error: Optional[str] = None
|
|
self._error_code: Optional[str] = None
|
|
self._latency_ms: Optional[float] = None
|
|
self._connected_since: Optional[float] = None
|
|
self._attempts = 0
|
|
self._task: Optional[asyncio.Task] = None
|
|
self._wake = asyncio.Event()
|
|
self._closing = False
|
|
|
|
# -- accessors ----------------------------------------------------------- #
|
|
@property
|
|
def adapter(self) -> RobotAdapter:
|
|
return self._adapter
|
|
|
|
@property
|
|
def state(self) -> RobotState:
|
|
return self._state
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return self._state == RobotState.CONNECTED and self._adapter.is_connected
|
|
|
|
def status_dict(self) -> Dict[str, Any]:
|
|
info = self._adapter.info
|
|
return {
|
|
"state": self._state.value,
|
|
"connected": self.is_connected,
|
|
"speaking": self._adapter.is_speaking,
|
|
"robot": info.to_dict(),
|
|
"latencyMs": round(self._latency_ms, 1) if self._latency_ms is not None else None,
|
|
"uptimeSeconds": (
|
|
round(time.time() - self._connected_since, 1) if self._connected_since else None
|
|
),
|
|
"attempts": self._attempts,
|
|
"error": self._error,
|
|
"errorCode": self._error_code,
|
|
"configIssues": [issue.to_dict() for issue in self._settings.issues],
|
|
}
|
|
|
|
# -- lifecycle ----------------------------------------------------------- #
|
|
async def start(self) -> None:
|
|
self._closing = False
|
|
self._task = asyncio.create_task(self._supervise(), name="robot-supervisor")
|
|
self._publish()
|
|
|
|
async def stop(self) -> None:
|
|
self._closing = True
|
|
self._wake.set()
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
self._task = None
|
|
try:
|
|
await self._adapter.disconnect()
|
|
except Exception: # pragma: no cover
|
|
logger.debug("adapter disconnect raised", exc_info=True)
|
|
self._set_state(RobotState.DISCONNECTED)
|
|
|
|
def request_reconnect(self) -> None:
|
|
"""Ask the supervisor to retry immediately (used by the API)."""
|
|
self._wake.set()
|
|
|
|
async def rebuild(self, settings: Settings) -> None:
|
|
"""Swap in a new configuration without restarting the process."""
|
|
await self.stop()
|
|
self._settings = settings
|
|
self._adapter = create_robot(settings)
|
|
self._error = None
|
|
self._error_code = None
|
|
self._latency_ms = None
|
|
self._attempts = 0
|
|
await self.start()
|
|
|
|
# -- supervisor ---------------------------------------------------------- #
|
|
async def _supervise(self) -> None:
|
|
delay = self._settings.robot.reconnect_min_delay
|
|
while not self._closing:
|
|
try:
|
|
if not self._adapter.is_connected:
|
|
self._attempts += 1
|
|
self._set_state(RobotState.CONNECTING)
|
|
started = time.perf_counter()
|
|
await self._adapter.connect()
|
|
self._latency_ms = (time.perf_counter() - started) * 1000
|
|
self._connected_since = time.time()
|
|
self._error = None
|
|
self._error_code = None
|
|
delay = self._settings.robot.reconnect_min_delay
|
|
self._set_state(RobotState.CONNECTED)
|
|
logger.info("robot connected (%.0f ms)", self._latency_ms)
|
|
|
|
await self._sleep_or_wake(self._settings.robot.health_interval)
|
|
if self._closing:
|
|
break
|
|
|
|
started = time.perf_counter()
|
|
healthy = await self._adapter.health_check()
|
|
elapsed = (time.perf_counter() - started) * 1000
|
|
if healthy:
|
|
# Smooth the latency reading so the UI badge does not flicker.
|
|
self._latency_ms = (
|
|
elapsed if self._latency_ms is None else self._latency_ms * 0.7 + elapsed * 0.3
|
|
)
|
|
if self._state != RobotState.CONNECTED:
|
|
self._set_state(RobotState.CONNECTED)
|
|
else:
|
|
self._publish() # keep latency fresh in the dashboard
|
|
else:
|
|
logger.warning("robot health check failed - reconnecting")
|
|
self._connected_since = None
|
|
self._error = "Lost connection to the robot."
|
|
self._error_code = "unreachable"
|
|
self._set_state(RobotState.DISCONNECTED)
|
|
await self._safe_disconnect()
|
|
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except RobotError as exc:
|
|
self._on_failure(exc.user_message, exc.code)
|
|
delay = await self._backoff(delay)
|
|
except Exception as exc: # never let the supervisor die
|
|
logger.exception("unexpected supervisor error")
|
|
self._on_failure("Unexpected robot error: {0}".format(exc), "internal")
|
|
delay = await self._backoff(delay)
|
|
|
|
def _on_failure(self, message: str, code: str) -> None:
|
|
self._connected_since = None
|
|
self._latency_ms = None
|
|
self._error = message
|
|
self._error_code = code
|
|
self._set_state(RobotState.ERROR if code == "not_configured" else RobotState.DISCONNECTED)
|
|
|
|
async def _backoff(self, delay: float) -> float:
|
|
await self._sleep_or_wake(delay)
|
|
return min(delay * 2, self._settings.robot.reconnect_max_delay)
|
|
|
|
async def _sleep_or_wake(self, seconds: float) -> None:
|
|
"""Sleep, but return early if someone calls request_reconnect()."""
|
|
try:
|
|
await asyncio.wait_for(self._wake.wait(), timeout=max(0.1, seconds))
|
|
except asyncio.TimeoutError:
|
|
return
|
|
finally:
|
|
self._wake.clear()
|
|
|
|
async def _safe_disconnect(self) -> None:
|
|
try:
|
|
await self._adapter.disconnect()
|
|
except Exception: # pragma: no cover
|
|
logger.debug("disconnect during recovery raised", exc_info=True)
|
|
|
|
# -- events -------------------------------------------------------------- #
|
|
def _set_state(self, state: RobotState) -> None:
|
|
changed = state != self._state
|
|
self._state = state
|
|
if changed:
|
|
logger.info("robot state -> %s", state.value)
|
|
self._publish()
|
|
|
|
def _publish(self) -> None:
|
|
self._bus.publish(EventType.ROBOT_STATUS, self.status_dict())
|