r202639: breaker position range

packages/schema-1/src/span_panel_api_schema_1/adapter.py

@@ -48,7 +48,7 @@
 from span_panel_api_schema_1.description import declared_settable, device_type, node_properties
 from span_panel_api_schema_1.devices import feed_circuit_ids
 from span_panel_api_schema_1.field_metadata import build_field_metadata
-from span_panel_api_schema_1.panel import integer, text
+from span_panel_api_schema_1.panel import integer, positions_outside_model, text
 from span_panel_api_schema_1.snapshot import TreeRoles, build_snapshot, harmonised_evse_keys
 from span_panel_api_schema_1.transport import ControllerRoutes, control_target
 
@@ -113,6 +113,7 @@ def __init__(
         self._clock = clock
         self._grace_since: dict[str, float] = {}
         self._grace_reported: set[str] = set()
+        self._positions_reported = False
         self._routes = ControllerRoutes()
         self._controller = Controller(root_device_id=serial_number, mqttc=self._routes)
         self._property_callbacks: list[Callable[[str, str, str, str | None], None]] = []
@@ -187,7 +188,34 @@ def is_ready(self) -> bool:
 
     def build_snapshot(self) -> SpanPanelSnapshot:
         root = self._require_root()
-        return build_snapshot(root, self._children())
+        snapshot = build_snapshot(root, self._children())
+        self._report_positions_outside_model(snapshot)
+        return snapshot
+
+    def _report_positions_outside_model(self, snapshot: SpanPanelSnapshot) -> None:
+        """Warn once per panel when its circuits occupy spaces outside its model's range.
+
+        The snapshot keeps reporting the model's range, so such a circuit lies
+        outside the positions a consumer lays out until the table is corrected.
+        """
+        if self._positions_reported:
+            return
+        outside = positions_outside_model(
+            snapshot.model, (tab for circuit in snapshot.circuits.values() for tab in circuit.tabs)
+        )
+        if outside is None:
+            return
+        (first, last), (lowest, highest) = outside
+        _LOGGER.warning(
+            "Panel model %s has breaker positions %d-%d, but its circuits occupy spaces %d-%d; "
+            "the snapshot reports the model's positions",
+            snapshot.model,
+            first,
+            last,
+            lowest,
+            highest,
+        )
+        self._positions_reported = True
 
     def build_field_metadata(self) -> dict[str, FieldMetadata]:
         root = self._controller.get_root(self._serial_number)

packages/schema-1/src/span_panel_api_schema_1/const.py

@@ -8,6 +8,9 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
+from typing import Final
+
 # -- Device classes ---------------------------------------------------------
 
 DEVICE_TYPE_PREFIX = "energy.ebus.device."
@@ -41,6 +44,7 @@
 # imports `panel` -- so the constant has to live below both of them.
 PROP_FEEDS_DEVICE_ID = "feeds-device-id"
 PROP_FEEDS_DEVICE_STATUS = "feeds-device-status"
+PROP_FEEDS_ROLE = "feeds-role"
 PROP_FED_BY_DEVICE_ID = "fed-by-device-id"
 PROP_FED_BY_DEVICE_STATUS = "fed-by-device-status"
 NODE_DOOR = "door"
@@ -192,6 +196,17 @@
     "MLO_48": 48,
 }
 
+# The first and last breaker position for each model. Keyed like
+# `PANEL_SIZE_BY_MODEL`, and like it the numbers are ours: the panel publishes
+# only which positions are occupied.
+PANEL_POSITIONS_BY_MODEL: Final[Mapping[str, tuple[int, int]]] = {
+    "MAIN_16": (9, 24),
+    "MLO_24": (1, 24),
+    "MAIN_32": (1, 32),
+    "MAIN_40": (9, 48),
+    "MLO_48": (1, 48),
+}
+
 # The Homie attribute that carries what the flat schema published as the
 # `never-backup` boolean. v1.0 retires the property and expresses it as
 # mutability: a circuit commissioned never-backup has its priority locked, and

packages/schema-1/src/span_panel_api_schema_1/panel.py

@@ -50,6 +50,7 @@
     NODE_SHED,
     NODE_SHED_FORECAST,
     NODE_STATUS,
+    PANEL_POSITIONS_BY_MODEL,
     PANEL_SIZE_BY_MODEL,
     PCS_ACTIVE_SUFFIX,
     PCS_ENABLEMENT_SUFFIX,
