1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
|
"""Trait for device network information."""
from __future__ import annotations
import logging
from roborock.data import NetworkInfo
from roborock.devices.cache import DeviceCache
from roborock.devices.traits.v1 import common
from roborock.roborock_typing import RoborockCommand
_LOGGER = logging.getLogger(__name__)
class NetworkInfoTrait(NetworkInfo, common.V1TraitMixin):
"""Trait for device network information.
This trait will always prefer reading from the cache if available. This
information is usually already fetched when creating the device local
connection, so reading from the cache avoids an unnecessary RPC call.
However, we have the fallback to reading from the device if the cache is
not populated for some reason.
"""
command = RoborockCommand.GET_NETWORK_INFO
def __init__(self, device_uid: str, device_cache: DeviceCache) -> None: # pylint: disable=super-init-not-called
"""Initialize the trait."""
self._device_uid = device_uid
self._device_cache = device_cache
self.ip = ""
async def refresh(self) -> None:
"""Refresh the network info from the cache."""
device_cache_data = await self._device_cache.get()
if device_cache_data.network_info:
_LOGGER.debug("Using cached network info for device %s", self._device_uid)
self._update_trait_values(device_cache_data.network_info)
return
# Load from device if not in cache
_LOGGER.debug("No cached network info for device %s, fetching from device", self._device_uid)
await super().refresh()
# Update the cache with the new network info
device_cache_data = await self._device_cache.get()
device_cache_data.network_info = self
await self._device_cache.set(device_cache_data)
def _parse_response(self, response: common.V1ResponseData) -> NetworkInfo:
"""Parse the response from the device into a NetworkInfo."""
if not isinstance(response, dict):
raise ValueError(f"Unexpected NetworkInfoTrait response format: {response!r}")
return NetworkInfo.from_dict(response)
|
