bhetherman/localthings
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: rewire bridge to batch-poll /device/0 + per-type registry

Browse Source 
Replace PollScheduler + OBSERVE with:
- _parse_device0_batch() extracts {href: rep} from /device/0 list response
- _discover_and_publish() uses by_type.for_device(oneUiVersion) for
  device-specific capability registry; falls back to global CAPABILITIES
- _poll_loop thread: batch GET /device/0 at active_interval_s or
  idle_interval_s; applies each rep to the state cache
- All OBSERVE machinery removed (subscribe, notifications, fetchback)
This commit is contained in:
Marc Billow committed 2026-06-30 16:33:36 -05:00
1 parent 15f462563d
commit 2b22673210
2 files changed
+98 -271
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

samsung_appliance/bridge.py
+94 -267
View File
@@ -1,23 +1,21 @@
"""Bridge: OCF CoAP-DTLS appliance → MQTT, polling-first with opportunistic OBSERVE.
"""Bridge: OCF CoAP-DTLS appliance → MQTT, batch-poll architecture.
Appliance ──CoAP DTLS─► PushBridge ──MQTT──► Home Assistant
State freshness comes from a tiered PollScheduler over a persistent DTLS
session. OBSERVE registrations are kept as an opportunistic freshness
accelerator — when the appliance has internet and pushes notifications,
the cache absorbs them; when it's air-gapped, polling carries the UX
unchanged.
State freshness comes from a periodic batch GET to /device/0 over a
persistent DTLS session. The poll interval switches between
active_interval_s and idle_interval_s based on whether the device
is in an active state.
DTLS-layer liveness is a separate KeepaliveTask (CoAP empty-CON ping
every PING_INTERVAL_S). Three consecutive failures publish MQTT
availability=offline.
Multiple PushBridges run concurrently — see main.py. They share one
MQTT client; each owns one DTLS session, one cache, one scheduler.
MQTT client; each owns one DTLS session, one cache, one poll thread.
"""
import json
import os
import random
import threading
import time
@@ -27,9 +25,8 @@ from .coap_dtls import DtlsCoapSession, fmt_code
from .config import ApplianceConfig, SharedConfig
from .keepalive import KeepaliveTask
from .logger import bridge_logger
from .observe_refresh import ObserveRefreshTask
from .poll_scheduler import PollScheduler
from .sensors import index_links
from .registry.by_type import for_device
from .registry.discovery import discover
from .state_cache import StateCache
@@ -44,24 +41,22 @@ def _href_to_segs(href: str) -> list[str]:
return [s for s in href.split('/') if s]
def _parse_device0_batch(device0: list) -> dict[str, dict]:
"""Extract {href: rep} from a /device/0 CBOR list response."""
out = {}
for entry in device0[1:]: # skip [0] (device-level rep)
if not isinstance(entry, dict):
continue
href = entry.get('href')
rep = entry.get('rep')
if href and isinstance(rep, dict) and set(rep.keys()) != {'href'}:
out[href] = rep
return out
SERIAL_PATH = '/information/vs/0'
SERIAL_FIELD = 'x.com.samsung.da.serialNum'
# If the keepalive watchdog flags the device unreachable for this long,
# force a session reconnect. Catches the half-open case where the DTLS
# socket is still writable but the peer has gone silent — without this,
# the bridge sits in offline state until the reader thread dies on its
# own (which may not happen at all if the OS sees no socket errors).
UNREACHABLE_RECONNECT_S = 120.0
# Periodic OBSERVE re-subscribe interval. Safety net for the case where
# the device stays reachable on the DTLS layer but Samsung's RT-OCF
# clears its observer table (e.g. during cloud auth blips). Without
# this, push delivery stays dead even after upstream connectivity
# recovers, since nothing triggers a fresh subscribe on the existing
# session.
OBSERVE_REFRESH_INTERVAL_S = 6 * 3600.0
def _encode(cfg: dict) -> bytes:
return json.dumps(cfg).encode()
@@ -72,11 +67,10 @@ def _bridge_diagnostic_discovery(topic_prefix: str,
device_name: str,
model: str = 'Bridge') -> list[tuple[str, bytes]]:
"""HA-discovery payloads for the per-bridge diagnostic entities
(push state, polling RTT, freshness). Returned as a list of
(polling stats, freshness). Returned as a list of
(topic, payload-bytes) tuples to be merged with entity discovery."""
avail_topic = f"{topic_prefix}/availability"
health_topic = f"{topic_prefix}/bridge/health"
push_topic = f"{topic_prefix}/bridge/push_active"
device = {
'identifiers': [topic_prefix],
'name': device_name,
@@ -107,41 +101,16 @@ def _bridge_diagnostic_discovery(topic_prefix: str,
f"{topic_prefix}/bridge_{slug}/config")
return topic, _encode(cfg)
push_active_cfg = {
'name': 'Push Active',
'unique_id': f"{topic_prefix}_bridge_push_active",
'object_id': f"{topic_prefix}_bridge_push_active",
'state_topic': push_topic,
'payload_on': 'online',
'payload_off': 'offline',
'availability': avail,
'device': device,
'entity_category': 'diagnostic',
'icon': 'mdi:rss',
}
push_active_topic = (f"{ha_discovery_prefix}/binary_sensor/"
f"{topic_prefix}/bridge_push_active/config")
return [
(push_active_topic, _encode(push_active_cfg)),
sensor('update_source', 'Last Update Source',
"{{ value_json.last_change_source | default('?') }}",
icon='mdi:transit-connection-variant'),
sensor('stalest_age_s', 'Stalest Resource Age',
"{{ value_json.stalest_age_s | default(0) }}",
unit='s', icon='mdi:clock-alert-outline'),
sensor('poll_max_rtt_ms', 'Poll Max RTT',
"{{ value_json.poll_window_max_rtt_ms | default(0) | int }}",
unit='ms', icon='mdi:speedometer'),
sensor('slow_polls', 'Slow Polls (window)',
"{{ value_json.poll_window_slow_count | default(0) }}",
icon='mdi:timer-sand'),
sensor('poll_errors', 'Poll Errors (window)',
"{{ value_json.poll_window_errors | default(0) }}",
icon='mdi:alert-circle-outline'),
sensor('observe_age_s', 'Last OBSERVE Age',
"{{ value_json.last_observe_age_s if value_json.last_observe_age_s is not none else 'never' }}",
icon='mdi:radar'),
]
@@ -161,9 +130,8 @@ class PushBridge:
self.port = app.ocf_port or 49154
self.session: DtlsCoapSession | None = None
self.scheduler: PollScheduler | None = None
self._poll_thread: threading.Thread | None = None
self.keepalive: KeepaliveTask | None = None
self.observe_refresh: ObserveRefreshTask | None = None
# descriptor is None until _discover_and_publish completes after seed.
self.descriptor = None
@@ -187,39 +155,20 @@ class PushBridge:
self.error_count = 0
self._publish_gate = False
self._last_change_source: str | None = None
self._last_observe_change_ts: float | None = None
self._last_push_active_pub: str | None = None
# Wall-clock timestamp the keepalive watchdog first reported the
# device unreachable on the current session. Cleared on recovery
# or session start. Drives the force-reconnect watchdog below.
self._unreachable_since: float | None = None
self._force_close_in_flight: bool = False
# Push is considered "active" if an OBSERVE-sourced change
# arrived within this window. Long enough that a quiet but
# working appliance doesn't flap to inactive; short enough that
# a genuinely silent push channel is visible within minutes.
self.push_active_window_s = 600.0
# Snapshot counters for the per-health-window deltas surfaced in
# publish_health's log summary.
# Poll counters — accumulated across reconnects.
self._poll_count = 0
self._poll_error_count = 0
self._win_prev_poll = 0
self._win_prev_poll_err = 0
self._win_prev_ping_fail = 0
# Per-href fetchback generation counter coalesces bursts of
# OBSERVE-block2 partial notifications: rapid changes scheduled
# many fetchbacks but only the latest actually publishes.
self._fetch_gen: dict[str, int] = {}
self._fetch_lock = threading.Lock()
p = app.topic_prefix
self.state_topic = f"{p}/state"
self.avail_topic = f"{p}/availability"
self.remote_topic = f"{p}/remote_available"
self.cycle_topic = f"{p}/cycle_active"
self.health_topic = f"{p}/bridge/health"
self.push_active_topic = f"{p}/bridge/push_active"
self.cmd_handlers = {}
self.cmd_topic_prefix = f"{p}/cmd/"
@@ -230,28 +179,8 @@ class PushBridge:
self.notif_count += 1
self.last_change_ts = time.time()
self._last_change_source = source
if source == 'observe':
self._last_observe_change_ts = self.last_change_ts
self.maybe_publish_state()
def _on_notification(self, href, payload_bytes):
"""Reader-thread callback for OBSERVE notifications. Large
resources (oven /mode/vs/0 ~9KB) arrive truncated with Block2.M=1
and we use cbor-decode failure as the partial signal."""
if not payload_bytes:
self._schedule_fetchback(href)
return
try:
rep = cbor2.loads(payload_bytes)
except Exception:
self._schedule_fetchback(href)
return
if not isinstance(rep, dict):
return
if DEBUG_BRIDGE:
self._debug_log_rep(href, rep)
self.cache.apply_rep(href, rep, source='observe')
def _debug_log_rep(self, href, rep):
if href == '/mode/vs/0' and isinstance(rep, dict):
self.log.info("mode modes=%r options=%r",
@@ -260,46 +189,6 @@ class PushBridge:
elif href in ('/operational/state/vs/0', '/oven/vs/0', '/power/vs/0'):
self.log.info("REP %s = %r", href, rep)
def _schedule_fetchback(self, href, delay_s: float = 0.0):
with self._fetch_lock:
gen = self._fetch_gen.get(href, 0) + 1
self._fetch_gen[href] = gen
threading.Thread(
target=self._fetch_back,
args=(href, delay_s, gen),
daemon=True,
name=f'fetch{href}',
).start()
def _fetch_back(self, href, delay_s: float, gen: int):
if delay_s > 0 and self.stop.wait(delay_s):
return
with self._fetch_lock:
if self._fetch_gen.get(href) != gen:
return
sess = self.session
if sess is None:
return
segs = _href_to_segs(href)
try:
code, payload = sess.get(segs, timeout=15.0)
except Exception as e:
self.log.warning("fetchback %s: %s", href, e)
return
with self._fetch_lock:
if self._fetch_gen.get(href) != gen:
return
if code != 0x45:
self.log.warning("fetchback %s: %s", href, fmt_code(code))
return
try:
rep = cbor2.loads(payload) if payload else {}
except Exception as e:
self.log.warning("fetchback %s cbor: %s", href, e)
return
if isinstance(rep, dict):
self.cache.apply_rep(href, rep, source='observe')
def _retag_logger_with_serial(self):
if self._serial is not None:
return
@@ -318,7 +207,6 @@ class PushBridge:
self.app.ip, self.port,
cert_path=self.shared.CERT_PATH,
key_path=self.shared.KEY_PATH,
on_notification=self._on_notification,
)
sess.connect()
self.session = sess
@@ -326,8 +214,6 @@ class PushBridge:
self.connect_count += 1
self.cache.descriptor_state.clear()
self._publish_gate = False
self._unreachable_since = None
self._force_close_in_flight = False
self.log.info("DTLS connected — port %d", self.port)
@@ -367,28 +253,12 @@ class PushBridge:
# on reconnect the descriptor is reused from the first session.
self._discover_and_publish(sess)
# Now subscribe to observe paths (descriptor populated above).
for path in self.descriptor.observe_paths:
sess.subscribe(path)
time.sleep(0.05)
self._publish_gate = True
self.maybe_publish_state(force=True)
self.set_availability(True)
self.log.info("seeded → %d links; sensors live",
len(self.cache.links))
scheduler = PollScheduler(
sess, self.cache,
tiers=self.descriptor.poll_tiers,
sweep_index_fn=index_links,
is_active_fn=self.descriptor.is_active,
logger=self.log,
)
# Half-open detection: if no successful poll lands inside this
# window the session is wedged, regardless of whether ping sends
# leave the socket. 60s gives ~60 hot-tier cycles of margin.
liveness_window_s = 60.0
keepalive = KeepaliveTask(
sess,
interval_s=float(self.shared.PING_INTERVAL_S),
@@ -396,38 +266,22 @@ class PushBridge:
on_reachable=self._on_reachable,
on_unreachable=self._on_unreachable,
logger=self.log,
liveness_fn=lambda: (time.monotonic() - scheduler.last_success_ts
) < liveness_window_s,
)
observe_refresh = ObserveRefreshTask(
sess,
paths=self.descriptor.observe_paths,
interval_s=OBSERVE_REFRESH_INTERVAL_S,
logger=self.log,
)
self.scheduler = scheduler
self.keepalive = keepalive
self.observe_refresh = observe_refresh
sched_t = threading.Thread(
target=scheduler.run_forever, args=(self.stop,),
poll_t = threading.Thread(
target=self._poll_loop, args=(sess,),
daemon=True, name=f'{self.app.klass}-poll')
ka_t = threading.Thread(
target=keepalive.run_forever, args=(self.stop,),
daemon=True, name=f'{self.app.klass}-ping')
ref_t = threading.Thread(
target=observe_refresh.run_forever, args=(self.stop,),
daemon=True, name=f'{self.app.klass}-obsref')
sched_t.start()
poll_t.start()
ka_t.start()
ref_t.start()
try:
sess.join()
finally:
self.scheduler = None
self.keepalive = None
self.observe_refresh = None
def _seed_from_device0(self, sess):
code, pl = sess.get(SEED_PATH, timeout=15.0)
@@ -443,7 +297,8 @@ class PushBridge:
# During the seed we want the cache populated without triggering
# a publish per resource — gate the on_change callback off until
# the publish gate opens just below.
for href, rep in index_links(body).items():
resources = _parse_device0_batch(body) if isinstance(body, list) else {}
for href, rep in resources.items():
if href not in self.cache.links:
self.cache.apply_rep(href, rep, source='seed')
self.last_seed_ts = time.time()
@@ -455,11 +310,26 @@ class PushBridge:
re-published via reassert_availability instead."""
if self._discovered:
return
from .registry import CAPABILITIES, discover, build_runtime_descriptor
from .registry import CAPABILITIES, build_runtime_descriptor
from .registry.identity import read_identity
snapshot = self.cache.snapshot()
bound = discover(snapshot, CAPABILITIES,
log=lambda m: self.log.debug("%s", m))
resources = self.cache.snapshot()
one_ui_version = (resources.get('/otninformation/vs/0', {})
.get('swVersionInfo', {})
.get('oneUiVersion', ''))
type_registry = for_device(one_ui_version) if one_ui_version else None
if type_registry is not None:
self.log.info("device type: %s (oneUiVersion=%r)",
type_registry.name, one_ui_version)
bound = discover(resources, type_registry.capabilities,
type_registry.pattern_capabilities,
log=lambda m: self.log.debug("%s", m))
else:
if one_ui_version:
self.log.warning(
"unknown device type oneUiVersion=%r; using common caps",
one_ui_version)
bound = discover(resources, CAPABILITIES,
log=lambda m: self.log.debug("%s", m))
ident = read_identity(sess, self._serial)
model = ident.model or self.app.klass.title() or 'Samsung Appliance'
device_name = self.app.device_name or ident.name or model
@@ -502,38 +372,47 @@ class PushBridge:
except Exception as e:
self.log.warning("oic/res get: %s", e)
def _poll_loop(self, sess: DtlsCoapSession) -> None:
"""Batch-poll /device/0 at active or idle interval until stop is set."""
while not self.stop.is_set():
rd = self.descriptor
if rd is not None:
snap = self.cache.snapshot()
active = rd.is_active is not None and rd.is_active(snap)
interval = rd.active_interval_s if active else rd.idle_interval_s
else:
interval = 30.0
if self.stop.wait(interval):
break
try:
code, payload = sess.get(SEED_PATH, timeout=15.0)
except Exception as e:
self.log.warning("poll error: %s", e)
self._poll_error_count += 1
continue
self._poll_count += 1
if code == 0x45 and payload:
try:
body = cbor2.loads(payload)
except Exception as e:
self.log.warning("poll cbor decode: %s", e)
self._poll_error_count += 1
continue
if isinstance(body, list):
new_resources = _parse_device0_batch(body)
for href, rep in new_resources.items():
self.cache.apply_rep(href, rep, source='poll')
elif code != 0x45:
self.log.warning("poll %s: %s", SEED_PATH, fmt_code(code))
self._poll_error_count += 1
def _on_reachable(self) -> None:
self._unreachable_since = None
self.set_availability(True)
self.reassert_availability()
def _on_unreachable(self) -> None:
if self._unreachable_since is None:
self._unreachable_since = time.time()
self.set_availability(False)
def _maybe_force_reconnect(self) -> None:
"""If the device has been unreachable for UNREACHABLE_RECONNECT_S,
close the DTLS session to break run_forever's session_once() out
of its sess.join() and trigger a fresh connect. Without this, a
half-open session (writable socket, silent peer) holds the bridge
in offline limbo until the OS surfaces a socket error."""
if self._unreachable_since is None or self._force_close_in_flight:
return
elapsed = time.time() - self._unreachable_since
if elapsed < UNREACHABLE_RECONNECT_S:
return
sess = self.session
if sess is None:
return
self.log.warning(
"unreachable for %.0fs — forcing session reconnect", elapsed)
self._force_close_in_flight = True
try:
sess.close()
except Exception as e:
self.log.warning("force-close: %s", e)
# ---- MQTT publishing --------------------------------------------
def maybe_publish_state(self, force=False):
@@ -616,10 +495,6 @@ class PushBridge:
if cycle_field is not None:
self.publish_cycle_active(
self.last_state_pub.get(cycle_field), force=True)
now = time.time()
active = (self._last_observe_change_ts is not None
and (now - self._last_observe_change_ts) <= self.push_active_window_s)
self.publish_push_active(active, force=True)
def set_availability(self, online):
value = 'online' if online else 'offline'
@@ -645,13 +520,6 @@ class PushBridge:
qos=1, retain=True)
except Exception:
pass
if not online:
self._last_push_active_pub = None
try:
self.mqtt.publish(self.push_active_topic, 'offline',
qos=1, retain=True)
except Exception:
pass
# ---- MQTT command handling --------------------------------------
@@ -676,40 +544,27 @@ class PushBridge:
self.log.warning("command %s: no DTLS session", topic)
return
href = '/' + '/'.join(path_segs)
sched = self.scheduler
defer_s = 4.0
if sched is not None:
sched.write_in_progress(href, settle_s=defer_s)
try:
code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=8.0)
except Exception as e:
self.log.warning("command %s POST failed: %s", topic, e)
return
defer_note = f" (poll-defer {href} {defer_s:.0f}s)" if sched is not None else ''
self.log.info("command %s payload=%r → %s%s",
suffix, payload, fmt_code(code), defer_note)
self.log.info("command %s payload=%r → %s",
suffix, payload, fmt_code(code))
if code >> 5 == 2:
# Optimistic local merge so HA sees the write reflected
# immediately. No Block2 fetchback — that triggers Samsung's
# 3-second revert (project_fetchback_revert_root_cause.md).
# The PollScheduler will reconcile on its next tier tick
# after the write_in_progress settle window expires.
# The batch poll will reconcile on its next tick.
self.cache.apply_optimistic(href, body)
def publish_health(self):
now = time.time()
sched = self.scheduler
ka = self.keepalive
last_obs_age = (round(now - self._last_observe_change_ts, 1)
if self._last_observe_change_ts else None)
push_active = (last_obs_age is not None
and last_obs_age <= self.push_active_window_s)
# Per-window deltas for the log summary AND the health topic
# (HA gets the same numbers without doing template arithmetic).
poll = sched.poll_count if sched else 0
poll_err = sched.poll_error_count if sched else 0
# Per-window deltas for the log summary AND the health topic.
poll = self._poll_count
poll_err = self._poll_error_count
ping_fail = ka.ping_fail_count if ka else 0
d_poll = poll - self._win_prev_poll
d_err = poll_err - self._win_prev_poll_err
@@ -717,12 +572,10 @@ class PushBridge:
self._win_prev_poll = poll
self._win_prev_poll_err = poll_err
self._win_prev_ping_fail = ping_fail
win_max_rtt, win_slow, win_timeouts = (
sched.take_window_stats() if sched else (0.0, 0, 0))
window_polls_ok = max(0, d_poll - d_err)
h = {
'mode': 'poll+observe',
'mode': 'poll',
'device_class': self.app.klass or 'appliance',
'serial': self._serial,
'connect_count': self.connect_count,
@@ -732,15 +585,10 @@ class PushBridge:
'poll_error_count': poll_err,
'poll_window_ok': window_polls_ok,
'poll_window_errors': d_err,
'poll_window_max_rtt_ms': round(win_max_rtt, 0),
'poll_window_slow_count': win_slow,
'poll_window_timeout_count': win_timeouts,
'ping_count': ka.ping_count if ka else 0,
'ping_fail_count': ping_fail,
'reachable': ka.reachable if ka else None,
'last_change_source': self._last_change_source,
'last_observe_age_s': last_obs_age,
'push_active': push_active,
'last_change_age_s': (round(now - self.last_change_ts, 1)
if self.last_change_ts else None),
'last_seed_age_s': (round(now - self.last_seed_ts, 1)
@@ -759,29 +607,12 @@ class PushBridge:
except Exception as e:
self.log.warning("health publish: %s", e)
self.publish_push_active(push_active)
self._maybe_force_reconnect()
if d_poll > 0 or d_err > 0 or d_ping_fail > 0:
self.log.info(
"poll-window: %d ok, %d err, %d ping-fail, "
"p_max=%.0fms, slow=%d, timeouts=%d (%ds)",
"poll-window: %d ok, %d err, %d ping-fail (%ds)",
window_polls_ok, d_err, d_ping_fail,
win_max_rtt, win_slow, win_timeouts,
self.shared.HEALTH_INTERVAL_S)
def publish_push_active(self, active: bool, force: bool = False) -> None:
value = 'online' if active else 'offline'
if not force and value == self._last_push_active_pub:
return
self._last_push_active_pub = value
try:
self.mqtt.publish(self.push_active_topic, value,
qos=1, retain=True)
self.log.info("push_active → %s", value)
except Exception as e:
self.log.warning("push_active publish: %s", e)
# ---- top-level loop ---------------------------------------------
def run_forever(self):
@@ -802,12 +633,8 @@ class PushBridge:
self.session_started_ts = None
if self.stop.is_set():
break
# Jitter the backoff so multiple bridges (dryer + oven) don't
# reconnect in lockstep after a router blip — synchronized
# storms make the broker / DTLS layer flap harder than need
# be. ±30% noise spreads the retry attempts.
wait = min(backoff, 30.0) * random.uniform(0.7, 1.3)
self.log.info("reconnect in %.1fs", wait)
wait = min(backoff, 30.0)
self.log.info("reconnect in %.0fs", wait)
if self.stop.wait(wait):
break
backoff = min(backoff * 2, 30.0)
tests/test_end_to_end.py
+4 -4
View File
@@ -10,16 +10,16 @@ from samsung_appliance.registry import (CAPABILITIES, discover,
('dishwasher', '10.0.0.129'),
('refrigerator', '10.0.0.254'),
])
def test_full_pipeline_produces_entities_and_observe_paths(name, ip):
def test_full_pipeline_produces_entities_and_intervals(name, ip):
resources = json.load(open(f'local-tools/dumps/{ip}.json'))['resources']
bound = discover(resources, CAPABILITIES)
rd = build_runtime_descriptor(
bound, topic_prefix=f'samsung_{name}', ha_prefix='homeassistant',
device_name=name.title(), model='M', name=name, default_port=49154)
assert rd.discovery_payloads, "no entities discovered"
assert rd.observe_paths, "no observe paths"
assert any(t.name == 'hot' for t in rd.poll_tiers), "no hot tier"
assert any(t.is_sweep for t in rd.poll_tiers), "no sweep tier"
assert rd.active_interval_s > 0, "active_interval_s must be positive"
assert rd.idle_interval_s > rd.active_interval_s, (
"idle_interval_s must be greater than active_interval_s")
# flatten runs cleanly over the real rep set
flat = rd.flatten(resources)
assert flat