python-kasa · sdb9696 · Jan 26, 2025 · Dec 4, 2024 · Dec 5, 2024 · Dec 5, 2024
diff --git a/kasa/smart/modules/alarm.py b/kasa/smart/modules/alarm.py
@@ -2,11 +2,27 @@
 
 from __future__ import annotations
 
-from typing import Literal
+from typing import TYPE_CHECKING, Annotated, Literal, TypeAlias
 
 from ...feature import Feature
+from ...module import FeatureAttribute
 from ..smartmodule import SmartModule
 
+DURATION_MAX = 10 * 60
+
+VOLUME_INT_TO_STR = {
+    0: "mute",
+    1: "low",
+    2: "normal",
+    3: "high",
+}
+
+VOLUME_STR_LIST = [v for v in VOLUME_INT_TO_STR.values()]
+VOLUME_INT_RANGE = (min(VOLUME_INT_TO_STR.keys()), max(VOLUME_INT_TO_STR.keys()))
+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):
     """Implementation of alarm module."""
@@ -21,10 +37,7 @@ def query(self) -> dict:
         }
 
     def _initialize_features(self) -> None:
-        """Initialize features.
-
-        This is implemented as some features depend on device responses.
-        """
+        """Initialize features."""
         device = self._device
         self._add_feature(
             Feature(
@@ -67,11 +80,37 @@ def _initialize_features(self) -> None:
                 id="alarm_volume",
                 name="Alarm volume",
                 container=self,
-                attribute_getter="alarm_volume",
+                attribute_getter="_alarm_volume_str",
                 attribute_setter="set_alarm_volume",
                 category=Feature.Category.Config,
                 type=Feature.Type.Choice,
-                choices_getter=lambda: ["low", "normal", "high"],
+                choices_getter=lambda: VOLUME_STR_LIST,
+            )
+        )
+        self._add_feature(
+            Feature(
+                device,
+                id="alarm_volume_level",
+                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_INT_RANGE,
+            )
+        )
+        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: (1, DURATION_MAX),
             )
         )
         self._add_feature(
@@ -96,15 +135,16 @@ def _initialize_features(self) -> None:
         )
 
     @property
-    def alarm_sound(self) -> str:
+    def alarm_sound(self) -> Annotated[str, FeatureAttribute()]:
         """Return current alarm sound."""
         return self.data["get_alarm_configure"]["type"]
 
-    async def set_alarm_sound(self, sound: str) -> dict:
+    async def set_alarm_sound(self, sound: str) -> Annotated[dict, FeatureAttribute()]:
         """Set alarm sound.
 
         See *alarm_sounds* for list of available sounds.
         """
+        self._check_sound(sound)
         payload = self.data["get_alarm_configure"].copy()
         payload["type"] = sound
         return await self.call("set_alarm_configure", payload)
@@ -115,16 +155,40 @@ def alarm_sounds(self) -> list[str]:
         return self.data["get_support_alarm_type_list"]["alarm_type_list"]
 
     @property
-    def alarm_volume(self) -> Literal["low", "normal", "high"]:
+    def alarm_volume(self) -> Annotated[int, FeatureAttribute("alarm_volume_level")]:
+        """Return alarm volume."""
+        return VOLUME_STR_TO_INT[self._alarm_volume_str]
+
+    @property
+    def _alarm_volume_str(
+        self,
+    ) -> Annotated[AlarmVolume, FeatureAttribute("alarm_volume")]:
         """Return alarm volume."""
         return self.data["get_alarm_configure"]["volume"]
 
-    async def set_alarm_volume(self, volume: Literal["low", "normal", "high"]) -> dict:
+    async def set_alarm_volume(
+        self, volume: AlarmVolume | int
+    ) -> Annotated[dict, FeatureAttribute()]:
         """Set alarm volume."""
+        self._check_and_convert_volume(volume)
         payload = self.data["get_alarm_configure"].copy()
         payload["volume"] = volume
         return await self.call("set_alarm_configure", payload)
 
