Add common alarm interface (#1479)

Add a common interface for the `alarm` module across `smart` and `smartcam` devices.

kasa/interfaces/__init__.py

@@ -1,5 +1,6 @@
 """Package for interfaces."""
 
+from .alarm import Alarm
 from .childsetup import ChildSetup
 from .energy import Energy
 from .fan import Fan
@@ -11,6 +12,7 @@ from .thermostat import Thermostat, ThermostatState
 from .time import Time
 
 __all__ = [
+    "Alarm",
     "ChildSetup",
     "Fan",
     "Energy",

kasa/interfaces/alarm.py

@@ -0,0 +1,75 @@
+"""Module for base alarm module."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Annotated
+
+from ..module import FeatureAttribute, Module
+
+
+class Alarm(Module, ABC):
+    """Base interface to represent an alarm module."""
+
+    @property
+    @abstractmethod
+    def alarm_sound(self) -> Annotated[str, FeatureAttribute()]:
+        """Return current alarm sound."""
+
+    @abstractmethod
+    async def set_alarm_sound(self, sound: str) -> Annotated[dict, FeatureAttribute()]:
+        """Set alarm sound.
+
+        See *alarm_sounds* for list of available sounds.
+        """
+
+    @property
+    @abstractmethod
+    def alarm_sounds(self) -> list[str]:
+        """Return list of available alarm sounds."""
+
+    @property
+    @abstractmethod
+    def alarm_volume(self) -> Annotated[int, FeatureAttribute()]:
+        """Return alarm volume."""
+
+    @abstractmethod
+    async def set_alarm_volume(
+        self, volume: int
+    ) -> Annotated[dict, FeatureAttribute()]:
+        """Set alarm volume."""
+
+    @property
+    @abstractmethod
+    def alarm_duration(self) -> Annotated[int, FeatureAttribute()]:
+        """Return alarm duration."""
+
+    @abstractmethod
+    async def set_alarm_duration(
+        self, duration: int
+    ) -> Annotated[dict, FeatureAttribute()]:
+        """Set alarm duration."""
+
+    @property
+    @abstractmethod
+    def active(self) -> bool:
+        """Return true if alarm is active."""
+
+    @abstractmethod
+    async def play(
+        self,
+        *,
+        duration: int | None = None,
+        volume: int | None = None,
+        sound: str | None = None,
+    ) -> dict:
+        """Play alarm.
+
+        The optional *duration*, *volume*, and *sound* to override the device settings.
+        *duration* is in seconds.
+        See *alarm_sounds* for the list of sounds available for the device.
+        """
+
+    @abstractmethod
+    async def stop(self) -> dict:
+        """Stop alarm."""

kasa/module.py

@@ -96,6 +96,7 @@ class Module(ABC):
     """
 
     # Common Modules
+    Alarm: Final[ModuleName[interfaces.Alarm]] = ModuleName("Alarm")
     ChildSetup: Final[ModuleName[interfaces.ChildSetup]] = ModuleName("ChildSetup")
     Energy: Final[ModuleName[interfaces.Energy]] = ModuleName("Energy")
     Fan: Final[ModuleName[interfaces.Fan]] = ModuleName("Fan")
@@ -116,7 +117,6 @@ class Module(ABC):
     IotCloud: Final[ModuleName[iot.Cloud]] = ModuleName("cloud")
 
     # SMART only Modules
-    Alarm: Final[ModuleName[smart.Alarm]] = ModuleName("Alarm")
     AutoOff: Final[ModuleName[smart.AutoOff]] = ModuleName("AutoOff")
     BatterySensor: Final[ModuleName[smart.BatterySensor]] = ModuleName("BatterySensor")
     Brightness: Final[ModuleName[smart.Brightness]] = ModuleName("Brightness")

kasa/smart/modules/alarm.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from typing import TYPE_CHECKING, Annotated, Literal, TypeAlias
 
 from ...feature import Feature
+from ...interfaces import Alarm as AlarmInterface
 from ...module import FeatureAttribute
 from ..smartmodule import SmartModule
 
@@ -24,7 +25,7 @@ VOLUME_STR_TO_INT = {v: k for k, v in VOLUME_INT_TO_STR.items()}
 AlarmVolume: TypeAlias = Literal["mute", "low", "normal", "high"]
 
 
-class Alarm(SmartModule):
+class Alarm(SmartModule, AlarmInterface):
     """Implementation of alarm module."""
 
     REQUIRED_COMPONENT = "alarm"

kasa/smartcam/modules/alarm.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from ...feature import Feature
+from ...interfaces import Alarm as AlarmInterface
 from ...smart.smartmodule import allow_update_after
 from ..smartcammodule import SmartCamModule
 
@@ -13,12 +14,9 @@ VOLUME_MIN = 0
 VOLUME_MAX = 10
 
 
-class Alarm(SmartCamModule):
+class Alarm(SmartCamModule, AlarmInterface):
     """Implementation of alarm module."""
 
-    # Needs a different name to avoid clashing with SmartAlarm
-    NAME = "SmartCamAlarm"
-
     REQUIRED_COMPONENT = "siren"
     QUERY_GETTER_NAME = "getSirenStatus"
     QUERY_MODULE_NAME = "siren"
@@ -117,11 +115,8 @@ class Alarm(SmartCamModule):
 
         See *alarm_sounds* for list of available sounds.
         """
-        if sound not in self.alarm_sounds:
+        config = self._validate_and_get_config(sound=sound)
-            raise ValueError(
+        return await self.call("setSirenConfig", {"siren": config})
-                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]:
@@ -139,9 +134,8 @@ class Alarm(SmartCamModule):
     @allow_update_after
     async def set_alarm_volume(self, volume: int) -> dict:
         """Set alarm volume."""
-        if volume < VOLUME_MIN or volume > VOLUME_MAX:
+        config = self._validate_and_get_config(volume=volume)
-            raise ValueError(f"volume must be between {VOLUME_MIN} and {VOLUME_MAX}")
+        return await self.call("setSirenConfig", {"siren": config})
-        return await self.call("setSirenConfig", {"siren": {"volume": str(volume)}})
 
     @property
     def alarm_duration(self) -> int:
@@ -151,20 +145,65 @@ class Alarm(SmartCamModule):
     @allow_update_after
     async def set_alarm_duration(self, duration: int) -> dict:
         """Set alarm volume."""
-        if duration < DURATION_MIN or duration > DURATION_MAX:
+        config = self._validate_and_get_config(duration=duration)
-            msg = f"duration must be between {DURATION_MIN} and {DURATION_MAX}"
+        return await self.call("setSirenConfig", {"siren": config})
-            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:
+    async def play(
-        """Play alarm."""
+        self,
+        *,
+        duration: int | None = None,
+        volume: int | None = None,
+        sound: str | None = None,
+    ) -> dict:
+        """Play alarm.
+
+        The optional *duration*, *volume*, and *sound* to override the device settings.
+        *duration* is in seconds.
+        See *alarm_sounds* for the list of sounds available for the device.
+        """
+        if config := self._validate_and_get_config(
+            duration=duration, volume=volume, sound=sound
+        ):
+            await self.call("setSirenConfig", {"siren": config})
+
         return await self.call("setSirenStatus", {"siren": {"status": "on"}})
 
     async def stop(self) -> dict:
         """Stop alarm."""
         return await self.call("setSirenStatus", {"siren": {"status": "off"}})
+
+    def _validate_and_get_config(
+        self,
+        *,
+        duration: int | None = None,
+        volume: int | None = None,
+        sound: str | None = None,
+    ) -> dict:
+        if sound and sound not in self.alarm_sounds:
+            raise ValueError(
+                f"sound must be one of {', '.join(self.alarm_sounds)}: {sound}"
+            )
+
+        if duration is not None and (
+            duration < DURATION_MIN or duration > DURATION_MAX
+        ):
+            msg = f"duration must be between {DURATION_MIN} and {DURATION_MAX}"
+            raise ValueError(msg)
+
+        if volume is not None and (volume < VOLUME_MIN or volume > VOLUME_MAX):
+            raise ValueError(f"volume must be between {VOLUME_MIN} and {VOLUME_MAX}")
+
+        config: dict[str, str | int] = {}
+        if sound:
+            config["siren_type"] = sound
+        if duration is not None:
+            config["duration"] = duration
+        if volume is not None:
+            config["volume"] = str(volume)
+
+        return config

tests/fakeprotocol_smartcam.py

@@ -276,12 +276,14 @@ class FakeSmartCamTransport(BaseTransport):
                 section = next(iter(val))
                 skey_val = val[section]
                 if not isinstance(skey_val, dict):  # single level query
-                    section_key = section
+                    updates = {
-                    section_val = skey_val
+                        k: v for k, v in val.items() if k in info.get(get_method, {})
-                    if (get_info := info.get(get_method)) and section_key in get_info:
+                    }
-                        get_info[section_key] = section_val
+                    if len(updates) != len(val):
-                    else:
+                        # All keys to update must already be in the getter
                         return {"error_code": -1}
+                    info[get_method] = {**info[get_method], **updates}
+
                     break
                 for skey, sval in skey_val.items():
                     section_key = skey

tests/smartcam/modules/test_alarm.py

@@ -4,14 +4,13 @@ from __future__ import annotations
 
 import pytest
 
-from kasa import Device
+from kasa import Device, Module
 from kasa.smartcam.modules.alarm import (
     DURATION_MAX,
     DURATION_MIN,
     VOLUME_MAX,
     VOLUME_MIN,
 )
-from kasa.smartcam.smartcammodule import SmartCamModule
 
 from ...conftest import hub_smartcam
 
@@ -19,7 +18,7 @@ from ...conftest import hub_smartcam
 @hub_smartcam
 async def test_alarm(dev: Device):
     """Test device alarm."""
-    alarm = dev.modules.get(SmartCamModule.SmartCamAlarm)
+    alarm = dev.modules.get(Module.Alarm)
     assert alarm
 
     original_duration = alarm.alarm_duration
@@ -63,6 +62,19 @@ async def test_alarm(dev: Device):
         await dev.update()
         assert alarm.alarm_sound == new_sound
 
+        # Test play parameters
+        await alarm.play(
+            duration=original_duration, volume=original_volume, sound=original_sound
+        )
+        await dev.update()
+        assert alarm.active
+        assert alarm.alarm_sound == original_sound
+        assert alarm.alarm_duration == original_duration
+        assert alarm.alarm_volume == original_volume
+        await alarm.stop()
+        await dev.update()
+        assert not alarm.active
+
     finally:
         await alarm.set_alarm_volume(original_volume)
         await alarm.set_alarm_duration(original_duration)
@@ -73,7 +85,7 @@ async def test_alarm(dev: Device):
 @hub_smartcam
 async def test_alarm_invalid_setters(dev: Device):
     """Test device alarm invalid setter values."""
-    alarm = dev.modules.get(SmartCamModule.SmartCamAlarm)
+    alarm = dev.modules.get(Module.Alarm)
     assert alarm
 
     # test set sound invalid
@@ -95,7 +107,7 @@ async def test_alarm_invalid_setters(dev: Device):
 @hub_smartcam
 async def test_alarm_features(dev: Device):
     """Test device alarm features."""
-    alarm = dev.modules.get(SmartCamModule.SmartCamAlarm)
+    alarm = dev.modules.get(Module.Alarm)
     assert alarm
 
     original_duration = alarm.alarm_duration