Repository navigation
Expand file tree
/
Copy pathcache.py
More file actions
60 lines (39 loc) · 1.58 KB
/
Copy pathcache.py
File metadata and controls
60 lines (39 loc) · 1.58 KB
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
"""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 Protocol
from roborock.containers import CombinedMapInfo, HomeData, NetworkInfo
@dataclass
class CacheData:
"""Data structure for caching device information."""
home_data: HomeData | None = None
"""Home data containing device and product information."""
network_info: dict[str, NetworkInfo] = field(default_factory=dict)
"""Network information indexed by device DUID."""
home_cache: dict[int, CombinedMapInfo] = field(default_factory=dict)
"""Home cache information indexed by map_flag."""
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."""
...
class InMemoryCache(Cache):
"""In-memory cache implementation."""
def __init__(self):
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