| """Daemon-managed device pairing for the face-tracking and music application.""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import os |
| import urllib.error |
| import urllib.request |
| from collections.abc import Mapping |
|
|
|
|
| class DaemonDeviceManager: |
| """Ask the Runtime to pair and report status; never connect to the device directly.""" |
|
|
| def __init__(self, status_url: str, *, timeout: float = 0.8) -> None: |
| self._status_url = status_url.rstrip("/") |
| self._timeout = timeout |
|
|
| @classmethod |
| def from_environment(cls) -> "DaemonDeviceManager": |
| status_url = os.environ.get("WATCHER_APP_DEVICE_STATUS_URL", "").strip() |
| if not status_url: |
| raise RuntimeError("Daemon device status URL is unavailable") |
| return cls(status_url) |
|
|
| def status(self) -> Mapping[str, object]: |
| return self._device_from_response(self._request(self._status_url, method="GET")) |
|
|
| def pair(self, pairing_code: str) -> Mapping[str, object]: |
| if len(pairing_code) != 6 or not pairing_code.isdigit(): |
| raise RuntimeError("Pairing code must contain six digits") |
| return self._device_from_response( |
| self._request( |
| f"{self._status_url}/pair", |
| method="POST", |
| body={"pairing_code": pairing_code, "target_mode": "python_sdk"}, |
| ) |
| ) |
|
|
| @staticmethod |
| def _device_from_response(payload: Mapping[str, object]) -> Mapping[str, object]: |
| device = payload.get("device") |
| if not isinstance(device, dict): |
| raise RuntimeError("Daemon returned an invalid device response") |
| return device |
|
|
| def _request( |
| self, |
| url: str, |
| *, |
| method: str, |
| body: Mapping[str, object] | None = None, |
| ) -> Mapping[str, object]: |
| data = json.dumps(body).encode("utf-8") if body is not None else None |
| request = urllib.request.Request( |
| url, |
| data=data, |
| headers={"Content-Type": "application/json"} if data else {}, |
| method=method, |
| ) |
| try: |
| with urllib.request.urlopen(request, timeout=self._timeout) as response: |
| payload = json.loads(response.read().decode("utf-8")) |
| except (urllib.error.URLError, urllib.error.HTTPError, UnicodeError, json.JSONDecodeError) as error: |
| raise RuntimeError("Daemon device request failed") from error |
| if not isinstance(payload, dict): |
| raise RuntimeError("Daemon response is invalid") |
| return payload |
|
|
