feat(observe): add subscribe-and-wait mode selection

custom_components/localthings/observe.py

@@ -14,6 +14,8 @@ import logging
 import threading
 import time
 
+import cbor2
+
 from smartthings_local.ocf.state_cache import StateCache
 
 _LOGGER = logging.getLogger(__name__)
@@ -22,6 +24,8 @@ MODE_OBSERVE = 'observe'
 MODE_POLL = 'poll'
 
 DEFAULT_SETTLE_S = 4.0
+GRACE_PERIOD_S = 15.0
+SUCCESS_FRACTION = 0.8
 
 
 class ObserveManager:
@@ -38,6 +42,8 @@ class ObserveManager:
         self.last_mode_change_wall = time.time()
         self._settle_until: dict[str, float] = {}
         self._settle_lock = threading.Lock()
+        self.subscribed_hrefs: set[str] = set()
+        self._notified: set[str] = set()
 
     def mark_write_pending(self, href: str, settle_s: float = DEFAULT_SETTLE_S) -> None:
         with self._settle_lock:
@@ -59,3 +65,56 @@ class ObserveManager:
             self.log.debug("dropping %s update for %s (settling)", source, href)
             return False
         return self.cache.apply_rep(href, rep, source=source)
+
+    def on_notification(self, href: str, payload: bytes) -> None:
+        """Wired as DtlsCoapSession.on_notification. Runs on the DTLS
+        reader thread — must not touch asyncio."""
+        try:
+            rep = cbor2.loads(payload)
+        except Exception as e:
+            self.log.warning("observe %s: cbor decode failed: %s", href, e)
+            return
+        if not isinstance(rep, dict):
+            return
+        self._notified.add(href)
+        self.apply(href, rep, source='observe')
+
+    def try_enter_observe_mode(
+        self, session, hrefs: list[str],
+        grace_period_s: float = GRACE_PERIOD_S,
+        success_fraction: float = SUCCESS_FRACTION,
+    ) -> bool:
+        """Blocking — subscribes to every href then sleeps for the whole
+        grace period. Caller must run this in an executor, never on the
+        event loop."""
+        subscribed: set[str] = set()
+        for href in hrefs:
+            segs = [s for s in href.strip('/').split('/') if s]
+            try:
+                session.subscribe(segs)
+                subscribed.add(href)
+            except Exception as e:
+                self.log.warning("subscribe %s failed: %s", href, e)
+        if not subscribed:
+            self._set_mode(MODE_POLL)
+            self.subscribed_hrefs = set()
+            return False
+
+        time.sleep(grace_period_s)
+
+        fraction = len(self._notified & subscribed) / len(subscribed)
+        if fraction >= success_fraction:
+            self.subscribed_hrefs = subscribed
+            self._set_mode(MODE_OBSERVE)
+            return True
+
+        self.subscribed_hrefs = set()
+        self._set_mode(MODE_POLL)
+        return False
+
+    def _set_mode(self, mode: str) -> None:
+        if mode != self.mode:
+            self.log.info("observe-mode transition: %s -> %s", self.mode, mode)
+            self.mode = mode
+            self.last_mode_change_ts = time.monotonic()
+            self.last_mode_change_wall = time.time()

tests/localthings/test_observe.py

@@ -3,6 +3,9 @@ from __future__ import annotations
 
 import time
 
+import cbor2
+import pytest
+
 from smartthings_local.ocf.state_cache import StateCache
 
 from custom_components.localthings.observe import ObserveManager, MODE_POLL
@@ -48,3 +51,88 @@ def test_apply_accepts_update_after_settle_window_elapses():
 
     assert result is True
     assert mgr.cache.get('/oven/vs/0') == {'a': 2}
+
+
+class _FakeSession:
+    """Minimal stand-in for DtlsCoapSession.subscribe(), for observe tests.
+
+    `notify_after_subscribe` maps href -> cbor-encodable rep. Call
+    `subscribe()` records the href; the test then calls
+    `mgr.on_notification(href, payload)` itself to simulate delivery,
+    since real notify delivery is async/threaded in production.
+    """
+    def __init__(self):
+        self.subscribed: list[str] = []
+        self.fail_hrefs: set[str] = set()
+
+    def subscribe(self, path_segs):
+        href = '/' + '/'.join(path_segs)
+        if href in self.fail_hrefs:
+            raise ConnectionError("subscribe failed")
+        self.subscribed.append(href)
+        return b'\x01'
+
+
+def test_try_enter_observe_mode_succeeds_when_all_hrefs_notify():
+    mgr = _manager()
+    session = _FakeSession()
+    hrefs = ['/oven/vs/0', '/power/vs/0']
+
+    def _notify_all():
+        for href in hrefs:
+            mgr.on_notification(href, cbor2.dumps({'x': 1}))
+
+    # Grace period is real time.sleep() in the manager; keep it tiny and
+    # notify synchronously beforehand so notifies are already recorded
+    # when the (short) sleep completes.
+    _notify_all()
+    entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.01)
+
+    assert entered is True
+    assert mgr.mode == 'observe'
+    assert mgr.subscribed_hrefs == set(hrefs)
+
+
+def test_try_enter_observe_mode_falls_back_when_no_notifies_arrive():
+    mgr = _manager()
+    session = _FakeSession()
+    hrefs = ['/oven/vs/0', '/power/vs/0']
+
+    entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.01)
+
+    assert entered is False
+    assert mgr.mode == 'poll'
+    assert mgr.subscribed_hrefs == set()
+
+
+def test_try_enter_observe_mode_falls_back_when_subscribe_fails_for_all():
+    mgr = _manager()
+    session = _FakeSession()
+    session.fail_hrefs = {'/oven/vs/0', '/power/vs/0'}
+    hrefs = ['/oven/vs/0', '/power/vs/0']
+
+    entered = mgr.try_enter_observe_mode(session, hrefs, grace_period_s=0.01)
+
+    assert entered is False
+    assert mgr.mode == 'poll'
+
+
+def test_try_enter_observe_mode_meets_success_fraction_with_partial_notifies():
+    mgr = _manager()
+    session = _FakeSession()
+    hrefs = ['/a/vs/0', '/b/vs/0', '/c/vs/0', '/d/vs/0']
+    for href in hrefs[:3]:  # 3/4 = 0.75, below default 0.8 threshold
+        mgr.on_notification(href, cbor2.dumps({'x': 1}))
+
+    entered = mgr.try_enter_observe_mode(
+        session, hrefs, grace_period_s=0.01, success_fraction=0.7,
+    )
+
+    assert entered is True
+    assert mgr.mode == 'observe'
+
+
+def test_on_notification_ignores_malformed_cbor():
+    mgr = _manager()
+    mgr.on_notification('/oven/vs/0', b'\xff\xff\xff not cbor')
+    assert mgr.cache.get('/oven/vs/0') is None