diff --git a/packages/schema-1/CHANGELOG.md b/packages/schema-1/CHANGELOG.md index f13cb0d..681241d 100644 --- a/packages/schema-1/CHANGELOG.md +++ b/packages/schema-1/CHANGELOG.md @@ -35,6 +35,8 @@ Requires `span-panel-api` **3.8.0 or newer**, whose snapshot fields accept the u - **The grid-forming device name falls back to vendor and model instead of showing a device id.** - **A panel with no MID reports `dsm_state` `DSM_ON_GRID` and `current_run_config` `PANEL_ON_GRID` whenever it is publishing**, where both dropped to `UNKNOWN` each time `power-flows/grid` read within 1 W of zero ([#215](https://github.com/SpanPanel/span-panel-api/issues/215)). +- **Each MQTT message costs the same few microseconds however many devices the panel declares**, where routing it tested every recorded subscription in turn and every readiness check walked the device tree. On a 30-device tree the work per message falls + from 90–250 µs to about 2 µs, measured on a desktop machine ([#223](https://github.com/SpanPanel/span-panel-api/issues/223)). ## [1.2.2] diff --git a/packages/schema-1/src/span_panel_api_schema_1/adapter.py b/packages/schema-1/src/span_panel_api_schema_1/adapter.py index 0ec1c94..f483f06 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/adapter.py +++ b/packages/schema-1/src/span_panel_api_schema_1/adapter.py @@ -118,7 +118,14 @@ def __init__( self._controller = Controller(root_device_id=serial_number, mqttc=self._routes) self._property_callbacks: list[Callable[[str, str, str, str | None], None]] = [] self._awaiting: tuple[str, ...] = () + # Readiness as last computed, or None once the tree has changed since. + self._ready: bool | None = None self._controller.set_on_property_changed_callback(self._on_property_changed) + # Every event that can change what `is_ready()` answers; see there. + self._controller.set_on_device_discovered_callback(self._tree_changed) + self._controller.set_on_device_state_changed_callback(self._tree_changed) + self._controller.set_on_device_removed_callback(self._tree_changed) + self._controller.set_on_description_received_callback(self._tree_changed) # Records the subscriptions the tree walk needs; nothing reaches the # wire, because this object has no connection to reach it with. self._controller.start_discovery() @@ -173,7 +180,22 @@ def is_ready(self) -> bool: panel's positions with none to lay out. Asking only for what the panel itself promised keeps a firmware that omits the property connectable — it falls back to the drift warning in `panel_size_from_model`. + + **The answer is remembered until the tree changes.** The transport asks + before and after every message, and walking the tree each time cost + more than routing the message did (#223). Only the Controller's own + events can move it: a device's first `$state`, a change of state, a + removal, a `$description`, and the root's `info/model` value. Each + forgets the answer and the next call works it out again. The events + never work it out themselves, because a description is announced before + the SDK has reconciled the children it declares. """ + if self._ready is None: + self._ready = self._tree_ready() + return self._ready + + def _tree_ready(self) -> bool: + """What `is_ready()` answers, worked out from the tree as it stands.""" root = self._controller.get_root(self._serial_number) if root is None or root.state != STATE_READY or not root.description: return False @@ -626,5 +648,12 @@ def _on_property_changed(self, device_id: str, node_id: str, property_id: str, v protocol has no place for the previous value, and consumers that need one keep it themselves. """ + # The one value readiness waits on. + if (device_id, node_id, property_id) == (self._serial_number, NODE_INFO, PROP_MODEL): + self._ready = None for callback in list(self._property_callbacks): callback(device_id, node_id, property_id, value) + + def _tree_changed(self, *_event: object) -> None: + """Forget the remembered readiness. Accepts any of the Controller's tree events.""" + self._ready = None diff --git a/packages/schema-1/src/span_panel_api_schema_1/transport.py b/packages/schema-1/src/span_panel_api_schema_1/transport.py index 839f6a2..827442c 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/transport.py +++ b/packages/schema-1/src/span_panel_api_schema_1/transport.py @@ -38,10 +38,11 @@ from __future__ import annotations +import itertools import logging -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, NamedTuple -from paho.mqtt.client import topic_matches_sub +from paho.mqtt.matcher import MQTTMatcher from span_panel_api.models import ControlTarget from span_panel_api_schema_1.const import HOMIE_DOMAIN, HOMIE_VERSION @@ -80,6 +81,13 @@ def control_target(device_id: str, node: str, prop: str) -> ControlTarget: MAX_HELD_MESSAGES = 4096 +class _Route(NamedTuple): + """A recorded callback, and when it was recorded, which decides an overlap.""" + + recorded: int + callback: Callable[[str, bytes], None] + + class ControllerRoutes: """Record `Controller`'s subscriptions and route messages back to them. @@ -88,8 +96,11 @@ class ControllerRoutes: """ def __init__(self) -> None: - # Insertion-ordered; dispatch walks it in reverse — see dispatch(). - self._routes: dict[str, Callable[[str, bytes], None]] = {} + self._routes: dict[str, _Route] = {} + # The same patterns, indexed for matching: each maps to itself, and + # dispatch resolves the patterns that match through `_routes`. + self._matcher = MQTTMatcher() + self._recordings = itertools.count() # Last payload per topic that matched no route yet, keyed by topic so a # newer value supersedes an older one — the same last-value-wins rule # the broker applies to the retained message this stands in for. @@ -132,27 +143,35 @@ def subscribe(self, sub: str, param: Any, qos: int = 1) -> None: # pylint: disa the adapter's behalf, not of a route recorded in a dict. """ # Pop before insert: assigning an existing key updates the value but - # keeps the key's original position, which would leave a re-registered - # pattern behind whatever was added after it in dispatch's match order. + # keeps the key's original position, and `routes` lists most recent last. self._routes.pop(sub, None) - self._routes[sub] = param + self._routes[sub] = _Route(next(self._recordings), param) + self._matcher[sub] = sub self._release(sub, param) def unsubscribe(self, sub: str) -> None: """Forget the callback for `sub`. The broad wire subscription stays. Messages for a device the SDK has - dropped simply stop matching a route, and dispatch discards them. + dropped simply stop matching a route, and dispatch holds them. """ - self._routes.pop(sub, None) + # Guarded because the matcher raises for a pattern it never held, and + # forgetting one that was never recorded is harmless. + if self._routes.pop(sub, None) is not None: + del self._matcher[sub] # -- our side ---------------------------------------------------------- def dispatch(self, topic: str, payload: str) -> None: """Deliver one message to the callback of the route that matches. - Walked most-recent-first, which is defensive rather than currently - required. Tree-rooted discovery subscribes four **device-scoped** + Matched through a prefix tree of the recorded patterns, so the cost + follows the topic's depth rather than the number of routes. Every + message takes this path, and a linear scan grew with each device the + panel declares, at four routes per device (#223). + + The most recently recorded match wins, which is defensive rather than + currently required. Tree-rooted discovery subscribes four **device-scoped** patterns per device — `$state`, `$description`, `+/+`, `+/+/$target` (`Controller._subscribe_device_topics`) — which cannot overlap each other or another device's, so today exactly one route matches any topic. @@ -172,11 +191,15 @@ def dispatch(self, topic: str, payload: str) -> None: The SDK hands callbacks `bytes`; the transport hands us `str`. """ - for sub in reversed(self._routes): - if topic_matches_sub(sub, topic): - self._routes[sub](topic, payload.encode()) - return - self._hold(topic, payload) + # The matcher is untyped and yields what it was given, the pattern + # string; `_routes` turns each back into a typed route. The list is + # complete before any callback runs, so a callback that subscribes + # changes a matcher nothing is still walking. + matched = [self._routes[sub] for sub in self._matcher.iter_match(topic)] + if not matched: + self._hold(topic, payload) + return + max(matched, key=lambda route: route.recorded).callback(topic, payload.encode()) def _hold(self, topic: str, payload: str) -> None: """Keep a message for a route that has not been registered yet.""" @@ -201,8 +224,13 @@ def _release(self, sub: str, callback: Callable[[str, bytes], None]) -> None: in turn, so a whole tree unfolds from one root subscription. The candidate list is therefore taken up front and each entry re-checked, since a nested release may have claimed it already. + + The pattern is indexed once and matched against every held topic, + rather than indexed afresh for each of them. """ - for topic in [held for held in self._held if topic_matches_sub(sub, held)]: + pattern = MQTTMatcher() + pattern[sub] = sub + for topic in [held for held in self._held if next(pattern.iter_match(held), None) is not None]: payload = self._held.pop(topic, None) if payload is not None: callback(topic, payload.encode()) diff --git a/pyproject.toml b/pyproject.toml index caab61c..e1bbb3c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -255,6 +255,12 @@ explicit_package_bases = true # `tests/` on `sys.path`; the support packages the mypy hook checks resolve the same # way here. mypy_path = "src:tests" +# paho ships `py.typed` but leaves its topic matcher unannotated, so every call +# into it would be an untyped call. The schema-1 route table resolves what the +# matcher yields through a typed dict at once, which keeps its `Any` from going +# further; `warn_return_any` does not police that for you (a union or a +# `yield from` passes it), so keep it to that shape. +untyped_calls_exclude = ["paho.mqtt.matcher"] # Excluded by name for the reason `[tool.ruff]` gives above: the capture script is # checked like the library, and the hand-run tools written before that was expected # are not. diff --git a/tests/test_schema_one_adapter.py b/tests/test_schema_one_adapter.py index 9f4f178..46bc7b3 100644 --- a/tests/test_schema_one_adapter.py +++ b/tests/test_schema_one_adapter.py @@ -9,9 +9,11 @@ import copy import json +import random import pytest +from ebus_sdk import Controller from reference_payloads.schema_one import RetainedTopicTree, parent_child_tree from span_panel_api.adapters import _derive_required_members from span_panel_api.models import V2HomieSchema @@ -52,14 +54,28 @@ def _feed( appears; `test_children_before_the_panel` covers the unfriendly order, which a broker is equally entitled to replay. """ + for topic, payload in _retained(device_ids, omit, tree): + adapter.handle_message(topic, payload) + + +def _retained( + device_ids: list[str] | None = None, + omit: tuple[str, ...] = (), + tree: RetainedTopicTree = _TREE, +) -> list[tuple[str, str]]: + """The messages `_feed` delivers, in the order it delivers them.""" + messages: list[tuple[str, str]] = [] for device_id in device_ids or [PANEL, *[d for d in tree if d != PANEL]]: topics = tree[device_id] prefix = f"ebus/5/{device_id}" - adapter.handle_message(f"{prefix}/$description", topics["$description"]) - adapter.handle_message(f"{prefix}/$state", topics["$state"]) - for topic, value in topics.items(): - if not topic.startswith("$") and topic not in omit: - adapter.handle_message(f"{prefix}/{topic}", value) + messages.append((f"{prefix}/$description", topics["$description"])) + messages.append((f"{prefix}/$state", topics["$state"])) + messages.extend( + (f"{prefix}/{topic}", value) + for topic, value in topics.items() + if not topic.startswith("$") and topic not in omit + ) + return messages @pytest.fixture(name="adapter") @@ -212,6 +228,152 @@ def test_a_panel_that_declares_no_model_still_connects() -> None: assert adapter.build_snapshot().panel_size == 0 +# --------------------------------------------------------------------------- +# Readiness is remembered between messages +# +# The transport asks `is_ready()` before and after every message, and walking +# the tree each time cost more than routing the message did (#223). The answer +# is kept until the tree changes. Each check below asks after every message, +# as the transport does, and compares the answer with an adapter that has seen +# the same messages and is asked for the first time. +# --------------------------------------------------------------------------- + + +def _readiness_after_each(messages: list[tuple[str, str]], *, from_message: int = 0) -> list[bool]: + """Readiness after each message past `from_message`, each checked against a fresh adapter.""" + adapter = SchemaOneAdapter(PANEL, _schema()) + answers: list[bool] = [] + for count, (topic, payload) in enumerate(messages, start=1): + adapter.handle_message(topic, payload) + remembered = adapter.is_ready() + if count <= from_message: + continue + fresh = SchemaOneAdapter(PANEL, _schema()) + for earlier in messages[:count]: + fresh.handle_message(*earlier) + assert remembered is fresh.is_ready(), f"readiness was stale after {topic} = {payload!r}" + answers.append(remembered) + return answers + + +def _panel_description(**changes: object) -> str: + description = json.loads(_TREE[PANEL]["$description"]) + description.update(changes) + return json.dumps(description) + + +def _shuffled(seed: int) -> list[tuple[str, str]]: + messages = _retained() + random.Random(seed).shuffle(messages) + return messages + + +@pytest.mark.parametrize( + "messages", + [ + pytest.param(_retained(), id="panel-first"), + pytest.param(_retained([*[d for d in _TREE if d != PANEL], PANEL]), id="children-first"), + *(pytest.param(_shuffled(seed), id=f"shuffled-{seed}") for seed in (1, 2, 3)), + ], +) +def test_readiness_follows_every_message_of_a_replay(messages: list[tuple[str, str]]) -> None: + """A broker replays its retained store in any order, `$state` before + `$description` and children before parents included.""" + answers = _readiness_after_each(messages) + + assert answers[-1] is True + + +def test_readiness_follows_the_panel_leaving_ready_and_returning() -> None: + base = _retained() + messages = [*base, (f"ebus/5/{PANEL}/$state", "init"), (f"ebus/5/{PANEL}/$state", "ready")] + + assert _readiness_after_each(messages, from_message=len(base)) == [False, True] + + +def test_readiness_follows_a_child_that_is_removed_and_returns() -> None: + """An empty retained `$state` removes a device; the panel still declares it.""" + base = _retained() + prefix = f"ebus/5/{KITCHEN_CIRCUIT}" + messages = [ + *base, + (f"{prefix}/$state", ""), + (f"{prefix}/$state", "ready"), + (f"{prefix}/$description", _TREE[KITCHEN_CIRCUIT]["$description"]), + ] + + assert _readiness_after_each(messages, from_message=len(base)) == [False, False, True] + + +def test_readiness_follows_the_panel_withdrawing_a_child_it_was_waiting_on() -> None: + """A device commissioned and decommissioned before it ever described itself.""" + base = _retained() + children = json.loads(_TREE[PANEL]["$description"])["children"] + messages = [ + *base, + (f"ebus/5/{PANEL}/$description", _panel_description(children=[*children, "circuit-38"])), + (f"ebus/5/{PANEL}/$description", _TREE[PANEL]["$description"]), + ] + + assert _readiness_after_each(messages, from_message=len(base)) == [False, True] + + +def test_readiness_follows_the_panel_redeclaring_its_model() -> None: + """Whether the model is waited for is itself read from the panel's description.""" + base = _retained(omit=("info/model",)) + description = json.loads(_TREE[PANEL]["$description"]) + del description["nodes"]["info"]["properties"]["model"] + messages = [ + *base, + (f"ebus/5/{PANEL}/$description", json.dumps(description)), + (f"ebus/5/{PANEL}/$description", _TREE[PANEL]["$description"]), + (f"ebus/5/{PANEL}/info/model", _TREE[PANEL]["info/model"]), + ] + + assert _readiness_after_each(messages, from_message=len(base)) == [True, False, True] + + +def test_readiness_follows_a_childless_panel_whose_state_arrives_last() -> None: + """The panel's first `$state` is the only change here, and nothing else + re-examines the tree after it.""" + panel = {topic: value for topic, value in _TREE[PANEL].items() if not topic.startswith("$")} + prefix = f"ebus/5/{PANEL}" + messages = [ + (f"{prefix}/$description", _panel_description(children=[])), + *((f"{prefix}/{topic}", value) for topic, value in panel.items()), + (f"{prefix}/$state", "ready"), + ] + + answers = _readiness_after_each(messages) + + assert answers[-1] is True + assert not any(answers[:-1]) + + +def test_a_reading_or_a_rename_does_not_walk_the_tree(adapter: SchemaOneAdapter, monkeypatch: pytest.MonkeyPatch) -> None: + """The per-message cost #223 measured: values change constantly and + readiness never depends on them.""" + assert adapter.is_ready() is True + walks = 0 + is_tree_complete = Controller.is_tree_complete + + def counting(controller: Controller, root_id: str) -> bool: + nonlocal walks + walks += 1 + return is_tree_complete(controller, root_id) + + monkeypatch.setattr(Controller, "is_tree_complete", counting) + prefix = f"ebus/5/{KITCHEN_CIRCUIT}" + + adapter.handle_message(f"{prefix}/meter/active-power", "-250.0") + assert adapter.is_ready() is True + adapter.handle_message(f"{prefix}/info/name", "Kitchen island") + assert adapter.is_ready() is True + + assert walks == 0 + assert adapter.build_snapshot().circuits[KITCHEN_CIRCUIT].name == "Kitchen island" + + def test_snapshot_is_built_from_the_discovered_tree(adapter: SchemaOneAdapter) -> None: snapshot = adapter.build_snapshot() diff --git a/tests/test_schema_one_transport.py b/tests/test_schema_one_transport.py index 52ceceb..f5f1296 100644 --- a/tests/test_schema_one_transport.py +++ b/tests/test_schema_one_transport.py @@ -147,6 +147,78 @@ def test_rerecording_a_pattern_replaces_it_and_moves_it_to_most_recent() -> None first.assert_not_called() +def test_recency_decides_an_overlap_rather_than_specificity() -> None: + """The tests above cannot tell the two rules apart. + + In each of them the more recent route is also the more specific one, so a + matcher that simply returns the most specific filter passes them. Recorded + the other way round, only recency gives the broad route the message. + """ + routes = ControllerRoutes() + narrow = MagicMock(name="device") + broad = MagicMock(name="wildcard") + routes.subscribe("ebus/5/child-a/$state", narrow) + routes.subscribe("ebus/5/+/$state", broad) + + routes.dispatch("ebus/5/child-a/$state", "ready") + + broad.assert_called_once() + narrow.assert_not_called() + + +def test_unsubscribing_a_route_leaves_its_sibling_routing() -> None: + """Routes that share a prefix share it in the matcher too, so removing + one must leave the other where it was.""" + routes = ControllerRoutes() + state = MagicMock(name="state") + description = MagicMock(name="description") + routes.subscribe("ebus/5/child-a/$state", state) + routes.subscribe("ebus/5/child-a/$description", description) + + routes.unsubscribe("ebus/5/child-a/$description") + routes.dispatch("ebus/5/child-a/$state", "ready") + routes.dispatch("ebus/5/child-a/$description", "{}") + + state.assert_called_once_with("ebus/5/child-a/$state", b"ready") + description.assert_not_called() + assert routes.held == 1 + + +def test_unsubscribing_a_route_leaves_a_longer_one_through_it_routing() -> None: + """The SDK's own shape: a device's `+/+/$target` extends its `+/+`.""" + routes = ControllerRoutes() + values = MagicMock(name="values") + targets = MagicMock(name="targets") + routes.subscribe("ebus/5/child-a/+/+", values) + routes.subscribe("ebus/5/child-a/+/+/$target", targets) + + routes.unsubscribe("ebus/5/child-a/+/+") + routes.dispatch("ebus/5/child-a/switch/relay/$target", "OPEN") + routes.dispatch("ebus/5/child-a/meter/active-power", "1.0") + + targets.assert_called_once_with("ebus/5/child-a/switch/relay/$target", b"OPEN") + values.assert_not_called() + assert routes.held == 1 + + +def test_a_route_recorded_again_after_it_was_dropped_routes_again() -> None: + """A device the SDK drops and later re-adopts: what arrived in between is + held, released to the returning route, and later messages reach it.""" + routes = ControllerRoutes() + callback = MagicMock() + routes.subscribe("ebus/5/child-a/+/+", callback) + routes.unsubscribe("ebus/5/child-a/+/+") + routes.dispatch("ebus/5/child-a/meter/active-power", "1.0") + + routes.subscribe("ebus/5/child-a/+/+", callback) + routes.dispatch("ebus/5/child-a/meter/active-power", "2.0") + + assert [call.args for call in callback.call_args_list] == [ + ("ebus/5/child-a/meter/active-power", b"1.0"), + ("ebus/5/child-a/meter/active-power", b"2.0"), + ] + + # --------------------------------------------------------------------------- # Arrival order #