+    @property
+    def alarm_duration(self) -> Annotated[int, FeatureAttribute()]:
+        """Return alarm duration."""
+        return self.data["get_alarm_configure"]["duration"]
+
+    async def set_alarm_duration(
+        self, duration: int
+    ) -> Annotated[dict, FeatureAttribute()]:
+        """Set alarm duration."""
+        self._check_duration(duration)
+        payload = self.data["get_alarm_configure"].copy()
+        payload["duration"] = duration
+        return await self.call("set_alarm_configure", payload)
+
     @property
     def active(self) -> bool:
         """Return true if alarm is active."""
@@ -136,10 +200,62 @@ def source(self) -> str | None:
         src = self._device.sys_info["in_alarm_source"]
         return src if src else None
 
-    async def play(self) -> dict:
-        """Play alarm."""
-        return await self.call("play_alarm")
+    async def play(
+        self,
+        *,
+        duration: int | None = None,
+        volume: int | AlarmVolume | None = None,
+        sound: str | None = None,
+    ) -> dict:
+        """Play alarm.
+
+        The optional *duration*, *volume*, and *sound* to override the device settings.
+        *volume* can be set to 'mute', 'low', 'normal', or 'high'.
+        *duration* is in seconds.
+        See *alarm_sounds* for the list of sounds available for the device.
+        """
+        params: dict[str, str | int] = {}
+
+        if duration is not None:
+            self._check_duration(duration)
+            params["alarm_duration"] = duration
+
+        if volume is not None:
+            target_volume = self._check_and_convert_volume(volume)
+            params["alarm_volume"] = target_volume
+
+        if sound is not None:
+            self._check_sound(sound)
+            params["alarm_type"] = sound
+
+        return await self.call("play_alarm", params)
 
     async def stop(self) -> dict:
         """Stop alarm."""
         return await self.call("stop_alarm")
+
+    def _check_and_convert_volume(self, volume: str | int) -> str:
+        """Raise an exception on invalid volume."""
+        if isinstance(volume, int):
+            volume = VOLUME_INT_TO_STR.get(volume, "invalid")
+
+        if TYPE_CHECKING:
+            assert isinstance(volume, str)
+
+        if volume not in VOLUME_INT_TO_STR.values():
+            raise ValueError(
+                f"Invalid volume {volume} "
+                f"available: {VOLUME_INT_TO_STR.keys()}, {VOLUME_INT_TO_STR.values()}"
+            )
+
+        return volume
+
+    def _check_duration(self, duration: int) -> None:
+        """Raise an exception on invalid duration."""
+        if duration < 1 or duration > DURATION_MAX:
+            raise ValueError(f"Invalid duration {duration} available: 1-600")
+
+    def _check_sound(self, sound: str) -> None:
+        """Raise an exception on invalid sound."""
+        if sound not in self.alarm_sounds:
+            raise ValueError(f"Invalid sound {sound} available: {self.alarm_sounds}")
diff --git a/tests/fakeprotocol_smart.py b/tests/fakeprotocol_smart.py
@@ -134,11 +134,9 @@ def credentials_hash(self):
         "get_alarm_configure": (
             "alarm",
             {
-                "get_alarm_configure": {
-                    "duration": 10,
-                    "type": "Doorbell Ring 2",
-                    "volume": "low",
-                }
+                "duration": 10,
+                "type": "Doorbell Ring 2",
+                "volume": "low",
             },
         ),
         "get_support_alarm_type_list": (
@@ -672,7 +670,7 @@ async def _send_request(self, request_dict: dict):
                     self.fixture_name, set()
                 ).add(method)
             return retval
-        elif method in ["set_qs_info", "fw_download"]:
+        elif method in ["set_qs_info", "fw_download", "play_alarm", "stop_alarm"]:
             return {"error_code": 0}
         elif method == "set_dynamic_light_effect_rule_enable":
             self._set_dynamic_light_effect(info, params)

