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

This commit is contained in:
Steven B. 2024-12-13 19:45:38 +00:00 committed by GitHub
parent 223f3318ea
commit 2ca6d3ebe9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 183 additions and 12 deletions

View File

@ -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")

View File

@ -23,6 +23,7 @@ from .lighteffect import LightEffect
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
@ -66,4 +67,5 @@ __all__ = [
"Thermostat",
"SmartLightEffect",
"OverheatProtection",
"Matter",
]

View File

@ -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

View File

@ -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]] = {}
@ -138,7 +138,9 @@ class SmartModule(Module):
Default implementation uses the raw query getter w/o parameters.
"""
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.

View File

@ -5,6 +5,7 @@ from .camera import Camera
from .childdevice import ChildDevice
from .device import DeviceModule
from .led import Led
from .matter import Matter
from .pantilt import PanTilt
from .time import Time
@ -16,4 +17,5 @@ __all__ = [
"Led",
"PanTilt",
"Time",
"Matter",
]

View File

@ -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

View File

@ -21,8 +21,6 @@ class SmartCamModule(SmartModule):
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
@ -37,6 +35,8 @@ class SmartCamModule(SmartModule):
Default implementation uses the raw query getter w/o parameters.
"""
if not self.QUERY_GETTER_NAME:
return {}
section_names = (
{"name": self.QUERY_SECTION_NAMES} if self.QUERY_SECTION_NAMES else {}
)
@ -86,7 +86,8 @@ class SmartCamModule(SmartModule):
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:

View File

@ -151,6 +151,13 @@ class FakeSmartTransport(BaseTransport):
"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):

View File

@ -44,6 +44,7 @@ class FakeSmartCamTransport(BaseTransport):
),
),
)
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.
@ -58,6 +59,13 @@ class FakeSmartCamTransport(BaseTransport):
# 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."""
@ -112,6 +120,15 @@ class FakeSmartCamTransport(BaseTransport):
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"): [
@ -217,7 +234,16 @@ class FakeSmartCamTransport(BaseTransport):
start_index : start_index + self.list_return_size
]
return {"result": result, "error_code": 0}
else:
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}

View File

@ -145,12 +145,21 @@ def filter_fixtures(
def _component_match(
fixture_data: FixtureInfo, component_filter: str | ComponentFilter
):
if (component_nego := fixture_data.data.get("component_nego")) is None:
return False
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
if isinstance(component_filter, str):
return component_filter in components
else:

View File

@ -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

View File

@ -533,3 +533,16 @@ async def test_initialize_modules_required_component(
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() == {}