8000 Use DeviceInfo consistently across devices by sdb9696 · Pull Request #1338 · python-kasa/python-kasa · GitHub
[go: up one dir, main page]

Skip to content

Use DeviceInfo consistently across devices #1338

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 8 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
10 changes: 8 additions & 2 deletions kasa/cli/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,14 @@ async def state(ctx, dev: Device):
echo(f"Device state: {dev.is_on}")

echo(f"Time: {dev.time} (tz: {dev.timezone})")
echo(f"Hardware: {dev.hw_info['hw_ver']}")
echo(f"Software: {dev.hw_info['sw_ver']}")
echo(
f"Hardware: {dev.device_info.hardware_version}"
f"{' (' + dev.region + ')' if dev.region else ''}"
)
echo(
f"Firmware: {dev.device_info.firmware_version}"
f" {dev.device_info.firmware_build}"
)
echo(f"MAC (rssi): {dev.mac} ({dev.rssi})")
if verbose:
echo(f"Location: {dev.location}")
Expand Down
22 changes: 17 additions & 5 deletions kasa/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
>>> dev.alias
Bedroom Lamp Plug
>>> dev.model
HS110(EU)
HS110
>>> dev.rssi
-71
>>> dev.mac
Expand Down Expand Up @@ -151,7 +151,7 @@ class WifiNetwork:


@dataclass
class _DeviceInfo:
class DeviceInfo:
"""Device Model Information."""

short_name: str
Expand Down Expand Up @@ -208,7 +208,7 @@ def __init__(
self.protocol: BaseProtocol = protocol or IotProtocol(
transport=XorTransport(config=config or DeviceConfig(host=host)),
)
self._last_update: Any = None
self._last_update: dict[str, Any] = {}
_LOGGER.debug("Initializing %s of type %s", host, type(self))
self._device_type = DeviceType.Unknown
# TODO: typing Any is just as using dict | None would require separate
Expand Down Expand Up @@ -334,9 +334,21 @@ def model(self) -> str:
"""Returns the device model."""

@property
def region(self) -> str | None:
"""Returns the device region."""
return self.device_info.region

@property
def device_info(self) -> DeviceInfo:
"""Return device info."""
return self._get_device_info(self._last_update, self._discovery_info)

@staticmethod
@abstractmethod
def _model_region(self) -> str:
"""Return device full model name and region."""
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> DeviceInfo:
"""Get device info."""

@property
@abstractmethod
Expand Down
12 changes: 6 additions & 6 deletions kasa/discover.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
>>>
>>> found_devices = await Discover.discover()
>>> [dev.model for dev in found_devices.values()]
['KP303(UK)', 'HS110(EU)', 'L530E', 'KL430(US)', 'HS220(US)']
['KP303', 'HS110', 'L530E', 'KL430', 'HS220']

You can pass username and password for devices requiring authentication

Expand Down Expand Up @@ -65,17 +65,17 @@
>>> print(f"Discovered {dev.alias} (model: {dev.model})")
>>>
>>> devices = await Discover.discover(on_discovered=print_dev_info, credentials=creds)
Discovered Bedroom Power Strip (model: KP303(UK))
Discovered Bedroom Lamp Plug (model: HS110(EU))
Discovered Bedroom Power Strip (model: KP303)
Discovered Bedroom Lamp Plug (model: HS110)
Discovered Living Room Bulb (model: L530)
Discovered Bedroom Lightstrip (model: KL430(US))
Discovered Living Room Dimmer Switch (model: HS220(US))
Discovered Bedroom Lightstrip (model: KL430)
Discovered Living Room Dimmer Switch (model: HS220)

Discovering a single device returns a kasa.Device object.

>>> device = await Discover.discover_single("127.0.0.1", credentials=creds)
>>> device.model
'KP303(UK)'
'KP303'

"""

Expand Down
33 changes: 15 additions & 18 deletions kasa/iot/iotdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
from typing import TYPE_CHECKING, Any, cast
from warnings import warn

from ..device import Device, WifiNetwork, _DeviceInfo
from ..device import Device, DeviceInfo, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import KasaException
Expand All @@ -43,7 +43,7 @@ def requires_update(f: Callable) -> Any:
@functools.wraps(f)
async def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if self._last_update is None and (
if not self._last_update and (
self._sys_info is None or f.__name__ not in self._sys_info
):
raise KasaException("You need to await update() to access the data")
Expand All @@ -54,7 +54,7 @@ async def wrapped(*args: Any, **kwargs: Any) -> Any:
@functools.wraps(f)
def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if self._last_update is None and (
if not self._last_update and (
self._sys_info is None or f.__name__ not in self._sys_info
):
raise KasaException("You need to await update() to access the data")
Expand Down Expand Up @@ -112,7 +112,7 @@ class IotDevice(Device):
>>> dev.alias
Bedroom Lamp Plug
>>> dev.model
HS110(EU)
HS110
>>> dev.rssi
-71
>>> dev.mac
Expand Down Expand Up @@ -310,7 +310,7 @@ async def update(self, update_children: bool = True) -> None:
# If this is the initial update, check only for the sysinfo
# This is necessary as some devices crash on unexpected modules
# See #105, #120, #161
if self._last_update is None:
if not self._last_update:
_LOGGER.debug("Performing the initial update to obtain sysinfo")
response = await self.protocol.query(req)
self._last_update = response
Expand Down Expand Up @@ -452,7 +452,9 @@ def update_from_discover_info(self, info: dict[str, Any]) -> None:
# This allows setting of some info properties directly
# from partial discovery info that will then be found
# by the requires_update decorator
self._set_sys_info(info)
discovery_model = info["device_model"]
no_region_model, _, _ = discovery_model.partition("(")
self._set_sys_info({**info, "model": no_region_model})

def _set_sys_info(self, sys_info: dict[str, Any]) -> None:
"""Set sys_info."""
Expand All @@ -471,18 +473,13 @@ class itself as @requires_update will be affected for other properties.
"""
return self._sys_info # type: ignore

@property # type: ignore
@requires_update
def model(self) -> str:
"""Return device model."""
sys_info = self._sys_info
return str(sys_info["model"])

@property
@requires_update
def _model_region(self) -> str:
"""Return device full model name and region."""
return self.model
def model(self) -> str:
"""Returns the device model."""
if self._last_update:
return self.device_info.short_name
return self._sys_info["model"]

@property # type: ignore
def alias(self) -> str | None:
Expand Down Expand Up @@ -748,7 +745,7 @@ def _get_device_type_from_sys_info(info: dict[str, Any]) -> DeviceType:
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
) -> DeviceInfo:
"""Get model information for a device."""
sys_info = _extract_sys_info(info)

