Add Switch platform for Harbor (#177511)

homeassistant/components/harbor/const.py

@@ -6,7 +6,7 @@ DOMAIN = "harbor"
 MANUFACTURER = "Harbor"
 MODEL = "Harbor Camera"
 
-PLATFORMS: list[Platform] = [Platform.SENSOR]
+PLATFORMS: list[Platform] = [Platform.SENSOR, Platform.SWITCH]
 
 CONF_CERT_PEM = "cert_pem"
 CONF_KEY_PEM = "key_pem"

homeassistant/components/harbor/coordinator.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import logging
-from typing import Any, override
+from typing import TYPE_CHECKING, Any, override
 from uuid import uuid4
 
 from harbor.config import HarborCameraConfig
@@ -148,6 +148,13 @@ class HarborCoordinator(DataUpdateCoordinator[HarborDeviceState]):
         self._unsubscribe_updates()
         self.device.shutdown()
 
+    @property
+    def _client(self) -> HarborMQTTClient:
+        """Return the active MQTT client."""
+        if TYPE_CHECKING:
+            assert self._mqtt_client is not None
+        return self._mqtt_client
+
     @property
     def device_info(self) -> DeviceInfo:
         """Return device info for the Harbor camera."""
@@ -161,6 +168,18 @@ class HarborCoordinator(DataUpdateCoordinator[HarborDeviceState]):
             sw_version=state.os_version,
         )
 
+    async def async_set_camera_on(self, camera_on: bool) -> None:
+        """Turn the camera stream on or off."""
+        await self._client.set_camera_on(camera_on)
+
+    async def async_set_video_flip(self, video_flip: bool) -> None:
+        """Rotate the camera image 180 degrees, or restore it upright."""
+        await self._client.set_video_flip(video_flip)
+
+    async def async_set_clock_display(self, clock_display: bool) -> None:
+        """Show or hide the clock overlay burned into the video."""
+        await self._client.set_clock_display(clock_display)
+
     def _handle_device_update(self, state: HarborDeviceState) -> None:
         """Mirror a library device update into Home Assistant."""
         self._data_event.set()

homeassistant/components/harbor/icons.json

@@ -10,6 +10,17 @@
       "wifi_strength": {
         "default": "mdi:wifi"
       }
+    },
+    "switch": {
+      "camera_on": {
+        "default": "mdi:cctv"
+      },
+      "clock_display": {
+        "default": "mdi:clock-outline"
+      },
+      "video_flip": {
+        "default": "mdi:flip-vertical"
+      }
     }
   }
 }

homeassistant/components/harbor/strings.json

@@ -49,11 +49,28 @@
         "name": "Wi-Fi strength",
         "unit_of_measurement": "bars"
       }
+    },
+    "switch": {
+      "camera_on": {
+        "name": "Camera"
+      },
+      "clock_display": {
+        "name": "Clock overlay"
+      },
+      "video_flip": {
+        "name": "Flip image"
+      }
     }
   },
   "exceptions": {
     "cannot_connect": {
       "message": "Could not connect to the Harbor camera. It may be offline or unreachable."
+    },
+    "switch_turn_off_failed": {
+      "message": "Failed to turn off {switch}."
+    },
+    "switch_turn_on_failed": {
+      "message": "Failed to turn on {switch}."
     }
   }
 }

homeassistant/components/harbor/switch.py

