Add Autofocus, IR lamp, and Wiper switches in ONVIF (#84317)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/onvif/__init__.py

@@ -44,6 +44,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     if device.capabilities.events:
         platforms += [Platform.BINARY_SENSOR, Platform.SENSOR]
 
+    if device.capabilities.imaging:
+        platforms += [Platform.SWITCH]
+
     await hass.config_entries.async_forward_entry_setups(entry, platforms)
 
     entry.async_on_unload(
@@ -60,8 +63,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     platforms = ["camera"]
 
     if device.capabilities.events and device.events.started:
-        platforms += ["binary_sensor", "sensor"]
+        platforms += [Platform.BINARY_SENSOR, Platform.SENSOR]
         await device.events.async_stop()
+    if device.capabilities.imaging:
+        platforms += [Platform.SWITCH]
 
     return await hass.config_entries.async_unload_platforms(entry, platforms)
 

homeassistant/components/onvif/device.py

@@ -296,7 +296,12 @@ class ONVIFDevice:
             self.device.get_definition("ptz")
             ptz = True
 
-        return Capabilities(snapshot, pullpoint, ptz)
+        imaging = False
+        with suppress(ONVIFError, Fault, RequestError):
+            self.device.create_imaging_service()
+            imaging = True
+
+        return Capabilities(snapshot, pullpoint, ptz, imaging)
 
     async def async_get_profiles(self) -> list[Profile]:
         """Obtain media profiles for this device."""
@@ -347,6 +352,12 @@ class ONVIFDevice:
                     # It's OK if Presets aren't supported
                     profile.ptz.presets = []
 
+            # Configure Imaging options
+            if self.capabilities.imaging and onvif_profile.VideoSourceConfiguration:
+                profile.video_source_token = (
+                    onvif_profile.VideoSourceConfiguration.SourceToken
+                )
+
             profiles.append(profile)
 
         return profiles
@@ -492,6 +503,63 @@ class ONVIFDevice:
             else:
                 LOGGER.error("Error trying to perform PTZ action: %s", err)
 
+    async def async_run_aux_command(
+        self,
+        profile: Profile,
+        cmd: str,
+    ) -> None:
+        """Execute a PTZ auxiliary command on the camera."""
+        if not self.capabilities.ptz:
+            LOGGER.warning("PTZ actions are not supported on device '%s'", self.name)
+            return
+
+        ptz_service = self.device.create_ptz_service()
+
+        LOGGER.debug(
+            "Running Aux Command | Cmd = %s",
+            cmd,
+        )
+        try:
+            req = ptz_service.create_type("SendAuxiliaryCommand")
+            req.ProfileToken = profile.token
+            req.AuxiliaryData = cmd
+            await ptz_service.SendAuxiliaryCommand(req)
+        except ONVIFError as err:
+            if "Bad Request" in err.reason:
+                LOGGER.warning("Device '%s' doesn't support PTZ", self.name)
+            else:
+                LOGGER.error("Error trying to send PTZ auxiliary command: %s", err)
+
+    async def async_set_imaging_settings(
+        self,
+        profile: Profile,
+        settings: dict,
+    ) -> None:
+        """Set an imaging setting on the ONVIF imaging service."""
+        # The Imaging Service is defined by ONVIF standard
+        # https://www.onvif.org/specs/srv/img/ONVIF-Imaging-Service-Spec-v210.pdf
+        if not self.capabilities.imaging:
+            LOGGER.warning(
+                "The imaging service is not supported on device '%s'", self.name
+            )
+            return
+
+        imaging_service = self.device.create_imaging_service()
+
+        LOGGER.debug("Setting Imaging Setting | Settings = %s", settings)
+        try:
+            req = imaging_service.create_type("SetImagingSettings")
+            req.VideoSourceToken = profile.video_source_token
+            req.ImagingSettings = settings
+            await imaging_service.SetImagingSettings(req)
+        except ONVIFError as err:
+            if "Bad Request" in err.reason:
+                LOGGER.warning(
+                    "Device '%s' doesn't support the Imaging Service", self.name
+                )
+            else:
+                LOGGER.error("Error trying to set Imaging settings: %s", err)
+
 
 def get_device(hass, host, port, username, password) -> ONVIFCamera:
     """Get ONVIFCamera instance."""

homeassistant/components/onvif/models.py

@@ -53,6 +53,7 @@ class Profile:
     name: str
     video: Video
     ptz: PTZ | None = None
+    video_source_token: str | None = None
 
 
 @dataclass
@@ -62,6 +63,7 @@ class Capabilities:
     snapshot: bool = False
     events: bool = False
     ptz: bool = False
+    imaging: bool = False
 
 
 @dataclass

homeassistant/components/onvif/switch.py

@@ -0,0 +1,110 @@
+"""ONVIF switches for controlling cameras."""
+from __future__ import annotations
+
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+from typing import Any
+
+from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .base import ONVIFBaseEntity
+from .const import DOMAIN
+from .device import ONVIFDevice
+from .models import Profile
+
+
+@dataclass
+class ONVIFSwitchEntityDescriptionMixin:
+    """Mixin for required keys."""
+
+    turn_on_fn: Callable[
+        [ONVIFDevice], Callable[[Profile, Any], Coroutine[Any, Any, None]]
+    ]
+    turn_off_fn: Callable[
+        [ONVIFDevice], Callable[[Profile, Any], Coroutine[Any, Any, None]]
+    ]
+    turn_on_data: Any
+    turn_off_data: Any
+
+
+@dataclass
+class ONVIFSwitchEntityDescription(
+    SwitchEntityDescription, ONVIFSwitchEntityDescriptionMixin
+):
+    """Describes ONVIF switch entity."""
+
+
+SWITCHES: tuple[ONVIFSwitchEntityDescription, ...] = (
+    ONVIFSwitchEntityDescription(
+        key="autofocus",
+        name="Autofocus",
+        icon="mdi:focus-auto",
+        turn_on_data={"Focus": {"AutoFocusMode": "AUTO"}},
+        turn_off_data={"Focus": {"AutoFocusMode": "MANUAL"}},
+        turn_on_fn=lambda device: device.async_set_imaging_settings,
+        turn_off_fn=lambda device: device.async_set_imaging_settings,
+    ),
+    ONVIFSwitchEntityDescription(
+        key="ir_lamp",
+        name="IR lamp",
+        icon="mdi:spotlight-beam",
+        turn_on_data={"IrCutFilter": "OFF"},
+        turn_off_data={"IrCutFilter": "ON"},
+        turn_on_fn=lambda device: device.async_set_imaging_settings,
+        turn_off_fn=lambda device: device.async_set_imaging_settings,
+    ),
+    ONVIFSwitchEntityDescription(
+        key="wiper",
+        name="Wiper",
+        icon="mdi:wiper",
+        turn_on_data="tt:Wiper|On",
+        turn_off_data="tt:Wiper|Off",
+        turn_on_fn=lambda device: device.async_run_aux_command,
+        turn_off_fn=lambda device: device.async_run_aux_command,
+    ),
+)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up a ONVIF switch platform."""
+    device = hass.data[DOMAIN][config_entry.unique_id]
+
+    async_add_entities(ONVIFSwitch(device, description) for description in SWITCHES)
+
+
+class ONVIFSwitch(ONVIFBaseEntity, SwitchEntity):
+    """An ONVIF switch."""
+
+    entity_description: ONVIFSwitchEntityDescription
+    _attr_has_entity_name = True
+
+    def __init__(
+        self, device: ONVIFDevice, description: ONVIFSwitchEntityDescription
+    ) -> None:
+        """Initialize the switch."""
+        super().__init__(device)
+        self._attr_unique_id = f"{self.mac_or_serial}_{description.key}"
+        self.entity_description = description
+
+    async def async_turn_on(self, **kwargs: Any) -> None:
+        """Turn on switch."""
+        self._attr_is_on = True
+        profile = self.device.profiles[0]
+        await self.entity_description.turn_on_fn(self.device)(
+            profile, self.entity_description.turn_on_data
+        )
+
+    async def async_turn_off(self, **kwargs: Any) -> None:
+        """Turn off switch."""
+        self._attr_is_on = False
+        profile = self.device.profiles[0]
+        await self.entity_description.turn_off_fn(self.device)(
+            profile, self.entity_description.turn_off_data
+        )

tests/components/onvif/__init__.py

@@ -6,7 +6,7 @@ from zeep.exceptions import Fault
 from homeassistant import config_entries
 from homeassistant.components.onvif import config_flow
 from homeassistant.components.onvif.const import CONF_SNAPSHOT_AUTH
-from homeassistant.components.onvif.models import Capabilities, DeviceInfo
+from homeassistant.components.onvif.models import Capabilities, DeviceInfo, Profile
 from homeassistant.const import HTTP_DIGEST_AUTHENTICATION
 
 from tests.common import MockConfigEntry
@@ -95,8 +95,16 @@ def setup_mock_device(mock_device):
         SERIAL_NUMBER,
         MAC,
     )
-    mock_device.capabilities = Capabilities()
-    mock_device.profiles = []
+    mock_device.capabilities = Capabilities(imaging=True)
+    profile1 = Profile(
+        index=0,
+        token="dummy",
+        name="profile1",
+        video=None,
+        ptz=None,
+        video_source_token=None,
+    )
+    mock_device.profiles = [profile1]
 
     def mock_constructor(hass, config):
         """Fake the controller constructor."""

tests/components/onvif/test_diagnostics.py

@@ -48,7 +48,21 @@ async def test_diagnostics(hass, hass_client):
                 "serial_number": SERIAL_NUMBER,
                 "mac": MAC,
             },
-            "capabilities": {"snapshot": False, "events": False, "ptz": False},
-            "profiles": [],
+            "capabilities": {
+                "snapshot": False,
+                "events": False,
+                "ptz": False,
+                "imaging": True,
+            },
+            "profiles": [
+                {
+                    "index": 0,
+                    "token": "dummy",
+                    "name": "profile1",
+                    "video": None,
+                    "ptz": None,
+                    "video_source_token": None,
+                }
+            ],
         },
     }

tests/components/onvif/test_switch.py

@@ -0,0 +1,161 @@
+"""Test switch platform of ONVIF integration."""
+from unittest.mock import AsyncMock
+
+from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
+from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNKNOWN
+from homeassistant.helpers import entity_registry as er
+
+from . import MAC, setup_onvif_integration
+
+
+async def test_wiper_switch(hass):
+    """Test states of the Wiper switch."""
+    _config, _camera, device = await setup_onvif_integration(hass)
+    device.profiles = device.async_get_profiles()
+
+    state = hass.states.get("switch.testcamera_wiper")
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+    registry = er.async_get(hass)
+    entry = registry.async_get("switch.testcamera_wiper")
+    assert entry
+    assert entry.unique_id == f"{MAC}_wiper"
+
+
+async def test_turn_wiper_switch_on(hass):
+    """Test Wiper switch turn on."""
+    _, _camera, device = await setup_onvif_integration(hass)
+    device.async_run_aux_command = AsyncMock(return_value=True)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: "switch.testcamera_wiper"},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    device.async_run_aux_command.assert_called_once()
+    state = hass.states.get("switch.testcamera_wiper")
+    assert state.state == STATE_ON
+
+
+async def test_turn_wiper_switch_off(hass):
+    """Test Wiper switch turn off."""
+    _, _camera, device = await setup_onvif_integration(hass)
+    device.async_run_aux_command = AsyncMock(return_value=True)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        "turn_off",
+        {ATTR_ENTITY_ID: "switch.testcamera_wiper"},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    device.async_run_aux_command.assert_called_once()
+    state = hass.states.get("switch.testcamera_wiper")
+    assert state.state == STATE_OFF
+
+
+async def test_autofocus_switch(hass):
+    """Test states of the autofocus switch."""
+    _config, _camera, device = await setup_onvif_integration(hass)
+    device.profiles = device.async_get_profiles()
+
+    state = hass.states.get("switch.testcamera_autofocus")
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+    registry = er.async_get(hass)
+    entry = registry.async_get("switch.testcamera_autofocus")
+    assert entry
+    assert entry.unique_id == f"{MAC}_autofocus"
+
+
+async def test_turn_autofocus_switch_on(hass):
+    """Test autofocus switch turn on."""
+    _, _camera, device = await setup_onvif_integration(hass)
+    device.async_set_imaging_settings = AsyncMock(return_value=True)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: "switch.testcamera_autofocus"},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    device.async_set_imaging_settings.assert_called_once()
+    state = hass.states.get("switch.testcamera_autofocus")
+    assert state.state == STATE_ON
+
+
+async def test_turn_autofocus_switch_off(hass):
+    """Test autofocus switch turn off."""
+    _, _camera, device = await setup_onvif_integration(hass)
+    device.async_set_imaging_settings = AsyncMock(return_value=True)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        "turn_off",
+        {ATTR_ENTITY_ID: "switch.testcamera_autofocus"},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    device.async_set_imaging_settings.assert_called_once()
+    state = hass.states.get("switch.testcamera_autofocus")
+    assert state.state == STATE_OFF
+
+
+async def test_infrared_switch(hass):
+    """Test states of the autofocus switch."""
+    _config, _camera, device = await setup_onvif_integration(hass)
+    device.profiles = device.async_get_profiles()
+
+    state = hass.states.get("switch.testcamera_ir_lamp")
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+    registry = er.async_get(hass)
+    entry = registry.async_get("switch.testcamera_ir_lamp")
+    assert entry
+    assert entry.unique_id == f"{MAC}_ir_lamp"
+
+
+async def test_turn_infrared_switch_on(hass):
+    """Test infrared switch turn on."""
+    _, _camera, device = await setup_onvif_integration(hass)
+    device.async_set_imaging_settings = AsyncMock(return_value=True)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        "turn_on",
+        {ATTR_ENTITY_ID: "switch.testcamera_ir_lamp"},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    device.async_set_imaging_settings.assert_called_once()
+    state = hass.states.get("switch.testcamera_ir_lamp")
+    assert state.state == STATE_ON
+
+
+async def test_turn_infrared_switch_off(hass):
+    """Test infrared switch turn off."""
+    _, _camera, device = await setup_onvif_integration(hass)
+    device.async_set_imaging_settings = AsyncMock(return_value=True)
+
+    await hass.services.async_call(
+        SWITCH_DOMAIN,
+        "turn_off",
+        {ATTR_ENTITY_ID: "switch.testcamera_ir_lamp"},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    device.async_set_imaging_settings.assert_called_once()
+    state = hass.states.get("switch.testcamera_ir_lamp")
+    assert state.state == STATE_OFF