diff --git a/pyproject.toml b/pyproject.toml index 8765872..715f110 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,9 @@ Issues = "https://github.com/SpanPanel/span-panel-api/issues" [project.scripts] format-markdown = "scripts.format_markdown:main" +[project.entry-points."span_panel_api.schema_adapters"] +schema_0 = "span_panel_api._impl.schema_0:SchemaZeroAdapter" + [dependency-groups] dev = [ "pytest>=9.0.2", diff --git a/src/span_panel_api/_impl/__init__.py b/src/span_panel_api/_impl/__init__.py new file mode 100644 index 0000000..b7981bd --- /dev/null +++ b/src/span_panel_api/_impl/__init__.py @@ -0,0 +1 @@ +"""Internal implementation packages. Not public API.""" diff --git a/src/span_panel_api/_impl/schema_0/__init__.py b/src/span_panel_api/_impl/schema_0/__init__.py new file mode 100644 index 0000000..a5314d1 --- /dev/null +++ b/src/span_panel_api/_impl/schema_0/__init__.py @@ -0,0 +1,9 @@ +"""Flat-schema adapter package (data-model-version absent).""" + +from span_panel_api._impl.schema_0.adapter import SchemaZeroAdapter + +# Inclusive lower bound, exclusive upper bound. The flat schema publishes no +# data-model-version, so it is treated as the synthetic version 0 range. +SUPPORTS_DATA_MODEL_VERSIONS: tuple[str, str] = (">=0", "<1.0") + +__all__ = ["SUPPORTS_DATA_MODEL_VERSIONS", "SchemaZeroAdapter"] diff --git a/src/span_panel_api/mqtt/accumulator.py b/src/span_panel_api/_impl/schema_0/accumulator.py similarity index 98% rename from src/span_panel_api/mqtt/accumulator.py rename to src/span_panel_api/_impl/schema_0/accumulator.py index a102ac8..8b82e7f 100644 --- a/src/span_panel_api/mqtt/accumulator.py +++ b/src/span_panel_api/_impl/schema_0/accumulator.py @@ -13,7 +13,8 @@ import logging import time -from .const import HOMIE_STATE_DISCONNECTED, HOMIE_STATE_LOST, HOMIE_STATE_READY, TOPIC_PREFIX +from span_panel_api._impl.schema_0.const import TOPIC_PREFIX +from span_panel_api.mqtt.const import HOMIE_STATE_DISCONNECTED, HOMIE_STATE_LOST, HOMIE_STATE_READY _LOGGER = logging.getLogger(__name__) diff --git a/src/span_panel_api/_impl/schema_0/adapter.py b/src/span_panel_api/_impl/schema_0/adapter.py new file mode 100644 index 0000000..e03a3c1 --- /dev/null +++ b/src/span_panel_api/_impl/schema_0/adapter.py @@ -0,0 +1,67 @@ +"""Flat-schema (data-model-version absent) adapter. + +Composes the existing accumulator + consumer and owns the flat wire format: +a single Homie device whose node ids are circuit UUIDs and capability names. +Nothing outside this package constructs a flat-schema topic. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING + +from span_panel_api._impl.schema_0.accumulator import HomiePropertyAccumulator +from span_panel_api._impl.schema_0.const import PROPERTY_SET_TOPIC_FMT, TYPE_CORE, WILDCARD_TOPIC_FMT +from span_panel_api._impl.schema_0.consumer import HomieDeviceConsumer +from span_panel_api._impl.schema_0.field_metadata import build_field_metadata + +if TYPE_CHECKING: + from span_panel_api.models import FieldMetadata, HomieSchemaTypes, SpanPanelSnapshot + + +class SchemaZeroAdapter: + """Parser for the flat single-device schema (firmware r202603-r202627).""" + + schema_major = "schema_0" + SUPPORTS_DATA_MODEL_VERSIONS: tuple[str, str] = (">=0", "<1.0") + + def __init__(self, serial_number: str, panel_size: int) -> None: + self._serial_number = serial_number + self._accumulator = HomiePropertyAccumulator(serial_number) + self._consumer = HomieDeviceConsumer(self._accumulator, panel_size) + + def topics_to_subscribe(self) -> list[str]: + return [WILDCARD_TOPIC_FMT.format(serial=self._serial_number)] + + def handle_message(self, topic: str, payload: str) -> None: + self._consumer.handle_message(topic, payload) + + def is_ready(self) -> bool: + return self._consumer.is_ready() + + def build_snapshot(self) -> SpanPanelSnapshot: + return self._consumer.build_snapshot() + + def build_field_metadata(self, schema_types: HomieSchemaTypes) -> dict[str, FieldMetadata]: + return build_field_metadata(schema_types) + + def circuit_nodes_missing_names(self) -> list[str]: + return self._consumer.circuit_nodes_missing_names() + + def find_node_by_type(self, type_str: str) -> str | None: + return self._consumer.find_node_by_type(type_str) + + def set_circuit_relay_topic(self, circuit_id: str) -> str: + return PROPERTY_SET_TOPIC_FMT.format(serial=self._serial_number, node=circuit_id, prop="relay") + + def set_circuit_priority_topic(self, circuit_id: str) -> str: + return PROPERTY_SET_TOPIC_FMT.format(serial=self._serial_number, node=circuit_id, prop="shed-priority") + + def set_dominant_power_source_topic(self) -> str | None: + core_node = self._consumer.find_node_by_type(TYPE_CORE) + if core_node is None: + return None + return PROPERTY_SET_TOPIC_FMT.format(serial=self._serial_number, node=core_node, prop="dominant-power-source") + + def register_property_callback(self, callback: Callable[[str, str, str, str | None], None]) -> Callable[[], None]: + return self._consumer.register_property_callback(callback) diff --git a/src/span_panel_api/_impl/schema_0/const.py b/src/span_panel_api/_impl/schema_0/const.py new file mode 100644 index 0000000..74b17e5 --- /dev/null +++ b/src/span_panel_api/_impl/schema_0/const.py @@ -0,0 +1,42 @@ +"""Constants for the flat-schema (Homie v5) parsing implementation.""" + +# Homie v5 topic structure +HOMIE_VERSION = 5 +HOMIE_DOMAIN = "ebus" +TOPIC_PREFIX = f"{HOMIE_DOMAIN}/{HOMIE_VERSION}" + +# Topic patterns (serial_number substituted at runtime) +DEVICE_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}" +STATE_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/$state" +DESCRIPTION_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/$description" +PROPERTY_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/{{node}}/{{prop}}" +PROPERTY_SET_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/{{node}}/{{prop}}/set" +WILDCARD_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/#" + +# Homie type strings from schema +TYPE_CORE = "energy.ebus.device.distribution-enclosure.core" +TYPE_LUGS = "energy.ebus.device.lugs" +TYPE_LUGS_UPSTREAM = "energy.ebus.device.lugs.upstream" +TYPE_LUGS_DOWNSTREAM = "energy.ebus.device.lugs.downstream" +TYPE_CIRCUIT = "energy.ebus.device.circuit" +TYPE_BESS = "energy.ebus.device.bess" +TYPE_PV = "energy.ebus.device.pv" +TYPE_EVSE = "energy.ebus.device.evse" +TYPE_PCS = "energy.ebus.device.pcs" +TYPE_POWER_FLOWS = "energy.ebus.device.power-flows" + +# Lugs direction values +LUGS_UPSTREAM = "UPSTREAM" +LUGS_DOWNSTREAM = "DOWNSTREAM" + + +def normalize_circuit_id(node_id: str) -> str: + """Strip dashes from Homie UUID for entity stability.""" + return node_id.replace("-", "") + + +def denormalize_circuit_id(circuit_id: str) -> str: + """Restore dashes to a 32-char dashless UUID (8-4-4-4-12 format).""" + if len(circuit_id) == 32 and "-" not in circuit_id: + return f"{circuit_id[:8]}-{circuit_id[8:12]}-{circuit_id[12:16]}-{circuit_id[16:20]}-{circuit_id[20:]}" + return circuit_id diff --git a/src/span_panel_api/mqtt/homie.py b/src/span_panel_api/_impl/schema_0/consumer.py similarity index 98% rename from src/span_panel_api/mqtt/homie.py rename to src/span_panel_api/_impl/schema_0/consumer.py index 88767e8..ba7a29c 100644 --- a/src/span_panel_api/mqtt/homie.py +++ b/src/span_panel_api/_impl/schema_0/consumer.py @@ -12,9 +12,8 @@ import time from typing import ClassVar -from ..models import SpanBatterySnapshot, SpanCircuitSnapshot, SpanEvseSnapshot, SpanPanelSnapshot, SpanPVSnapshot -from .accumulator import HomiePropertyAccumulator -from .const import ( +from span_panel_api._impl.schema_0.accumulator import HomiePropertyAccumulator +from span_panel_api._impl.schema_0.const import ( LUGS_DOWNSTREAM, LUGS_UPSTREAM, TYPE_BESS, @@ -28,6 +27,13 @@ TYPE_PV, normalize_circuit_id, ) +from span_panel_api.models import ( + SpanBatterySnapshot, + SpanCircuitSnapshot, + SpanEvseSnapshot, + SpanPanelSnapshot, + SpanPVSnapshot, +) _LOGGER = logging.getLogger(__name__) diff --git a/src/span_panel_api/mqtt/field_metadata.py b/src/span_panel_api/_impl/schema_0/field_metadata.py similarity index 78% rename from src/span_panel_api/mqtt/field_metadata.py rename to src/span_panel_api/_impl/schema_0/field_metadata.py index a888506..ab83d68 100644 --- a/src/span_panel_api/mqtt/field_metadata.py +++ b/src/span_panel_api/_impl/schema_0/field_metadata.py @@ -15,10 +15,7 @@ from __future__ import annotations -import logging - -from ..models import FieldMetadata, HomieSchemaTypes -from .const import ( +from span_panel_api._impl.schema_0.const import ( TYPE_BESS, TYPE_CIRCUIT, TYPE_CORE, @@ -29,8 +26,7 @@ TYPE_POWER_FLOWS, TYPE_PV, ) - -_LOGGER = logging.getLogger(__name__) +from span_panel_api.models import FieldMetadata, HomieSchemaTypes # --------------------------------------------------------------------------- # Static mapping: (node_type, property_id) → snapshot field path @@ -177,53 +173,3 @@ def build_field_metadata( result[field_path] = FieldMetadata(unit=unit, datatype=datatype) return result - - -def log_schema_drift( - previous: HomieSchemaTypes, - current: HomieSchemaTypes, -) -> None: - """Log property-level differences between two schema versions. - - Called by the client when the schema hash changes between connections. - All Homie-specific detail stays in this module — the integration never - sees this output, only the transport-agnostic field metadata. - """ - prev_types = set(previous.keys()) - curr_types = set(current.keys()) - - for node_type in sorted(curr_types - prev_types): - _LOGGER.debug("Schema drift: new node type '%s'", node_type) - - for node_type in sorted(prev_types - curr_types): - _LOGGER.debug("Schema drift: removed node type '%s'", node_type) - - for node_type in sorted(prev_types & curr_types): - prev_props = previous[node_type] - curr_props = current[node_type] - if not isinstance(prev_props, dict) or not isinstance(curr_props, dict): - continue - - for prop_id in sorted(set(curr_props) - set(prev_props)): - _LOGGER.debug("Schema drift: new property '%s/%s'", node_type, prop_id) - - for prop_id in sorted(set(prev_props) - set(curr_props)): - _LOGGER.debug("Schema drift: removed property '%s/%s'", node_type, prop_id) - - for prop_id in sorted(set(prev_props) & set(curr_props)): - prev_def = prev_props[prop_id] - curr_def = curr_props[prop_id] - if not isinstance(prev_def, dict) or not isinstance(curr_def, dict): - continue - for attr in ("datatype", "unit", "format"): - old_val = prev_def.get(attr) - new_val = curr_def.get(attr) - if old_val != new_val: - _LOGGER.debug( - "Schema drift: '%s/%s' %s changed: '%s' → '%s'", - node_type, - prop_id, - attr, - old_val, - new_val, - ) diff --git a/src/span_panel_api/adapters.py b/src/span_panel_api/adapters.py new file mode 100644 index 0000000..173992a --- /dev/null +++ b/src/span_panel_api/adapters.py @@ -0,0 +1,41 @@ +"""Adapter discovery via the `span_panel_api.schema_adapters` entry-point group. + +Called once per process on the first create_span_client(). A venv change needs a +process restart regardless, so a process-lifetime cache is correct. +""" + +from __future__ import annotations + +from importlib.metadata import entry_points +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from span_panel_api.protocol import SchemaAdapter + +_LOGGER = logging.getLogger(__name__) +_ENTRY_POINT_GROUP = "span_panel_api.schema_adapters" +_REGISTRY: dict[str, type[SchemaAdapter]] | None = None + + +def discover_adapters() -> dict[str, type[SchemaAdapter]]: + """Load and cache every adapter class registered under the entry-point group.""" + global _REGISTRY # pylint: disable=global-statement # process-lifetime cache by design + if _REGISTRY is None: + registry: dict[str, type[SchemaAdapter]] = {} + for ep in entry_points(group=_ENTRY_POINT_GROUP): + if ep.name in registry: + _LOGGER.warning("Duplicate schema adapter entry point %r; keeping the first found", ep.name) + continue + try: + registry[ep.name] = ep.load() + except Exception: # pylint: disable=broad-exception-caught + _LOGGER.exception("Failed to load schema adapter entry point %r", ep.name) + _REGISTRY = registry + return _REGISTRY + + +def _reset_adapter_cache() -> None: + """Test hook. Not public API.""" + global _REGISTRY # pylint: disable=global-statement # test hook for the cache above + _REGISTRY = None diff --git a/src/span_panel_api/exceptions.py b/src/span_panel_api/exceptions.py index 24f1a56..16323a0 100644 --- a/src/span_panel_api/exceptions.py +++ b/src/span_panel_api/exceptions.py @@ -40,3 +40,17 @@ class SpanPanelStaleDataError(SpanPanelError): but data cannot be trusted right now (broker disconnected, or the Homie device has declared $state=disconnected/lost). """ + + +class SpanPanelAdapterMissingError(SpanPanelError): + """No installed adapter covers the schema this panel publishes.""" + + def __init__(self, needed: str, reason: str, available: list[str]) -> None: + self.needed = needed + self.reason = reason + self.available = available + super().__init__( + f"Panel requires adapter {needed!r} (reason: {reason}); " + f"installed adapters: {sorted(available)}. " + "Update the integration or install the missing adapter package." + ) diff --git a/src/span_panel_api/factory.py b/src/span_panel_api/factory.py index a36bef2..f8006e2 100644 --- a/src/span_panel_api/factory.py +++ b/src/span_panel_api/factory.py @@ -7,17 +7,51 @@ from __future__ import annotations import logging +import re +from .adapters import discover_adapters from .auth import register_v2 from .detection import detect_api_version -from .exceptions import SpanPanelAuthError +from .exceptions import SpanPanelAdapterMissingError, SpanPanelAuthError from .mqtt.client import SpanMqttClient from .mqtt.models import MqttClientConfig +from .protocol import SchemaAdapter _LOGGER = logging.getLogger(__name__) _V2_CLIENT_NAME = "span-panel-api" +_DMV_PATTERN = re.compile(r"^(\d+)\.\d+(?:\.\d+)?$") + + +def _select_adapter_key(data_model_version: str | None) -> tuple[str, str]: + """Tier 1 dispatch: the panel's data-model-version selects the adapter major. + + Absence is the flat-schema signal — the property was introduced by the same + firmware that introduced the parent/child model, so a panel that does not + publish it is speaking the flat schema. + """ + if data_model_version is None: + return "schema_0", "data-model-version absent (flat schema)" + + match = _DMV_PATTERN.match(data_model_version) + if match is None: + return "schema_0", f"unrecognised data-model-version={data_model_version!r}, assuming flat" + + return ( + f"schema_{int(match.group(1))}", + f"data-model-version={data_model_version!r}", + ) + + +def _resolve_adapter_cls(key: str, reason: str) -> type[SchemaAdapter]: + """Look up the discovered adapter class for `key`, or raise with the installed list.""" + registry = discover_adapters() + adapter_cls = registry.get(key) + if adapter_cls is None: + raise SpanPanelAdapterMissingError(needed=key, reason=reason, available=sorted(registry)) + return adapter_cls + async def create_span_client( host: str, @@ -68,6 +102,15 @@ async def create_span_client( if serial_number is None: raise SpanPanelAuthError("serial_number is required for MQTT transport but could not be determined") - client = SpanMqttClient(host, serial_number, mqtt_config, panel_http_port=port) + # Phase 0: the factory does not fetch the Homie schema, so no panel can + # report a data-model-version yet. `None` is the correct observation for + # every panel currently in the field — Phase 1 adds the fetch. + data_model_version: str | None = None + adapter_key, dispatch_reason = _select_adapter_key(data_model_version) + adapter_cls = _resolve_adapter_cls(adapter_key, dispatch_reason) + + client = SpanMqttClient(host, serial_number, mqtt_config, panel_http_port=port, adapter_factory=adapter_cls) + client._data_model_version = data_model_version # pylint: disable=protected-access + client._schema_dispatch_reason = dispatch_reason # pylint: disable=protected-access await client.connect() return client diff --git a/src/span_panel_api/mqtt/__init__.py b/src/span_panel_api/mqtt/__init__.py index 6eaebeb..8be5e51 100644 --- a/src/span_panel_api/mqtt/__init__.py +++ b/src/span_panel_api/mqtt/__init__.py @@ -1,10 +1,11 @@ """SPAN Panel MQTT/Homie transport.""" -from .accumulator import HomieLifecycle, HomiePropertyAccumulator +from span_panel_api._impl.schema_0.accumulator import HomieLifecycle, HomiePropertyAccumulator +from span_panel_api._impl.schema_0.consumer import HomieDeviceConsumer + from .async_client import AsyncMQTTClient from .client import SpanMqttClient from .connection import AsyncMqttBridge -from .homie import HomieDeviceConsumer from .models import MqttClientConfig __all__ = [ diff --git a/src/span_panel_api/mqtt/client.py b/src/span_panel_api/mqtt/client.py index af52623..72132ba 100644 --- a/src/span_panel_api/mqtt/client.py +++ b/src/span_panel_api/mqtt/client.py @@ -1,6 +1,6 @@ """SPAN Panel MQTT client. -Composes AsyncMqttBridge and HomieDeviceConsumer to implement +Composes AsyncMqttBridge and a SchemaAdapter to implement SpanPanelClientProtocol, CircuitControlProtocol, PanelControlProtocol, and StreamingCapableProtocol. """ @@ -10,18 +10,20 @@ import asyncio from collections.abc import Awaitable, Callable import contextlib +from importlib.metadata import version import logging import time +from span_panel_api._impl.schema_0 import SchemaZeroAdapter +from span_panel_api.schema_drift import log_schema_drift + +from ..adapters import discover_adapters from ..auth import get_homie_schema from ..exceptions import SpanPanelConnectionError, SpanPanelServerError, SpanPanelStaleDataError from ..models import FieldMetadata, HomieSchemaTypes, SpanPanelSnapshot -from ..protocol import PanelCapability -from .accumulator import HomiePropertyAccumulator +from ..protocol import PanelCapability, SchemaAdapter from .connection import AsyncMqttBridge -from .const import MQTT_READY_TIMEOUT_S, PROPERTY_SET_TOPIC_FMT, TYPE_CORE, WILDCARD_TOPIC_FMT -from .field_metadata import build_field_metadata, log_schema_drift -from .homie import HomieDeviceConsumer +from .const import MQTT_READY_TIMEOUT_S from .models import MqttClientConfig _LOGGER = logging.getLogger(__name__) @@ -42,16 +44,17 @@ def __init__( broker_config: MqttClientConfig, snapshot_interval: float = 1.0, panel_http_port: int = 80, + adapter_factory: Callable[[str, int], SchemaAdapter] = SchemaZeroAdapter, ) -> None: self._host = host self._serial_number = serial_number self._broker_config = broker_config self._snapshot_interval = snapshot_interval self._panel_http_port = panel_http_port + self._adapter_factory = adapter_factory self._bridge: AsyncMqttBridge | None = None - self._accumulator: HomiePropertyAccumulator | None = None - self._homie: HomieDeviceConsumer | None = None + self._adapter: SchemaAdapter | None = None self._streaming = False self._snapshot_callbacks: list[Callable[[SpanPanelSnapshot], Awaitable[None]]] = [] self._connection_callbacks: list[Callable[[bool], None]] = [] @@ -67,12 +70,56 @@ def __init__( # Homie accumulator with the same panel size after a transport-level # rebuild. Schema cannot change within a session, so caching is safe. self._panel_size: int | None = None + # Diagnostics — the factory overwrites these after adapter selection. + # Defaults describe a client built directly (bypassing create_span_client). + self._data_model_version: str | None = None + self._schema_dispatch_reason: str = "not dispatched" + + def _build_adapter(self, panel_size: int) -> SchemaAdapter: + """Construct the parser for this session. - def _require_homie(self) -> HomieDeviceConsumer: - """Return the HomieDeviceConsumer, raising if not yet connected.""" - if self._homie is None: + Called from connect() and from the reconnect path — the only two + places a parser is built today. + """ + self._adapter = self._adapter_factory(self._serial_number, panel_size) + return self._adapter + + @property + def adapter(self) -> SchemaAdapter | None: + """Return the active schema adapter, or None before connect(). + + On transport rebuild (see ``_on_pre_rebuild``), the adapter instance + is replaced with a fresh one — any callback registered via + ``adapter.register_property_callback(...)`` on the old instance does + not survive the rebuild and must be re-registered on the new one. + """ + return self._adapter + + @property + def schema_major(self) -> str | None: + """Return the active adapter's schema major, or None before connect().""" + return self._adapter.schema_major if self._adapter is not None else None + + @property + def data_model_version(self) -> str | None: + """Return the panel's observed data-model-version, or None if absent/not yet dispatched.""" + return self._data_model_version + + @property + def schema_dispatch_reason(self) -> str: + """Return the human-readable reason the active adapter was selected.""" + return self._schema_dispatch_reason + + @property + def available_adapters(self) -> list[str]: + """Return the sorted keys of every schema adapter discovered in this process.""" + return sorted(discover_adapters()) + + def _require_adapter(self) -> SchemaAdapter: + """Return the SchemaAdapter, raising if not yet connected.""" + if self._adapter is None: raise SpanPanelConnectionError("Client not connected — call connect() first") - return self._homie + return self._adapter # -- SpanPanelClientProtocol ------------------------------------------- @@ -107,7 +154,7 @@ async def connect(self) -> None: 1. Fetch Homie schema to determine panel size 2. Create AsyncMqttBridge with broker credentials 3. Connect to MQTT broker - 4. Subscribe to ebus/5/{serial}/# + 4. Subscribe to the adapter's topics 5. Wait for $state==ready and $description parsed Raises: @@ -120,8 +167,16 @@ async def connect(self) -> None: # Fetch schema to determine panel size and build field metadata schema = await get_homie_schema(self._host, port=self._panel_http_port) self._panel_size = schema.panel_size - self._accumulator = HomiePropertyAccumulator(self._serial_number) - self._homie = HomieDeviceConsumer(self._accumulator, schema.panel_size) + adapter = self._build_adapter(schema.panel_size) + + _LOGGER.info( + "MQTT adapter selected: %s (span-panel-api %s)\n data-model-version: %r\n reason: %s\n available: %s", + adapter.schema_major, + version("span-panel-api"), + self._data_model_version, + self._schema_dispatch_reason, + sorted(discover_adapters()), + ) # Detect schema drift from previous connection new_hash = schema.types_schema_hash @@ -137,7 +192,7 @@ async def connect(self) -> None: self._previous_schema_types = schema.types # Build transport-agnostic field metadata from schema - self._field_metadata = build_field_metadata(schema.types) + self._field_metadata = self._require_adapter().build_field_metadata(schema.types) _LOGGER.debug( "MQTT: Creating bridge to %s:%s (serial=%s)", @@ -174,9 +229,10 @@ async def connect(self) -> None: _LOGGER.debug("MQTT: Broker connected, subscribing...") # Subscribe to all device topics - wildcard = WILDCARD_TOPIC_FMT.format(serial=self._serial_number) - self._bridge.subscribe(wildcard, qos=0) - _LOGGER.debug("MQTT: Subscribed to %s, waiting for Homie ready...", wildcard) + topics = self._require_adapter().topics_to_subscribe() + for topic in topics: + self._bridge.subscribe(topic, qos=0) + _LOGGER.debug("MQTT: Subscribed to %s, waiting for Homie ready...", topics) # Wait for Homie ready state try: @@ -203,14 +259,13 @@ async def close(self) -> None: if self._bridge is not None: await self._bridge.disconnect() self._bridge = None - self._accumulator = None self._live = False async def ping(self) -> bool: """Check if MQTT connection is alive and device is ready.""" - if self._bridge is None or self._homie is None: + if self._bridge is None or self._adapter is None: return False - return self._bridge.is_connected() and self._homie.is_ready() + return self._bridge.is_connected() and self._adapter.is_ready() def register_connection_callback(self, callback: Callable[[bool], None]) -> Callable[[], None]: """Subscribe to broker connection state transitions. @@ -242,13 +297,13 @@ async def get_snapshot(self) -> SpanPanelSnapshot: No network call — snapshot is built from in-memory property values when the liveness checks pass. """ - if self._bridge is None or self._homie is None: + if self._bridge is None or self._adapter is None: raise SpanPanelStaleDataError("Client not connected — call connect() first") if not self._bridge.is_connected(): raise SpanPanelStaleDataError("MQTT broker disconnected") - if not self._homie.is_ready(): + if not self._adapter.is_ready(): raise SpanPanelStaleDataError("Homie device not ready") - return self._homie.build_snapshot() + return self._adapter.build_snapshot() # -- CircuitControlProtocol -------------------------------------------- @@ -259,33 +314,32 @@ async def set_circuit_relay(self, circuit_id: str, state: str) -> None: circuit_id: Dashless UUID (matches wire format) state: "OPEN" or "CLOSED" """ - topic = PROPERTY_SET_TOPIC_FMT.format(serial=self._serial_number, node=circuit_id, prop="relay") + topic = self._require_adapter().set_circuit_relay_topic(circuit_id) if self._bridge is not None: self._bridge.publish(topic, state, qos=1) async def set_circuit_priority(self, circuit_id: str, priority: str) -> None: - """Publish shed-priority change for a circuit. + """Publish a circuit priority change. Args: circuit_id: Dashless UUID (matches wire format) priority: v2 enum value (NEVER, SOC_THRESHOLD, OFF_GRID) """ - topic = PROPERTY_SET_TOPIC_FMT.format(serial=self._serial_number, node=circuit_id, prop="shed-priority") + topic = self._require_adapter().set_circuit_priority_topic(circuit_id) if self._bridge is not None: self._bridge.publish(topic, priority, qos=1) # -- PanelControlProtocol ---------------------------------------------- async def set_dominant_power_source(self, value: str) -> None: - """Publish dominant-power-source change to the core node. + """Publish a dominant power source change for the panel. Args: value: DPS enum value (GRID, BATTERY, NONE, GENERATOR, PV) """ - core_node = self._require_homie().find_node_by_type(TYPE_CORE) - if core_node is None: + topic = self._require_adapter().set_dominant_power_source_topic() + if topic is None: raise SpanPanelServerError("Core node not found in panel topology") - topic = PROPERTY_SET_TOPIC_FMT.format(serial=self._serial_number, node=core_node, prop="dominant-power-source") if self._bridge is not None: self._bridge.publish(topic, value, qos=1) @@ -322,18 +376,18 @@ async def stop_streaming(self) -> None: def _on_message(self, topic: str, payload: str) -> None: """Handle incoming MQTT message (called from asyncio loop).""" - homie = self._homie - if homie is None: + adapter = self._adapter + if adapter is None: return - was_ready = homie.is_ready() - homie.handle_message(topic, payload) + was_ready = adapter.is_ready() + adapter.handle_message(topic, payload) # Check if device just became ready - if not was_ready and homie.is_ready() and self._ready_event is not None: + if not was_ready and adapter.is_ready() and self._ready_event is not None: self._ready_event.set() # Dispatch snapshot callbacks if streaming - if self._streaming and homie.is_ready() and self._loop is not None: + if self._streaming and adapter.is_ready() and self._loop is not None: if self._snapshot_interval <= 0: # Real-time mode — dispatch immediately, no debounce. self._create_dispatch_task() @@ -358,9 +412,9 @@ def _on_connection_change(self, connected: bool) -> None: # edge-only (see the guard after this block). if connected: _LOGGER.debug("MQTT connection established") - if self._bridge is not None: - wildcard = WILDCARD_TOPIC_FMT.format(serial=self._serial_number) - self._bridge.subscribe(wildcard, qos=0) + if self._bridge is not None and self._adapter is not None: + for topic in self._adapter.topics_to_subscribe(): + self._bridge.subscribe(topic, qos=0) else: _LOGGER.debug("MQTT connection lost") # Cancel any pending snapshot-debounce timer so it cannot @@ -400,27 +454,26 @@ def _on_pre_rebuild(self) -> None: # because connect() never completed. return _LOGGER.debug("Pre-rebuild — resetting Homie accumulator") - self._accumulator = HomiePropertyAccumulator(self._serial_number) - self._homie = HomieDeviceConsumer(self._accumulator, self._panel_size) + self._build_adapter(self._panel_size) async def _wait_for_circuit_names(self, timeout: float) -> None: """Wait for all circuit-like nodes to have a ``name`` property. Retained MQTT messages may arrive after the Homie device transitions - to ready. This polls the HomieDeviceConsumer at short intervals and + to ready. This polls the schema adapter at short intervals and returns as soon as all circuit names are populated, or when the timeout elapses (non-fatal — entities will use fallback names). """ - homie = self._require_homie() + adapter = self._require_adapter() deadline = time.monotonic() + timeout while time.monotonic() < deadline: - missing = homie.circuit_nodes_missing_names() + missing = adapter.circuit_nodes_missing_names() if not missing: _LOGGER.debug("All circuit names received") return await asyncio.sleep(_CIRCUIT_NAMES_POLL_INTERVAL_S) - still_missing = homie.circuit_nodes_missing_names() + still_missing = adapter.circuit_nodes_missing_names() if still_missing: _LOGGER.warning( "Timed out waiting for circuit names (%d still missing): %s", @@ -473,15 +526,15 @@ async def _dispatch_snapshot(self) -> None: snapshot to subscribers after the fact. """ bridge = self._bridge - homie = self._homie - if bridge is None or not bridge.is_connected() or homie is None or not homie.is_ready(): + adapter = self._adapter + if bridge is None or not bridge.is_connected() or adapter is None or not adapter.is_ready(): _LOGGER.debug( "Skipping stale snapshot dispatch (bridge_connected=%s, homie_ready=%s)", bridge is not None and bridge.is_connected(), - homie is not None and homie.is_ready(), + adapter is not None and adapter.is_ready(), ) return - snapshot = homie.build_snapshot() + snapshot = adapter.build_snapshot() for cb in list(self._snapshot_callbacks): try: await cb(snapshot) diff --git a/src/span_panel_api/mqtt/const.py b/src/span_panel_api/mqtt/const.py index b5bc893..ac49f40 100644 --- a/src/span_panel_api/mqtt/const.py +++ b/src/span_panel_api/mqtt/const.py @@ -1,18 +1,5 @@ """Constants for SPAN Panel MQTT/Homie transport.""" -# Homie v5 topic structure -HOMIE_VERSION = 5 -HOMIE_DOMAIN = "ebus" -TOPIC_PREFIX = f"{HOMIE_DOMAIN}/{HOMIE_VERSION}" - -# Topic patterns (serial_number substituted at runtime) -DEVICE_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}" -STATE_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/$state" -DESCRIPTION_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/$description" -PROPERTY_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/{{node}}/{{prop}}" -PROPERTY_SET_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/{{node}}/{{prop}}/set" -WILDCARD_TOPIC_FMT = f"{TOPIC_PREFIX}/{{serial}}/#" - # Homie device states HOMIE_STATE_INIT = "init" HOMIE_STATE_READY = "ready" @@ -21,18 +8,6 @@ HOMIE_STATE_LOST = "lost" HOMIE_STATE_ALERT = "alert" -# Homie type strings from schema -TYPE_CORE = "energy.ebus.device.distribution-enclosure.core" -TYPE_LUGS = "energy.ebus.device.lugs" -TYPE_LUGS_UPSTREAM = "energy.ebus.device.lugs.upstream" -TYPE_LUGS_DOWNSTREAM = "energy.ebus.device.lugs.downstream" -TYPE_CIRCUIT = "energy.ebus.device.circuit" -TYPE_BESS = "energy.ebus.device.bess" -TYPE_PV = "energy.ebus.device.pv" -TYPE_EVSE = "energy.ebus.device.evse" -TYPE_PCS = "energy.ebus.device.pcs" -TYPE_POWER_FLOWS = "energy.ebus.device.power-flows" - # MQTT connection defaults MQTT_DEFAULT_MQTTS_PORT = 8883 MQTT_DEFAULT_WS_PORT = 9001 @@ -53,19 +28,3 @@ # going through HA's config_entry teardown. Resets after every rebuild attempt so the cadence holds # throughout extended outages. MQTT_FULL_REBUILD_AFTER_FAILURES = 3 - -# Lugs direction values -LUGS_UPSTREAM = "UPSTREAM" -LUGS_DOWNSTREAM = "DOWNSTREAM" - - -def normalize_circuit_id(node_id: str) -> str: - """Strip dashes from Homie UUID for entity stability.""" - return node_id.replace("-", "") - - -def denormalize_circuit_id(circuit_id: str) -> str: - """Restore dashes to a 32-char dashless UUID (8-4-4-4-12 format).""" - if len(circuit_id) == 32 and "-" not in circuit_id: - return f"{circuit_id[:8]}-{circuit_id[8:12]}-{circuit_id[12:16]}-{circuit_id[16:20]}-{circuit_id[20:]}" - return circuit_id diff --git a/src/span_panel_api/protocol.py b/src/span_panel_api/protocol.py index 11f7e97..9675293 100644 --- a/src/span_panel_api/protocol.py +++ b/src/span_panel_api/protocol.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Protocol, runtime_checkable if TYPE_CHECKING: - from .models import FieldMetadata, SpanPanelSnapshot + from .models import FieldMetadata, HomieSchemaTypes, SpanPanelSnapshot class PanelCapability(Flag): @@ -77,3 +77,40 @@ def register_snapshot_callback( async def start_streaming(self) -> None: ... async def stop_streaming(self) -> None: ... + + +@runtime_checkable +class SchemaAdapter(Protocol): + """Parser for a single data-model-major schema. + + Frozen within a major version of this package. Most methods here are + called by SpanMqttClient, which is the only bootstrap code that knows + the wire format; ``find_node_by_type`` and ``register_property_callback`` + are not called by the bootstrap at all — they exist for external + consumers of the active adapter. + """ + + schema_major: str + SUPPORTS_DATA_MODEL_VERSIONS: tuple[str, str] + + def topics_to_subscribe(self) -> list[str]: ... + + def handle_message(self, topic: str, payload: str) -> None: ... + + def is_ready(self) -> bool: ... + + def build_snapshot(self) -> SpanPanelSnapshot: ... + + def build_field_metadata(self, schema_types: HomieSchemaTypes) -> dict[str, FieldMetadata]: ... + + def circuit_nodes_missing_names(self) -> list[str]: ... + + def find_node_by_type(self, type_str: str) -> str | None: ... + + def set_circuit_relay_topic(self, circuit_id: str) -> str: ... + + def set_circuit_priority_topic(self, circuit_id: str) -> str: ... + + def set_dominant_power_source_topic(self) -> str | None: ... + + def register_property_callback(self, callback: Callable[[str, str, str, str | None], None]) -> Callable[[], None]: ... diff --git a/src/span_panel_api/schema_drift.py b/src/span_panel_api/schema_drift.py new file mode 100644 index 0000000..13f3c62 --- /dev/null +++ b/src/span_panel_api/schema_drift.py @@ -0,0 +1,65 @@ +"""Diagnostic logging for Homie schema drift between panel sessions. + +Schema-agnostic: operates purely on ``HomieSchemaTypes`` dicts (a mapping of +node type to property definitions) and has no dependency on flat-schema +(schema_0) internals. Lives at the bootstrap level so ``span_panel_api.mqtt`` +can call it without importing anything from ``_impl/schema_0``. +""" + +from __future__ import annotations + +import logging + +from span_panel_api.models import HomieSchemaTypes + +_LOGGER = logging.getLogger(__name__) + + +def log_schema_drift( + previous: HomieSchemaTypes, + current: HomieSchemaTypes, +) -> None: + """Log property-level differences between two schema versions. + + Called by the client when the schema hash changes between connections. + All Homie-specific detail stays in this module — the integration never + sees this output, only the transport-agnostic field metadata. + """ + prev_types = set(previous.keys()) + curr_types = set(current.keys()) + + for node_type in sorted(curr_types - prev_types): + _LOGGER.debug("Schema drift: new node type '%s'", node_type) + + for node_type in sorted(prev_types - curr_types): + _LOGGER.debug("Schema drift: removed node type '%s'", node_type) + + for node_type in sorted(prev_types & curr_types): + prev_props = previous[node_type] + curr_props = current[node_type] + if not isinstance(prev_props, dict) or not isinstance(curr_props, dict): + continue + + for prop_id in sorted(set(curr_props) - set(prev_props)): + _LOGGER.debug("Schema drift: new property '%s/%s'", node_type, prop_id) + + for prop_id in sorted(set(prev_props) - set(curr_props)): + _LOGGER.debug("Schema drift: removed property '%s/%s'", node_type, prop_id) + + for prop_id in sorted(set(prev_props) & set(curr_props)): + prev_def = prev_props[prop_id] + curr_def = curr_props[prop_id] + if not isinstance(prev_def, dict) or not isinstance(curr_def, dict): + continue + for attr in ("datatype", "unit", "format"): + old_val = prev_def.get(attr) + new_val = curr_def.get(attr) + if old_val != new_val: + _LOGGER.debug( + "Schema drift: '%s/%s' %s changed: '%s' → '%s'", + node_type, + prop_id, + attr, + old_val, + new_val, + ) diff --git a/tests/conftest.py b/tests/conftest.py index 8a80b68..96e1458 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,7 +14,7 @@ import span_panel_api._http as _http_mod from span_panel_api.models import V2HomieSchema -from span_panel_api.mqtt.const import TOPIC_PREFIX, TYPE_CORE +from span_panel_api._impl.schema_0.const import TOPIC_PREFIX, TYPE_CORE @pytest.fixture(autouse=True) diff --git a/tests/test_accumulator.py b/tests/test_accumulator.py index bd038fe..4e6d216 100644 --- a/tests/test_accumulator.py +++ b/tests/test_accumulator.py @@ -21,8 +21,8 @@ import pytest -from span_panel_api.mqtt.accumulator import HomieLifecycle, HomiePropertyAccumulator -from span_panel_api.mqtt.const import TOPIC_PREFIX +from span_panel_api._impl.schema_0.accumulator import HomieLifecycle, HomiePropertyAccumulator +from span_panel_api._impl.schema_0.const import TOPIC_PREFIX SERIAL = "nj-2316-XXXX" PREFIX = f"{TOPIC_PREFIX}/{SERIAL}" diff --git a/tests/test_adapters_discovery.py b/tests/test_adapters_discovery.py new file mode 100644 index 0000000..3ad0433 --- /dev/null +++ b/tests/test_adapters_discovery.py @@ -0,0 +1,16 @@ +from __future__ import annotations + +from span_panel_api.adapters import _reset_adapter_cache, discover_adapters + + +def test_discovers_the_self_registered_schema_zero_adapter() -> None: + _reset_adapter_cache() + registry = discover_adapters() + + assert "schema_0" in registry + assert registry["schema_0"].__name__ == "SchemaZeroAdapter" + + +def test_registry_is_cached_across_calls() -> None: + _reset_adapter_cache() + assert discover_adapters() is discover_adapters() diff --git a/tests/test_auth_and_homie_helpers.py b/tests/test_auth_and_homie_helpers.py index a65fb67..cf73a1a 100644 --- a/tests/test_auth_and_homie_helpers.py +++ b/tests/test_auth_and_homie_helpers.py @@ -8,10 +8,10 @@ import httpx import pytest +from span_panel_api._impl.schema_0.accumulator import HomiePropertyAccumulator +from span_panel_api._impl.schema_0.consumer import HomieDeviceConsumer, _parse_int from span_panel_api.auth import _int, download_ca_cert, get_homie_schema from span_panel_api.exceptions import SpanPanelConnectionError, SpanPanelTimeoutError -from span_panel_api.mqtt.accumulator import HomiePropertyAccumulator -from span_panel_api.mqtt.homie import HomieDeviceConsumer, _parse_int # --------------------------------------------------------------------------- diff --git a/tests/test_factory_dispatch.py b/tests/test_factory_dispatch.py new file mode 100644 index 0000000..d6c4456 --- /dev/null +++ b/tests/test_factory_dispatch.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from span_panel_api._impl.schema_0 import SchemaZeroAdapter +from span_panel_api.adapters import _reset_adapter_cache +from span_panel_api.exceptions import SpanPanelAdapterMissingError +from span_panel_api.factory import _select_adapter_key +from span_panel_api.mqtt.client import SpanMqttClient +from span_panel_api.mqtt.models import MqttClientConfig + +from conftest import MINIMAL_DESCRIPTION, SERIAL, TOPIC_PREFIX_SERIAL + + +def test_absent_data_model_version_selects_schema_zero() -> None: + key, reason = _select_adapter_key(None) + assert key == "schema_0" + assert "absent" in reason + + +@pytest.mark.parametrize("dmv", ["1.0", "1.4", "2.0"]) +def test_present_data_model_version_requests_a_numbered_adapter(dmv: str) -> None: + key, reason = _select_adapter_key(dmv) + assert key == f"schema_{dmv.split('.')[0]}" + assert dmv in reason + + +def test_missing_adapter_raises_with_the_installed_list() -> None: + from span_panel_api.factory import _resolve_adapter_cls + + _reset_adapter_cache() + with pytest.raises(SpanPanelAdapterMissingError) as exc: + _resolve_adapter_cls("schema_1", "data-model-version='1.0'") + + assert exc.value.needed == "schema_1" + assert "schema_0" in exc.value.available + + +# --------------------------------------------------------------------------- +# create_span_client — wiring the selected adapter class into SpanMqttClient +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_create_span_client_wires_schema_zero_adapter_and_diagnostics() -> None: + """The factory must pass the resolved adapter *class* as adapter_factory, + and assign the dispatch diagnostics onto the constructed client before + connect() runs.""" + from span_panel_api.factory import create_span_client + + _reset_adapter_cache() + config = MqttClientConfig(broker_host="broker.local", username="user", password="pass") + + with patch("span_panel_api.factory.SpanMqttClient") as mock_cls: + mock_client = mock_cls.return_value + mock_client.connect = AsyncMock() + + result = await create_span_client( + "192.168.1.1", + mqtt_config=config, + serial_number="test-serial", + ) + + assert result is mock_client + _, kwargs = mock_cls.call_args + assert kwargs["adapter_factory"] is SchemaZeroAdapter + mock_client.connect.assert_awaited_once() + # Diagnostics were assigned directly on the instance ahead of connect(). + assert mock_client._data_model_version is None # pylint: disable=protected-access + assert "absent" in mock_client._schema_dispatch_reason # pylint: disable=protected-access + + +# --------------------------------------------------------------------------- +# SpanMqttClient diagnostics properties — before and after connect() +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_diagnostics_properties_before_and_after_connect(mqtt_client_mock: MagicMock) -> None: + _reset_adapter_cache() + config = MqttClientConfig(broker_host="broker.local", username="user", password="pass") + client = SpanMqttClient(host="192.168.1.1", serial_number=SERIAL, broker_config=config) + + # Before connect(): no adapter yet. Defaults describe a client built + # directly, bypassing create_span_client. + assert client.adapter is None + assert client.schema_major is None + assert client.data_model_version is None + assert client.schema_dispatch_reason == "not dispatched" + assert "schema_0" in client.available_adapters + + # Simulate what create_span_client does after adapter selection, ahead of connect(). + client._data_model_version = None # pylint: disable=protected-access + client._schema_dispatch_reason = "data-model-version absent (flat schema)" # pylint: disable=protected-access + + connect_task = asyncio.create_task(client.connect()) + await asyncio.sleep(0.05) + client._on_message(f"{TOPIC_PREFIX_SERIAL}/$description", MINIMAL_DESCRIPTION) + client._on_message(f"{TOPIC_PREFIX_SERIAL}/$state", "ready") + await asyncio.wait_for(connect_task, timeout=5.0) + + assert isinstance(client.adapter, SchemaZeroAdapter) + assert client.schema_major == "schema_0" + assert client.data_model_version is None + assert client.schema_dispatch_reason == "data-model-version absent (flat schema)" + + await client.close() diff --git a/tests/test_field_metadata.py b/tests/test_field_metadata.py index 72dcb40..333776f 100644 --- a/tests/test_field_metadata.py +++ b/tests/test_field_metadata.py @@ -5,7 +5,8 @@ import logging from span_panel_api.models import FieldMetadata -from span_panel_api.mqtt.field_metadata import build_field_metadata, log_schema_drift +from span_panel_api._impl.schema_0.field_metadata import build_field_metadata +from span_panel_api.schema_drift import log_schema_drift def _make_schema_types() -> dict[str, dict[str, object]]: diff --git a/tests/test_mqtt_client_connection.py b/tests/test_mqtt_client_connection.py index 581b5a6..a7742fc 100644 --- a/tests/test_mqtt_client_connection.py +++ b/tests/test_mqtt_client_connection.py @@ -8,10 +8,9 @@ from span_panel_api.exceptions import SpanPanelError, SpanPanelStaleDataError from span_panel_api.models import SpanPanelSnapshot +from span_panel_api._impl.schema_0.const import WILDCARD_TOPIC_FMT from span_panel_api.mqtt.client import SpanMqttClient from span_panel_api.mqtt.connection import AsyncMqttBridge -from span_panel_api.mqtt.const import WILDCARD_TOPIC_FMT -from span_panel_api.mqtt.homie import HomieDeviceConsumer from span_panel_api.mqtt.models import MqttClientConfig @@ -44,15 +43,15 @@ def subscribe(self, topic: str, qos: int = 0) -> None: self.subscribed_topics.append((topic, qos)) -class _FakeHomie(HomieDeviceConsumer): - """Minimal Homie stub for get_snapshot() tests. +class _FakeAdapter: + """Minimal SchemaAdapter stub for get_snapshot()/resubscribe tests. - Bypasses HomieDeviceConsumer.__init__ — only is_ready() and - build_snapshot() are invoked on this stub. + Only the methods SpanMqttClient actually calls on the adapter are + implemented: is_ready() and build_snapshot() for liveness/dispatch + tests, topics_to_subscribe() for resubscribe tests. """ def __init__(self, ready: bool = True, snapshot: SpanPanelSnapshot | None = None) -> None: - # Intentionally do not call super().__init__ — avoids accumulator setup. self._ready_flag = ready self._snapshot = snapshot @@ -61,9 +60,12 @@ def is_ready(self) -> bool: def build_snapshot(self) -> SpanPanelSnapshot: if self._snapshot is None: - raise RuntimeError("_FakeHomie: no snapshot configured") + raise RuntimeError("_FakeAdapter: no snapshot configured") return self._snapshot + def topics_to_subscribe(self) -> list[str]: + return [WILDCARD_TOPIC_FMT.format(serial="test-serial")] + class TestRegisterConnectionCallback: """Callback subscription API — structural only (fan-out is tested in Task 4).""" @@ -210,6 +212,7 @@ def test_reconnect_triggers_resubscribe_and_callback(self) -> None: client = _make_client() bridge = _FakeBridge(connected=True) client._bridge = bridge + client._adapter = _FakeAdapter() client._live = False # was offline calls: list[bool] = [] client.register_connection_callback(calls.append) @@ -231,6 +234,7 @@ def test_resubscribe_fires_even_on_duplicate_true(self) -> None: client = _make_client() bridge = _FakeBridge(connected=True) client._bridge = bridge + client._adapter = _FakeAdapter() client._live = True # already online calls: list[bool] = [] client.register_connection_callback(calls.append) @@ -278,7 +282,7 @@ class TestGetSnapshotLiveness: async def test_raises_stale_when_bridge_none(self) -> None: client = _make_client() client._bridge = None - client._homie = _FakeHomie(ready=True) + client._adapter = _FakeAdapter(ready=True) with pytest.raises(SpanPanelStaleDataError) as exc_info: await client.get_snapshot() @@ -287,7 +291,7 @@ async def test_raises_stale_when_bridge_none(self) -> None: async def test_raises_stale_when_homie_none(self) -> None: client = _make_client() client._bridge = _FakeBridge(connected=True) - client._homie = None + client._adapter = None with pytest.raises(SpanPanelStaleDataError) as exc_info: await client.get_snapshot() @@ -296,7 +300,7 @@ async def test_raises_stale_when_homie_none(self) -> None: async def test_raises_stale_when_broker_disconnected(self) -> None: client = _make_client() client._bridge = _FakeBridge(connected=False) - client._homie = _FakeHomie(ready=True) + client._adapter = _FakeAdapter(ready=True) with pytest.raises(SpanPanelStaleDataError) as exc_info: await client.get_snapshot() @@ -305,7 +309,7 @@ async def test_raises_stale_when_broker_disconnected(self) -> None: async def test_raises_stale_when_homie_not_ready(self) -> None: client = _make_client() client._bridge = _FakeBridge(connected=True) - client._homie = _FakeHomie(ready=False) + client._adapter = _FakeAdapter(ready=False) with pytest.raises(SpanPanelStaleDataError) as exc_info: await client.get_snapshot() @@ -315,7 +319,7 @@ async def test_returns_snapshot_when_fully_live(self) -> None: sentinel = _make_sentinel_snapshot() client = _make_client() client._bridge = _FakeBridge(connected=True) - client._homie = _FakeHomie(ready=True, snapshot=sentinel) + client._adapter = _FakeAdapter(ready=True, snapshot=sentinel) snapshot = await client.get_snapshot() assert snapshot is sentinel @@ -323,7 +327,7 @@ async def test_returns_snapshot_when_fully_live(self) -> None: async def test_raised_exception_is_span_panel_error(self) -> None: client = _make_client() client._bridge = None - client._homie = None + client._adapter = None with pytest.raises(SpanPanelError): await client.get_snapshot() @@ -355,7 +359,7 @@ async def test_dispatch_snapshot_bails_when_bridge_disconnected(self, caplog: py snapshot_sentinel = _make_sentinel_snapshot() client = _make_client() client._bridge = _FakeBridge(connected=False) - client._homie = _FakeHomie(ready=True, snapshot=snapshot_sentinel) + client._adapter = _FakeAdapter(ready=True, snapshot=snapshot_sentinel) calls: list[SpanPanelSnapshot] = [] @@ -375,7 +379,7 @@ async def test_dispatch_snapshot_bails_when_homie_not_ready(self) -> None: snapshot_sentinel = _make_sentinel_snapshot() client = _make_client() client._bridge = _FakeBridge(connected=True) - client._homie = _FakeHomie(ready=False, snapshot=snapshot_sentinel) + client._adapter = _FakeAdapter(ready=False, snapshot=snapshot_sentinel) calls: list[SpanPanelSnapshot] = [] @@ -393,7 +397,7 @@ async def test_dispatch_snapshot_delivers_when_live(self) -> None: snapshot_sentinel = _make_sentinel_snapshot() client = _make_client() client._bridge = _FakeBridge(connected=True) - client._homie = _FakeHomie(ready=True, snapshot=snapshot_sentinel) + client._adapter = _FakeAdapter(ready=True, snapshot=snapshot_sentinel) calls: list[SpanPanelSnapshot] = [] @@ -429,3 +433,56 @@ def cancel(self) -> None: assert handle.cancelled is True assert client._snapshot_timer is None + + +def test_adapter_is_none_before_connect() -> None: + """The parser needs panel_size, which only connect() knows, so there is no + adapter until then — mirroring today's `self._homie = None`.""" + from span_panel_api.mqtt.client import SpanMqttClient + from span_panel_api.mqtt.models import MqttClientConfig + + client = SpanMqttClient( + "192.0.2.10", "sim-40t-001", MqttClientConfig(broker_host="192.0.2.10", username="test", password="test") + ) + + assert client.adapter is None + + +def test_client_defaults_to_the_schema_zero_factory() -> None: + from span_panel_api._impl.schema_0 import SchemaZeroAdapter + from span_panel_api.mqtt.client import SpanMqttClient + from span_panel_api.mqtt.models import MqttClientConfig + + client = SpanMqttClient( + "192.0.2.10", "sim-40t-001", MqttClientConfig(broker_host="192.0.2.10", username="test", password="test") + ) + + assert client._adapter_factory is SchemaZeroAdapter + + +def test_injected_factory_receives_serial_and_panel_size() -> None: + """The factory must be called with the panel_size discovered at connect, + not a placeholder — panel_size drives unmapped-tab computation.""" + from span_panel_api._impl.schema_0 import SchemaZeroAdapter + from span_panel_api.mqtt.client import SpanMqttClient + from span_panel_api.mqtt.models import MqttClientConfig + + seen: list[tuple[str, int]] = [] + + def factory(serial_number: str, panel_size: int) -> SchemaZeroAdapter: + seen.append((serial_number, panel_size)) + return SchemaZeroAdapter(serial_number=serial_number, panel_size=panel_size) + + client = SpanMqttClient( + "192.0.2.10", + "sim-40t-001", + MqttClientConfig(broker_host="192.0.2.10", username="test", password="test"), + adapter_factory=factory, + ) + + # Exercise the construction path directly rather than standing up a broker. + client._panel_size = 40 + client._build_adapter(40) + + assert seen == [("sim-40t-001", 40)] + assert isinstance(client.adapter, SchemaZeroAdapter) diff --git a/tests/test_mqtt_connect_flow.py b/tests/test_mqtt_connect_flow.py index 8ac7371..6297218 100644 --- a/tests/test_mqtt_connect_flow.py +++ b/tests/test_mqtt_connect_flow.py @@ -725,7 +725,8 @@ class TestSpanMqttClientAccumulatorReset: @pytest.mark.asyncio async def test_pre_rebuild_resets_accumulator(self, mqtt_client_mock: MagicMock) -> None: - """`_on_pre_rebuild` replaces accumulator and consumer with fresh instances.""" + """`_on_pre_rebuild` replaces the adapter (and its internal accumulator/ + consumer) with a fresh instance.""" client = _make_span_client() connect_task = asyncio.create_task(client.connect()) @@ -734,21 +735,18 @@ async def test_pre_rebuild_resets_accumulator(self, mqtt_client_mock: MagicMock) client._on_message(f"{TOPIC_PREFIX_SERIAL}/$state", "ready") await asyncio.wait_for(connect_task, timeout=5.0) - original_accumulator = client._accumulator - original_homie = client._homie - assert original_accumulator is not None - assert original_homie is not None - # Accumulator is in a ready-ish state from the simulated Homie messages. - assert original_homie.is_ready() is True + original_adapter = client._adapter + assert original_adapter is not None + # Adapter is in a ready-ish state from the simulated Homie messages. + assert original_adapter.is_ready() is True # Trigger the pre-rebuild hook directly — same call the bridge makes. client._on_pre_rebuild() - # New accumulator / consumer instances, fresh state. - assert client._accumulator is not original_accumulator - assert client._homie is not original_homie - assert client._homie is not None - assert client._homie.is_ready() is False + # New adapter instance, fresh state. + assert client._adapter is not original_adapter + assert client._adapter is not None + assert client._adapter.is_ready() is False await client.close() @@ -785,8 +783,7 @@ async def test_pre_rebuild_before_connect_is_noop(self) -> None: # _panel_size is None because connect() never ran. client._on_pre_rebuild() # No exception, no state changes. - assert client._accumulator is None - assert client._homie is None + assert client._adapter is None # --------------------------------------------------------------------------- diff --git a/tests/test_mqtt_homie.py b/tests/test_mqtt_homie.py index fa92a0e..e732eea 100644 --- a/tests/test_mqtt_homie.py +++ b/tests/test_mqtt_homie.py @@ -22,11 +22,9 @@ import pytest -from span_panel_api.mqtt.const import ( - HOMIE_STATE_READY, - MQTT_DEFAULT_MQTTS_PORT, - MQTT_DEFAULT_WS_PORT, - MQTT_DEFAULT_WSS_PORT, +from span_panel_api._impl.schema_0 import SchemaZeroAdapter +from span_panel_api._impl.schema_0.accumulator import HomiePropertyAccumulator +from span_panel_api._impl.schema_0.const import ( TOPIC_PREFIX, TYPE_BESS, TYPE_CIRCUIT, @@ -38,9 +36,9 @@ TYPE_POWER_FLOWS, TYPE_PV, ) -from span_panel_api.mqtt.accumulator import HomiePropertyAccumulator +from span_panel_api._impl.schema_0.consumer import HomieDeviceConsumer +from span_panel_api.mqtt.const import HOMIE_STATE_READY, MQTT_DEFAULT_MQTTS_PORT, MQTT_DEFAULT_WS_PORT, MQTT_DEFAULT_WSS_PORT from span_panel_api.mqtt.connection import AsyncMqttBridge -from span_panel_api.mqtt.homie import HomieDeviceConsumer from span_panel_api.mqtt.models import MqttClientConfig from span_panel_api.protocol import ( PanelCapability, @@ -180,18 +178,18 @@ def test_ignores_set_topics(self): class TestHomieCircuitSnapshot: def test_circuit_id_normalization(self): - from span_panel_api.mqtt.const import normalize_circuit_id + from span_panel_api._impl.schema_0.const import normalize_circuit_id assert normalize_circuit_id("aabbccdd-1122-3344-5566-778899001122") == "aabbccdd11223344556677889900112" + "2" def test_circuit_id_denormalization(self): - from span_panel_api.mqtt.const import denormalize_circuit_id + from span_panel_api._impl.schema_0.const import denormalize_circuit_id result = denormalize_circuit_id("aabbccdd11223344556677889900112" + "2") assert result == "aabbccdd-1122-3344-5566-778899001122" def test_denormalize_non_uuid(self): - from span_panel_api.mqtt.const import denormalize_circuit_id + from span_panel_api._impl.schema_0.const import denormalize_circuit_id # Non-32-char strings pass through unchanged assert denormalize_circuit_id("short") == "short" @@ -1019,6 +1017,7 @@ async def test_set_circuit_relay_publishes(self): config = MqttClientConfig(broker_host="h", username="u", password="p") client = SpanMqttClient(host="192.168.1.1", serial_number=SERIAL, broker_config=config) + client._adapter = SchemaZeroAdapter(serial_number=SERIAL, panel_size=32) mock_bridge = MagicMock() client._bridge = mock_bridge @@ -1037,6 +1036,7 @@ async def test_set_circuit_priority_publishes(self): config = MqttClientConfig(broker_host="h", username="u", password="p") client = SpanMqttClient(host="192.168.1.1", serial_number=SERIAL, broker_config=config) + client._adapter = SchemaZeroAdapter(serial_number=SERIAL, panel_size=32) mock_bridge = MagicMock() client._bridge = mock_bridge @@ -1055,13 +1055,12 @@ async def test_set_dominant_power_source_publishes(self): config = MqttClientConfig(broker_host="h", username="u", password="p") client = SpanMqttClient(host="192.168.1.1", serial_number=SERIAL, broker_config=config) - client._accumulator = HomiePropertyAccumulator(SERIAL) - client._homie = HomieDeviceConsumer(client._accumulator, panel_size=32) + client._adapter = SchemaZeroAdapter(serial_number=SERIAL, panel_size=32) # Populate the homie description so core node is known desc = _make_description(_core_description()) - client._homie.handle_message(f"{PREFIX}/$state", HOMIE_STATE_READY) - client._homie.handle_message(f"{PREFIX}/$description", desc) + client._adapter.handle_message(f"{PREFIX}/$state", HOMIE_STATE_READY) + client._adapter.handle_message(f"{PREFIX}/$description", desc) mock_bridge = MagicMock() client._bridge = mock_bridge @@ -1081,8 +1080,7 @@ async def test_set_dominant_power_source_no_core_node_raises(self): config = MqttClientConfig(broker_host="h", username="u", password="p") client = SpanMqttClient(host="192.168.1.1", serial_number=SERIAL, broker_config=config) - client._accumulator = HomiePropertyAccumulator(SERIAL) - client._homie = HomieDeviceConsumer(client._accumulator, panel_size=32) + client._adapter = SchemaZeroAdapter(serial_number=SERIAL, panel_size=32) # No description loaded — core node not found with pytest.raises(SpanPanelServerError, match="Core node not found"): @@ -1101,14 +1099,13 @@ async def test_get_snapshot_returns_homie_state(self): config = MqttClientConfig(broker_host="h", username="u", password="p") client = SpanMqttClient(host="192.168.1.1", serial_number=SERIAL, broker_config=config) - client._accumulator = HomiePropertyAccumulator(SERIAL) - client._homie = HomieDeviceConsumer(client._accumulator, panel_size=32) + client._adapter = SchemaZeroAdapter(serial_number=SERIAL, panel_size=32) client._bridge = _ConnectedBridge() - # Manually ready the homie consumer - client._homie.handle_message(f"{PREFIX}/$state", "ready") - client._homie.handle_message(f"{PREFIX}/$description", _make_description(_core_description())) - client._homie.handle_message(f"{PREFIX}/core/software-version", "test-fw") + # Manually ready the adapter + client._adapter.handle_message(f"{PREFIX}/$state", "ready") + client._adapter.handle_message(f"{PREFIX}/$description", _make_description(_core_description())) + client._adapter.handle_message(f"{PREFIX}/core/software-version", "test-fw") snapshot = await client.get_snapshot() assert snapshot.serial_number == SERIAL @@ -1132,11 +1129,10 @@ async def test_ping_true_when_connected_and_ready(self): mock_bridge = MagicMock() mock_bridge.is_connected.return_value = True client._bridge = mock_bridge - client._accumulator = HomiePropertyAccumulator(SERIAL) - client._homie = HomieDeviceConsumer(client._accumulator, panel_size=32) + client._adapter = SchemaZeroAdapter(serial_number=SERIAL, panel_size=32) - client._homie.handle_message(f"{PREFIX}/$state", "ready") - client._homie.handle_message(f"{PREFIX}/$description", _make_description(_core_description())) + client._adapter.handle_message(f"{PREFIX}/$state", "ready") + client._adapter.handle_message(f"{PREFIX}/$description", _make_description(_core_description())) assert await client.ping() is True diff --git a/tests/test_protocol_conformance.py b/tests/test_protocol_conformance.py index b3b8b52..dc7eb0e 100644 --- a/tests/test_protocol_conformance.py +++ b/tests/test_protocol_conformance.py @@ -46,3 +46,46 @@ def test_satisfies_panel_control_protocol(self) -> None: def test_satisfies_streaming_protocol(self) -> None: if not issubclass(SpanMqttClient, StreamingCapableProtocol): raise TypeError("SpanMqttClient does not satisfy StreamingCapableProtocol") + + +def test_schema_adapter_declares_its_methods() -> None: + """The protocol must name every method SpanMqttClient calls on its parser.""" + from span_panel_api.protocol import SchemaAdapter + + for name in ( + "topics_to_subscribe", + "handle_message", + "is_ready", + "build_snapshot", + "build_field_metadata", + "circuit_nodes_missing_names", + "find_node_by_type", + "set_circuit_relay_topic", + "set_circuit_priority_topic", + "set_dominant_power_source_topic", + "register_property_callback", + ): + assert hasattr(SchemaAdapter, name), f"SchemaAdapter is missing method {name}" + + +def test_schema_adapter_declares_its_class_attributes() -> None: + """`schema_major` and `SUPPORTS_DATA_MODEL_VERSIONS` are annotation-only members. + + A bare annotation on a Protocol creates no class attribute, so `hasattr` is + False for them even when correctly declared — they must be checked through + `__annotations__` instead. + """ + from span_panel_api.protocol import SchemaAdapter + + for name in ("schema_major", "SUPPORTS_DATA_MODEL_VERSIONS"): + assert name in SchemaAdapter.__annotations__, f"SchemaAdapter is missing attribute {name}" + + +def test_adapter_missing_error_reports_what_is_installed() -> None: + from span_panel_api.exceptions import SpanPanelAdapterMissingError + + err = SpanPanelAdapterMissingError(needed="schema_1", reason="data-model-version='1.0'", available=["schema_0"]) + assert err.needed == "schema_1" + assert err.available == ["schema_0"] + assert "schema_1" in str(err) + assert "schema_0" in str(err) diff --git a/tests/test_public_api_unchanged.py b/tests/test_public_api_unchanged.py new file mode 100644 index 0000000..f228cc2 --- /dev/null +++ b/tests/test_public_api_unchanged.py @@ -0,0 +1,81 @@ +"""Guard: Phase 0 is a restructure, so the public surface must not move. + +The HA integration pins span-panel-api and imports these names directly. If this +test fails, the change is no longer Phase 0 — it is a breaking release. +""" + +from __future__ import annotations + +import span_panel_api + +# Source of truth: src/span_panel_api/__init__.py __all__ (transcribed in full, +# not trimmed, per Phase 0 Task 7's instruction to reconcile against the real file +# rather than an earlier hand-transcribed listing). +EXPECTED_PUBLIC_API = { + # Protocols + "CircuitControlProtocol", + "PanelCapability", + "PanelControlProtocol", + "SpanPanelClientProtocol", + "StreamingCapableProtocol", + # Metadata + "FieldMetadata", + "HomieSchemaTypes", + # Snapshots + "SpanBatterySnapshot", + "SpanCircuitSnapshot", + "SpanEvseSnapshot", + "SpanPVSnapshot", + "SpanPanelSnapshot", + # Factory + "create_span_client", + # Detection + "DetectionResult", + "detect_api_version", + # v2 auth + "V2AuthResponse", + "V2HomieSchema", + "V2StatusInfo", + "delete_fqdn", + "download_ca_cert", + "get_fqdn", + "get_homie_schema", + "get_v2_status", + "register_fqdn", + "regenerate_passphrase", + "register_v2", + # Transport + "HomieLifecycle", + "HomiePropertyAccumulator", + "MqttClientConfig", + "SpanMqttClient", + # Phase validation + "PhaseDistribution", + "are_tabs_opposite_phase", + "get_phase_distribution", + "get_tab_phase", + "suggest_balanced_pairing", + "validate_solar_tabs", + # Exceptions + "SpanPanelAPIError", + "SpanPanelAuthError", + "SpanPanelConnectionError", + "SpanPanelError", + "SpanPanelServerError", + "SpanPanelStaleDataError", + "SpanPanelTimeoutError", + "SpanPanelValidationError", +} + + +def test_all_is_unchanged() -> None: + missing = EXPECTED_PUBLIC_API - set(span_panel_api.__all__) + assert not missing, f"Phase 0 removed public names: {sorted(missing)}" + + extra = set(span_panel_api.__all__) - EXPECTED_PUBLIC_API + assert not extra, f"Phase 0 added undocumented public names: {sorted(extra)}" + + +def test_every_exported_name_is_importable() -> None: + for name in span_panel_api.__all__: + assert hasattr(span_panel_api, name), f"{name} is in __all__ but not importable" diff --git a/tests/test_schema_zero_adapter.py b/tests/test_schema_zero_adapter.py new file mode 100644 index 0000000..0c6f2d8 --- /dev/null +++ b/tests/test_schema_zero_adapter.py @@ -0,0 +1,53 @@ +"""SchemaZeroAdapter contract tests. + +The adapter owns every piece of flat-schema knowledge that used to live in +SpanMqttClient: which topics to subscribe to, and how to address a settable +property. These tests pin the exact topic strings, because the flat wire format +is fixed by shipped firmware and must not drift. +""" + +from __future__ import annotations + +import pytest + +from span_panel_api._impl.schema_0 import SchemaZeroAdapter +from span_panel_api.protocol import SchemaAdapter + +SERIAL = "sim-40t-001" + + +@pytest.fixture +def adapter() -> SchemaZeroAdapter: + return SchemaZeroAdapter(serial_number=SERIAL, panel_size=40) + + +def test_satisfies_the_protocol(adapter: SchemaZeroAdapter) -> None: + assert isinstance(adapter, SchemaAdapter) + + +def test_declares_its_dispatch_key_and_range(adapter: SchemaZeroAdapter) -> None: + assert adapter.schema_major == "schema_0" + assert adapter.SUPPORTS_DATA_MODEL_VERSIONS == (">=0", "<1.0") + + +def test_subscribes_to_the_single_panel_wildcard(adapter: SchemaZeroAdapter) -> None: + """Flat schema is one device, so one wildcard captures everything.""" + assert adapter.topics_to_subscribe() == [f"ebus/5/{SERIAL}/#"] + + +def test_circuit_setter_topics_address_the_panel_device(adapter: SchemaZeroAdapter) -> None: + circuit = "ac3dccda46a94b98878a227df6fed588" + assert adapter.set_circuit_relay_topic(circuit) == f"ebus/5/{SERIAL}/{circuit}/relay/set" + assert adapter.set_circuit_priority_topic(circuit) == f"ebus/5/{SERIAL}/{circuit}/shed-priority/set" + + +def test_dominant_power_source_topic_is_none_before_the_core_node_is_known( + adapter: SchemaZeroAdapter, +) -> None: + """The core node id is discovered from $description, so it is unavailable + until a description has been routed through handle_message.""" + assert adapter.set_dominant_power_source_topic() is None + + +def test_is_not_ready_before_any_message(adapter: SchemaZeroAdapter) -> None: + assert adapter.is_ready() is False