@@ -0,0 +1,111 @@
+"""Switch entities for Harbor."""
+
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+from typing import Any, override
+
+from harbor import HarborCommandError
+
+from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
+from homeassistant.const import EntityCategory
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import DOMAIN
+from .coordinator import HarborConfigEntry, HarborCoordinator
+from .entity import HarborEntity
+
+# Commands are sent over a single MQTT session to one camera, and each settings
+# write is followed by a settings refresh, so they are serialized.
+PARALLEL_UPDATES = 1
+
+
+@dataclass(frozen=True, kw_only=True)
+class HarborSwitchEntityDescription(SwitchEntityDescription):
+    """Describes a Harbor switch entity."""
+
+    turn_on_fn: Callable[[HarborCoordinator], Coroutine[Any, Any, None]]
+    turn_off_fn: Callable[[HarborCoordinator], Coroutine[Any, Any, None]]
+
+
+CAMERA_SWITCHES: tuple[HarborSwitchEntityDescription, ...] = (
+    HarborSwitchEntityDescription(
+        key="camera_on",
+        translation_key="camera_on",
+        turn_on_fn=lambda coordinator: coordinator.async_set_camera_on(True),
+        turn_off_fn=lambda coordinator: coordinator.async_set_camera_on(False),
+    ),
+    HarborSwitchEntityDescription(
+        key="video_flip",
+        translation_key="video_flip",
+        entity_category=EntityCategory.CONFIG,
+        turn_on_fn=lambda coordinator: coordinator.async_set_video_flip(True),
+        turn_off_fn=lambda coordinator: coordinator.async_set_video_flip(False),
+    ),
+    HarborSwitchEntityDescription(
+        key="clock_display",
+        translation_key="clock_display",
+        entity_category=EntityCategory.CONFIG,
+        turn_on_fn=lambda coordinator: coordinator.async_set_clock_display(True),
+        turn_off_fn=lambda coordinator: coordinator.async_set_clock_display(False),
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: HarborConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Harbor switches from a config entry."""
+    coordinator = entry.runtime_data
+    async_add_entities(
+        HarborSwitch(coordinator, description) for description in CAMERA_SWITCHES
+    )
+
+
+class HarborSwitch(HarborEntity, SwitchEntity):
+    """A Harbor switch entity."""
+
+    entity_description: HarborSwitchEntityDescription
+
+    def __init__(
+        self,
+        coordinator: HarborCoordinator,
+        description: HarborSwitchEntityDescription,
+    ) -> None:
+        """Initialize the Harbor switch."""
+        self.entity_description = description
+        super().__init__(coordinator, description.key)
+
+    @override
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if the switch is on."""
+        return self.coordinator.data.values.get(self.entity_description.key)
+
+    @override
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn the switch on."""
+        await self._async_call(self.entity_description.turn_on_fn, "turn_on")
+
+    @override
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn the switch off."""
+        await self._async_call(self.entity_description.turn_off_fn, "turn_off")
+
+    async def _async_call(
+        self,
+        action: Callable[[HarborCoordinator], Coroutine[Any, Any, None]],
+        translation_key: str,
+    ) -> None:
+        """Run a switch command and translate library errors."""
+        try:
+            await action(self.coordinator)
+        except (HarborCommandError, TimeoutError, ConnectionError) as err:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key=f"switch_{translation_key}_failed",
+                translation_placeholders={"switch": self.entity_description.key},
+            ) from err

tests/components/harbor/conftest.py

@@ -23,7 +23,15 @@ KEY_PEM = "-----BEGIN PRIVATE KEY-----\nMIIBdummy\n-----END PRIVATE KEY-----"
 
 HEARTBEAT_TOPIC = f"cameras/{SERIAL}/events/heartbeat"
 LIVEKIT_TOPIC = f"cameras/{SERIAL}/events/local_livekit_heartbeat"
+SETTINGS_TOPIC = f"cameras/{SERIAL}/responses/get-settings"
 
+SETTINGS_PAYLOAD: dict[str, Any] = {
+    "settings": {
+        "preference_stream_paused": False,
+        "preference_video_flip": True,
+        "preference_video_has_clock_display": False,
+    },
+}
 HEARTBEAT_PAYLOAD: dict[str, Any] = {
     "temperature": 98.6,
     "os_version": "1.2.3",

tests/components/harbor/snapshots/test_switch.ambr

@@ -0,0 +1,151 @@
+# serializer version: 1
+# name: test_switches[switch.harbor_camera_1234567890_camera-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': list([
+      None,
+    ]),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': None,
+    'entity_id': 'switch.harbor_camera_1234567890_camera',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Camera',
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Camera',
+    'platform': 'harbor',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': 'camera_on',
+    'unique_id': '1234567890_camera_on',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_switches[switch.harbor_camera_1234567890_camera-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      <EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Harbor Camera 1234567890 Camera',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.harbor_camera_1234567890_camera',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'on',
+  })
+# ---
+# name: test_switches[switch.harbor_camera_1234567890_clock_overlay-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': list([
+      None,
+    ]),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': <EntityCategory.CONFIG: 'config'>,
+    'entity_id': 'switch.harbor_camera_1234567890_clock_overlay',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Clock overlay',
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Clock overlay',
+    'platform': 'harbor',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': 'clock_display',
+    'unique_id': '1234567890_clock_display',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_switches[switch.harbor_camera_1234567890_clock_overlay-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      <EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Harbor Camera 1234567890 Clock overlay',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.harbor_camera_1234567890_clock_overlay',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---
+# name: test_switches[switch.harbor_camera_1234567890_flip_image-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': list([
+      None,
+    ]),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'switch',
+    'entity_category': <EntityCategory.CONFIG: 'config'>,
+    'entity_id': 'switch.harbor_camera_1234567890_flip_image',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Flip image',
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Flip image',
+    'platform': 'harbor',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': 'video_flip',
+    'unique_id': '1234567890_video_flip',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_switches[switch.harbor_camera_1234567890_flip_image-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      <EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Harbor Camera 1234567890 Flip image',
+    }),
+    'context': <ANY>,
+    'entity_id': 'switch.harbor_camera_1234567890_flip_image',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'on',
+  })
+# ---

tests/components/harbor/test_sensor.py