diff --git a/tests/smart/modules/test_alarm.py b/tests/smart/modules/test_alarm.py
@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import pytest
+from pytest_mock import MockerFixture
+
+from kasa import Module
+from kasa.smart import SmartDevice
+from kasa.smart.modules import Alarm
+
+from ...device_fixtures import get_parent_and_child_modules, parametrize
+
+alarm = parametrize("has alarm", component_filter="alarm", protocol_filter={"SMART"})
+
+
+@alarm
+@pytest.mark.parametrize(
+    ("feature", "prop_name", "type"),
+    [
+        ("alarm", "active", bool),
+        ("alarm_source", "source", str | None),
+        ("alarm_sound", "alarm_sound", str),
+        ("alarm_volume", "_alarm_volume_str", str),
+        ("alarm_volume_level", "alarm_volume", int),
+    ],
+)
+async def test_features(dev: SmartDevice, feature: str, prop_name: str, type: type):
+    """Test that features are registered and work as expected."""
+    alarm = next(get_parent_and_child_modules(dev, Module.Alarm))
+    assert alarm is not None
+
+    prop = getattr(alarm, prop_name)
+    assert isinstance(prop, type)
+
+    feat = alarm._device.features[feature]
+    assert feat.value == prop
+    assert isinstance(feat.value, type)
+
+
+@alarm
+async def test_volume_feature(dev: SmartDevice):
+    """Test that volume features have correct choices and range."""
+    alarm = next(get_parent_and_child_modules(dev, Module.Alarm))
+    assert alarm is not None
+
+    volume_str_feat = alarm.get_feature("_alarm_volume_str")
+    assert volume_str_feat
+
+    assert volume_str_feat.choices == ["mute", "low", "normal", "high"]
+
+    volume_int_feat = alarm.get_feature("alarm_volume")
+    assert volume_int_feat.minimum_value == 0
+    assert volume_int_feat.maximum_value == 3
+
+
+@alarm
+@pytest.mark.parametrize(
+    ("kwargs", "request_params"),
+    [
+        pytest.param({"volume": "low"}, {"alarm_volume": "low"}, id="volume"),
+        pytest.param({"volume": 0}, {"alarm_volume": "mute"}, id="volume-integer"),
+        pytest.param({"duration": 1}, {"alarm_duration": 1}, id="duration"),
+        pytest.param(
+            {"sound": "Doorbell Ring 1"}, {"alarm_type": "Doorbell Ring 1"}, id="sound"
+        ),
+    ],
+)
+async def test_play(dev: SmartDevice, kwargs, request_params, mocker: MockerFixture):
+    """Test that play parameters are handled correctly."""
+    alarm: Alarm = next(get_parent_and_child_modules(dev, Module.Alarm))
+    call_spy = mocker.spy(alarm, "call")
+    await alarm.play(**kwargs)
+
+    call_spy.assert_called_with("play_alarm", request_params)
+
+    with pytest.raises(ValueError, match="Invalid duration"):
+        await alarm.play(duration=-1)
+
+    with pytest.raises(ValueError, match="Invalid sound"):
+        await alarm.play(sound="unknown")
+
+    with pytest.raises(ValueError, match="Invalid volume"):
+        await alarm.play(volume="unknown")  # type: ignore[arg-type]
+
+    with pytest.raises(ValueError, match="Invalid volume"):
+        await alarm.play(volume=-1)
+
+
+@alarm
+async def test_stop(dev: SmartDevice, mocker: MockerFixture):
+    """Test that stop creates the correct call."""
+    alarm: Alarm = next(get_parent_and_child_modules(dev, Module.Alarm))
+    call_spy = mocker.spy(alarm, "call")
+    await alarm.stop()
+
+    call_spy.assert_called_with("stop_alarm")
+
+
+@alarm
+@pytest.mark.parametrize(
+    ("method", "value", "target_key"),
+    [
+        pytest.param(
+            "set_alarm_sound", "Doorbell Ring 1", "type", id="set_alarm_sound"
+        ),
+        pytest.param("set_alarm_volume", "low", "volume", id="set_alarm_volume"),
+        pytest.param("set_alarm_duration", 10, "duration", id="set_alarm_duration"),
+    ],
+)
+async def test_set_alarm_configure(
+    dev: SmartDevice,
+    mocker: MockerFixture,
+    method: str,
+    value: str | int,
+    target_key: str,
+):
+    """Test that set_alarm_sound creates the correct call."""
+    alarm: Alarm = next(get_parent_and_child_modules(dev, Module.Alarm))
+    call_spy = mocker.spy(alarm, "call")
+    await getattr(alarm, method)(value)
+
+    expected_params = {"duration": mocker.ANY, "type": mocker.ANY, "volume": mocker.ANY}
+    expected_params[target_key] = value
+
+    call_spy.assert_called_with("set_alarm_configure", expected_params)