diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 8c0438d9b..c274bb979 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -21,6 +21,11 @@ repos: hooks: - id: mypy additional_dependencies: [types-click] + exclude: | + (?x)^( + kasa/modulemapping\.py| + )$ + - repo: https://github.com/PyCQA/doc8 rev: 'v1.1.1' diff --git a/devtools/create_module_fixtures.py b/devtools/create_module_fixtures.py index 8372bfff5..ed881a88b 100644 --- a/devtools/create_module_fixtures.py +++ b/devtools/create_module_fixtures.py @@ -19,7 +19,7 @@ def create_fixtures(dev: IotDevice, outputdir: Path): """Iterate over supported modules and create version-specific fixture files.""" for name, module in dev.modules.items(): - module_dir = outputdir / name + module_dir = outputdir / str(name) if not module_dir.exists(): module_dir.mkdir(exist_ok=True, parents=True) diff --git a/kasa/__init__.py b/kasa/__init__.py index 0e0a29d1a..e9f64c708 100755 --- a/kasa/__init__.py +++ b/kasa/__init__.py @@ -40,6 +40,7 @@ IotProtocol, _deprecated_TPLinkSmartHomeProtocol, # noqa: F401 ) +from kasa.module import Module from kasa.protocol import BaseProtocol from kasa.smartprotocol import SmartProtocol @@ -60,6 +61,7 @@ "Device", "Bulb", "Plug", + "Module", "KasaException", "AuthenticationError", "DeviceError", diff --git a/kasa/device.py b/kasa/device.py index ea358a8de..8150352d9 100644 --- a/kasa/device.py +++ b/kasa/device.py @@ -6,7 +6,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime -from typing import Any, Mapping, Sequence, overload +from typing import TYPE_CHECKING, Any, Mapping, Sequence from .credentials import Credentials from .device_type import DeviceType @@ -15,10 +15,13 @@ from .exceptions import KasaException from .feature import Feature from .iotprotocol import IotProtocol -from .module import Module, ModuleT +from .module import Module from .protocol import BaseProtocol from .xortransport import XorTransport +if TYPE_CHECKING: + from .modulemapping import ModuleMapping + @dataclass class WifiNetwork: @@ -113,21 +116,9 @@ async def disconnect(self): @property @abstractmethod - def modules(self) -> Mapping[str, Module]: + def modules(self) -> ModuleMapping[Module]: """Return the device modules.""" - @overload - @abstractmethod - def get_module(self, module_type: type[ModuleT]) -> ModuleT | None: ... - - @overload - @abstractmethod - def get_module(self, module_type: str) -> Module | None: ... - - @abstractmethod - def get_module(self, module_type: type[ModuleT] | str) -> ModuleT | Module | None: - """Return the module from the device modules or None if not present.""" - @property @abstractmethod def is_on(self) -> bool: diff --git a/kasa/modules/ledmodule.py b/kasa/interfaces/ledinterface.py similarity index 58% rename from kasa/modules/ledmodule.py rename to kasa/interfaces/ledinterface.py index 0ea35e4b3..e7f920120 100644 --- a/kasa/modules/ledmodule.py +++ b/kasa/interfaces/ledinterface.py @@ -2,17 +2,15 @@ from __future__ import annotations +from abc import ABC, abstractmethod + from ..feature import Feature from ..module import Module -class LedModule(Module): +class LedInterface(Module, ABC): """Base interface to represent a LED module.""" - # This needs to implement abstract methods for typing to work with - # overload get_module(type[ModuleT]) -> ModuleT: - # https://discuss.python.org/t/add-abstracttype-to-the-typing-module/21996 - def _initialize_features(self): """Initialize features.""" device = self._device @@ -22,7 +20,7 @@ def _initialize_features(self): container=self, name="LED", id="led", - icon="mdi:led-{state}", + icon="mdi:led", attribute_getter="led", attribute_setter="set_led", type=Feature.Type.Switch, @@ -31,19 +29,10 @@ def _initialize_features(self): ) @property + @abstractmethod def led(self) -> bool: """Return current led status.""" - raise NotImplementedError() + @abstractmethod async def set_led(self, enable: bool) -> None: """Set led.""" - raise NotImplementedError() - - def query(self) -> dict: - """Query to execute during the update cycle.""" - raise NotImplementedError() - - @property - def data(self): - """Query to execute during the update cycle.""" - raise NotImplementedError() diff --git a/kasa/modules/lighteffectmodule.py b/kasa/interfaces/lighteffectinterface.py similarity index 74% rename from kasa/modules/lighteffectmodule.py rename to kasa/interfaces/lighteffectinterface.py index 795a44652..625834408 100644 --- a/kasa/modules/lighteffectmodule.py +++ b/kasa/interfaces/lighteffectinterface.py @@ -2,17 +2,15 @@ from __future__ import annotations +from abc import ABC, abstractmethod + from ..feature import Feature from ..module import Module -class LightEffectModule(Module): +class LightEffectInterface(Module, ABC): """Interface to represent a light effect module.""" - # This needs to implement abstract methods for typing to work with - # overload get_module(type[ModuleT]) -> ModuleT: - # https://discuss.python.org/t/add-abstracttype-to-the-typing-module/21996 - LIGHT_EFFECTS_OFF = "Off" def _initialize_features(self): @@ -33,24 +31,25 @@ def _initialize_features(self): ) @property + @abstractmethod def has_custom_effects(self) -> bool: """Return True if the device supports setting custom effects.""" - raise NotImplementedError() @property + @abstractmethod def effect(self) -> str: """Return effect state or name.""" - raise NotImplementedError() @property + @abstractmethod def effect_list(self) -> list[str]: """Return built-in effects list. Example: ['Aurora', 'Bubbling Cauldron', ...] """ - raise NotImplementedError() + @abstractmethod async def set_effect( self, effect: str, @@ -70,7 +69,6 @@ async def set_effect( :param int brightness: The wanted brightness :param int transition: The wanted transition time """ - raise NotImplementedError() async def set_custom_effect( self, @@ -80,13 +78,3 @@ async def set_custom_effect( :param str effect_dict: The custom effect dict to set """ - raise NotImplementedError() - - def query(self) -> dict: - """Query to execute during the update cycle.""" - raise NotImplementedError() - - @property - def data(self): - """Query to execute during the update cycle.""" - raise NotImplementedError() diff --git a/kasa/iot/iotdevice.py b/kasa/iot/iotdevice.py index b7f3ce438..1bf6cb992 100755 --- a/kasa/iot/iotdevice.py +++ b/kasa/iot/iotdevice.py @@ -19,14 +19,15 @@ import inspect import logging from datetime import datetime, timedelta -from typing import Any, Mapping, Sequence, cast, overload +from typing import Any, Mapping, Sequence, cast from ..device import Device, WifiNetwork from ..deviceconfig import DeviceConfig from ..emeterstatus import EmeterStatus from ..exceptions import KasaException from ..feature import Feature -from ..module import Module, ModuleT +from ..module import Module +from ..modulemapping import ModuleMapping, ModuleName from ..protocol import BaseProtocol from .iotmodule import IotModule from .modules import Emeter, Time @@ -190,7 +191,7 @@ def __init__( self._supported_modules: dict[str, IotModule] | None = None self._legacy_features: set[str] = set() self._children: Mapping[str, IotDevice] = {} - self._modules: dict[str, IotModule] = {} + self._modules: dict[str | ModuleName[Module], IotModule] = {} @property def children(self) -> Sequence[IotDevice]: @@ -198,36 +199,18 @@ def children(self) -> Sequence[IotDevice]: return list(self._children.values()) @property - def modules(self) -> dict[str, IotModule]: + def modules(self) -> ModuleMapping[IotModule]: """Return the device modules.""" - return self._modules + return cast(ModuleMapping[IotModule], self._modules) - @overload - def get_module(self, module_type: type[ModuleT]) -> ModuleT | None: ... - - @overload - def get_module(self, module_type: str) -> IotModule | None: ... - - def get_module(self, module_type: type[ModuleT] | str) -> ModuleT | Module | None: - """Return the module from the device modules or None if not present.""" - if isinstance(module_type, str): - module_name = module_type.lower() - elif issubclass(module_type, Module): - module_name = module_type.__name__.lower() - else: - return None - if module_name in self.modules: - return self.modules[module_name] - return None - - def add_module(self, name: str, module: IotModule): + def add_module(self, name: str | ModuleName[Module], module: IotModule): """Register a module.""" if name in self.modules: _LOGGER.debug("Module %s already registered, ignoring..." % name) return _LOGGER.debug("Adding module %s", module) - self.modules[name] = module + self._modules[name] = module def _create_request( self, target: str, cmd: str, arg: dict | None = None, child_ids=None @@ -289,11 +272,11 @@ def features(self) -> dict[str, Feature]: @property # type: ignore @requires_update - def supported_modules(self) -> list[str]: + def supported_modules(self) -> list[str | ModuleName[Module]]: """Return a set of modules supported by the device.""" # TODO: this should rather be called `features`, but we don't want to break # the API now. Maybe just deprecate it and point the users to use this? - return list(self.modules.keys()) + return list(self._modules.keys()) @property # type: ignore @requires_update @@ -368,7 +351,7 @@ async def _modular_update(self, req: dict) -> None: # making separate handling for this unnecessary if self._supported_modules is None: supported = {} - for module in self.modules.values(): + for module in self._modules.values(): if module.is_supported: supported[module._module] = module @@ -376,7 +359,7 @@ async def _modular_update(self, req: dict) -> None: request_list = [] est_response_size = 1024 if "system" in req else 0 - for module in self.modules.values(): + for module in self._modules.values(): if not module.is_supported: _LOGGER.debug("Module %s not supported, skipping" % module) continue diff --git a/kasa/iot/iotlightstrip.py b/kasa/iot/iotlightstrip.py index 422985b56..a120be7a7 100644 --- a/kasa/iot/iotlightstrip.py +++ b/kasa/iot/iotlightstrip.py @@ -4,6 +4,7 @@ from ..device_type import DeviceType from ..deviceconfig import DeviceConfig +from ..module import Module from ..protocol import BaseProtocol from .effects import EFFECT_NAMES_V1 from .iotbulb import IotBulb @@ -55,10 +56,10 @@ def __init__( ) -> None: super().__init__(host=host, config=config, protocol=protocol) self._device_type = DeviceType.LightStrip - self._light_effect_module = LightEffectModule( - self, "smartlife.iot.lighting_effect" + self.add_module( + Module.LightEffect, + LightEffectModule(self, "smartlife.iot.lighting_effect"), ) - self.add_module("lighteffectmodule", self._light_effect_module) @property # type: ignore @requires_update @@ -79,7 +80,7 @@ def effect(self) -> dict: 'name': ''} """ # LightEffectModule returns the current effect name - # so return the dict here for backwards compatability + # so return the dict here for backwards compatibility return self.sys_info["lighting_effect_state"] @property # type: ignore @@ -91,7 +92,7 @@ def effect_list(self) -> list[str] | None: ['Aurora', 'Bubbling Cauldron', ...] """ # LightEffectModule returns effect names along with a LIGHT_EFFECTS_OFF value - # so return the original effect names here for backwards compatability + # so return the original effect names here for backwards compatibility return EFFECT_NAMES_V1 if self.has_effects else None @requires_update @@ -114,7 +115,7 @@ async def set_effect( :param int brightness: The wanted brightness :param int transition: The wanted transition time """ - await self._light_effect_module.set_effect( + await self.modules[Module.LightEffect].set_effect( effect, brightness=brightness, transition=transition ) @@ -129,4 +130,4 @@ async def set_custom_effect( """ if not self.has_effects: raise KasaException("Bulb does not support effects.") - await self._light_effect_module.set_custom_effect(effect_dict) + await self.modules[Module.LightEffect].set_custom_effect(effect_dict) diff --git a/kasa/iot/iotplug.py b/kasa/iot/iotplug.py index fe1b691e8..22238c7a5 100644 --- a/kasa/iot/iotplug.py +++ b/kasa/iot/iotplug.py @@ -6,6 +6,7 @@ from ..device_type import DeviceType from ..deviceconfig import DeviceConfig +from ..module import Module from ..protocol import BaseProtocol from .iotdevice import IotDevice, requires_update from .modules import Antitheft, Cloud, LedModule, Schedule, Time, Usage @@ -57,8 +58,7 @@ def __init__( self.add_module("antitheft", Antitheft(self, "anti_theft")) self.add_module("time", Time(self, "time")) self.add_module("cloud", Cloud(self, "cnCloud")) - self._led_module = LedModule(self, "system") - self.add_module("ledmodule", self._led_module) + self.add_module(Module.Led, LedModule(self, "system")) @property # type: ignore @requires_update @@ -79,11 +79,11 @@ async def turn_off(self, **kwargs): @requires_update def led(self) -> bool: """Return the state of the led.""" - return self._led_module.led + return self.modules[Module.Led].led async def set_led(self, state: bool): """Set the state of the led (night mode).""" - return await self._led_module.set_led(state) + return await self.modules[Module.Led].set_led(state) class IotWallSwitch(IotPlug): diff --git a/kasa/iot/iotstrip.py b/kasa/iot/iotstrip.py index 9e99a0748..ab14abb0a 100755 --- a/kasa/iot/iotstrip.py +++ b/kasa/iot/iotstrip.py @@ -253,7 +253,6 @@ def __init__(self, host: str, parent: IotStrip, child_id: str) -> None: self._last_update = parent._last_update self._set_sys_info(parent.sys_info) self._device_type = DeviceType.StripSocket - self._modules = {} self.protocol = parent.protocol # Must use the same connection as the parent self.add_module("time", Time(self, "time")) diff --git a/kasa/iot/modules/ledmodule.py b/kasa/iot/modules/ledmodule.py index bed0273a2..60341a18d 100644 --- a/kasa/iot/modules/ledmodule.py +++ b/kasa/iot/modules/ledmodule.py @@ -2,16 +2,13 @@ from __future__ import annotations -from ...modules.ledmodule import LedModule as BaseLedModule +from ...interfaces.ledinterface import LedInterface from ..iotmodule import IotModule -class LedModule(IotModule, BaseLedModule): +class LedModule(IotModule, LedInterface): """Implementation of led controls.""" - REQUIRED_COMPONENT = "led" - QUERY_GETTER_NAME = "get_led_info" - def query(self) -> dict: """Query to execute during the update cycle.""" return {} diff --git a/kasa/iot/modules/lighteffectmodule.py b/kasa/iot/modules/lighteffectmodule.py index b1035ac16..1ec09a0b8 100644 --- a/kasa/iot/modules/lighteffectmodule.py +++ b/kasa/iot/modules/lighteffectmodule.py @@ -2,12 +2,12 @@ from __future__ import annotations -from ...modules.lighteffectmodule import LightEffectModule as BaseLightEffectModule +from ...interfaces.lighteffectinterface import LightEffectInterface from ..effects import EFFECT_MAPPING_V1, EFFECT_NAMES_V1 from ..iotmodule import IotModule -class LightEffectModule(IotModule, BaseLightEffectModule): +class LightEffectModule(IotModule, LightEffectInterface): """Implementation of dynamic light effects.""" @property diff --git a/kasa/module.py b/kasa/module.py index 3da0c1ad2..3ce3cf11a 100644 --- a/kasa/module.py +++ b/kasa/module.py @@ -6,14 +6,20 @@ from abc import ABC, abstractmethod from typing import ( TYPE_CHECKING, + Final, TypeVar, ) from .exceptions import KasaException from .feature import Feature +from .modulemapping import ModuleName if TYPE_CHECKING: - from .device import Device + from .device import Device as DeviceType # avoid name clash with Device module + from .interfaces.ledinterface import LedInterface + from .interfaces.lighteffectinterface import LightEffectInterface + from .iot import modules as iot + from .smart import modules as smart _LOGGER = logging.getLogger(__name__) @@ -27,7 +33,61 @@ class Module(ABC): executed during the regular update cycle. """ - def __init__(self, device: Device, module: str): + # Common Modules + LightEffect: Final[ModuleName[LightEffectInterface]] = ModuleName( + "LightEffectModule" + ) + Led: Final[ModuleName[LedInterface]] = ModuleName("LedModule") + + # IOT only Modules + IotAmbientLight: Final[ModuleName[iot.AmbientLight]] = ModuleName("ambient") + IotAntitheft: Final[ModuleName[iot.Antitheft]] = ModuleName("anti_theft") + IotCountdown: Final[ModuleName[iot.Countdown]] = ModuleName("countdown") + IotEmeter: Final[ModuleName[iot.Emeter]] = ModuleName("emeter") + IotMotion: Final[ModuleName[iot.Motion]] = ModuleName("motion") + IotSchedule: Final[ModuleName[iot.Schedule]] = ModuleName("schedule") + IotUsage: Final[ModuleName[iot.Usage]] = ModuleName("usage") + IotCloud: Final[ModuleName[iot.Cloud]] = ModuleName("cloud") + IotTime: Final[ModuleName[iot.Time]] = ModuleName("time") + + # SMART only Modules + Alarm: Final[ModuleName[smart.AlarmModule]] = ModuleName("AlarmModule") + AutoOff: Final[ModuleName[smart.AutoOffModule]] = ModuleName("AutoOffModule") + BatterySensor: Final[ModuleName[smart.BatterySensor]] = ModuleName("BatterySensor") + Brightness: Final[ModuleName[smart.Brightness]] = ModuleName("Brightness") + ChildDevice: Final[ModuleName[smart.ChildDeviceModule]] = ModuleName( + "ChildDeviceModule" + ) + Cloud: Final[ModuleName[smart.CloudModule]] = ModuleName("CloudModule") + Color: Final[ModuleName[smart.ColorModule]] = ModuleName("ColorModule") + ColorTemp: Final[ModuleName[smart.ColorTemperatureModule]] = ModuleName( + "ColorTemperatureModule" + ) + ContactSensor: Final[ModuleName[smart.ContactSensor]] = ModuleName("ContactSensor") + Device: Final[ModuleName[smart.DeviceModule]] = ModuleName("DeviceModule") + Energy: Final[ModuleName[smart.EnergyModule]] = ModuleName("EnergyModule") + Fan: Final[ModuleName[smart.FanModule]] = ModuleName("FanModule") + Firmware: Final[ModuleName[smart.Firmware]] = ModuleName("Firmware") + FrostProtection: Final[ModuleName[smart.FrostProtectionModule]] = ModuleName( + "FrostProtectionModule" + ) + Humidity: Final[ModuleName[smart.HumiditySensor]] = ModuleName("HumiditySensor") + LightTransition: Final[ModuleName[smart.LightTransitionModule]] = ModuleName( + "LightTransitionModule" + ) + Report: Final[ModuleName[smart.ReportModule]] = ModuleName("ReportModule") + Temperature: Final[ModuleName[smart.TemperatureSensor]] = ModuleName( + "TemperatureSensor" + ) + TemperatureSensor: Final[ModuleName[smart.TemperatureControl]] = ModuleName( + "TemperatureControl" + ) + Time: Final[ModuleName[smart.TimeModule]] = ModuleName("TimeModule") + WaterleakSensor: Final[ModuleName[smart.WaterleakSensor]] = ModuleName( + "WaterleakSensor" + ) + + def __init__(self, device: DeviceType, module: str): self._device = device self._module = module self._module_features: dict[str, Feature] = {} diff --git a/kasa/modulemapping.py b/kasa/modulemapping.py new file mode 100644 index 000000000..06ba86190 --- /dev/null +++ b/kasa/modulemapping.py @@ -0,0 +1,25 @@ +"""Module for Implementation for ModuleMapping and ModuleName types. + +Custom dict for getting typed modules from the module dict. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Generic, TypeVar + +if TYPE_CHECKING: + from .module import Module + +_ModuleT = TypeVar("_ModuleT", bound="Module") + + +class ModuleName(str, Generic[_ModuleT]): + """Generic Module name type. + + At runtime this is a generic subclass of str. + """ + + __slots__ = () + + +ModuleMapping = dict diff --git a/kasa/modulemapping.pyi b/kasa/modulemapping.pyi new file mode 100644 index 000000000..8d110d39f --- /dev/null +++ b/kasa/modulemapping.pyi @@ -0,0 +1,96 @@ +"""Typing stub file for ModuleMapping.""" + +from abc import ABCMeta +from collections.abc import Mapping +from typing import Generic, TypeVar, overload + +from .module import Module + +__all__ = [ + "ModuleMapping", + "ModuleName", +] + +_ModuleT = TypeVar("_ModuleT", bound=Module, covariant=True) +_ModuleBaseT = TypeVar("_ModuleBaseT", bound=Module, covariant=True) + +class ModuleName(Generic[_ModuleT]): + """Class for typed Module names. At runtime delegated to str.""" + + def __init__(self, value: str, /) -> None: ... + def __len__(self) -> int: ... + def __hash__(self) -> int: ... + def __eq__(self, other: object) -> bool: ... + def __getitem__(self, index: int) -> str: ... + +class ModuleMapping( + Mapping[ModuleName[_ModuleBaseT] | str, _ModuleBaseT], metaclass=ABCMeta +): + """Custom dict type to provide better value type hints for Module key types.""" + + @overload + def __getitem__(self, key: ModuleName[_ModuleT], /) -> _ModuleT: ... + @overload + def __getitem__(self, key: str, /) -> _ModuleBaseT: ... + @overload + def __getitem__( + self, key: ModuleName[_ModuleT] | str, / + ) -> _ModuleT | _ModuleBaseT: ... + @overload # type: ignore[override] + def get(self, key: ModuleName[_ModuleT], /) -> _ModuleT | None: ... + @overload + def get(self, key: str, /) -> _ModuleBaseT | None: ... + @overload + def get( + self, key: ModuleName[_ModuleT] | str, / + ) -> _ModuleT | _ModuleBaseT | None: ... + +def _test_module_mapping_typing() -> None: + """Test ModuleMapping overloads work as intended. + + This is tested during the mypy run and needs to be in this file. + """ + from typing import Any, NewType, cast + + from typing_extensions import assert_type + + from .iot.iotmodule import IotModule + from .module import Module + from .smart.smartmodule import SmartModule + + NewCommonModule = NewType("NewCommonModule", Module) + NewIotModule = NewType("NewIotModule", IotModule) + NewSmartModule = NewType("NewSmartModule", SmartModule) + NotModule = NewType("NotModule", list) + + NEW_COMMON_MODULE: ModuleName[NewCommonModule] = ModuleName("NewCommonModule") + NEW_IOT_MODULE: ModuleName[NewIotModule] = ModuleName("NewIotModule") + NEW_SMART_MODULE: ModuleName[NewSmartModule] = ModuleName("NewSmartModule") + + # TODO Enable --warn-unused-ignores + NOT_MODULE: ModuleName[NotModule] = ModuleName("NotModule") # type: ignore[type-var] # noqa: F841 + NOT_MODULE_2 = ModuleName[NotModule]("NotModule2") # type: ignore[type-var] # noqa: F841 + + device_modules: ModuleMapping[Module] = cast(ModuleMapping[Module], {}) + assert_type(device_modules[NEW_COMMON_MODULE], NewCommonModule) + assert_type(device_modules[NEW_IOT_MODULE], NewIotModule) + assert_type(device_modules[NEW_SMART_MODULE], NewSmartModule) + assert_type(device_modules["foobar"], Module) + assert_type(device_modules[3], Any) # type: ignore[call-overload] + + assert_type(device_modules.get(NEW_COMMON_MODULE), NewCommonModule | None) + assert_type(device_modules.get(NEW_IOT_MODULE), NewIotModule | None) + assert_type(device_modules.get(NEW_SMART_MODULE), NewSmartModule | None) + assert_type(device_modules.get(NEW_COMMON_MODULE, default=[1, 2]), Any) # type: ignore[call-overload] + + iot_modules: ModuleMapping[IotModule] = cast(ModuleMapping[IotModule], {}) + smart_modules: ModuleMapping[SmartModule] = cast(ModuleMapping[SmartModule], {}) + + assert_type(smart_modules["foobar"], SmartModule) + assert_type(iot_modules["foobar"], IotModule) + + # Test for covariance + device_modules_2: ModuleMapping[Module] = iot_modules # noqa: F841 + device_modules_3: ModuleMapping[Module] = smart_modules # noqa: F841 + NEW_MODULE: ModuleName[Module] = NEW_SMART_MODULE # noqa: F841 + NEW_MODULE_2: ModuleName[Module] = NEW_IOT_MODULE # noqa: F841 diff --git a/kasa/modules/__init__.py b/kasa/modules/__init__.py deleted file mode 100644 index 4e1dbafb4..000000000 --- a/kasa/modules/__init__.py +++ /dev/null @@ -1,78 +0,0 @@ -"""Sub Package for device family independant modules.""" - -# Iot Modules -from ..iot.modules.ambientlight import AmbientLight -from ..iot.modules.antitheft import Antitheft -from ..iot.modules.cloud import Cloud -from ..iot.modules.countdown import Countdown -from ..iot.modules.emeter import Emeter -from ..iot.modules.motion import Motion -from ..iot.modules.rulemodule import Rule, RuleModule -from ..iot.modules.schedule import Schedule -from ..iot.modules.time import Time -from ..iot.modules.usage import Usage - -# Smart Modules -from ..smart.modules.alarmmodule import AlarmModule -from ..smart.modules.autooffmodule import AutoOffModule -from ..smart.modules.battery import BatterySensor -from ..smart.modules.brightness import Brightness -from ..smart.modules.childdevicemodule import ChildDeviceModule -from ..smart.modules.cloudmodule import CloudModule -from ..smart.modules.colormodule import ColorModule -from ..smart.modules.colortemp import ColorTemperatureModule -from ..smart.modules.devicemodule import DeviceModule -from ..smart.modules.energymodule import EnergyModule -from ..smart.modules.fanmodule import FanModule -from ..smart.modules.firmware import Firmware -from ..smart.modules.frostprotection import FrostProtectionModule -from ..smart.modules.humidity import HumiditySensor -from ..smart.modules.lighttransitionmodule import LightTransitionModule -from ..smart.modules.reportmodule import ReportModule -from ..smart.modules.temperature import TemperatureSensor -from ..smart.modules.temperaturecontrol import TemperatureControl -from ..smart.modules.timemodule import TimeModule -from ..smart.modules.waterleak import WaterleakSensor - -# Common Modules -from .ledmodule import LedModule -from .lighteffectmodule import LightEffectModule - -__all__ = [ - # Common modules - "LightEffectModule", - "LedModule", - # Iot Modules - "AmbientLight", - "Antitheft", - "Cloud", - "Countdown", - "Emeter", - "Motion", - "Rule", - "RuleModule", - "Schedule", - "Time", - "Usage", - # Smart Modules - "AlarmModule", - "TimeModule", - "EnergyModule", - "DeviceModule", - "ChildDeviceModule", - "BatterySensor", - "HumiditySensor", - "TemperatureSensor", - "TemperatureControl", - "ReportModule", - "AutoOffModule", - "Brightness", - "FanModule", - "Firmware", - "CloudModule", - "LightTransitionModule", - "ColorTemperatureModule", - "ColorModule", - "WaterleakSensor", - "FrostProtectionModule", -] diff --git a/kasa/smart/modules/ledmodule.py b/kasa/smart/modules/ledmodule.py index f600647bc..20f69a2f3 100644 --- a/kasa/smart/modules/ledmodule.py +++ b/kasa/smart/modules/ledmodule.py @@ -2,11 +2,11 @@ from __future__ import annotations -from ...modules.ledmodule import LedModule as BaseLedModule +from ...interfaces.ledinterface import LedInterface from ..smartmodule import SmartModule -class LedModule(SmartModule, BaseLedModule): +class LedModule(SmartModule, LedInterface): """Implementation of led controls.""" REQUIRED_COMPONENT = "led" diff --git a/kasa/smart/modules/lighteffectmodule.py b/kasa/smart/modules/lighteffectmodule.py index 667455f5d..eacf5bba4 100644 --- a/kasa/smart/modules/lighteffectmodule.py +++ b/kasa/smart/modules/lighteffectmodule.py @@ -6,14 +6,14 @@ import copy from typing import TYPE_CHECKING, Any -from ...modules.lighteffectmodule import LightEffectModule as BaseLightEffectModule +from ...interfaces.lighteffectinterface import LightEffectInterface from ..smartmodule import SmartModule if TYPE_CHECKING: from ..smartdevice import SmartDevice -class LightEffectModule(SmartModule, BaseLightEffectModule): +class LightEffectModule(SmartModule, LightEffectInterface): """Implementation of dynamic light effects.""" REQUIRED_COMPONENT = "light_effect" diff --git a/kasa/smart/smartdevice.py b/kasa/smart/smartdevice.py index 891129414..066ddf65b 100644 --- a/kasa/smart/smartdevice.py +++ b/kasa/smart/smartdevice.py @@ -5,7 +5,7 @@ import base64 import logging from datetime import datetime, timedelta -from typing import Any, Mapping, Sequence, cast, overload +from typing import Any, Mapping, Sequence, cast from ..aestransport import AesTransport from ..bulb import HSV, Bulb, BulbPreset, ColorTempRange @@ -16,7 +16,8 @@ from ..exceptions import AuthenticationError, DeviceError, KasaException, SmartErrorCode from ..fan import Fan from ..feature import Feature -from ..module import Module, ModuleT +from ..module import Module +from ..modulemapping import ModuleMapping, ModuleName from ..smartprotocol import SmartProtocol from .modules import ( Brightness, @@ -61,7 +62,7 @@ def __init__( self._components_raw: dict[str, Any] | None = None self._components: dict[str, int] = {} self._state_information: dict[str, Any] = {} - self._modules: dict[str, SmartModule] = {} + self._modules: dict[str | ModuleName[Module], SmartModule] = {} self._exposes_child_modules = False self._parent: SmartDevice | None = None self._children: Mapping[str, SmartDevice] = {} @@ -102,9 +103,17 @@ def children(self) -> Sequence[SmartDevice]: return list(self._children.values()) @property - def modules(self) -> dict[str, SmartModule]: + def modules(self) -> ModuleMapping[SmartModule]: """Return the device modules.""" - return self._modules + if self._exposes_child_modules: + modules = {k: v for k, v in self._modules.items()} + for child in self._children.values(): + for k, v in child._modules.items(): + if k not in modules: + modules[k] = v + return cast(ModuleMapping[SmartModule], modules) + + return cast(ModuleMapping[SmartModule], self._modules) def _try_get_response(self, responses: dict, request: str, default=None) -> dict: response = responses.get(request) @@ -315,28 +324,6 @@ async def _initialize_features(self): for feat in module._module_features.values(): self._add_feature(feat) - @overload - def get_module(self, module_type: type[ModuleT]) -> ModuleT | None: ... - - @overload - def get_module(self, module_type: str) -> SmartModule | None: ... - - def get_module(self, module_type: type[ModuleT] | str) -> ModuleT | Module | None: - """Return the module from the device modules or None if not present.""" - if isinstance(module_type, str): - module_name = module_type - elif issubclass(module_type, Module): - module_name = module_type.__name__ - else: - return None - if module_name in self.modules: - return self.modules[module_name] - elif self._exposes_child_modules: - for child in self._children.values(): - if module_name in child.modules: - return child.modules[module_name] - return None - @property def is_cloud_connected(self): """Returns if the device is connected to the cloud.""" diff --git a/kasa/tests/fakeprotocol_smart.py b/kasa/tests/fakeprotocol_smart.py index ef3dacd8c..f43e5f8ba 100644 --- a/kasa/tests/fakeprotocol_smart.py +++ b/kasa/tests/fakeprotocol_smart.py @@ -189,6 +189,11 @@ def _set_light_effect(self, info, params): if "current_rule_id" in info["get_dynamic_light_effect_rules"]: del info["get_dynamic_light_effect_rules"]["current_rule_id"] + def _set_led_info(self, info, params): + """Set or remove values as per the device behaviour.""" + info["get_led_info"]["led_status"] = params["led_rule"] != "never" + info["get_led_info"]["led_rule"] = params["led_rule"] + def _send_request(self, request_dict: dict): method = request_dict["method"] params = request_dict["params"] @@ -242,8 +247,7 @@ def _send_request(self, request_dict: dict): self._set_light_effect(info, params) return {"error_code": 0} elif method == "set_led_info": - info["get_led_info"]["led_status"] = params["led_rule"] != "never" - info["get_led_info"]["led_rule"] = params["led_rule"] + self._set_led_info(info, params) return {"error_code": 0} elif method[:4] == "set_": target_method = f"get_{method[4:]}" diff --git a/kasa/tests/smart/features/test_brightness.py b/kasa/tests/smart/features/test_brightness.py index 02a396aae..3c00a4d11 100644 --- a/kasa/tests/smart/features/test_brightness.py +++ b/kasa/tests/smart/features/test_brightness.py @@ -10,7 +10,7 @@ @brightness async def test_brightness_component(dev: SmartDevice): """Test brightness feature.""" - brightness = dev.get_module("Brightness") + brightness = dev.modules.get("Brightness") assert brightness assert isinstance(dev, SmartDevice) assert "brightness" in dev._components diff --git a/kasa/tests/smart/modules/test_contact.py b/kasa/tests/smart/modules/test_contact.py index fc3375450..88677c58f 100644 --- a/kasa/tests/smart/modules/test_contact.py +++ b/kasa/tests/smart/modules/test_contact.py @@ -1,7 +1,6 @@ import pytest -from kasa import SmartDevice -from kasa.smart.modules import ContactSensor +from kasa import Module, SmartDevice from kasa.tests.device_fixtures import parametrize contact = parametrize( @@ -18,7 +17,7 @@ ) async def test_contact_features(dev: SmartDevice, feature, type): """Test that features are registered and work as expected.""" - contact = dev.get_module(ContactSensor) + contact = dev.modules.get(Module.ContactSensor) assert contact is not None prop = getattr(contact, feature) diff --git a/kasa/tests/smart/modules/test_fan.py b/kasa/tests/smart/modules/test_fan.py index 372459510..9597471b6 100644 --- a/kasa/tests/smart/modules/test_fan.py +++ b/kasa/tests/smart/modules/test_fan.py @@ -1,8 +1,8 @@ import pytest from pytest_mock import MockerFixture +from kasa import Module from kasa.smart import SmartDevice -from kasa.smart.modules import FanModule from kasa.tests.device_fixtures import parametrize fan = parametrize("has fan", component_filter="fan_control", protocol_filter={"SMART"}) @@ -11,7 +11,7 @@ @fan async def test_fan_speed(dev: SmartDevice, mocker: MockerFixture): """Test fan speed feature.""" - fan = dev.get_module(FanModule) + fan = dev.modules.get(Module.Fan) assert fan level_feature = fan._module_features["fan_speed_level"] @@ -36,7 +36,7 @@ async def test_fan_speed(dev: SmartDevice, mocker: MockerFixture): @fan async def test_sleep_mode(dev: SmartDevice, mocker: MockerFixture): """Test sleep mode feature.""" - fan = dev.get_module(FanModule) + fan = dev.modules.get(Module.Fan) assert fan sleep_feature = fan._module_features["fan_sleep_mode"] assert isinstance(sleep_feature.value, bool) @@ -55,7 +55,7 @@ async def test_sleep_mode(dev: SmartDevice, mocker: MockerFixture): async def test_fan_interface(dev: SmartDevice, mocker: MockerFixture): """Test fan speed on device interface.""" assert isinstance(dev, SmartDevice) - fan = dev.get_module(FanModule) + fan = dev.modules.get(Module.Fan) assert fan device = fan._device assert device.is_fan diff --git a/kasa/tests/smart/modules/test_light_effect.py b/kasa/tests/smart/modules/test_light_effect.py index ba1b22934..cc0eee8a9 100644 --- a/kasa/tests/smart/modules/test_light_effect.py +++ b/kasa/tests/smart/modules/test_light_effect.py @@ -1,12 +1,11 @@ from __future__ import annotations from itertools import chain -from typing import cast import pytest from pytest_mock import MockerFixture -from kasa import Device, Feature +from kasa import Device, Feature, Module from kasa.smart.modules import LightEffectModule from kasa.tests.device_fixtures import parametrize @@ -18,8 +17,8 @@ @light_effect async def test_light_effect(dev: Device, mocker: MockerFixture): """Test light effect.""" - light_effect = cast(LightEffectModule, dev.modules.get("LightEffectModule")) - assert light_effect + light_effect = dev.modules.get(Module.LightEffect) + assert isinstance(light_effect, LightEffectModule) feature = light_effect._module_features["light_effect"] assert feature.type == Feature.Type.Choice diff --git a/kasa/tests/test_common_modules.py b/kasa/tests/test_common_modules.py index af9dd475b..8f7def957 100644 --- a/kasa/tests/test_common_modules.py +++ b/kasa/tests/test_common_modules.py @@ -1,8 +1,7 @@ import pytest from pytest_mock import MockerFixture -from kasa import Device -from kasa.modules import LedModule, LightEffectModule +from kasa import Device, Module from kasa.tests.device_fixtures import ( lightstrip, parametrize, @@ -24,7 +23,7 @@ @led async def test_led_module(dev: Device, mocker: MockerFixture): """Test fan speed feature.""" - led_module = dev.get_module(LedModule) + led_module = dev.modules.get(Module.Led) assert led_module feat = led_module._module_features["led"] @@ -51,7 +50,7 @@ async def test_led_module(dev: Device, mocker: MockerFixture): @light_effect async def test_light_effect_module(dev: Device, mocker: MockerFixture): """Test fan speed feature.""" - light_effect_module = dev.get_module(LightEffectModule) + light_effect_module = dev.modules[Module.LightEffect] assert light_effect_module feat = light_effect_module._module_features["light_effect"] diff --git a/kasa/tests/test_iotdevice.py b/kasa/tests/test_iotdevice.py index b4d56291e..d5c76192b 100644 --- a/kasa/tests/test_iotdevice.py +++ b/kasa/tests/test_iotdevice.py @@ -16,7 +16,7 @@ Schema, ) -from kasa import KasaException +from kasa import KasaException, Module from kasa.iot import IotDevice from .conftest import get_device_for_fixture_protocol, handle_turn_on, turn_on @@ -261,27 +261,26 @@ async def test_modules_not_supported(dev: IotDevice): async def test_get_modules(): - """Test get_modules for child and parent modules.""" + """Test getting modules for child and parent modules.""" dummy_device = await get_device_for_fixture_protocol( "HS100(US)_2.0_1.5.6.json", "IOT" ) from kasa.iot.modules import Cloud - from kasa.smart.modules import CloudModule # Modules on device - module = dummy_device.get_module("Cloud") + module = dummy_device.modules.get("cloud") assert module assert module._device == dummy_device assert isinstance(module, Cloud) - module = dummy_device.get_module(Cloud) + module = dummy_device.modules.get(Module.IotCloud) assert module assert module._device == dummy_device assert isinstance(module, Cloud) # Invalid modules - module = dummy_device.get_module("DummyModule") + module = dummy_device.modules.get("DummyModule") assert module is None - module = dummy_device.get_module(CloudModule) + module = dummy_device.modules.get(Module.Cloud) assert module is None diff --git a/kasa/tests/test_smartdevice.py b/kasa/tests/test_smartdevice.py index bb2f81bf0..a0af2cb12 100644 --- a/kasa/tests/test_smartdevice.py +++ b/kasa/tests/test_smartdevice.py @@ -9,7 +9,7 @@ import pytest from pytest_mock import MockerFixture -from kasa import KasaException +from kasa import KasaException, Module from kasa.exceptions import SmartErrorCode from kasa.smart import SmartDevice @@ -123,40 +123,39 @@ async def test_update_module_queries(dev: SmartDevice, mocker: MockerFixture): async def test_get_modules(): - """Test get_modules for child and parent modules.""" + """Test getting modules for child and parent modules.""" dummy_device = await get_device_for_fixture_protocol( "KS240(US)_1.0_1.0.5.json", "SMART" ) - from kasa.iot.modules import AmbientLight - from kasa.smart.modules import CloudModule, FanModule + from kasa.smart.modules import CloudModule # Modules on device - module = dummy_device.get_module("CloudModule") + module = dummy_device.modules.get("CloudModule") assert module assert module._device == dummy_device assert isinstance(module, CloudModule) - module = dummy_device.get_module(CloudModule) + module = dummy_device.modules.get(Module.Cloud) assert module assert module._device == dummy_device assert isinstance(module, CloudModule) # Modules on child - module = dummy_device.get_module("FanModule") + module = dummy_device.modules.get("FanModule") assert module assert module._device != dummy_device assert module._device._parent == dummy_device - module = dummy_device.get_module(FanModule) + module = dummy_device.modules.get(Module.Fan) assert module assert module._device != dummy_device assert module._device._parent == dummy_device # Invalid modules - module = dummy_device.get_module("DummyModule") + module = dummy_device.modules.get("DummyModule") assert module is None - module = dummy_device.get_module(AmbientLight) + module = dummy_device.modules.get(Module.IotAmbientLight) assert module is None