roborock.devices.traits.v1.network_info

Trait for device network information.

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

Converter for NetworkInfo objects.

def convert( self, response: dict | list | int | str) -> roborock.data.v1.v1_containers.NetworkInfo:
17    def convert(self, response: common.V1ResponseData) -> NetworkInfo:
18        """Parse the response from the device into a NetworkInfoConverter instance."""
19        if not isinstance(response, dict):
20            raise ValueError(f"Unexpected NetworkInfoTrait response format: {response!r}")
21        return NetworkInfo.from_dict(response)

Parse the response from the device into a NetworkInfoConverter instance.

24class NetworkInfoTrait(NetworkInfo, common.V1TraitMixin):
25    """Trait for device network information.
26
27    This trait will always prefer reading from the cache if available. This
28    information is usually already fetched when creating the device local
29    connection, so reading from the cache avoids an unnecessary RPC call.
30    However, we have the fallback to reading from the device if the cache is
31    not populated for some reason.
32    """
33
34    command = RoborockCommand.GET_NETWORK_INFO
35    converter = NetworkInfoConverter()
36
37    def __init__(self, device_uid: str, device_cache: DeviceCache) -> None:  # pylint: disable=super-init-not-called
38        """Initialize the trait."""
39        self._device_uid = device_uid
40        self._device_cache = device_cache
41        self.ip = ""
42
43    async def refresh(self) -> None:
44        """Refresh the network info from the cache."""
45
46        device_cache_data = await self._device_cache.get()
47        if device_cache_data.network_info:
48            _LOGGER.debug("Using cached network info for device %s", self._device_uid)
49            common.merge_trait_values(self, device_cache_data.network_info)
50            return
51
52        # Load from device if not in cache
53        _LOGGER.debug("No cached network info for device %s, fetching from device", self._device_uid)
54        await super().refresh()
55
56        # Update the cache with the new network info
57        device_cache_data = await self._device_cache.get()
58        device_cache_data.network_info = self
59        await self._device_cache.set(device_cache_data)

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.

NetworkInfoTrait(device_uid: str, device_cache: roborock.devices.cache.DeviceCache)
37    def __init__(self, device_uid: str, device_cache: DeviceCache) -> None:  # pylint: disable=super-init-not-called
38        """Initialize the trait."""
39        self._device_uid = device_uid
40        self._device_cache = device_cache
41        self.ip = ""

Initialize the trait.

command = <RoborockCommand.GET_NETWORK_INFO: 'get_network_info'>

The RoborockCommand used to fetch the trait data from the device (internal only).

converter = NetworkInfoConverter

The converter used to parse the response from the device (internal only).

ip
async def refresh(self) -> None:
43    async def refresh(self) -> None:
44        """Refresh the network info from the cache."""
45
46        device_cache_data = await self._device_cache.get()
47        if device_cache_data.network_info:
48            _LOGGER.debug("Using cached network info for device %s", self._device_uid)
49            common.merge_trait_values(self, device_cache_data.network_info)
50            return
51
52        # Load from device if not in cache
53        _LOGGER.debug("No cached network info for device %s, fetching from device", self._device_uid)
54        await super().refresh()
55
56        # Update the cache with the new network info
57        device_cache_data = await self._device_cache.get()
58        device_cache_data.network_info = self
59        await self._device_cache.set(device_cache_data)

Refresh the network info from the cache.