diff --git a/custom_components/localthings/observe.py b/custom_components/localthings/observe.py index bed99b1..e82800f 100644 --- a/custom_components/localthings/observe.py +++ b/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() diff --git a/tests/localthings/test_observe.py b/tests/localthings/test_observe.py index 102020e..f9cafb2 100644 --- a/tests/localthings/test_observe.py +++ b/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