8000 Add alarm module for smartcamera hubs (#1258) · python-kasa/python-kasa@7f94b0e · GitHub
[go: up one dir, main page]
Skip to content

Commit 7f94b0e

Browse files
sdb9696rytilahti
sdb9696
authored and
rytilahti
committed
Add alarm module for smartcamera hubs (#1258)
1 parent e18bcb2 commit 7f94b0e

File tree

6 files changed

+372
-43
lines changed
  • kasa/smartcamera
    • modules
  • smartcameramodule.py
  • tests

    • 6 files changed

    +372
    -43
    lines changed

    ‎kasa/smartcamera/modules/__init__.py

    Lines changed: 2 additions & 0 deletions
    Original file line numberDiff line numberDiff line change
    @@ -1,12 +1,14 @@
    11
    """Modules for SMARTCAMERA devices."""
    22

    3+
    from .alarm import Alarm
    34
    from .camera import Camera
    45
    from .childdevice import ChildDevice
    56
    from .device import DeviceModule
    67
    from .led import Led
    78
    from .time import Time
    89

    910
    __all__ = [
    11+
    "Alarm",
    1012
    "Camera",
    1113
    "ChildDevice",
    1214
    "DeviceModule",

    ‎kasa/smartcamera/modules/alarm.py

    Lines changed: 166 additions & 0 deletions
    Original file line numberDiff line numberDiff line change
    @@ -0,0 +1,166 @@
    1+
    """Implementation of alarm module."""
    2+
    3+
    from __future__ import annotations
    4+
    5+
    from ...feature import Feature
    6+
    from ..smartcameramodule import SmartCameraModule
    7+
    8+
    DURATION_MIN = 0
    9+
    DURATION_MAX = 6000
    10+
    11+
    VOLUME_MIN = 0
    12+
    VOLUME_MAX = 10
    13+
    14+
    15+
    class Alarm(SmartCameraModule):
    16+
    """Implementation of alarm module."""
    17+
    18+
    # Needs a different name to avoid clashing with SmartAlarm
    19+
    NAME = "SmartCameraAlarm"
    20+
    21+
    REQUIRED_COMPONENT = "siren"
    22+
    QUERY_GETTER_NAME = "getSirenStatus"
    23+
    QUERY_MODULE_NAME = "siren"
    24+
    25+
    def query(self) -> dict:
    26+
    """Query to execute during the update cycle."""
    27+
    q = super().query()
    28+
    q["getSirenConfig"] = {self.QUERY_MODULE_NAME: {}}
    29+
    q["getSirenTypeList"] = {self.QUERY_MODULE_NAME: {}}
    30+
    31+
    return q
    32+
    33+
    def _initialize_features(self) -> None:
    34+
    """Initialize features."""
    35+
    device = self._device
    36+
    self._add_feature(
    37+
    Feature(
    38+
    device,
    39+
    id="alarm",
    40+
    name="Alarm",
    41+
    container=self,
    42+
    attribute_getter="active",
    43+
    icon="mdi:bell",
    44+
    category=Feature.Category.Debug,
    45+
    type=Feature.Type.BinarySensor,
    46+
    )
    47+
    )
    48+
    self._add_feature(
    49+
    Feature(
    50+
    device,
    51+
    id="alarm_sound",
    52+
    name="Alarm sound",
    53+
    container=self,
    54+
    attribute_getter="alarm_sound",
    55+
    attribute_setter="set_alarm_sound",
    56+
    category=Feature.Category.Config,
    57+
    type=Feature.Type.Choice,
    58+
    choices_getter="alarm_sounds",
    59+
    )
    60+
    )
    61+
    self._add_feature(
    62+
    Feature(
    63+
    device,
    64+
    id="alarm_volume",
    65+
    name="Alarm volume",
    66+
    container=self,
    67+
    attribute_getter="alarm_volume",
    68+
    attribute_setter="set_alarm_volume",
    69+
    category=Feature.Category.Config,
    70+
    type=Feature.Type.Number,
    71+
    range_getter=lambda: (VOLUME_MIN, VOLUME_MAX),
    72+
    )
    73+
    )
    74+
    self._add_feature(
    75+
    Feature(
    76+
    device,
    77+
    id="alarm_duration",
    78+
    name="Alarm duration",
    79+
    container=self,
    80+
    attribute_getter="alarm_duration",
    81+
    attribute_setter="set_alarm_duration",
    82+
    category=Feature.Category.Config,
    83+
    type=Feature.Type.Number,
    84+
    range_getter=lambda: (DURATION_MIN, DURATION_MAX),
    85+
    )
    86+
    )
    87+
    self._add_feature(
    88+
    Feature(
    89+
    device,
    90+
    id="test_alarm",
    91+
    name="Test alarm",
    92+
    container=self,
    93+
    attribute_setter="play",
    94+
    type=Feature.Type.Action,
    95+
    )
    96+
    )
    97+
    self._add_feature(
    98+
    Feature(
    99+
    device,
    100+
    id="stop_alarm",
    101+
    name="Stop alarm",
    102+
    container=self,
    103+
    attribute_setter="stop",
    104+
    type=Feature.Type.Action,
    105+
    )
    106+
    )
    107+
    108+
    @property
    109+
    def alarm_sound(self) -> str:
    110+
    """Return current alarm sound."""
    111+
    return self.data["getSirenConfig"]["siren_type"]
    112+
    113+
    async def set_alarm_sound(self, sound: str) -> dict:
    114+
    """Set alarm sound.
    115+
    116+
    See *alarm_sounds* for list of available sounds.
    117+
    """
    118+
    if sound not in self.alarm_sounds:
    119+
    raise ValueError(
    120+
    f"sound must be one of {', '.join(self.alarm_sounds)}: {sound}"
    121+
    )
    122+
    return await self.call("setSirenConfig", {"siren": {"siren_type": sound}})
    123+
    124+
    @property
    125+
    def alarm_sounds(self) -> list[str]:
    126+
    """Return list of available alarm sounds."""
    127+
    return self.data["getSirenTypeList"]["siren_type_list"]
    128+
    129+
    @property
    130+
    def alarm_volume(self) -> int:
    131+
    """Return alarm volume.
    132+
    133+
    Unlike duration the device expects/returns a string for volume.
    134+
    """
    135+
    return int(self.data["getSirenConfig"]["volume"])
    136+
    137+
    async def set_alarm_volume(self, volume: int) -> dict:
    138+
    """Set alarm volume."""
    139+
    if volume < VOLUME_MIN or volume > VOLUME_MAX:
    140+
    raise ValueError(f"volume must be between {VOLUME_MIN} and {VOLUME_MAX}")
    141+
    return await self.call("setSirenConfig", {"siren": {"volume": str(volume)}})
    142+
    143+
    @property
    144+
    def alarm_duration(self) -> int:
    145+
    """Return alarm duration."""
    146+
    return self.data["getSirenConfig"]["duration"]
    147+
    148+
    async def set_alarm_duration(self, duration: int) -> dict:
    149+
    """Set alarm volume."""
    150+
    if duration < DURATION_MIN or duration > DURATION_MAX:
    151+
    msg = f"duration must be between {DURATION_MIN} and {DURATION_MAX}"
    152+
    raise ValueError(msg)
    153+
    return await self.call("setSirenConfig", {"siren": {"duration": duration}})
    154+
    155+
    @property
    156+
    def active(self) -> bool:
    157+
    """Return true if alarm is active."""
    158+
    return self.data["getSirenStatus"]["status"] != "off"
    159+
    160+
    async def play(self) -> dict:
    161+
    """Play alarm."""
    162+
    return await self.call("setSirenStatus", {"siren": {"status": "on"}})
    163+
    164+
    async def stop(self) -> dict:
    165+
    """Stop alarm."""
    166+
    return await self.call("setSirenStatus", {"siren": {"status": "off"}})

    ‎kasa/smartcamera/smartcameramodule.py

    Lines changed: 10 additions & 7 deletions
    Original file line numberDiff line numberDiff line change
    @@ -3,12 +3,14 @@
    33
    from __future__ import annotations
    44

    55
    import logging
    6-
    from typing import TYPE_CHECKING, Any, cast
    6+
    from typing import TYPE_CHECKING, Any, Final, cast
    77

    88
    from ..exceptions import DeviceError, KasaException, SmartErrorCode
    9+
    from ..modulemapping import ModuleName
    910
    from ..smart.smartmodule import SmartModule
    1011

    1112
    if TYPE_CHECKING:
    13+
    from . import modules
    1214
    from .smartcamera import SmartCamera
    1315

    1416
    _LOGGER = logging.getLogger(__name__)
    @@ -17,12 +19,14 @@
    1719
    class SmartCameraModule(SmartModule):
    1820
    """Base class for SMARTCAMERA modules."""
    1921

    22+
    SmartCameraAlarm: Final[ModuleName[modules.Alarm]] = ModuleName("SmartCameraAlarm")
    23+
    2024
    #: Query to execute during the main update cycle
    2125
    QUERY_GETTER_NAME: str
    2226
    #: Module name to be queried
    2327
    QUERY_MODULE_NAME: str
    2428
    #: Section name or names to be queried
    25-
    QUERY_SECTION_NAMES: str | list[str]
    29+
    QUERY_SECTION_NAMES: str | list[str] | None = None
    2630

    2731
    REGISTERED_MODULES = {}
    2832

    @@ -33,11 +37,10 @@ def query(self) -> dict:
    3337
    3438
    Default implementation uses the raw query getter w/o parameters.
    3539
    """
    36-
    return {
    37-
    self.QUERY_GETTER_NAME: {
    38-
    self.QUERY_MODULE_NAME: {"name": self.QUERY_SECTION_NAMES}
    39-
    }
    40-
    }
    40+
    section_names = (
    41+
    {"name": self.QUERY_SECTION_NAMES} if self.QUERY_SECTION_NAMES else {}
    42+
    )
    43+
    return {self.QUERY_GETTER_NAME: {self.QUERY_MODULE_NAME: section_names}}
    4144

    4245
    async def call(self, method: str, params: dict | None = None) -> dict:
    4346
    """Call a method.

    ‎tests/fakeprotocol_smartcamera.py

    Lines changed: 35 additions & 36 deletions
    Original file line numberDiff line numberDiff line change
    @@ -105,43 +105,28 @@ def _get_param_set_value(info: dict, set_keys: list[str], value):
    105105
    info = info[key]
    106106
    info[set_keys[-1]] = value
    107107

    108+
    # Setters for when there's not a simple mapping of setters to getters
    108109
    SETTERS = {
    109110
    ("system", "sys", "dev_alias"): [
    110111
    "getDeviceInfo",
    111112
    "device_info",
    112113
    "basic_info",
    113114
    "device_alias",
    114115
    ],
    115-
    ("lens_mask", "lens_mask_info", "enabled"): [
    116-
    "getLensMaskConfig",
    117-
    "lens_mask",
    118-
    "lens_mask_info",
    119-
    "enabled",
    120-
    ],
    116+
    # setTimezone maps to getClockStatus
    121117
    ("system", "clock_status", "seconds_from_1970"): [
    122118
    "getClockStatus",
    123119
    "system",
    124120
    "clock_status",
    125121
    "seconds_from_1970",
    126122
    ],
    123+
    # setTimezone maps to getClockStatus
    127124
    ("system", "clock_status", "local_time"): [
    128125
    "getClockStatus",
    129126
    "system",
    130127
    "clock_status",
    131128
    "local_time",
    132129
    ],
    133-
    ("system", "basic", "zone_id"): [
    134-
    "getTimezone",
    135-
    "system",
    136-
    "basic",
    137-
    "zone_id",
    138-
    ],
    139-
    ("led", "config", "enabled"): [
    140-
    "getLedStatus",
    141-
    "led",
    142-
    "config",
    143-
    "enabled",
    144-
    ],
    145130
    }
    146131

    147132
    async def _send_request(self, request_dict: dict):
    @@ -154,27 +139,41 @@ async def _send_request(self, request_dict: dict):
    154139
    )
    155140

    156141
    if method[:3] == "set":
    142+
    get_method = "g" + method[1:]
    157143
    for key, val in request_dict.items():
    158-
    if key != "method":
    159-
    # key is params for multi request and the actual params
    160-
    # for single requests
    161-
    if key == "params":
    162-
    module = next(iter(val))
    163-
    val = val[module]
    144+
    if key == "method":
    145+
    continue
    146+
    # key is params for multi request and the actual params
    147+
    # for single requests
    148+
    if key == "params":
    149+
    module = next(iter(val))
    150+
    val = val[module]
    151+
    else:
    152+
    module = key
    153+
    section = next(iter(val))
    154+
    skey_val = val[section]
    155+
    if not isinstance(skey_val, dict): # single level query
    156+
    section_key = section
    157+
    section_val = skey_val
    158+
    if (get_info := info.get(get_method)) and section_key in get_info:
    159+
    get_info[section_key] = section_val
    164160
    else:
    165-
    module = key
    166-
    section = next(iter(val))
    167-
    skey_val = val[section]
    168-
    for skey, sval in skey_val.items():
    169-
    section_key = skey
    170-
    section_value = sval
    171-
    if setter_keys := self.SETTERS.get(
    172-
    (module, section, section_key)
    173-
    ):
    174-
    self._get_param_set_value(info, setter_keys, section_value)
    175-
    else:
    176-
    return {"error_code": -1}
    161+
    return {"error_code": -1}
    177162
    break
    163+
    for skey, sval in skey_val.items():
    164+
    section_key = skey
    165+
    section_value = sval
    166+
    if setter_keys := self.SETTERS.get((module, section, section_key)):
    167+
    self._get_param_set_value(info, setter_keys, section_value)
    168+
    elif (
    169+
    section := info.get(get_method, {})
    170+
    .get(module, {})
    171+
    .get(section, {})
    172+
    ) and section_key in section:
    173+
    section[section_key] = section_value
    174+
    else:
    175+
    return {"error_code": -1}
    176+
    break
    178177
    return {"error_code": 0}
    179178
    elif method[:3] == "get":
    180179
    params = request_dict.get("params")

    ‎tests/smartcamera/modules/__init__.py

    Whitespace-only changes.

    0 commit comments

    Comments
     (0)
    0