roborock.devices.file_cache

This module implements a file-backed cache for device information.

This module provides a FileCache class that implements the Cache protocol to store and retrieve cached device information from a file on disk. This allows persistent caching of device data across application restarts.

 1"""This module implements a file-backed cache for device information.
 2
 3This module provides a `FileCache` class that implements the `Cache` protocol
 4to store and retrieve cached device information from a file on disk. This allows
 5persistent caching of device data across application restarts.
 6"""
 7
 8import asyncio
 9import pathlib
10import pickle
11from collections.abc import Callable
12from typing import Any
13
14from .cache import Cache, CacheData
15
16
17class FileCache(Cache):
18    """File backed cache implementation."""
19
20    def __init__(
21        self,
22        file_path: pathlib.Path,
23        init_fn: Callable[[], CacheData] = CacheData,
24        serialize_fn: Callable[[Any], bytes] = pickle.dumps,
25        deserialize_fn: Callable[[bytes], Any] = pickle.loads,
26    ) -> None:
27        """Initialize the file cache with the given file path."""
28        self._init_fn = init_fn
29        self._file_path = file_path
30        self._cache_data: CacheData | None = None
31        self._serialize_fn = serialize_fn
32        self._deserialize_fn = deserialize_fn
33
34    async def get(self) -> CacheData:
35        """Get cached value."""
36        if self._cache_data is not None:
37            return self._cache_data
38        data = await load_value(self._file_path, self._deserialize_fn)
39        if data is not None and not isinstance(data, CacheData):
40            raise TypeError(f"Invalid cache data loaded from {self._file_path}")
41
42        self._cache_data = data or self._init_fn()
43        return self._cache_data
44
45    async def set(self, value: CacheData) -> None:  # type: ignore[override]
46        """Set value in the cache."""
47        self._cache_data = value
48
49    async def flush(self) -> None:
50        """Flush the cache to disk."""
51        if self._cache_data is None:
52            return
53        await store_value(self._file_path, self._cache_data, self._serialize_fn)
54
55
56async def store_value(file_path: pathlib.Path, value: Any, serialize_fn: Callable[[Any], bytes] = pickle.dumps) -> None:
57    """Store a value to the given file path."""
58
59    def _store_to_disk(file_path: pathlib.Path, value: Any) -> None:
60        with open(file_path, "wb") as f:
61            data = serialize_fn(value)
62            f.write(data)
63
64    await asyncio.to_thread(_store_to_disk, file_path, value)
65
66
67async def load_value(file_path: pathlib.Path, deserialize_fn: Callable[[bytes], Any] = pickle.loads) -> Any | None:
68    """Load a value from the given file path."""
69
70    def _load_from_disk(file_path: pathlib.Path) -> Any | None:
71        if not file_path.exists():
72            return None
73        with open(file_path, "rb") as f:
74            data = f.read()
75            return deserialize_fn(data)
76
77    return await asyncio.to_thread(_load_from_disk, file_path)
class FileCache(roborock.devices.cache.Cache): 
18class FileCache(Cache):
19    """File backed cache implementation."""
20
21    def __init__(
22        self,
23        file_path: pathlib.Path,
24        init_fn: Callable[[], CacheData] = CacheData,
25        serialize_fn: Callable[[Any], bytes] = pickle.dumps,
26        deserialize_fn: Callable[[bytes], Any] = pickle.loads,
27    ) -> None:
28        """Initialize the file cache with the given file path."""
29        self._init_fn = init_fn
30        self._file_path = file_path
31        self._cache_data: CacheData | None = None
32        self._serialize_fn = serialize_fn
33        self._deserialize_fn = deserialize_fn
34
35    async def get(self) -> CacheData:
36        """Get cached value."""
37        if self._cache_data is not None:
38            return self._cache_data
39        data = await load_value(self._file_path, self._deserialize_fn)
40        if data is not None and not isinstance(data, CacheData):
41            raise TypeError(f"Invalid cache data loaded from {self._file_path}")
42
43        self._cache_data = data or self._init_fn()
44        return self._cache_data
45
46    async def set(self, value: CacheData) -> None:  # type: ignore[override]
47        """Set value in the cache."""
48        self._cache_data = value
49
50    async def flush(self) -> None:
51        """Flush the cache to disk."""
52        if self._cache_data is None:
53            return
54        await store_value(self._file_path, self._cache_data, self._serialize_fn)

File backed cache implementation.

FileCache( file_path: pathlib.Path, init_fn: Callable[[], roborock.devices.cache.CacheData] = <class 'roborock.devices.cache.CacheData'>, serialize_fn: Callable[[typing.Any], bytes] = <built-in function dumps>, deserialize_fn: Callable[[bytes], typing.Any] = <built-in function loads>) 
21    def __init__(
22        self,
23        file_path: pathlib.Path,
24        init_fn: Callable[[], CacheData] = CacheData,
25        serialize_fn: Callable[[Any], bytes] = pickle.dumps,
26        deserialize_fn: Callable[[bytes], Any] = pickle.loads,
27    ) -> None:
28        """Initialize the file cache with the given file path."""
29        self._init_fn = init_fn
30        self._file_path = file_path
31        self._cache_data: CacheData | None = None
32        self._serialize_fn = serialize_fn
33        self._deserialize_fn = deserialize_fn

Initialize the file cache with the given file path.

async def get(self) -> roborock.devices.cache.CacheData: 
35    async def get(self) -> CacheData:
36        """Get cached value."""
37        if self._cache_data is not None:
38            return self._cache_data
39        data = await load_value(self._file_path, self._deserialize_fn)
40        if data is not None and not isinstance(data, CacheData):
41            raise TypeError(f"Invalid cache data loaded from {self._file_path}")
42
43        self._cache_data = data or self._init_fn()
44        return self._cache_data

Get cached value.

async def set(self, value: roborock.devices.cache.CacheData) -> None: 
46    async def set(self, value: CacheData) -> None:  # type: ignore[override]
47        """Set value in the cache."""
48        self._cache_data = value

Set value in the cache.

async def flush(self) -> None: 
50    async def flush(self) -> None:
51        """Flush the cache to disk."""
52        if self._cache_data is None:
53            return
54        await store_value(self._file_path, self._cache_data, self._serialize_fn)

Flush the cache to disk.

async def store_value( file_path: pathlib.Path, value: Any, serialize_fn: Callable[[typing.Any], bytes] = <built-in function dumps>) -> None: 
57async def store_value(file_path: pathlib.Path, value: Any, serialize_fn: Callable[[Any], bytes] = pickle.dumps) -> None:
58    """Store a value to the given file path."""
59
60    def _store_to_disk(file_path: pathlib.Path, value: Any) -> None:
61        with open(file_path, "wb") as f:
62            data = serialize_fn(value)
63            f.write(data)
64
65    await asyncio.to_thread(_store_to_disk, file_path, value)

Store a value to the given file path.

async def load_value( file_path: pathlib.Path, deserialize_fn: Callable[[bytes], typing.Any] = <built-in function loads>) -> typing.Any | None: 
68async def load_value(file_path: pathlib.Path, deserialize_fn: Callable[[bytes], Any] = pickle.loads) -> Any | None:
69    """Load a value from the given file path."""
70
71    def _load_from_disk(file_path: pathlib.Path) -> Any | None:
72        if not file_path.exists():
73            return None
74        with open(file_path, "rb") as f:
75            data = f.read()
76            return deserialize_fn(data)
77
78    return await asyncio.to_thread(_load_from_disk, file_path)

Load a value from the given file path.