From cf77128853d75dc51e8721d109b8bc8797f33b4d Mon Sep 17 00:00:00 2001 From: "Steven B." <51370195+sdb9696@users.noreply.github.com> Date: Fri, 15 Nov 2024 10:19:40 +0000 Subject: [PATCH] Add alarm module for smartcamera hubs (#1258) --- kasa/smartcamera/modules/__init__.py | 2 + kasa/smartcamera/modules/alarm.py | 166 ++++++++++++++++++++++++ kasa/smartcamera/smartcameramodule.py | 17 ++- tests/fakeprotocol_smartcamera.py | 71 +++++----- tests/smartcamera/modules/__init__.py | 0 tests/smartcamera/modules/test_alarm.py | 159 +++++++++++++++++++++++ 6 files changed, 372 insertions(+), 43 deletions(-) create mode 100644 kasa/smartcamera/modules/alarm.py create mode 100644 tests/smartcamera/modules/__init__.py create mode 100644 tests/smartcamera/modules/test_alarm.py diff --git a/kasa/smartcamera/modules/__init__.py b/kasa/smartcamera/modules/__init__.py index cf2b4377..f3e36cc3 100644 --- a/kasa/smartcamera/modules/__init__.py +++ b/kasa/smartcamera/modules/__init__.py @@ -1,5 +1,6 @@ """Modules for SMARTCAMERA devices.""" +from .alarm import Alarm from .camera import Camera from .childdevice import ChildDevice from .device import DeviceModule @@ -7,6 +8,7 @@ from .led import Led from .time import Time __all__ = [ + "Alarm", "Camera", "ChildDevice", "DeviceModule", diff --git a/kasa/smartcamera/modules/alarm.py b/kasa/smartcamera/modules/alarm.py new file mode 100644 index 00000000..bf7ce1a5 --- /dev/null +++ b/kasa/smartcamera/modules/alarm.py @@ -0,0 +1,166 @@ +"""Implementation of alarm module.""" + +from __future__ import annotations + +from ...feature import Feature +from ..smartcameramodule import SmartCameraModule + +DURATION_MIN = 0 +DURATION_MAX = 6000 + +VOLUME_MIN = 0 +VOLUME_MAX = 10 + + +class Alarm(SmartCameraModule): + """Implementation of alarm module.""" + + # Needs a different name to avoid clashing with SmartAlarm + NAME = "SmartCameraAlarm" + + REQUIRED_COMPONENT = "siren" + QUERY_GETTER_NAME = "getSirenStatus" + QUERY_MODULE_NAME = "siren" + + def query(self) -> dict: + """Query to execute during the update cycle.""" + q = super().query() + q["getSirenConfig"] = {self.QUERY_MODULE_NAME: {}} + q["getSirenTypeList"] = {self.QUERY_MODULE_NAME: {}} + + return q + + def _initialize_features(self) -> None: + """Initialize features.""" + device = self._device + self._add_feature( + Feature( + device, + id="alarm", + name="Alarm", + container=self, + attribute_getter="active", + icon="mdi:bell", + category=Feature.Category.Debug, + type=Feature.Type.BinarySensor, + ) + ) + self._add_feature( + Feature( + device, + id="alarm_sound", + name="Alarm sound", + container=self, + attribute_getter="alarm_sound", + attribute_setter="set_alarm_sound", + category=Feature.Category.Config, + type=Feature.Type.Choice, + choices_getter="alarm_sounds", + ) + ) + self._add_feature( + Feature( + device, + id="alarm_volume", + name="Alarm volume", + container=self, + attribute_getter="alarm_volume", + attribute_setter="set_alarm_volume", + category=Feature.Category.Config, + type=Feature.Type.Number, + range_getter=lambda: (VOLUME_MIN, VOLUME_MAX), + ) + ) + self._add_feature( + Feature( + device, + id="alarm_duration", + name="Alarm duration", + container=self, + attribute_getter="alarm_duration", + attribute_setter="set_alarm_duration", + category=Feature.Category.Config, + type=Feature.Type.Number, + range_getter=lambda: (DURATION_MIN, DURATION_MAX), + ) + ) + self._add_feature( + Feature( + device, + id="test_alarm", + name="Test alarm", + container=self, + attribute_setter="play", + type=Feature.Type.Action, + ) + ) + self._add_feature( + Feature( + device, + id="stop_alarm", + name="Stop alarm", + container=self, + attribute_setter="stop", + type=Feature.Type.Action, + ) + ) + + @property + def alarm_sound(self) -> str: + """Return current alarm sound.""" + return self.data["getSirenConfig"]["siren_type"] + + async def set_alarm_sound(self, sound: str) -> dict: + """Set alarm sound. + + See *alarm_sounds* for list of available sounds. + """ + if sound not in self.alarm_sounds: + raise ValueError( + f"sound must be one of {', '.join(self.alarm_sounds)}: {sound}" + ) + return await self.call("setSirenConfig", {"siren": {"siren_type": sound}}) + + @property + def alarm_sounds(self) -> list[str]: + """Return list of available alarm sounds.""" + return self.data["getSirenTypeList"]["siren_type_list"] + + @property + def alarm_volume(self) -> int: + """Return alarm volume. + + Unlike duration the device expects/returns a string for volume. + """ + return int(self.data["getSirenConfig"]["volume"]) + + async def set_alarm_volume(self, volume: int) -> dict: + """Set alarm volume.""" + if volume < VOLUME_MIN or volume > VOLUME_MAX: + raise ValueError(f"volume must be between {VOLUME_MIN} and {VOLUME_MAX}") + return await self.call("setSirenConfig", {"siren": {"volume": str(volume)}}) + + @property + def alarm_duration(self) -> int: + """Return alarm duration.""" + return self.data["getSirenConfig"]["duration"] + + async def set_alarm_duration(self, duration: int) -> dict: + """Set alarm volume.""" + if duration < DURATION_MIN or duration > DURATION_MAX: + msg = f"duration must be between {DURATION_MIN} and {DURATION_MAX}" + raise ValueError(msg) + return await self.call("setSirenConfig", {"siren": {"duration": duration}}) + + @property + def active(self) -> bool: + """Return true if alarm is active.""" + return self.data["getSirenStatus"]["status"] != "off" + + async def play(self) -> dict: + """Play alarm.""" + return await self.call("setSirenStatus", {"siren": {"status": "on"}}) + + async def stop(self) -> dict: + """Stop alarm.""" + return await self.call("setSirenStatus", {"siren": {"status": "off"}}) diff --git a/kasa/smartcamera/smartcameramodule.py b/kasa/smartcamera/smartcameramodule.py index 217b69e7..4b1bd36e 100644 --- a/kasa/smartcamera/smartcameramodule.py +++ b/kasa/smartcamera/smartcameramodule.py @@ -3,12 +3,14 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Final, cast from ..exceptions import DeviceError, KasaException, SmartErrorCode +from ..modulemapping import ModuleName from ..smart.smartmodule import SmartModule if TYPE_CHECKING: + from . import modules from .smartcamera import SmartCamera _LOGGER = logging.getLogger(__name__) @@ -17,12 +19,14 @@ _LOGGER = logging.getLogger(__name__) class SmartCameraModule(SmartModule): """Base class for SMARTCAMERA modules.""" + SmartCameraAlarm: Final[ModuleName[modules.Alarm]] = ModuleName("SmartCameraAlarm") + #: Query to execute during the main update cycle QUERY_GETTER_NAME: str #: Module name to be queried QUERY_MODULE_NAME: str #: Section name or names to be queried - QUERY_SECTION_NAMES: str | list[str] + QUERY_SECTION_NAMES: str | list[str] | None = None REGISTERED_MODULES = {} @@ -33,11 +37,10 @@ class SmartCameraModule(SmartModule): Default implementation uses the raw query getter w/o parameters. """ - return { - self.QUERY_GETTER_NAME: { - self.QUERY_MODULE_NAME: {"name": self.QUERY_SECTION_NAMES} - } - } + section_names = ( + {"name": self.QUERY_SECTION_NAMES} if self.QUERY_SECTION_NAMES else {} + ) + return {self.QUERY_GETTER_NAME: {self.QUERY_MODULE_NAME: section_names}} async def call(self, method: str, params: dict | None = None) -> dict: """Call a method. diff --git a/tests/fakeprotocol_smartcamera.py b/tests/fakeprotocol_smartcamera.py index d6751f7d..e84b5bf9 100644 --- a/tests/fakeprotocol_smartcamera.py +++ b/tests/fakeprotocol_smartcamera.py @@ -105,6 +105,7 @@ class FakeSmartCameraTransport(BaseTransport): info = info[key] info[set_keys[-1]] = value + # Setters for when there's not a simple mapping of setters to getters SETTERS = { ("system", "sys", "dev_alias"): [ "getDeviceInfo", @@ -112,36 +113,20 @@ class FakeSmartCameraTransport(BaseTransport): "basic_info", "device_alias", ], - ("lens_mask", "lens_mask_info", "enabled"): [ - "getLensMaskConfig", - "lens_mask", - "lens_mask_info", - "enabled", - ], + # setTimezone maps to getClockStatus ("system", "clock_status", "seconds_from_1970"): [ "getClockStatus", "system", "clock_status", "seconds_from_1970", ], + # setTimezone maps to getClockStatus ("system", "clock_status", "local_time"): [ "getClockStatus", "system", "clock_status", "local_time", ], - ("system", "basic", "zone_id"): [ - "getTimezone", - "system", - "basic", - "zone_id", - ], - ("led", "config", "enabled"): [ - "getLedStatus", - "led", - "config", - "enabled", - ], } async def _send_request(self, request_dict: dict): @@ -154,27 +139,41 @@ class FakeSmartCameraTransport(BaseTransport): ) if method[:3] == "set": + get_method = "g" + method[1:] for key, val in request_dict.items(): - if key != "method": - # key is params for multi request and the actual params - # for single requests - if key == "params": - module = next(iter(val)) - val = val[module] + if key == "method": + continue + # key is params for multi request and the actual params + # for single requests + if key == "params": + module = next(iter(val)) + val = val[module] + else: + module = key + section = next(iter(val)) + skey_val = val[section] + if not isinstance(skey_val, dict): # single level query + section_key = section + section_val = skey_val + if (get_info := info.get(get_method)) and section_key in get_info: + get_info[section_key] = section_val else: - module = key - section = next(iter(val)) - skey_val = val[section] - for skey, sval in skey_val.items(): - section_key = skey - section_value = sval - if setter_keys := self.SETTERS.get( - (module, section, section_key) - ): - self._get_param_set_value(info, setter_keys, section_value) - else: - return {"error_code": -1} + return {"error_code": -1} break + for skey, sval in skey_val.items(): + section_key = skey + section_value = sval + if setter_keys := self.SETTERS.get((module, section, section_key)): + self._get_param_set_value(info, setter_keys, section_value) + elif ( + section := info.get(get_method, {}) + .get(module, {}) + .get(section, {}) + ) and section_key in section: + section[section_key] = section_value + else: + return {"error_code": -1} + break return {"error_code": 0} elif method[:3] == "get": params = request_dict.get("params") diff --git a/tests/smartcamera/modules/__init__.py b/tests/smartcamera/modules/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/smartcamera/modules/test_alarm.py b/tests/smartcamera/modules/test_alarm.py new file mode 100644 index 00000000..2301a2be --- /dev/null +++ b/tests/smartcamera/modules/test_alarm.py @@ -0,0 +1,159 @@ +"""Tests for smart camera devices.""" + +from __future__ import annotations + +import pytest + +from kasa import Device +from kasa.smartcamera.modules.alarm import ( + DURATION_MAX, + DURATION_MIN, + VOLUME_MAX, + VOLUME_MIN, +) +from kasa.smartcamera.smartcameramodule import SmartCameraModule + +from ...conftest import hub_smartcamera + + +@hub_smartcamera +async def test_alarm(dev: Device): + """Test device alarm.""" + alarm = dev.modules.get(SmartCameraModule.SmartCameraAlarm) + assert alarm + + original_duration = alarm.alarm_duration + assert original_duration is not None + original_volume = alarm.alarm_volume + assert original_volume is not None + original_sound = alarm.alarm_sound + + try: + # test volume + new_volume = original_volume - 1 if original_volume > 1 else original_volume + 1 + await alarm.set_alarm_volume(new_volume) # type: ignore[arg-type] + await dev.update() + assert alarm.alarm_volume == new_volume + + # test duration + new_duration = ( + original_duration - 1 if original_duration > 1 else original_duration + 1 + ) + await alarm.set_alarm_duration(new_duration) + await dev.update() + assert alarm.alarm_duration == new_duration + + # test start + await alarm.play() + await dev.update() + assert alarm.active + + # test stop + await alarm.stop() + await dev.update() + assert not alarm.active + + # test set sound + new_sound = ( + alarm.alarm_sounds[0] + if alarm.alarm_sound != alarm.alarm_sounds[0] + else alarm.alarm_sounds[1] + ) + await alarm.set_alarm_sound(new_sound) + await dev.update() + assert alarm.alarm_sound == new_sound + + finally: + await alarm.set_alarm_volume(original_volume) + await alarm.set_alarm_duration(original_duration) + await alarm.set_alarm_sound(original_sound) + await dev.update() + + +@hub_smartcamera +async def test_alarm_invalid_setters(dev: Device): + """Test device alarm invalid setter values.""" + alarm = dev.modules.get(SmartCameraModule.SmartCameraAlarm) + assert alarm + + # test set sound invalid + msg = f"sound must be one of {', '.join(alarm.alarm_sounds)}: foobar" + with pytest.raises(ValueError, match=msg): + await alarm.set_alarm_sound("foobar") + + # test volume invalid + msg = f"volume must be between {VOLUME_MIN} and {VOLUME_MAX}" + with pytest.raises(ValueError, match=msg): + await alarm.set_alarm_volume(-3) + + # test duration invalid + msg = f"duration must be between {DURATION_MIN} and {DURATION_MAX}" + with pytest.raises(ValueError, match=msg): + await alarm.set_alarm_duration(-3) + + +@hub_smartcamera +async def test_alarm_features(dev: Device): + """Test device alarm features.""" + alarm = dev.modules.get(SmartCameraModule.SmartCameraAlarm) + assert alarm + + original_duration = alarm.alarm_duration + assert original_duration is not None + original_volume = alarm.alarm_volume + assert original_volume is not None + original_sound = alarm.alarm_sound + + try: + # test volume + new_volume = original_volume - 1 if original_volume > 1 else original_volume + 1 + feature = dev.features.get("alarm_volume") + assert feature + await feature.set_value(new_volume) # type: ignore[arg-type] + await dev.update() + assert feature.value == new_volume + + # test duration + feature = dev.features.get("alarm_duration") + assert feature + new_duration = ( + original_duration - 1 if original_duration > 1 else original_duration + 1 + ) + await feature.set_value(new_duration) + await dev.update() + assert feature.value == new_duration + + # test start + feature = dev.features.get("test_alarm") + assert feature + await feature.set_value(None) + await dev.update() + feature = dev.features.get("alarm") + assert feature + assert feature.value is True + + # test stop + feature = dev.features.get("stop_alarm") + assert feature + await feature.set_value(None) + await dev.update() + assert dev.features["alarm"].value is False + + # test set sound + feature = dev.features.get("alarm_sound") + assert feature + new_sound = ( + alarm.alarm_sounds[0] + if alarm.alarm_sound != alarm.alarm_sounds[0] + else alarm.alarm_sounds[1] + ) + await feature.set_value(new_sound) + await alarm.set_alarm_sound(new_sound) + await dev.update() + assert feature.value == new_sound + + finally: + await alarm.set_alarm_volume(original_volume) + await alarm.set_alarm_duration(original_duration) + await alarm.set_alarm_sound(original_sound) + await dev.update()