diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index d454bc2..fa8985e 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -3,18 +3,20 @@ from __future__ import annotations import asyncio import logging +import time from datetime import timedelta from typing import Any import cbor2 from homeassistant.config_entries import ConfigEntry -from homeassistant.core import HomeAssistant +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers import issue_registry as ir from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.device_registry import DeviceInfo from smartthings_local.protocol.dtls_session import DtlsCoapSession +from smartthings_local.ocf.state_cache import StateCache from .registry.batch import parse_device0_batch from .registry.by_type import for_device @@ -22,6 +24,7 @@ from .registry.discovery import discover, BoundEntity from .registry import CAPABILITIES from .registry.adapter import flatten from .registry.identity import read_identity, DeviceIdentity +from .observe import ObserveManager, MODE_OBSERVE, MODE_POLL from .const import ( DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, @@ -33,6 +36,17 @@ _LOGGER = logging.getLogger(__name__) _SEED_PATH = ['device', '0'] +class _NoOpDescriptor: + """StateCache requires a descriptor with an on_observation hook. This + integration doesn't use per-capability observation hooks, so this is a + deliberate no-op, not a placeholder for missing functionality.""" + def on_observation(self, state: dict, href: str, rep: dict) -> None: + return None + + +_RECOVERY_RETRY_S = 600.0 # re-attempt observe mode this often while polling + + class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): """Manages one Samsung appliance: session, discovery, polling.""" @@ -52,7 +66,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): self._identity: DeviceIdentity | None = None self._discovered = False self.bound = [] - self._last_resources: dict[str, dict] = {} + self._cache = StateCache(_NoOpDescriptor()) + self._cache.set_on_change(self._on_cache_changed) + self._observe = ObserveManager(self._cache, logger=_LOGGER) self.device_serial = entry.data[CONF_HOST] # placeholder until first poll self.device_info = DeviceInfo( identifiers={(DOMAIN, entry.data[CONF_HOST])}, @@ -73,7 +89,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): @property def last_resources(self) -> dict: - return self._last_resources + return self._cache.snapshot() + + @property + def observe_mode(self) -> str: + return self._observe.mode def _connect_session(self) -> None: host = self._entry.data[CONF_HOST] @@ -81,7 +101,8 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): cert_pem = self._entry.data[CONF_LEAF_CERT_PEM] key_pem = self._entry.data[CONF_LEAF_KEY_PEM] - sess = DtlsCoapSession(host, port, cert_pem=cert_pem, key_pem=key_pem) + sess = DtlsCoapSession(host, port, cert_pem=cert_pem, key_pem=key_pem, + on_notification=self._observe.on_notification) sess.connect() sess.start_reader() self._session = sess @@ -105,8 +126,23 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): if self._subpoll_task is not None: self._subpoll_task.cancel() self._subpoll_task = None + self._observe.close() await self.hass.async_add_executor_job(self._close_session) + def _on_cache_changed(self, changed: bool, source: str) -> None: + """StateCache.set_on_change callback. Runs on whatever thread + applied the update (DTLS reader thread for observe notifications, + an executor thread for poll/sweep) — never the event loop, so the + HA push must be scheduled thread-safely.""" + if not changed: + return + self.hass.add_job(self._push_cache_snapshot) + + @callback + def _push_cache_snapshot(self) -> None: + if self.bound: + self.async_set_updated_data(flatten(self.bound, self._cache.snapshot())) + def _poll_once(self) -> dict[str, dict]: """GET /device/0, return parsed resources. Blocking.""" if self._session is None: @@ -142,6 +178,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): if code == 0x45 and payload: rep = cbor2.loads(payload) if isinstance(rep, dict): + self._observe.apply(href, rep, source='poll') results[href] = rep except Exception as e: _LOGGER.debug("sub-poll %s: %s", href, e) @@ -152,13 +189,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): # ------------------------------------------------------------------ async def _run_subpolls(self) -> None: - """Poll hot/warm hrefs in the gaps between summary polls. - - Schedule over a SUMMARY_INTERVAL_S window: - hot hrefs every slot (10x, ~3 s apart) - warm hrefs every other slot (5x, ~6 s apart) - The lock prevents overlap with the summary poll or a concurrent sub-poll. - """ + """Poll hot/warm hrefs in the gaps between summary polls. Only + runs in poll-only mode — in observe-primary mode those hrefs are + already covered by push notifications.""" + if self._observe.mode == MODE_OBSERVE: + return hot = self._hot_hrefs warm = self._warm_hrefs if not hot and not warm: @@ -169,15 +204,11 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): hrefs = list(hot) + (list(warm) if i % 2 == 0 else []) async with self._session_lock: try: - updates = await self.hass.async_add_executor_job( + await self.hass.async_add_executor_job( self._poll_hrefs_blocking, hrefs ) except Exception as e: _LOGGER.debug("sub-poll batch failed: %s", e) - continue - if updates: - self._last_resources.update(updates) - self.async_set_updated_data(flatten(self.bound, self._last_resources)) # ------------------------------------------------------------------ # Discovery (runs once on first successful poll) @@ -262,6 +293,30 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): else: ir.async_delete_issue(self.hass, DOMAIN, issue_id) + async def _attempt_observe_mode(self) -> None: + """Called once, right after first discovery. Blocking (sleeps for + the whole grace period) — must run in an executor.""" + hrefs = self._hot_hrefs + self._warm_hrefs + if not hrefs: + return + if self._session is None: + # _poll_once already connects on a real poll; this only fires + # if the session was closed out from under us concurrently. + await self.hass.async_add_executor_job(self._connect_session) + sess = self._session + if sess is None: + return + await self.hass.async_add_executor_job( + self._observe.try_enter_observe_mode, sess, hrefs + ) + + async def _maybe_retry_observe_mode(self) -> None: + """While in poll-only mode, periodically re-attempt observe mode + so a device that gains internet access recovers push automatically.""" + if time.monotonic() - self._observe.last_mode_change_ts < _RECOVERY_RETRY_S: + return + await self._attempt_observe_mode() + # ------------------------------------------------------------------ # DataUpdateCoordinator hook # ------------------------------------------------------------------ @@ -285,23 +340,32 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): resources = await self.hass.async_add_executor_job(self._poll_once) except Exception as e2: _LOGGER.error("poll failed after reconnect: %s", e2) - if self._last_resources: + snapshot = self._cache.snapshot() + if snapshot: _LOGGER.debug("Full error:", exc_info=e2) - return flatten(self.bound, self._last_resources) + return flatten(self.bound, snapshot) raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2 - self._last_resources = resources + source = 'sweep' if self._discovered else 'poll' + if self._observe.mode == MODE_OBSERVE and self._observe.check_sweep_for_misses(resources): + self._observe.downgrade_to_poll() + for href, rep in resources.items(): + self._observe.apply(href, rep, source=source) if not self._discovered: self._run_discovery(resources) + await self._attempt_observe_mode() + elif self._observe.mode == MODE_POLL: + await self._maybe_retry_observe_mode() - # Schedule sub-polls for hot/warm hrefs between summary polls. + # Schedule sub-polls for hot/warm hrefs between summary polls + # (no-op in observe-primary mode; _run_subpolls checks the mode). if self._hot_hrefs or self._warm_hrefs: self._subpoll_task = self.hass.async_create_task( self._run_subpolls(), name="localthings_subpoll" ) - return flatten(self.bound, resources) + return flatten(self.bound, self._cache.snapshot()) # ------------------------------------------------------------------ # Command dispatch (called by entity platforms in Task 5) @@ -315,7 +379,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): if write_fn is None: return href = bound_entity.href - rep = self._last_resources.get(href or '', {}) + rep = self._cache.get(href or '') or {} result = write_fn(payload, rep, href) if result is None: _LOGGER.warning("write_fn rejected payload %r for %s", payload, href) @@ -326,6 +390,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): sess = self._session if sess is None: raise RuntimeError("no session") + self._observe.mark_write_pending(href) code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0) _LOGGER.info("PUT %s → code %#04x", href, code) diff --git a/tests/localthings/conftest.py b/tests/localthings/conftest.py index 295800e..c19fc1e 100644 --- a/tests/localthings/conftest.py +++ b/tests/localthings/conftest.py @@ -115,6 +115,55 @@ def mock_coordinator_session(fridge_resources): yield +class FakeObserveSession: + """Stand-in for DtlsCoapSession that supports subscribe()/on_notification + for coordinator-level observe tests, without a real DTLS connection.""" + + def __init__(self, on_notification=None): + self.on_notification = on_notification + self.subscribed: list[str] = [] + self.fail_hrefs: set[str] = set() + self.closed = False + + 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 refresh_observes(self, paths): + return None + + def close(self): + self.closed = True + + +@pytest.fixture +def mock_coordinator_observe_session(fridge_resources): + """Like mock_coordinator_session, but _connect_session installs a + FakeObserveSession on coordinator._session so subscribe()/on_notification + wiring can be exercised.""" + fake = FakeObserveSession() + + def _connect(self): + fake.on_notification = self._observe.on_notification + self._session = fake + + with ( + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._connect_session', + _connect, + ), + patch( + 'custom_components.localthings.coordinator.LocalThingsCoordinator._poll_once', + return_value=fridge_resources, + ), + patch('custom_components.localthings.coordinator.LocalThingsCoordinator._close_session'), + ): + yield fake + + @pytest.fixture def mock_entry(hass): """A MockConfigEntry added to hass, ready for async_setup.""" diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index fb273d6..5067a90 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -1,9 +1,12 @@ """Tests for the LocalThingsCoordinator.""" from __future__ import annotations +import threading +import time from datetime import timedelta from unittest.mock import patch +import cbor2 import pytest from homeassistant.core import HomeAssistant from homeassistant.helpers import issue_registry as ir @@ -12,6 +15,7 @@ from custom_components.localthings.const import ( DOMAIN, SUMMARY_INTERVAL_S, ) from custom_components.localthings.coordinator import LocalThingsCoordinator +from custom_components.localthings.observe import MODE_OBSERVE, MODE_POLL from .conftest import ENTRY_DATA, MOCK_SERIAL @@ -158,3 +162,89 @@ def test_update_coverage_gap_issue_clears_previous_issue( coordinator._update_coverage_gap_issue(False, [], 'Test Appliance') assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None + + +# --------------------------------------------------------------------------- +# StateCache / ObserveManager wiring +# --------------------------------------------------------------------------- + +async def test_last_resources_backed_by_state_cache( + hass: HomeAssistant, mock_entry, mock_coordinator_session +) -> None: + """coordinator.last_resources still returns a plain href->rep dict.""" + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + resources = coordinator.last_resources + assert isinstance(resources, dict) + assert all(isinstance(v, dict) for v in resources.values()) + assert resources # fridge fixture is non-empty + + +async def test_enters_poll_mode_when_no_notifies_arrive( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """No fake notifies are sent, so the grace period should time out + and the coordinator should stay in poll mode (unchanged behavior).""" + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + assert coordinator.observe_mode == MODE_POLL + + +async def test_enters_observe_mode_when_hot_warm_hrefs_notify( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + hrefs = coordinator._hot_hrefs + coordinator._warm_hrefs + + # try_enter_observe_mode clears prior notifications as soon as it + # starts, so notifies must land *during* its grace-period sleep (same + # pattern test_observe.py uses), not before the call. + def _notify_during_grace_period(): + time.sleep(0.005) + for href in hrefs: + fake.on_notification(href, cbor2.dumps({'notified': True})) + + notifier = threading.Thread(target=_notify_during_grace_period, daemon=True) + notifier.start() + entered = await hass.async_add_executor_job( + coordinator._observe.try_enter_observe_mode, + fake, hrefs, 0.02, 0.8, + ) + notifier.join() + + assert entered is True + assert coordinator._observe.mode == MODE_OBSERVE + + +async def test_write_marks_href_pending_before_post( + hass: HomeAssistant, mock_entry, mock_coordinator_observe_session +) -> None: + """async_send_command must call mark_write_pending before POSTing so a + slow-to-settle device can't immediately revert the optimistic write.""" + from custom_components.localthings.registry.discovery import BoundEntity + from custom_components.localthings.registry.entities import NumberDesc + + fake = mock_coordinator_observe_session + await hass.config_entries.async_setup(mock_entry.entry_id) + await hass.async_block_till_done() + coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id] + + def _write_fn(payload, rep, href): + return (['some', 'path'], {'value': payload}) + + desc = NumberDesc(key='test', field='value', write_fn=_write_fn) + bound = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc) + + with patch.object(fake, 'subscribe'): + fake.post = lambda *a, **k: (0x44, b'') + await coordinator.async_send_command(bound, 5) + + assert coordinator._observe._settle_until.get('/test/vs/0') is not None