roborock.devices.cache

  1"""This module provides caching functionality for the Roborock device management system.
  2
  3This module defines a cache interface that you may use to cache device
  4information to avoid unnecessary API calls. Callers may implement
  5this interface to provide their own caching mechanism.
  6"""
  7
  8from dataclasses import dataclass, field
  9from typing import Any, Protocol
 10
 11from roborock.data import CombinedMapInfo, HomeData, NetworkInfo, RoborockBase
 12from roborock.device_features import DeviceFeatures
 13
 14
 15@dataclass
 16class DeviceCacheData(RoborockBase):
 17    """Data structure for caching device information."""
 18
 19    network_info: NetworkInfo | None = None
 20    """Network information for the device"""
 21
 22    home_map_info: dict[int, CombinedMapInfo] | None = None
 23    """Home map information for the device by map_flag."""
 24
 25    home_map_content_base64: dict[int, str] | None = None
 26    """Home cache content for the device (encoded base64) by map_flag."""
 27
 28    device_features: DeviceFeatures | None = None
 29    """Device features information."""
 30
 31    trait_data: dict[str, Any] | None = None
 32    """Trait-specific cached data used internally for caching device features."""
 33
 34
 35@dataclass
 36class CacheData(RoborockBase):
 37    """Data structure for caching device information."""
 38
 39    home_data: HomeData | None = None
 40    """Home data containing device and product information."""
 41
 42    device_info: dict[str, DeviceCacheData] = field(default_factory=dict)
 43    """Per-device cached information indexed by device DUID."""
 44
 45    network_info: dict[str, NetworkInfo] = field(default_factory=dict)
 46    """Network information indexed by device DUID.
 47
 48    This is deprecated. Use the per-device `network_info` field instead.
 49    """
 50
 51    home_map_info: dict[int, CombinedMapInfo] = field(default_factory=dict)
 52    """Home map information indexed by map_flag.
 53
 54    This is deprecated. Use the per-device `home_map_info` field instead.
 55    """
 56
 57    home_map_content: dict[int, bytes] = field(default_factory=dict)
 58    """Home cache content for each map data indexed by map_flag.
 59
 60    This is deprecated. Use the per-device `home_map_content_base64` field instead.
 61    """
 62
 63    home_map_content_base64: dict[int, str] = field(default_factory=dict)
 64    """Home cache content for each map data (encoded base64) indexed by map_flag.
 65
 66    This is deprecated. Use the per-device `home_map_content_base64` field instead.
 67    """
 68
 69    device_features: DeviceFeatures | None = None
 70    """Device features information.
 71
 72    This is deprecated. Use the per-device `device_features` field instead.
 73    """
 74
 75    trait_data: dict[str, Any] | None = None
 76    """Trait-specific cached data used internally for caching device features.
 77
 78    This is deprecated. Use the per-device `trait_data` field instead.
 79    """
 80
 81
 82class Cache(Protocol):
 83    """Protocol for a cache that can store and retrieve values."""
 84
 85    async def get(self) -> CacheData:
 86        """Get cached value."""
 87        ...
 88
 89    async def set(self, value: CacheData) -> None:
 90        """Set value in the cache."""
 91        ...
 92
 93
 94@dataclass
 95class DeviceCache(RoborockBase):
 96    """Provides a cache interface for a specific device.
 97
 98    This is a convenience wrapper around a general Cache implementation to
 99    provide device-specific caching functionality.
100    """
101
102    def __init__(self, duid: str, cache: Cache) -> None:
103        """Initialize the device cache with the given cache implementation."""
104        self._duid = duid
105        self._cache = cache
106
107    async def get(self) -> DeviceCacheData:
108        """Get cached device-specific information."""
109        cache_data = await self._cache.get()
110        if self._duid not in cache_data.device_info:
111            cache_data.device_info[self._duid] = DeviceCacheData()
112            await self._cache.set(cache_data)
113        return cache_data.device_info[self._duid]
114
115    async def set(self, device_cache_data: DeviceCacheData) -> None:
116        """Set cached device-specific information."""
117        cache_data = await self._cache.get()
118        cache_data.device_info[self._duid] = device_cache_data
119        await self._cache.set(cache_data)
120
121
122class InMemoryCache(Cache):
123    """In-memory cache implementation."""
124
125    def __init__(self) -> None:
126        """Initialize the in-memory cache."""
127        self._data = CacheData()
128
129    async def get(self) -> CacheData:
130        return self._data
131
132    async def set(self, value: CacheData) -> None:
133        self._data = value
134
135
136class NoCache(Cache):
137    """No-op cache implementation."""
138
139    async def get(self) -> CacheData:
140        return CacheData()
141
142    async def set(self, value: CacheData) -> None:
143        pass