Cache miIO protocol sequence IDs across CLI invocations

Devices track message sequence IDs and ignore duplicates. Without
persisting the counter, restarting the CLI resets it to 0, causing
devices to silently drop messages and time out.

This generalizes the approach already used by the roborock vacuum CLI
into the shared DeviceGroup infrastructure, so all device types benefit.

Cache files are stored per-device (keyed by hashed IP) under the
platformdirs user cache directory.

Closes #1751

Author: Acrobot

miio/click_common.py

@@ -13,6 +13,7 @@
 
 import click
 
+from .device_cache import read_cache, write_cache
 from .exceptions import DeviceError
 
 try:
@@ -265,8 +266,21 @@ def group_callback(self, ctx, *args, **kwargs):
         gco = ctx.find_object(GlobalContextObject)
         if gco:
             kwargs["debug"] = gco.debug
+
+        ip = kwargs.get("ip")
+        if ip:
+            cached = read_cache(ip)
+            kwargs.setdefault("start_id", cached["seq"])
+
         ctx.obj = self.device_class(*args, **kwargs)
 
+        if ip:
+
+            def _save_cache() -> None:
+                write_cache(ip, {"seq": ctx.obj.raw_id})
+
+            ctx.call_on_close(_save_cache)
+
     def command_callback(self, miio_command, miio_device, *args, **kwargs):
         return miio_command.call(miio_device, *args, **kwargs)
 

miio/device_cache.py

@@ -0,0 +1,61 @@
+"""Cache for device connection state.
+
+Persists the miIO protocol message sequence counter between CLI invocations.
+Without this, restarting the CLI resets the counter to 0, and devices ignore
+messages with sequence IDs they've already seen, causing timeouts.
+"""
+
+import hashlib
+import json
+import logging
+from pathlib import Path
+from typing import TypedDict
+
+from platformdirs import user_cache_dir
+
+_LOGGER = logging.getLogger(__name__)
+
+CACHE_DIR = Path(user_cache_dir("python-miio"))
+
+
+class DeviceState(TypedDict):
+    """Cached state for a single device.
+
+    seq: The miIO protocol message sequence counter. Each message sent to a
+    device increments this counter, and the device tracks seen IDs to
+    deduplicate. Persisting it avoids ID reuse across CLI invocations.
+    """
+
+    seq: int
+
+
+def _cache_path(ip: str) -> Path:
+    """Return the cache file path for a device IP.
+
+    Uses a hash of the IP to avoid filesystem issues with IPv6 colons.
+    """
+    ip_hash = hashlib.sha256(ip.encode()).hexdigest()[:16]
+    return CACHE_DIR / f"{ip_hash}.json"
+
+
+def read_cache(ip: str) -> DeviceState:
+    """Read cached connection state for a device."""
+    path = _cache_path(ip)
+    try:
+        data = json.loads(path.read_text())
+        seq = int(data["seq"])
+        _LOGGER.debug("Loaded cache for %s: seq=%d", ip, seq)
+        return DeviceState(seq=seq)
+    except FileNotFoundError:
+        return DeviceState(seq=0)
+    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as ex:
+        _LOGGER.warning("Corrupt cache for %s, ignoring: %s", ip, ex)
+        return DeviceState(seq=0)
+
+
+def write_cache(ip: str, state: DeviceState) -> None:
+    """Write connection state to cache for a device."""
+    path = _cache_path(ip)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(state))
+    _LOGGER.debug("Wrote cache for %s: %s", ip, state)

miio/tests/test_device_cache.py

@@ -0,0 +1,81 @@
+import json
+from pathlib import Path
+
+import pytest
+
+from miio.device_cache import DeviceState, _cache_path, read_cache, write_cache
+
+
+@pytest.fixture()
+def cache_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Override CACHE_DIR to use a temporary directory."""
+    monkeypatch.setattr("miio.device_cache.CACHE_DIR", tmp_path)
+    return tmp_path
+
+
+class TestCachePath:
+    def test_ipv4_produces_valid_path(self, cache_dir: Path) -> None:
+        path = _cache_path("192.168.1.1")
+        assert path.parent == cache_dir
+        assert path.suffix == ".json"
+
+    def test_ipv6_produces_valid_path(self, cache_dir: Path) -> None:
+        path = _cache_path("fe80::1")
+        assert path.parent == cache_dir
+        assert path.suffix == ".json"
+        assert ":" not in path.name
+
+    def test_different_ips_get_different_paths(self, cache_dir: Path) -> None:
+        assert _cache_path("192.168.1.1") != _cache_path("192.168.1.2")
+
+    def test_same_ip_gets_same_path(self, cache_dir: Path) -> None:
+        assert _cache_path("192.168.1.1") == _cache_path("192.168.1.1")
+
+
+class TestReadCache:
+    def test_missing_file_returns_zero(self, cache_dir: Path) -> None:
+        state: DeviceState = read_cache("192.168.1.1")
+        assert state["seq"] == 0
+
+    def test_reads_written_data(self, cache_dir: Path) -> None:
+        write_cache("192.168.1.1", DeviceState(seq=42))
+        state: DeviceState = read_cache("192.168.1.1")
+        assert state["seq"] == 42
+
+    def test_corrupt_json_returns_zero(self, cache_dir: Path) -> None:
+        path = _cache_path("192.168.1.1")
+        path.write_text("not json")
+        state: DeviceState = read_cache("192.168.1.1")
+        assert state["seq"] == 0
+
+    def test_missing_seq_key_returns_zero(self, cache_dir: Path) -> None:
+        path = _cache_path("192.168.1.1")
+        path.write_text(json.dumps({"other": 123}))
+        state: DeviceState = read_cache("192.168.1.1")
+        assert state["seq"] == 0
+
+    def test_non_int_seq_returns_zero(self, cache_dir: Path) -> None:
+        path = _cache_path("192.168.1.1")
+        path.write_text(json.dumps({"seq": "not_a_number"}))
+        state: DeviceState = read_cache("192.168.1.1")
+        assert state["seq"] == 0
+
+
+class TestWriteCache:
+    def test_creates_directory(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        nested = tmp_path / "sub" / "dir"
+        monkeypatch.setattr("miio.device_cache.CACHE_DIR", nested)
+        write_cache("192.168.1.1", DeviceState(seq=5))
+        assert nested.exists()
+
+    def test_overwrites_existing(self, cache_dir: Path) -> None:
+        write_cache("192.168.1.1", DeviceState(seq=10))
+        write_cache("192.168.1.1", DeviceState(seq=20))
+        state: DeviceState = read_cache("192.168.1.1")
+        assert state["seq"] == 20
+
+    def test_written_file_is_valid_json(self, cache_dir: Path) -> None:
+        write_cache("192.168.1.1", DeviceState(seq=99))
+        path = _cache_path("192.168.1.1")
+        data: dict = json.loads(path.read_text())
+        assert data == {"seq": 99}