diff --git a/custom_components/localthings/const.py b/custom_components/localthings/const.py index 0fa612c..dbe10ca 100644 --- a/custom_components/localthings/const.py +++ b/custom_components/localthings/const.py @@ -11,5 +11,4 @@ CONF_LEAF_KEY_PEM = "leaf_key_pem" PROBE_PORTS = [49154, 49155] -ACTIVE_INTERVAL_S = 5.0 -IDLE_INTERVAL_S = 30.0 +SUMMARY_INTERVAL_S = 30.0 diff --git a/custom_components/localthings/coordinator.py b/custom_components/localthings/coordinator.py index aa28157..d1b6576 100644 --- a/custom_components/localthings/coordinator.py +++ b/custom_components/localthings/coordinator.py @@ -18,12 +18,12 @@ from .ocf.coap_dtls import DtlsCoapSession from .ocf.registry.by_type import for_device from .ocf.registry.discovery import discover, BoundEntity from .ocf.registry import CAPABILITIES -from .ocf.registry.adapter import flatten, is_active +from .ocf.registry.adapter import flatten from .ocf.registry.identity import read_identity, DeviceIdentity from .const import ( DOMAIN, CONF_HOST, CONF_PORT, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, - ACTIVE_INTERVAL_S, IDLE_INTERVAL_S, + SUMMARY_INTERVAL_S, ) _LOGGER = logging.getLogger(__name__) @@ -43,7 +43,7 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): hass, _LOGGER, name=f"{DOMAIN}_{entry.data[CONF_HOST]}", - update_interval=timedelta(seconds=IDLE_INTERVAL_S), + update_interval=timedelta(seconds=SUMMARY_INTERVAL_S), ) self._entry = entry self._session: DtlsCoapSession | None = None @@ -57,6 +57,10 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): name=f"Samsung Appliance ({entry.data[CONF_HOST]})", manufacturer="Samsung", ) + self._session_lock = asyncio.Lock() + self._subpoll_task: asyncio.Task | None = None + self._hot_hrefs: list[str] = [] + self._warm_hrefs: list[str] = [] # ------------------------------------------------------------------ # Session management (all blocking — must run in executor) @@ -93,6 +97,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): pass async def async_close(self) -> None: + if self._subpoll_task is not None: + self._subpoll_task.cancel() + self._subpoll_task = None await self.hass.async_add_executor_job(self._close_session) def _poll_once(self) -> dict[str, dict]: @@ -114,6 +121,59 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): raise RuntimeError(f"poll cbor decode: {e}") from e return parse_device0_batch(body) if isinstance(body, list) else {} + def _poll_hrefs_blocking(self, hrefs: list[str]) -> dict[str, dict]: + """GET individual hrefs sequentially. Does not reconnect on failure. Blocking.""" + if self._session is None: + return {} + results = {} + first = True + for href in hrefs: + if not first: + self._session.pace() + first = False + try: + path = [s for s in href.strip('/').split('/')] + code, payload = self._session.get(path, timeout=10.0) + if code == 0x45 and payload: + rep = cbor2.loads(payload) + if isinstance(rep, dict): + results[href] = rep + except Exception as e: + _LOGGER.debug("sub-poll %s: %s", href, e) + return results + + # ------------------------------------------------------------------ + # Sub-poll loop (runs between summary polls) + # ------------------------------------------------------------------ + + 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. + """ + hot = self._hot_hrefs + warm = self._warm_hrefs + if not hot and not warm: + return + step = SUMMARY_INTERVAL_S / 10 # 3.0 s + for i in range(1, 10): # slots 1..9 (T+3 s … T+27 s) + await asyncio.sleep(step) + hrefs = list(hot) + (list(warm) if i % 2 == 0 else []) + async with self._session_lock: + try: + updates = 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) # ------------------------------------------------------------------ @@ -148,37 +208,56 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]): manufacturer=mfr, model=model, ) + hot, warm = set(), set() + for be in bound: + tier = be.capability.poll_tier + if tier == 'hot': + hot.add(be.href) + elif tier == 'warm': + warm.add(be.href) + self._hot_hrefs = sorted(hot) + self._warm_hrefs = sorted(warm) + self._discovered = True - _LOGGER.info("discovered %d entities (serial=%s)", len(bound), serial) + _LOGGER.info( + "discovered %d entities (serial=%s) hot=%s warm=%s", + len(bound), serial, self._hot_hrefs, self._warm_hrefs, + ) # ------------------------------------------------------------------ # DataUpdateCoordinator hook # ------------------------------------------------------------------ async def _async_update_data(self) -> dict[str, Any]: - try: - resources = await self.hass.async_add_executor_job(self._poll_once) - except Exception as e: - # One reconnect attempt — pause briefly so the device can - # clean up its DTLS session state before we knock again. - _LOGGER.warning("poll failed, reconnecting: %s", e) - await self.hass.async_add_executor_job(self._close_session) - await asyncio.sleep(5.0) + # Stop any in-flight sub-poll before taking the session lock. + if self._subpoll_task is not None: + self._subpoll_task.cancel() + self._subpoll_task = None + + async with self._session_lock: try: resources = await self.hass.async_add_executor_job(self._poll_once) - except Exception as e2: - raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2 + except Exception as e: + # One reconnect attempt — pause briefly so the device can + # clean up its DTLS session state before we knock again. + _LOGGER.warning("poll failed, reconnecting: %s", e) + await self.hass.async_add_executor_job(self._close_session) + await asyncio.sleep(5.0) + try: + resources = await self.hass.async_add_executor_job(self._poll_once) + except Exception as e2: + raise UpdateFailed(f"poll failed after reconnect: {e2}") from e2 self._last_resources = resources if not self._discovered: self._run_discovery(resources) - # Adjust polling interval dynamically - active = is_active(self.bound, resources) - self.update_interval = timedelta( - seconds=ACTIVE_INTERVAL_S if active else IDLE_INTERVAL_S - ) + # Schedule sub-polls for hot/warm hrefs between summary polls. + 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) diff --git a/custom_components/localthings/ocf/coap_dtls.py b/custom_components/localthings/ocf/coap_dtls.py index f19b4f8..85b73bc 100644 --- a/custom_components/localthings/ocf/coap_dtls.py +++ b/custom_components/localthings/ocf/coap_dtls.py @@ -102,6 +102,13 @@ BLOCK_SZX = 6 _BLOCK_MAX_ATTEMPTS = 3 _BLOCK_ACK_TIMEOUT = 4.0 +# Inter-request pacing: minimum seconds between CoAP CON sends on one session. +# Samsung's RT-OCF stacks drop requests when hit faster than their firmware +# ceiling (dryer ~14 req/s, oven ~8 req/s, dishwasher unknown). 5 req/s +# (200 ms) is conservative enough for all tested devices; tune per device +# once the ceiling is measured empirically. +_DEFAULT_RATE_LIMIT_RPS = 5.0 + def _vlen(v): """Variable-length integer encoder used in option deltas + lengths.""" @@ -222,13 +229,15 @@ class DtlsCoapSession: MAX_BLOCKS = 32 # safety bound for Block2 fetches def __init__(self, host, port, cert_pem, key_pem, - on_notification=None, mtu=1200): + on_notification=None, mtu=1200, + rate_limit_rps: float = _DEFAULT_RATE_LIMIT_RPS): self.host = host self.port = port self.cert_pem = cert_pem self.key_pem = key_pem self.on_notification = on_notification # fn(href, payload_bytes) self.mtu = mtu + self._min_req_interval = 1.0 / rate_limit_rps self.sock = None self.conn = None @@ -254,6 +263,10 @@ class DtlsCoapSession: self._stop = threading.Event() self._reader_thread = None + def pace(self) -> None: + """Sleep one rate-limit interval. Uses _stop so session teardown wakes it.""" + self._stop.wait(self._min_req_interval) + # ---- lifecycle --------------------------------------------------- def connect(self): @@ -566,8 +579,8 @@ class DtlsCoapSession: raise ConnectionError("DTLS session closed") tok = self._next_tok() logger.debug( - "GET /%s token=%s timeout=%.1fs", - '/'.join(path_segs), tok.hex(), timeout, + "GET %s /%s token=%s timeout=%.1fs", + self.host, '/'.join(path_segs), tok.hex(), timeout, ) blob = b'' num = 0 @@ -576,6 +589,8 @@ class DtlsCoapSession: last_opts = [] deadline = time.time() + timeout while True: + if num > 0: + self.pace() container = {} for attempt in range(_BLOCK_MAX_ATTEMPTS): ev = threading.Event() @@ -598,14 +613,14 @@ class DtlsCoapSession: remaining = deadline - time.time() if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1: logger.debug( - "GET /%s block %d: timed out after %d attempt(s)", - '/'.join(path_segs), num, attempt + 1, + "GET %s /%s block %d: timed out after %d attempt(s)", + self.host, '/'.join(path_segs), num, attempt + 1, ) raise TimeoutError( f"GET /{'/'.join(path_segs)} block {num} timeout") logger.debug( - "GET /%s block %d: attempt %d/%d timeout, retrying", - '/'.join(path_segs), num, + "GET %s /%s block %d: attempt %d/%d timeout, retrying", + self.host, '/'.join(path_segs), num, attempt + 1, _BLOCK_MAX_ATTEMPTS, ) finally: @@ -620,8 +635,8 @@ class DtlsCoapSession: ) else: logger.debug( - "GET /%s block %d: %s (%d bytes)", - '/'.join(path_segs), num, + "GET %s /%s block %d: %s (%d bytes)", + self.host, '/'.join(path_segs), num, fmt_code(code_seen), len(container.get('payload', b'')), ) diff --git a/custom_components/localthings/ocf/registry/capabilities/common.py b/custom_components/localthings/ocf/registry/capabilities/common.py index 5539674..904508e 100644 --- a/custom_components/localthings/ocf/registry/capabilities/common.py +++ b/custom_components/localthings/ocf/registry/capabilities/common.py @@ -73,6 +73,7 @@ POWER = Capability( ALARMS = Capability( href='/alarms/vs/0', + poll_tier='hot', entities=( SensorDesc(key='alarm_code', field='x.com.samsung.da.items', name='Alarm code', icon='mdi:alert', diff --git a/custom_components/localthings/ocf/registry/capability.py b/custom_components/localthings/ocf/registry/capability.py index 9d44f85..b3d99a9 100644 --- a/custom_components/localthings/ocf/registry/capability.py +++ b/custom_components/localthings/ocf/registry/capability.py @@ -11,7 +11,7 @@ from .entities import SamsungEntityDescription class Capability: href: Optional[str] = None entities: tuple[SamsungEntityDescription, ...] = () - poll_tier: str = 'warm' # 'hot' | 'warm' | 'cold' + poll_tier: str = 'cold' # 'hot' | 'warm' | 'cold' rt_filter: Optional[str] = None # bind only if rt_filter in rep.get('rt', ()) href_prefix: Optional[str] = None # pattern caps only: bind only if href starts with this strip_prefix_in_key: bool = False # strip href_prefix segs before building key_override diff --git a/tests/localthings/test_coordinator.py b/tests/localthings/test_coordinator.py index 0ae6d21..46f919c 100644 --- a/tests/localthings/test_coordinator.py +++ b/tests/localthings/test_coordinator.py @@ -8,7 +8,7 @@ import pytest from homeassistant.core import HomeAssistant from custom_components.localthings.const import ( - ACTIVE_INTERVAL_S, DOMAIN, IDLE_INTERVAL_S, + DOMAIN, SUMMARY_INTERVAL_S, ) from custom_components.localthings.coordinator import LocalThingsCoordinator @@ -42,15 +42,15 @@ async def test_device_info_populated( ) -async def test_idle_interval_by_default( +async def test_summary_interval( hass: HomeAssistant, mock_entry, mock_coordinator_session ) -> None: - """Interval defaults to idle when no active_when fires.""" + """Summary poll interval is always SUMMARY_INTERVAL_S.""" 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.update_interval == timedelta(seconds=IDLE_INTERVAL_S) + assert coordinator.update_interval == timedelta(seconds=SUMMARY_INTERVAL_S) async def test_update_failed_on_persistent_poll_error( diff --git a/tests/test_capability.py b/tests/test_capability.py index a1c2fcf..0e27111 100644 --- a/tests/test_capability.py +++ b/tests/test_capability.py @@ -5,7 +5,7 @@ from custom_components.localthings.ocf.registry.entities import BinarySensorDesc def test_capability_defaults(): c = Capability(href='/kidslock/vs/0', entities=(BinarySensorDesc(key='child_lock', field='x.com.samsung.da.kidsLock'),)) - assert c.poll_tier == 'warm' + assert c.poll_tier == 'cold' assert c.active_when is None assert len(c.entities) == 1