Skip to content

Add bare-bones matter modules to smart and smartcam devices #1371

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Dec 13, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions kasa/module.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,8 @@ class Module(ABC):
)
TriggerLogs: Final[ModuleName[smart.TriggerLogs]] = ModuleName("TriggerLogs")

Matter: Final[ModuleName[smart.Matter]] = ModuleName("Matter")

# SMARTCAM only modules
Camera: Final[ModuleName[smartcam.Camera]] = ModuleName("Camera")

Expand Down
2 changes: 2 additions & 0 deletions kasa/smart/modules/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from .lightpreset import LightPreset
from .lightstripeffect import LightStripEffect
from .lighttransition import LightTransition
from .matter import Matter
from .motionsensor import MotionSensor
from .overheatprotection import OverheatProtection
from .reportmode import ReportMode
Expand Down Expand Up @@ -66,4 +67,5 @@
"Thermostat",
"SmartLightEffect",
"OverheatProtection",
"Matter",
]
43 changes: 43 additions & 0 deletions kasa/smart/modules/matter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
"""Implementation of matter module."""

from __future__ import annotations

from ...feature import Feature
from ..smartmodule import SmartModule


class Matter(SmartModule):
"""Implementation of matter module."""

QUERY_GETTER_NAME: str = "get_matter_setup_info"
REQUIRED_COMPONENT = "matter"

