feat: add get_setting command and run it on startup

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",

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,10 +56,45 @@ 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.
         Dispatches messages to all interested devices.
         """
         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

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."""
 

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."""

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)

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