diff --git a/harbor/__init__.py b/harbor/__init__.py index 7be13e1..89ac2f7 100644 --- a/harbor/__init__.py +++ b/harbor/__init__.py @@ -1,6 +1,7 @@ from .config import HarborCameraConfig from .core import Harbor from .data.mqtt_models import ( + GetCameraSettingsRequest, HeartbeatEvent, LocalLivekitHeartbeatEvent, MotionDetectedEvent, @@ -49,6 +50,7 @@ "MotionDetectedUpdate", "ViewerInfo", "parse_message", + "GetCameraSettingsRequest", "HeartbeatEvent", "LocalLivekitHeartbeatEvent", "SettingsEvent", diff --git a/harbor/core.py b/harbor/core.py index af6909f..6200c42 100644 --- a/harbor/core.py +++ b/harbor/core.py @@ -2,8 +2,9 @@ from typing import Any from .config import HarborCameraConfig +from .data.mqtt_models import SettingsEvent from .device import HarborDevice -from .mqtt import HarborMQTTClient +from .mqtt import DEFAULT_INITIAL_COMMANDS, HarborMQTTClient _LOGGER = logging.getLogger(__name__) @@ -32,12 +33,13 @@ def add_camera_connection(self, config: HarborCameraConfig) -> None: if config.serial in self._clients: _LOGGER.warning("Camera connection already exists: %s", config.serial) return - topics = list(self._topics_cache) + topics = list({*self._topics_cache, f"cameras/{config.serial}/responses/#"}) client = HarborMQTTClient( config=config, topics=topics, message_handler=self.handle_message, client_id=f"harbor-client-{config.serial}", + initial_commands=DEFAULT_INITIAL_COMMANDS, ) self._clients[config.serial] = client _LOGGER.info("Added MQTT client for camera: %s", config.serial) @@ -54,6 +56,35 @@ async def stop(self) -> None: for device in self._devices.values(): device.shutdown() + async def publish_camera_command( + self, + serial: str, + command: str, + payload: Any, + ) -> None: + """Publish a command to a camera.""" + await self._get_client(serial).publish_command(command, payload) + + async def request_camera_command( + self, + serial: str, + command: str, + payload: dict[str, Any], + *, + timeout: float = 10.0, + ) -> Any: + """Publish a command to a camera and wait for the matching response.""" + return await self._get_client(serial).request_command(command, payload, timeout=timeout) + + async def get_camera_settings( + self, + serial: str, + *, + timeout: float = 10.0, + ) -> SettingsEvent: + """Request the camera settings payload.""" + return await self._get_client(serial).get_settings(timeout=timeout) + async def handle_message(self, topic: str, payload: Any) -> None: """ Central message handler. @@ -61,3 +92,9 @@ async def handle_message(self, topic: str, payload: Any) -> None: """ for device in self._devices.values(): await device.handle_message(topic, payload) + + def _get_client(self, serial: str) -> HarborMQTTClient: + try: + return self._clients[serial] + except KeyError as exc: + raise KeyError(f"No camera connection exists for serial {serial!r}") from exc diff --git a/harbor/data/mqtt_models.py b/harbor/data/mqtt_models.py index e81fcf9..0a66444 100644 --- a/harbor/data/mqtt_models.py +++ b/harbor/data/mqtt_models.py @@ -8,7 +8,7 @@ class HarborMQTTPayload(BaseModel): """Base model for Harbor MQTT payloads.""" - model_config = ConfigDict(extra="allow") + model_config = ConfigDict(extra="allow", populate_by_name=True) class LocalLivekitHeartbeatEvent(HarborMQTTPayload): @@ -82,14 +82,22 @@ class SettingsEvent(HarborMQTTPayload): """Payload for a settings event.""" client: str | None = None - is_updating: bool | None = None + is_updating: bool | None = Field(default=None, alias="isUpdating") seq: str | None = None settings: Settings | None = None state: SettingsState | None = None - triggered_by: str | None = None + triggered_by: str | None = Field(default=None, alias="triggeredBy") updated: dict[str, Any] = Field(default_factory=dict) +class GetCameraSettingsRequest(HarborMQTTPayload): + """Payload for the get-settings camera command.""" + + seq: str + client: str + triggered_by: str = Field(alias="triggeredBy") + + class ViewerJoinedEvent(HarborMQTTPayload): """Payload for a viewer joined event.""" diff --git a/harbor/devices/camera.py b/harbor/devices/camera.py index 821d71f..d703198 100644 --- a/harbor/devices/camera.py +++ b/harbor/devices/camera.py @@ -46,7 +46,10 @@ def __init__(self, config: HarborCameraConfig) -> None: def get_topics(self) -> list[str]: """Return topics that should be subscribed for this device.""" - return [f"cameras/{self.serial}/events/#"] + return [ + f"cameras/{self.serial}/events/#", + f"cameras/{self.serial}/responses/#", + ] def _apply_event(self, event: HarborEvent) -> None: """Apply a Harbor event to camera state, including camera-only events.""" diff --git a/harbor/events.py b/harbor/events.py index 5ef854f..822155a 100644 --- a/harbor/events.py +++ b/harbor/events.py @@ -178,7 +178,7 @@ def parse_topic( return None, None, None root, serial, event_root, *rest = parts - if event_root != "events" or not rest: + if event_root not in {"events", "responses"} or not rest: return None, None, None if root == "cameras": @@ -336,7 +336,7 @@ def parse_message( ) return RawEventUpdate(payload=raw_payload, **base_kwargs) - if event_key == "settings": + if event_key in {"settings", "get_settings"}: if typed_payload := _validate_payload(SettingsEvent, raw_payload, topic): return SettingsUpdate(payload=typed_payload, **base_kwargs) return RawEventUpdate(payload=raw_payload, **base_kwargs) diff --git a/harbor/mqtt.py b/harbor/mqtt.py index c9f3e01..42377ef 100644 --- a/harbor/mqtt.py +++ b/harbor/mqtt.py @@ -4,15 +4,21 @@ import sys from collections.abc import Awaitable, Callable from typing import Any +from uuid import uuid4 from aiomqtt import Client, MqttError from .config import HarborCameraConfig +from .data.mqtt_models import GetCameraSettingsRequest, SettingsEvent from .utils import get_camera_host, get_ssl_cache_key, get_ssl_context _LOGGER = logging.getLogger(__name__) DEFAULT_CONNECTION_GRACE_PERIOD = 90.0 +DEFAULT_COMMAND_QOS = 2 +DEFAULT_REQUEST_TIMEOUT = 10.0 +GET_SETTINGS_COMMAND = "get-settings" +DEFAULT_INITIAL_COMMANDS = (GET_SETTINGS_COMMAND,) class HarborMQTTClient: @@ -25,6 +31,7 @@ def __init__( ssl_context_cache: dict | None = None, on_connection_change: Callable[[bool], Awaitable[None]] | None = None, connection_grace_period: float = DEFAULT_CONNECTION_GRACE_PERIOD, + initial_commands: list[str] | tuple[str, ...] | None = None, ) -> None: """Initialize the MQTT client. @@ -42,9 +49,12 @@ def __init__( self.ssl_context_cache = ssl_context_cache or {} self.on_connection_change = on_connection_change self.connection_grace_period = connection_grace_period + self.initial_commands = tuple(initial_commands or ()) self.connected: bool = False self._stop_event = asyncio.Event() self._task: asyncio.Task | None = None + self._client: Client | None = None + self._pending_responses: dict[str, asyncio.Future[Any]] = {} self._reported_connected: bool | None = None self._disconnect_grace_task: asyncio.Task | None = None @@ -55,6 +65,23 @@ async def _handle_message(self, topic: str, payload_raw: str) -> None: payload = payload_raw await self.message_handler(topic, payload) + self._resolve_pending_response(topic, payload) + + def _resolve_pending_response(self, topic: str, payload: Any) -> None: + """Resolve a pending request when a camera response echoes its seq.""" + if not topic.startswith(f"cameras/{self.config.serial}/responses/"): + return + if not isinstance(payload, dict): + return + + seq = payload.get("seq") + if not isinstance(seq, str): + return + + future = self._pending_responses.pop(seq, None) + if future is None or future.done(): + return + future.set_result(payload) async def _set_connected(self, connected: bool) -> None: """Update the raw connection flag and debounce listener notifications.""" @@ -108,6 +135,12 @@ async def _notify_connection_change(self, connected: bool) -> None: def _invalidate_ssl_cache(self) -> None: self.ssl_context_cache.pop(get_ssl_cache_key(self.config), None) + def _fail_pending_responses(self, exc: Exception) -> None: + for future in self._pending_responses.values(): + if not future.done(): + future.set_exception(exc) + self._pending_responses.clear() + async def run(self) -> None: reconnect_delay = 2 @@ -143,6 +176,7 @@ async def run(self) -> None: timeout=10, identifier=self.client_id, ) as client: + self._client = client _LOGGER.info( "Harbor: MQTT connected to %s:%s for camera %s", host, @@ -159,6 +193,8 @@ async def run(self) -> None: self.config.serial, ) + await self._publish_initial_commands() + async for message in client.messages: if self._stop_event.is_set(): break @@ -177,6 +213,8 @@ async def run(self) -> None: reconnect_delay = 2 finally: + self._client = None + self._fail_pending_responses(ConnectionError("Harbor MQTT client disconnected")) await self._set_connected(False) except TimeoutError as e: @@ -247,9 +285,131 @@ async def stop(self) -> None: _LOGGER.info("Harbor: MQTT client stopped for camera %s", self.config.serial) # An intentional stop is a stable disconnect: skip the grace period. self._cancel_disconnect_grace() + self._client = None + self._fail_pending_responses(ConnectionError("Harbor MQTT client stopped")) if self._reported_connected: await self._notify_connection_change(False) + async def publish( + self, + topic: str, + payload: Any, + *, + qos: int = DEFAULT_COMMAND_QOS, + retain: bool = False, + ) -> None: + """Publish a JSON-compatible payload to a Harbor MQTT topic.""" + client = self._client + if client is None or not self.connected: + raise ConnectionError(f"Harbor MQTT client is not connected for camera {self.config.serial}") + + if isinstance(payload, str): + payload_raw = payload + else: + payload_raw = json.dumps(payload, separators=(",", ":")) + + _LOGGER.debug( + "Harbor: MQTT publishing to topic '%s' for camera %s: %s", + topic, + self.config.serial, + payload_raw, + ) + await client.publish(topic, payload_raw, qos=qos, retain=retain) + + async def publish_command( + self, + command: str, + payload: Any, + *, + qos: int = DEFAULT_COMMAND_QOS, + ) -> None: + """Publish a camera command using the app's command topic layout.""" + await self.publish(f"cameras/{self.config.serial}/{command}", payload, qos=qos) + + async def _publish_initial_commands(self) -> None: + """Publish configured one-shot state population commands after connect.""" + for command in self.initial_commands: + try: + if command == GET_SETTINGS_COMMAND: + await self.publish_command(command, self._build_get_settings_payload()) + else: + _LOGGER.warning( + "Harbor: skipping unsupported initial command %s for camera %s", + command, + self.config.serial, + ) + except Exception: + _LOGGER.exception( + "Harbor: failed to publish initial command %s for camera %s", + command, + self.config.serial, + ) + + def _build_get_settings_payload( + self, + *, + client: str | None = None, + triggered_by: str | None = None, + seq: str | None = None, + ) -> dict[str, Any]: + request = GetCameraSettingsRequest( + seq=seq or _generate_seq(), + client=client or self.client_id or f"harbor-client-{self.config.serial}", + triggered_by=triggered_by or "harbor-python", + ) + return request.model_dump(by_alias=True) + + async def request_command( + self, + command: str, + payload: dict[str, Any], + *, + seq: str | None = None, + timeout: float = DEFAULT_REQUEST_TIMEOUT, + qos: int = DEFAULT_COMMAND_QOS, + ) -> Any: + """Publish a command and wait for a response carrying the same seq.""" + request_seq = seq or _generate_seq() + payload = {**payload, "seq": request_seq} + loop = asyncio.get_running_loop() + future: asyncio.Future[Any] = loop.create_future() + self._pending_responses[request_seq] = future + + try: + await self.publish_command(command, payload, qos=qos) + return await asyncio.wait_for(future, timeout=timeout) + except Exception: + pending = self._pending_responses.pop(request_seq, None) + if pending is not None and not pending.done(): + pending.cancel() + raise + + async def get_settings( + self, + *, + client: str | None = None, + triggered_by: str | None = None, + timeout: float = DEFAULT_REQUEST_TIMEOUT, + ) -> SettingsEvent: + """Request camera settings via the get-settings command.""" + seq = _generate_seq() + response = await self.request_command( + GET_SETTINGS_COMMAND, + self._build_get_settings_payload( + seq=seq, + client=client, + triggered_by=triggered_by, + ), + seq=seq, + timeout=timeout, + ) + return SettingsEvent.model_validate(response) + def __del__(self) -> None: if self._stop_event and not self._stop_event.is_set(): self._stop_event.set() + + +def _generate_seq() -> str: + """Generate a request sequence string echoed by Harbor responses.""" + return uuid4().hex diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..49face1 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from harbor.config import HarborCameraConfig +from harbor.core import Harbor +from harbor.mqtt import DEFAULT_INITIAL_COMMANDS + + +def test_camera_connection_populates_settings_on_connect() -> None: + """Camera connections should request initial state after connecting.""" + + harbor = Harbor() + config = HarborCameraConfig( + serial="TEST123", + cert_path="/path/to/cert.pem", + key_path="/path/to/key.pem", + ) + + harbor.add_camera_connection(config) + + assert harbor._clients["TEST123"].initial_commands == DEFAULT_INITIAL_COMMANDS diff --git a/tests/test_mqtt.py b/tests/test_mqtt.py index de98979..d746f23 100644 --- a/tests/test_mqtt.py +++ b/tests/test_mqtt.py @@ -1,9 +1,10 @@ from __future__ import annotations import asyncio +import json from harbor.config import HarborCameraConfig -from harbor.mqtt import HarborMQTTClient +from harbor.mqtt import GET_SETTINGS_COMMAND, HarborMQTTClient def _create_config() -> HarborCameraConfig: @@ -137,3 +138,137 @@ async def test_stop_flushes_pending_disconnect() -> None: await client.stop() assert changes == [True, False] + + +class _FakePublishClient: + def __init__(self) -> None: + self.published: list[tuple[str, str, int, bool]] = [] + + async def publish(self, topic: str, payload: str, *, qos: int, retain: bool) -> None: + self.published.append((topic, payload, qos, retain)) + + +async def test_request_command_publishes_and_waits_for_matching_response() -> None: + """Requests should publish to camera commands and resolve from response seq.""" + + messages: list[tuple[str, object]] = [] + + async def message_handler(topic: str, payload: object) -> None: + messages.append((topic, payload)) + + client = HarborMQTTClient( + config=_create_config(), + topics=[], + message_handler=message_handler, + ) + fake_client = _FakePublishClient() + client.connected = True + client._client = fake_client + + task = asyncio.create_task( + client.request_command( + "get-settings", + {"seq": "seq-1", "client": "test-client", "triggeredBy": "harbor-python"}, + seq="seq-1", + timeout=1, + ) + ) + await asyncio.sleep(0) + + assert fake_client.published == [ + ( + "cameras/TEST123/get-settings", + '{"seq":"seq-1","client":"test-client","triggeredBy":"harbor-python"}', + 2, + False, + ) + ] + + response = { + "seq": "seq-1", + "client": "test-client", + "isUpdating": False, + "settings": {"preference_display_name": "Nursery"}, + } + await client._handle_message("cameras/TEST123/responses/get-settings", json.dumps(response)) + + assert await task == response + assert messages == [("cameras/TEST123/responses/get-settings", response)] + + +async def test_get_settings_uses_app_payload_shape() -> None: + """The get-settings helper should use the APK's command topic and field names.""" + + async def message_handler(topic: str, payload: object) -> None: + pass + + client = HarborMQTTClient( + config=_create_config(), + topics=[], + message_handler=message_handler, + ) + fake_client = _FakePublishClient() + client.connected = True + client._client = fake_client + + task = asyncio.create_task(client.get_settings(client="test-client", triggered_by="users/user1", timeout=1)) + await asyncio.sleep(0) + + topic, payload_raw, qos, retain = fake_client.published[0] + payload = json.loads(payload_raw) + assert topic == "cameras/TEST123/get-settings" + assert payload["client"] == "test-client" + assert payload["triggeredBy"] == "users/user1" + assert isinstance(payload["seq"], str) + assert qos == 2 + assert retain is False + + await client._handle_message( + "cameras/TEST123/responses/get-settings", + json.dumps( + { + "seq": payload["seq"], + "client": "test-client", + "triggeredBy": "users/user1", + "isUpdating": False, + "settings": {"preference_display_name": "Nursery"}, + } + ), + ) + + settings = await task + assert settings.seq == payload["seq"] + assert settings.triggered_by == "users/user1" + assert settings.is_updating is False + assert settings.settings is not None + assert settings.settings.preference_display_name == "Nursery" + + +async def test_initial_commands_publish_get_settings_without_waiting() -> None: + """Initial populate commands should request settings after connection.""" + + async def message_handler(topic: str, payload: object) -> None: + pass + + client = HarborMQTTClient( + config=_create_config(), + topics=[], + message_handler=message_handler, + client_id="test-client", + initial_commands=[GET_SETTINGS_COMMAND], + ) + fake_client = _FakePublishClient() + client.connected = True + client._client = fake_client + + await client._publish_initial_commands() + + assert len(fake_client.published) == 1 + topic, payload_raw, qos, retain = fake_client.published[0] + payload = json.loads(payload_raw) + assert topic == "cameras/TEST123/get-settings" + assert payload["client"] == "test-client" + assert payload["triggeredBy"] == "harbor-python" + assert isinstance(payload["seq"], str) + assert qos == 2 + assert retain is False diff --git a/tests/test_subscription.py b/tests/test_subscription.py index 0a5730e..0cc1bbd 100644 --- a/tests/test_subscription.py +++ b/tests/test_subscription.py @@ -9,6 +9,7 @@ HarborEventBus, HeartbeatUpdate, MotionDetectedUpdate, + SettingsUpdate, ViewerJoinedUpdate, extract_explicit_event_state, ) @@ -104,6 +105,33 @@ async def on_any_event(event: HarborEvent) -> None: unsubscribe() +async def test_get_settings_response_updates_camera_display_name() -> None: + """responses/get-settings should parse as settings and update friendly name.""" + + camera = _create_camera() + events_received: list[SettingsUpdate] = [] + camera.subscribe(SettingsUpdate, lambda event: events_received.append(event)) + + await camera.handle_message( + "cameras/TEST123/responses/get-settings", + { + "seq": "seq-1", + "client": "test-client", + "triggeredBy": "users/user1", + "isUpdating": False, + "settings": {"preference_display_name": "Nursery"}, + "state": {"network_bars": 3, "temperature": 22.5}, + }, + ) + + assert len(events_received) == 1 + assert events_received[0].event_type is EventType.SETTINGS + assert events_received[0].event_key == "get_settings" + assert camera.state.display_name == "Nursery" + assert camera.state.values["wifi_strength"] == 3 + assert camera.state.values["temperature"] == 22.5 + + def test_event_bus_parses_generic_camera_events() -> None: """The package should normalize generic camera events."""