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