@@ -1,12 +1,12 @@
 """Test the Harbor sensors."""
 
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, patch
 
 import pytest
 from syrupy.assertion import SnapshotAssertion
 
 from homeassistant.config_entries import ConfigEntryState
-from homeassistant.const import STATE_UNKNOWN
+from homeassistant.const import STATE_UNKNOWN, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 
@@ -31,7 +31,8 @@ async def test_sensors(
     snapshot: SnapshotAssertion,
 ) -> None:
     """Test the Harbor sensors report their values."""
-    await setup_integration(hass, mock_config_entry)
+    with patch("homeassistant.components.harbor.PLATFORMS", [Platform.SENSOR]):
+        await setup_integration(hass, mock_config_entry)
     assert mock_config_entry.state is ConfigEntryState.LOADED
 
     await emit_message(mock_mqtt_client, HEARTBEAT_TOPIC, HEARTBEAT_PAYLOAD)

tests/components/harbor/test_switch.py

@@ -0,0 +1,133 @@
+"""Test the Harbor switches."""
+
+from unittest.mock import AsyncMock, patch
+
+from harbor import HarborCommandError
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+
+from . import setup_integration
+from .conftest import SETTINGS_PAYLOAD, SETTINGS_TOPIC, emit_message
+
+from tests.common import MockConfigEntry, snapshot_platform
+
+CAMERA_ON_ENTITY = "switch.harbor_camera_1234567890_camera"
+VIDEO_FLIP_ENTITY = "switch.harbor_camera_1234567890_flip_image"
+CLOCK_DISPLAY_ENTITY = "switch.harbor_camera_1234567890_clock_overlay"
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_switches(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_config_entry: MockConfigEntry,
+    mock_mqtt_client: AsyncMock,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the Harbor switches report their state."""
+    with patch("homeassistant.components.harbor.PLATFORMS", [Platform.SWITCH]):
+        await setup_integration(hass, mock_config_entry)
+    assert mock_config_entry.state is ConfigEntryState.LOADED
+
+    await emit_message(mock_mqtt_client, SETTINGS_TOPIC, SETTINGS_PAYLOAD)
+    await hass.async_block_till_done()
+
+    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+
+
+@pytest.mark.parametrize(
+    ("entity_id", "library_method"),
+    [
+        pytest.param(CAMERA_ON_ENTITY, "set_camera_on", id="camera_on"),
+        pytest.param(VIDEO_FLIP_ENTITY, "set_video_flip", id="video_flip"),
+        pytest.param(CLOCK_DISPLAY_ENTITY, "set_clock_display", id="clock_display"),
+    ],
+)
+@pytest.mark.parametrize(
+    ("service", "expected"),
+    [
+        pytest.param(SERVICE_TURN_ON, True, id="turn_on"),
+        pytest.param(SERVICE_TURN_OFF, False, id="turn_off"),
+    ],
+)
+async def test_turn_on_and_off(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_mqtt_client: AsyncMock,
+    entity_id: str,
+    library_method: str,
+    service: str,
+    expected: bool,
+) -> None:
+    """Test turning each switch on and off calls the library."""
+    await setup_integration(hass, mock_config_entry)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        service,
+        {ATTR_ENTITY_ID: entity_id},
+        blocking=True,
+    )
+
+    mock_method = getattr(mock_mqtt_client.return_value, library_method)
+    mock_method.assert_awaited_once_with(expected)
+
+
+@pytest.mark.parametrize(
+    ("entity_id", "library_method"),
+    [
+        pytest.param(CAMERA_ON_ENTITY, "set_camera_on", id="camera_on"),
+        pytest.param(VIDEO_FLIP_ENTITY, "set_video_flip", id="video_flip"),
+        pytest.param(CLOCK_DISPLAY_ENTITY, "set_clock_display", id="clock_display"),
+    ],
+)
+@pytest.mark.parametrize(
+    "service",
+    [
+        pytest.param(SERVICE_TURN_ON, id="turn_on"),
+        pytest.param(SERVICE_TURN_OFF, id="turn_off"),
+    ],
+)
+@pytest.mark.parametrize(
+    "error",
+    [
+        pytest.param(
+            HarborCommandError("command", {"error": "rejected"}), id="command"
+        ),
+        pytest.param(TimeoutError, id="timeout"),
+        pytest.param(ConnectionError, id="connection"),
+    ],
+)
+async def test_command_failure_raises(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_mqtt_client: AsyncMock,
+    entity_id: str,
+    library_method: str,
+    service: str,
+    error: Exception | type[Exception],
+) -> None:
+    """Test a failed camera command surfaces as a HomeAssistantError."""
+    await setup_integration(hass, mock_config_entry)
+
+    getattr(mock_mqtt_client.return_value, library_method).side_effect = error
+
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            SWITCH_DOMAIN,
+            service,
+            {ATTR_ENTITY_ID: entity_id},
+            blocking=True,
+        )