diff --git a/homeassistant/components/onvif/__init__.py b/homeassistant/components/onvif/__init__.py index ac20a564c8a..77a5c6d1bd8 100644 --- a/homeassistant/components/onvif/__init__.py +++ b/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) diff --git a/homeassistant/components/onvif/device.py b/homeassistant/components/onvif/device.py index 3398c905d58..3d74f1e4664 100644 --- a/homeassistant/components/onvif/device.py +++ b/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.""" diff --git a/homeassistant/components/onvif/models.py b/homeassistant/components/onvif/models.py index 6cefa6332e2..6da8595e2f6 100644 --- a/homeassistant/components/onvif/models.py +++ b/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 diff --git a/homeassistant/components/onvif/switch.py b/homeassistant/components/onvif/switch.py new file mode 100644 index 00000000000..c72a826a79c --- /dev/null +++ b/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 + ) diff --git a/tests/components/onvif/__init__.py b/tests/components/onvif/__init__.py index 5bfc2ca53fd..d90ec02159f 100644 --- a/tests/components/onvif/__init__.py +++ b/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.""" diff --git a/tests/components/onvif/test_diagnostics.py b/tests/components/onvif/test_diagnostics.py index ae22838b8c5..a27ec0146c0 100644 --- a/tests/components/onvif/test_diagnostics.py +++ b/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, + } + ], }, } diff --git a/tests/components/onvif/test_switch.py b/tests/components/onvif/test_switch.py new file mode 100644 index 00000000000..bb518d3afad --- /dev/null +++ b/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