def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
self._device,
id="matter_setup_code",
name="Matter setup code",
container=self,
attribute_getter=lambda x: x.info["setup_code"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
self._add_feature(
Feature(
self._device,
id="matter_setup_payload",
name="Matter setup payload",
container=self,
attribute_getter=lambda x: x.info["setup_payload"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)

@property
def info(self) -> dict[str, str]:
"""Matter setup info."""
return self.data
6 changes: 4 additions & 2 deletions kasa/smart/smartmodule.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ class SmartModule(Module):
#: Module is initialized, if any of the given keys exists in the sysinfo
SYSINFO_LOOKUP_KEYS: list[str] = []
#: Query to execute during the main update cycle
QUERY_GETTER_NAME: str
QUERY_GETTER_NAME: str = ""

REGISTERED_MODULES: dict[str, type[SmartModule]] = {}

Expand Down Expand Up @@ -138,7 +138,9 @@ def query(self) -> dict:

Default implementation uses the raw query getter w/o parameters.
"""
return {self.QUERY_GETTER_NAME: None}
if self.QUERY_GETTER_NAME:
return {self.QUERY_GETTER_NAME: None}
return {}

async def call(self, method: str, params: dict | None = None) -> dict:
"""Call a method.
Expand Down
2 changes: 2 additions & 0 deletions kasa/smartcam/modules/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from .childdevice import ChildDevice
from .device import DeviceModule
from .led import Led
from .matter import Matter
from .pantilt import PanTilt
from .time import Time

Expand All @@ -16,4 +17,5 @@
"Led",
"PanTilt",
"Time",
"Matter",
]
44 changes: 44 additions & 0 deletions kasa/smartcam/modules/matter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
"""Implementation of matter module."""

from __future__ import annotations

from ...feature import Feature
from ..smartcammodule import SmartCamModule


class Matter(SmartCamModule):
"""Implementation of matter module."""

QUERY_GETTER_NAME = "getMatterSetupInfo"
QUERY_MODULE_NAME = "matter"
REQUIRED_COMPONENT = "matter"

def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
self._device,
id="matter_setup_code",
name="Matter setup code",
container=self,
attribute_getter=lambda x: x.info["setup_code"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
self._add_feature(
Feature(
self._device,
id="matter_setup_payload",
name="Matter setup payload",
container=self,
attribute_getter=lambda x: x.info["setup_payload"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)

@property
def info(self) -> dict[str, str]:
"""Matter setup info."""
return self.data
7 changes: 4 additions & 3 deletions kasa/smartcam/smartcammodule.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,6 @@

SmartCamAlarm: Final[ModuleName[modules.Alarm]] = ModuleName("SmartCamAlarm")

#: 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
Expand All @@ -37,6 +35,8 @@

Default implementation uses the raw query getter w/o parameters.
"""
if not self.QUERY_GETTER_NAME:
return {}

Check warning on line 39 in kasa/smartcam/smartcammodule.py

View check run for this annotation

Codecov / codecov/patch

kasa/smartcam/smartcammodule.py#L39

Added line #L39 was not covered by tests
section_names = (
{"name": self.QUERY_SECTION_NAMES} if self.QUERY_SECTION_NAMES else {}
)
Expand Down Expand Up @@ -86,7 +86,8 @@
f" for '{self._module}'"
)

return query_resp.get(self.QUERY_MODULE_NAME)
# Some calls return the data under the module, others not
return query_resp.get(self.QUERY_MODULE_NAME, query_resp)
else:
found = {key: val for key, val in dev._last_update.items() if key in q}
for key in q:
Expand Down
7 changes: 7 additions & 0 deletions tests/fakeprotocol_smart.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,13 @@ def credentials_hash(self):
"energy_monitoring",
{"igain": 10861, "vgain": 118657},
),
"get_matter_setup_info": (
"matter",
{
"setup_code": "00000000000",
"setup_payload": "00:0000000-0000.00.000",
},
),
}

async def send(self, request: str):
Expand Down
30 changes: 28 additions & 2 deletions tests/fakeprotocol_smartcam.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ def __init__(
),
),
)

self.fixture_name = fixture_name
# When True verbatim will bypass any extra processing of missing
# methods and is used to test the fixture creation itself.
Expand All @@ -58,6 +59,13 @@ def __init__(
# self.child_protocols = self._get_child_protocols()
self.list_return_size = list_return_size

self.components = {
comp["name"]: comp["version"]
for comp in self.info["getAppComponentList"]["app_component"][
"app_component_list"
]
}

@property
def default_port(self):
"""Default port for the transport."""
Expand Down Expand Up @@ -112,6 +120,15 @@ def _get_param_set_value(info: dict, set_keys: list[str], value):
info = info[key]
info[set_keys[-1]] = value

FIXTURE_MISSING_MAP = {
"getMatterSetupInfo": (
"matter",
{
"setup_code": "00000000000",
"setup_payload": "00:0000000-0000.00.000",
},
)
}
# Setters for when there's not a simple mapping of setters to getters
SETTERS = {
("system", "sys", "dev_alias"): [
Expand Down Expand Up @@ -217,8 +234,17 @@ async def _send_request(self, request_dict: dict):
start_index : start_index + self.list_return_size
]
return {"result": result, "error_code": 0}
else:
return {"error_code": -1}
if (
# FIXTURE_MISSING is for service calls not in place when
# SMART fixtures started to be generated
missing_result := self.FIXTURE_MISSING_MAP.get(method)
) and missing_result[0] in self.components:
# Copy to info so it will work with update methods
info[method] = copy.deepcopy(missing_result[1])
result = copy.deepcopy(info[method])
return {"result": result, "error_code": 0}

return {"error_code": -1}
return {"error_code": -1}

async def close(self) -> None:
Expand Down
19 changes: 14 additions & 5 deletions tests/fixtureinfo.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,12 +145,21 @@ def _model_startswith_match(fixture_data: FixtureInfo, starts_with: str):
def _component_match(
fixture_data: FixtureInfo, component_filter: str | ComponentFilter
):
if (component_nego := fixture_data.data.get("component_nego")) is None:
components = {}
if component_nego := fixture_data.data.get("component_nego"):
components = {
component["id"]: component["ver_code"]
for component in component_nego["component_list"]
}
if get_app_component_list := fixture_data.data.get("getAppComponentList"):
components = {
component["name"]: component["version"]
for component in get_app_component_list["app_component"][
"app_component_list"
]
}
if not components:
return False
components = {
component["id"]: component["ver_code"]
for component in component_nego["component_list"]
}
if isinstance(component_filter, str):
return component_filter in components
else:
Expand Down
20 changes: 20 additions & 0 deletions tests/smart/modules/test_matter.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
from kasa import Module
from kasa.smart import SmartDevice

from ...device_fixtures import parametrize

matter = parametrize(
"has matter", component_filter="matter", protocol_filter={"SMART", "SMARTCAM"}
)


@matter
async def test_info(dev: SmartDevice):
"""Test matter info."""
matter = dev.modules.get(Module.Matter)
assert matter
assert matter.info
setup_code = dev.features.get("matter_setup_code")
assert setup_code
setup_payload = dev.features.get("matter_setup_payload")
assert setup_payload
13 changes: 13 additions & 0 deletions tests/smart/test_smartdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -533,3 +533,16 @@ class NonExistingComponent(SmartModule):

assert "AvailableComponent" in dev.modules
assert "NonExistingComponent" not in dev.modules


async def test_smartmodule_query():
"""Test that a module that doesn't set QUERY_GETTER_NAME has empty query."""

class DummyModule(SmartModule):
pass

dummy_device = await get_device_for_fixture_protocol(
"KS240(US)_1.0_1.0.5.json", "SMART"
)
mod = DummyModule(dummy_device, "dummy")
assert mod.query() == {}
Loading