Expand All @@ -766,7 +763,7 @@ def _get_device_info(
firmware_version, firmware_build = fw_version_full.split(" ", maxsplit=1)
auth = bool(discovery_info and ("mgt_encrypt_schm" in discovery_info))

return _DeviceInfo(
return DeviceInfo(
short_name=long_name,
long_name=long_name,
brand="kasa",
Expand Down
17 changes: 17 additions & 0 deletions kasa/smart/smartchilddevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import time
from typing import Any

from ..device import DeviceInfo
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
Expand Down Expand Up @@ -50,6 +51,22 @@ def __init__(
self._components_raw = component_info_raw
self._components = self._parse_components(self._components_raw)

@property
def device_info(self) -> DeviceInfo:
"""Return device info.

Child device does not have it info and components in _last_update so
this overrides the base implementation to call _get_device_info with
info and components combined as they would be in _last_update.
"""
return self._get_device_info(
{
"get_device_info": self._info,
"component_nego": self._components_raw,
},
None,
)

async def update(self, update_children: bool = True) -> None:
"""Update child module info.

Expand Down
24 changes: 9 additions & 15 deletions kasa/smart/smartdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
from datetime import UTC, datetime, timedelta, tzinfo
from typing import TYPE_CHECKING, Any, TypeAlias, cast

from ..device import Device, WifiNetwork, _DeviceInfo
from ..device import Device, DeviceInfo, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import AuthenticationError, DeviceError, KasaException, SmartErrorCode
Expand Down Expand Up @@ -69,7 +69,6 @@ def __init__(
self._modules: dict[str | ModuleName[Module], SmartModule] = {}
self._parent: SmartDevice | None = None
self._children: Mapping[str, SmartDevice] = {}
self._last_update = {}
self._last_update_time: float | None = None
self._on_since: datetime | None = None
self._info: dict[str, Any] = {}
Expand Down Expand Up @@ -497,18 +496,13 @@ def sys_info(self) -> dict[str, Any]:
@property
def model(self) -> str:
"""Returns the device model."""
return str(self._info.get("model"))
# If update hasn't been called self._device_info can't be used
if self._last_update:
return self.device_info.short_name

@property
def _model_region(self) -> str:
"""Return device full model name and region."""
if (disco := self._discovery_info) and (
disco_model := disco.get("device_model")
):
return disco_model
# Some devices have the region in the specs element.
region = f"({specs})" if (specs := self._info.get("specs")) else ""
return f"{self.model}{region}"
disco_model = str(self._info.get("device_model"))
long_name, _, _ = disco_model.partition("(")
return long_name

@property
def alias(self) -> str | None:
Expand Down Expand Up @@ -808,7 +802,7 @@ def _get_device_type_from_components(
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
) -> DeviceInfo:
"""Get model information for a device."""
di = info["get_device_info"]
components = [comp["id"] for comp in info["component_nego"]["component_list"]]
Expand Down Expand Up @@ -837,7 +831,7 @@ def _get_device_info(
# Brand inferred from SMART.KASAPLUG/SMART.TAPOPLUG etc.
brand = devicetype[:4].lower()

return _DeviceInfo(
return DeviceInfo(
short_name=short_name,
long_name=long_name,
brand=brand,
Expand Down
10 changes: 5 additions & 5 deletions kasa/smartcam/smartcamdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import logging
from typing import Any, cast

from ..device import _DeviceInfo
from ..device import DeviceInfo
from ..device_type import DeviceType
from ..module import Module
from ..protocols.smartcamprotocol import _ChildCameraProtocolWrapper
Expand Down Expand Up @@ -37,15 +37,15 @@ def _get_device_type_from_sysinfo(sysinfo: dict[str, Any]) -> DeviceType:
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
) -> DeviceInfo:
"""Get model information for a device."""
basic_info = info["getDeviceInfo"]["device_info"]["basic_info"]
short_name = basic_info["device_model"]
long_name = discovery_info["device_model"] if discovery_info else short_name
device_type = SmartCamDevice._get_device_type_from_sysinfo(basic_info)
fw_version_full = basic_info["sw_version"]
firmware_version, firmware_build = fw_version_full.split(" ", maxsplit=1)
return _DeviceInfo(
return DeviceInfo(
short_name=basic_info["device_model"],
long_name=long_name,
brand="tapo",
Expand Down Expand Up @@ -248,8 +248,8 @@ async def set_alias(self, alias: str) -> dict:
def hw_info(self) -> dict:
"""Return hardware info for the device."""
return {
"sw_ver": self._info.get("hw_ver"),
"hw_ver": self._info.get("fw_ver"),
"sw_ver": self._info.get("fw_ver"),
"hw_ver": self._info.get("hw_ver"),
"mac": self._info.get("mac"),
"type": self._info.get("type"),
"hwId": self._info.get("hwId"),
Expand Down
8 changes: 6 additions & 2 deletions tests/device_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -473,8 +473,12 @@ def get_nearest_fixture_to_ip(dev):
assert protocol_fixtures, "Unknown device type"

# This will get the best fixture with a match on model region
if model_region_fixtures := filter_fixtures(
"", model_filter={dev._model_region}, fixture_list=protocol_fixtures
if (di := dev.device_info) and (
model_region_fixtures := filter_fixtures(
"",
model_filter={di.long_name + (f"({di.region})" if di.region else "")},
fixture_list=protocol_fixtures,
)
):
return next(iter(model_region_fixtures))

Expand Down
4 changes: 2 additions & 2 deletions tests/iot/test_iotdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ async def test_invalid_connection(mocker, dev):
@has_emeter_iot
async def test_initial_update_emeter(dev, mocker):
"""Test that the initial update performs second query if emeter is available."""
dev._last_update = None
dev._last_update = {}
dev._legacy_features = set()
spy = mocker.spy(dev.protocol, "query")
await dev.update()
Expand All @@ -111,7 +111,7 @@ async def test_initial_update_emeter(dev, mocker):
@no_emeter_iot
async def test_initial_update_no_emeter(dev, mocker):
"""Test that the initial update performs second query if emeter is available."""
dev._last_update = None
dev._last_update = {}
dev._legacy_features = set()
spy = mocker.spy(dev.protocol, "query")
await dev.update()
Expand Down
7 changes: 5 additions & 2 deletions tests/smart/test_smartdevice.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,19 +62,22 @@ async def test_device_type_no_update(discovery_mock, caplog: pytest.LogCaptureFi
assert repr(dev) == f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - update() needed>"

discovery_result = copy.deepcopy(discovery_mock.discovery_data["result"])

disco_model = discovery_result["device_model"]
short_model, _, _ = disco_model.partition("(")
dev.update_from_discover_info(discovery_result)
assert dev.device_type is DeviceType.Unknown
assert (
repr(dev)
== f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - None (None) - update() needed>"
== f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - None ({short_model}) - update() needed>"
)
discovery_result["device_type"] = "SMART.FOOBAR"
dev.update_from_discover_info(discovery_result)
dev._components = {"dummy": 1}
assert dev.device_type is DeviceType.Plug
assert (
repr(dev)
== f"<DeviceType.Plug at {DISCOVERY_MOCK_IP} - None (None) - update() needed>"
== f"<DeviceType.Plug at {DISCOVERY_MOCK_IP} - None ({short_model}) - update() needed>"
)
assert "Unknown device type, falling back to plug" in caplog.text

Expand Down
Loading
Loading
0