From b2809778693c15654589c4de518946b2e86df68b Mon Sep 17 00:00:00 2001
From: chenwei <chenwei@bany.tech>
Date: Thu, 30 Jul 2026 15:53:50 +0000
Subject: [PATCH] =?UTF-8?q?feat(mcp):=20DeviceManager=E2=80=94=E2=80=94?=
 =?UTF-8?q?=E6=8C=89=E9=9C=80=E8=BF=9E=E6=8E=A5+=E7=A9=BA=E9=97=B2?=
 =?UTF-8?q?=E4=BF=9D=E6=8C=81+=E4=B8=B2=E8=A1=8C=E9=94=81+=E6=8E=89?=
 =?UTF-8?q?=E7=BA=BF=E9=87=8D=E8=BF=9E=EF=BC=88=E6=96=B9=E6=A1=88=20C?=
 =?UTF-8?q?=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 src/ppclock/device_manager.py | 113 +++++++++++++++++++++++++
 tests/test_device_manager.py  | 154 ++++++++++++++++++++++++++++++++++
 2 files changed, 267 insertions(+)
 create mode 100644 src/ppclock/device_manager.py
 create mode 100644 tests/test_device_manager.py

diff --git a/src/ppclock/device_manager.py b/src/ppclock/device_manager.py
new file mode 100644
index 0000000..a161270
--- /dev/null
+++ b/src/ppclock/device_manager.py
@@ -0,0 +1,113 @@
+"""设备连接管理器（方案 C：按需连接 + 空闲保持，全操作串行）。
+
+- 一把 asyncio.Lock 串行所有设备操作（BLE 适配器独占）
+- 首次操作守候连接（connect_timeout 应覆盖设备 30-60s 广播窗口）
+- 操作后保持连接；空闲 idle_timeout 秒自动断开让设备睡眠
+- 操作中 TransportError（连接丢失）→ 重连并重试一次，再失败向上抛
+"""
+from __future__ import annotations
+
+import asyncio
+import time
+
+from .client import PPClient
+from .transports.base import TransportError
+
+
+class DeviceManager:
+    def __init__(self, mac: str | None = None, connect_timeout: float = 90.0,
+                 idle_timeout: float = 300.0, client_factory=None):
+        self._mac = mac
+        self._connect_timeout = connect_timeout
+        self._idle_timeout = idle_timeout
+        self._factory = client_factory or (
+            lambda mac, timeout: PPClient.via_local(mac=mac, timeout=timeout))
+        self._client = None
+        self._lock = asyncio.Lock()
+        self._last_activity: float | None = None
+        self._idle_task: asyncio.Task | None = None
+        self._connect_count = 0
+
+    @property
+    def connected(self) -> bool:
+        return self._client is not None
+
+    async def run(self, op):
+        """串行执行 op(client)；掉线重连重试一次。"""
+        async with self._lock:
+            try:
+                return await self._run_with_retry(op)
+            finally:
+                self._touch()
+
+    async def status(self) -> dict:
+        async with self._lock:
+            await self._ensure_connected()
+            self._touch()
+            info = {
+                "connected": self.connected,
+                "mac": self._mac,
+                "idle_seconds": None if self._last_activity is None
+                else round(time.monotonic() - self._last_activity, 1),
+                "connect_count": self._connect_count,
+                "connect_timeout": self._connect_timeout,
+                "idle_timeout": self._idle_timeout,
+            }
+            try:
+                info["device_id"] = await self._client.get_device_id()
+            except Exception:  # noqa: BLE001 - 状态查询尽力而为
+                pass
+            return info
+
+    async def close(self) -> None:
+        if self._idle_task is not None:
+            self._idle_task.cancel()
+        async with self._lock:
+            await self._drop()
+
+    # ---------- 内部 ----------
+
+    async def _run_with_retry(self, op):
+        await self._ensure_connected()
+        try:
+            return await op(self._client)
+        except TransportError:
+            await self._drop()
+            await self._ensure_connected()
+            return await op(self._client)
+
+    async def _ensure_connected(self):
+        if self._client is not None:
+            return
+        client = self._factory(self._mac, self._connect_timeout)
+        await client.__aenter__()
+        self._client = client
+        self._connect_count += 1
+        if self._mac is None:  # 自动扫描后记住实际地址
+            self._mac = getattr(client._t, "address", None)
+
+    async def _drop(self):
+        client, self._client = self._client, None
+        if client is not None:
+            try:
+                await client.__aexit__(None, None, None)
+            except Exception:  # noqa: BLE001 - 断开失败不影响状态
+                pass
+
+    def _touch(self):
+        self._last_activity = time.monotonic()
+        if self._idle_task is not None:
+            self._idle_task.cancel()
+            self._idle_task = None
+        if self.connected:
+            self._idle_task = asyncio.create_task(self._idle_watch())
+
+    async def _idle_watch(self):
+        try:
+            await asyncio.sleep(self._idle_timeout)
+            async with self._lock:
+                if (self._last_activity is not None
+                        and time.monotonic() - self._last_activity >= self._idle_timeout):
+                    await self._drop()
+        except asyncio.CancelledError:
+            pass
diff --git a/tests/test_device_manager.py b/tests/test_device_manager.py
new file mode 100644
index 0000000..66fbc2a
--- /dev/null
+++ b/tests/test_device_manager.py
@@ -0,0 +1,154 @@
+"""DeviceManager 测试：按需连接/串行/空闲超时/掉线重连。"""
+import asyncio
+
+import pytest
+
+from ppclock.client import PPClient
+from ppclock.device_manager import DeviceManager
+from ppclock.transports.base import TransportError
+
+
+class FakeTransport:
+    def __init__(self):
+        self.address = "AA:BB:CC:DD:EE:FF"
+        self.enters = 0
+        self.exits = 0
+        self.rxtx_writes = []
+        self.fail_next_write = False
+
+    async def __aenter__(self):
+        self.enters += 1
+        return self
+
+    async def __aexit__(self, *exc):
+        self.exits += 1
+        return False
+
+    async def write_epd(self, data, response=True):
+        pass
+
+    async def write_rxtx(self, data, response=True):
+        if self.fail_next_write:
+            self.fail_next_write = False
+            raise TransportError("connection lost")
+        self.rxtx_writes.append(data)
+
+    async def read_rxtx(self):
+        return b"\x00"
+
+    async def request_device_id(self, timeout=8.0):
+        return "81233F3C267112"
+
+    async def run_ota(self, image, on_progress=None):
+        return {}
+
+    async def delay(self, seconds):
+        pass
+
+
+def make_manager(transports, **kw):
+    """client_factory 依次弹出预置 transport 包成 PPClient。"""
+    def factory(mac, timeout):
+        return PPClient(transports.pop(0))
+    kw.setdefault("client_factory", factory)
+    return DeviceManager(**kw)
+
+
+class TestConnect:
+    @pytest.mark.asyncio
+    async def test_connect_on_demand_and_reuse(self):
+        t = FakeTransport()
+        m = make_manager([t])
+        assert not m.connected
+        await m.run(lambda dev: dev.set_mode("clock1"))
+        await m.run(lambda dev: dev.set_mode("clock2"))
+        assert t.enters == 1  # 第二次复用连接
+        assert m.connected
+        await m.close()
+
+    @pytest.mark.asyncio
+    async def test_remembers_address(self):
+        t = FakeTransport()
+        m = make_manager([t], mac=None)
+        await m.run(lambda dev: dev.set_mode("clock1"))
+        assert (await m.status())["mac"] == "AA:BB:CC:DD:EE:FF"
+        await m.close()
+
+
+class TestSerialization:
+    @pytest.mark.asyncio
+    async def test_ops_are_serialized(self):
+        t = FakeTransport()
+        m = make_manager([t])
+        concurrent = 0
+        peak = 0
+
+        async def op(dev):
+            nonlocal concurrent, peak
+            concurrent += 1
+            peak = max(peak, concurrent)
+            await asyncio.sleep(0.02)
+            concurrent -= 1
+            return 1
+
+        await asyncio.gather(*(m.run(op) for _ in range(5)))
+        assert peak == 1
+        await m.close()
+
+
+class TestIdleTimeout:
+    @pytest.mark.asyncio
+    async def test_idle_disconnect(self):
+        t = FakeTransport()
+        m = make_manager([t], idle_timeout=0.05)
+        await m.run(lambda dev: dev.set_mode("clock1"))
+        assert m.connected
+        await asyncio.sleep(0.15)
+        assert not m.connected
+        assert t.exits == 1
+
+    @pytest.mark.asyncio
+    async def test_activity_resets_idle(self):
+        t = FakeTransport()
+        m = make_manager([t], idle_timeout=0.1)
+        await m.run(lambda dev: dev.set_mode("clock1"))
+        await asyncio.sleep(0.06)
+        await m.run(lambda dev: dev.set_mode("clock2"))
+        await asyncio.sleep(0.06)
+        assert m.connected  # 第二次活动续期
+        await m.close()
+
+
+class TestReconnect:
+    @pytest.mark.asyncio
+    async def test_reconnect_once_on_transport_error(self):
+        t1, t2 = FakeTransport(), FakeTransport()
+        t1.fail_next_write = True
+        m = make_manager([t1, t2])
+        await m.run(lambda dev: dev.set_mode("clock1"))
+        assert t1.enters == 1 and t2.enters == 1  # 掉线重连一次并成功
+        await m.close()
+
+    @pytest.mark.asyncio
+    async def test_retry_exhaustion_raises(self):
+        t1, t2 = FakeTransport(), FakeTransport()
+        t1.fail_next_write = True
+        t2.fail_next_write = True
+        m = make_manager([t1, t2])
+        with pytest.raises(TransportError):
+            await m.run(lambda dev: dev.set_mode("clock1"))
+        await m.close()
+
+
+class TestStatus:
+    @pytest.mark.asyncio
+    async def test_status_fields(self):
+        t = FakeTransport()
+        m = make_manager([t], connect_timeout=90.0, idle_timeout=300.0)
+        s = await m.status()
+        assert s["connected"] is True  # status 自身触发连接
+        assert s["device_id"] == "81233F3C267112"
+        assert s["connect_timeout"] == 90.0
+        assert s["idle_timeout"] == 300.0
+        assert isinstance(s["idle_seconds"], float)
+        await m.close()