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

Commit cf77128

Browse files
sdb9696sdb9696
authored
Add alarm module for smartcamera hubs (#1258)
1 parent 5fe75ca commit cf77128

File tree

6 files changed

+372
-43
lines changed

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