@@ -92,7 +93,7 @@
 )
 
 if TYPE_CHECKING:
-    from collections.abc import Mapping, Sequence
+    from collections.abc import Iterable, Mapping, Sequence
 
     from ebus_sdk.homie import DiscoveredDevice
 
@@ -206,11 +207,13 @@ def panel_size_from_model(model: str) -> int:
 
     Unknown models return 0 and log, because inventing a size is worse than
     reporting none: a wrong total lays out positions that are not there, or
-    hides real ones.
+    hides real ones. `UNKNOWN` is a member of the model enum that names no
+    model, so it returns 0 without logging on every snapshot.
     """
-    size = PANEL_SIZE_BY_MODEL.get(model.strip().upper())
+    normalized = model.strip().upper()
+    size = PANEL_SIZE_BY_MODEL.get(normalized)
     if size is None:
-        if model:
+        if model and normalized != UNKNOWN:
             _LOGGER.warning(
                 "Unknown panel model %r; panel size unavailable. Known models: %s",
                 model,
@@ -220,6 +223,45 @@ def panel_size_from_model(model: str) -> int:
     return size
 
 
+def panel_positions(model: str | None, occupied: Iterable[int]) -> tuple[int | None, int | None]:
+    """The panel's first and last breaker positions.
+
+    The reported model's range, exactly as `PANEL_POSITIONS_BY_MODEL` gives it,
+    where the table knows the model; an occupied space outside that range does
+    not move it (`positions_outside_model` says when one does). With no known
+    model, the lowest and highest occupied spaces, and `(None, None)` when none
+    is occupied. Only the range: nothing is filled in over it.
+    """
+    known = _model_positions(model)
+    if known is not None:
+        return known
+    spaces = set(occupied)
+    if not spaces:
+        return None, None
+    return min(spaces), max(spaces)
+
+
+def positions_outside_model(model: str | None, occupied: Iterable[int]) -> tuple[tuple[int, int], tuple[int, int]] | None:
+    """The model's position range and the occupied extremes, where a space lies outside the range.
+
+    `None` where the table does not know the model, or every occupied space lies
+    inside its range. A panel that answers otherwise disagrees with the table,
+    which is ours to correct; the range reported stays the table's.
+    """
+    known = _model_positions(model)
+    spaces = set(occupied)
+    if known is None or not spaces:
+        return None
+    extremes = (min(spaces), max(spaces))
+    if known[0] <= extremes[0] and extremes[1] <= known[1]:
+        return None
+    return known, extremes
+
+
+def _model_positions(model: str | None) -> tuple[int, int] | None:
+    return PANEL_POSITIONS_BY_MODEL.get(model.strip().upper()) if model else None
+
+
 def panel_model_drift(panel: DiscoveredDevice) -> tuple[str, ...]:
     """Models the panel says are valid that we have no size for.
 
@@ -239,10 +281,11 @@ def panel_model_drift(panel: DiscoveredDevice) -> tuple[str, ...]:
     advertised = str(definition.get("format", ""))
     if not advertised:
         return ()
+    # `UNKNOWN` is a member that names no model, so there is no size to miss.
     unknown = [
         value.strip()
         for value in advertised.split(",")
-        if value.strip() and value.strip().upper() not in PANEL_SIZE_BY_MODEL
+        if value.strip() and value.strip().upper() not in {*PANEL_SIZE_BY_MODEL, UNKNOWN}
     ]
     if unknown:
         _LOGGER.warning(
@@ -386,7 +429,10 @@ def __init__(
         # one: the consumer owns the fallback text it has always shown, and a
         # default invented here would replace it with a different invention.
         self.vendor_name = text(panel, NODE_INFO, PROP_VENDOR_NAME) or None
-        self.model = text(panel, NODE_INFO, PROP_MODEL) or None
+        # `UNKNOWN` is a member of the model enum that names no model, so it
+        # reads as absent and the consumer keeps its own fallback.
+        model = text(panel, NODE_INFO, PROP_MODEL)
+        self.model = model if model and model.strip().upper() != UNKNOWN else None
         self.hardware_version = text(panel, NODE_INFO, PROP_HARDWARE_VERSION) or None
         self.main_relay_state = text(panel, NODE_STATUS, PROP_RELAY, UNKNOWN)
         self.door_state = text(panel, NODE_DOOR, PROP_STATE, UNKNOWN)