refactor: simplify RuntimeDescriptor — drop observe/poll_tiers, add intervals

Remove observe_paths and poll_tiers from RuntimeDescriptor; add
active_interval_s (default 5s) and idle_interval_s (default 30s).
Delete _make_observe_paths() and _make_poll_tiers() helpers.
Bridge rewire in Task 6 will consume the new fields.
This commit is contained in:
Marc Billow
2026-06-30 16:32:42 -05:00
parent 1b36ec4a59
commit 15f462563d
2 changed files with 25 additions and 50 deletions
+17 -43
View File
@@ -7,7 +7,6 @@ import json
from dataclasses import dataclass, field
from typing import Callable, Optional
from ..poll_scheduler import PollTier
from .discovery import BoundEntity
from .entities import (
BinarySensorDesc, ButtonDesc, NumberDesc, PLATFORM_OF, SelectDesc,
@@ -39,17 +38,17 @@ class RuntimeDescriptor:
name: str
seed_path: list[str]
default_observe_port: int
observe_paths: list[list[str]]
poll_tiers: list[PollTier]
is_active: Optional[Callable[[dict], bool]]
flatten: Callable[[dict], dict]
discovery_payloads: list[tuple[str, bytes]]
command_handlers: Callable[[], dict]
project: Optional[Callable[[dict, dict], dict]]
on_observation: Optional[Callable[[dict, str, dict], None]]
remote_available_field: Optional[str]
cycle_active_field: Optional[str]
log_state_change: Optional[Callable[[dict], str]]
active_interval_s: float = 5.0
idle_interval_s: float = 30.0
is_active: Optional[Callable[[dict], bool]] = None
flatten: Optional[Callable[[dict], dict]] = None
discovery_payloads: list[tuple[str, bytes]] = field(default_factory=list)
command_handlers: Optional[Callable[[], dict]] = None
project: Optional[Callable[[dict, dict], dict]] = None
on_observation: Optional[Callable[[dict, str, dict], None]] = None
remote_available_field: Optional[str] = None
cycle_active_field: Optional[str] = None
log_state_change: Optional[Callable[[dict], str]] = None
def _make_flatten(bound: list[BoundEntity]) -> Callable[[dict], dict]:
@@ -67,31 +66,6 @@ def _make_flatten(bound: list[BoundEntity]) -> Callable[[dict], dict]:
return flatten
def _make_observe_paths(bound: list[BoundEntity]) -> list[list[str]]:
seen, out = set(), []
for b in bound:
if b.href in seen:
continue
seen.add(b.href)
out.append(_segs(b.href))
return out
def _make_poll_tiers(bound: list[BoundEntity]) -> list[PollTier]:
groups: dict[str, set[str]] = {}
for b in bound:
groups.setdefault(b.capability.poll_tier, set()).add(b.href)
tiers: list[PollTier] = []
for name, hrefs in groups.items():
interval, active = _TIER_CADENCE.get(name, (15.0, None))
tiers.append(PollTier(
name=name, interval_s=interval, active_interval_s=active,
paths=tuple(tuple(_segs(h)) for h in sorted(hrefs))))
tiers.append(PollTier(name='sweep', interval_s=300.0,
paths=(('device', '0'),), is_sweep=True))
return tiers
def _make_is_active(bound):
caps = {b.href: b.capability for b in bound if b.capability.active_when}
if not caps:
@@ -200,8 +174,7 @@ def _discovery_payloads(bound, topic_prefix, ha_prefix, device_name, model):
cfg['state_topic'] = state_topic
cfg['value_template'] = f"{{{{ value_json.{key} }}}}"
cfg['command_topic'] = f"{topic_prefix}/cmd/{key}"
opts = d.options
cfg['options'] = list(opts) if isinstance(opts, (list, tuple)) else []
cfg['options'] = list(d.options) if d.options else []
elif isinstance(d, SwitchDesc):
cfg['state_topic'] = state_topic
cfg['value_template'] = f"{{% if value_json.{key} %}}On{{% else %}}Off{{% endif %}}"
@@ -229,14 +202,15 @@ def _discovery_payloads(bound, topic_prefix, ha_prefix, device_name, model):
def build_runtime_descriptor(bound, *, topic_prefix, ha_prefix, device_name,
model, name, default_port) -> RuntimeDescriptor:
model, name, default_port=None,
active_interval_s=5.0, idle_interval_s=30.0) -> RuntimeDescriptor:
produced = {_key(b) for b in bound}
return RuntimeDescriptor(
name=name,
seed_path=['device', '0'],
default_observe_port=default_port,
observe_paths=_make_observe_paths(bound),
poll_tiers=_make_poll_tiers(bound),
default_observe_port=default_port or 0,
active_interval_s=active_interval_s,
idle_interval_s=idle_interval_s,
is_active=_make_is_active(bound),
flatten=_make_flatten(bound),
discovery_payloads=_discovery_payloads(
+8 -7
View File
@@ -24,13 +24,6 @@ def test_flatten_applies_value_fn(dishwasher_resources):
assert isinstance(flat['child_lock'], bool)
def test_observe_paths_are_segment_lists(dishwasher_resources):
rd = build_runtime_descriptor(
_bound(dishwasher_resources), topic_prefix='t', ha_prefix='homeassistant',
device_name='Dishwasher', model='M', name='dishwasher', default_port=49154)
assert ['kidslock', 'vs', '0'] in rd.observe_paths
def test_discovery_payloads_have_unique_ids(dishwasher_resources):
rd = build_runtime_descriptor(
_bound(dishwasher_resources), topic_prefix='samsung_dishwasher',
@@ -85,3 +78,11 @@ def test_bound_entity_key_override_takes_precedence():
# key_override with instance suffix should preserve instance suffix
b3 = BoundEntity(href='/power/vs/0', capability=cap, desc=desc, instance='_1', key_override='bar')
assert _key(b3) == 'bar_1'
def test_runtime_descriptor_has_active_and_idle_intervals(dishwasher_resources):
rd = build_runtime_descriptor(
_bound(dishwasher_resources), topic_prefix='t', ha_prefix='homeassistant',
device_name='Dishwasher', model='M', name='dishwasher', default_port=49154)
assert rd.active_interval_s == 5.0
assert rd.idle_interval_s == 30.0