r202639: link status of a circuit-fed battery

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

@@ -241,6 +241,8 @@ def build_battery(
     bess: DiscoveredDevice | None,
     owners: list[DiscoveredDevice],
     *,
+    feeds: Mapping[str, str] | None = None,
+    feed_statuses: Mapping[str, str] | None = None,
     firmware_version: str | None = None,
     power_flow_battery: float | None = None,
 ) -> SpanBatterySnapshot:
@@ -249,11 +251,20 @@ def build_battery(
     `firmware_version` is the enclosure's ``info/firmware-version`` and
     `power_flow_battery` its ``power-flows/battery``; together they pick the
     BESS meter's wire frame. Omitted, they leave the old frame assumed.
+
+    The link status is the owner's ``fed-by-*`` record where a device claims
+    the battery that way, as the upstream lugs do for a battery ahead of them.
+    Otherwise it is the record of the circuit that feeds the battery, from
+    `feeds` and `feed_statuses` (``feed_circuit_ids`` and
+    ``feed_connection_statuses``), as it is for an inverter or a charger.
     """
     if bess is None:
         return SpanBatterySnapshot(present=False)
 
+    feed_circuit_id = (feeds or {}).get(bess.device_id)
     status = connection_status_for(bess.device_id, owners)
+    if status is None:
+        status = (feed_statuses or {}).get(bess.device_id)
     raw_power_w = number(bess, NODE_METER, PROP_ACTIVE_POWER)
     charge_positive_on_wire = _bess_meter_is_charge_positive(firmware_version, raw_power_w, power_flow_battery)
 
@@ -280,6 +291,7 @@ def build_battery(
         # would have to pick one.
         communication_state=_optional(text(bess, NODE_STATUS, PROP_COMMUNICATION_STATE)),
         present=True,
+        feed_circuit_id=feed_circuit_id,
     )
 
 

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

@@ -295,6 +295,8 @@ def build_snapshot(panel: DiscoveredDevice, children: list[DiscoveredDevice]) ->
         battery=build_battery(
             roles.bess,
             owners,
+            feeds=feeds,
+            feed_statuses=feed_statuses,
             firmware_version=fields.firmware_version,
             power_flow_battery=fields.power_flow_battery,
         ),

src/span_panel_api/models.py

@@ -494,6 +494,13 @@ class SpanBatterySnapshot:
     # its battery values whether or not one is installed.
     present: bool | None = None
 
+    # The circuit whose `connection/feeds-device-id` names this battery, v1.0
+    # only. Where it is set and no device claims the battery through a
+    # `fed-by-*` record, `connected` is that circuit's `feeds-device-status`.
+    # None where no circuit feeds the battery, as when it sits ahead of the
+    # upstream lugs, and on the flat schema.
+    feed_circuit_id: str | None = None
+
 
 @dataclass(frozen=True, slots=True)
 class FieldMetadata:

tests/test_circuit_fed_battery_captures.py

@@ -0,0 +1,21 @@
+"""A battery fed by a circuit reports that circuit's link status, over every reference capture."""
+
+from __future__ import annotations
+
+import pytest
+
+from reference_payloads.captures import CAPTURES, capture_snapshot, capture_tree, circuit_fed_batteries
+
+
+@pytest.mark.parametrize("stem", CAPTURES)
+def test_a_circuit_fed_battery_follows_its_circuit(stem: str) -> None:
+    tree, snapshot = capture_tree(stem), capture_snapshot(stem)
+    for circuit_id in circuit_fed_batteries(tree).values():
+        status = tree.value(circuit_id, "connection/feeds-device-status")
+        assert snapshot.battery.connected == (None if status is None else status == "OK")
+        assert snapshot.battery.feed_circuit_id == circuit_id
+        assert snapshot.circuits[circuit_id].device_type == "circuit"
+
+
+def test_the_captures_hold_a_circuit_fed_battery() -> None:
+    assert any(circuit_fed_batteries(capture_tree(stem)) for stem in CAPTURES)

tests/test_schema_one_circuit_fed_battery.py

@@ -0,0 +1,61 @@
+"""A battery fed by a circuit takes its link status from that circuit.
+
+Where a circuit's `connection/feeds-device-id` names the battery, the
+circuit's `feeds-device-status` is the panel's view of the link, exactly as it
+is for an inverter or a charger: `SpanBatterySnapshot.connected` follows it and
+`feed_circuit_id` names the circuit. The circuit stays a circuit. A battery
+claimed by the upstream lugs keeps reading its link from them.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from conftest import flat_schema
+from reference_payloads.schema_one import parent_child_tree, replay
+from reference_payloads.synthetic_trees import PANEL_ID, SyntheticDevice, circuit_fed_battery, panel, tree, without_values
+from span_panel_api import SpanPanelSnapshot
+from span_panel_api_schema_0 import SchemaZeroAdapter
+
+_REFERENCE_PANEL = "example-40t-001"
+
+
+def _snapshot(*children: SyntheticDevice) -> SpanPanelSnapshot:
+    return replay(tree(panel(), *children), PANEL_ID).build_snapshot()
+
+
+@pytest.mark.parametrize(("status", "connected"), [("OK", True), ("LOST", False), ("DEGRADED", False)])
+def test_the_feeding_circuits_status_is_the_link(status: str, connected: bool) -> None:
+    snapshot = _snapshot(*circuit_fed_battery("c-1", "battery-a", status=status))
+
+    assert snapshot.battery.connected is connected
+    assert snapshot.battery.feed_circuit_id == "c-1"
+    assert snapshot.circuits["c-1"].device_type == "circuit"
+
+
+def test_an_unpublished_status_leaves_the_link_unknown() -> None:
+    circuit, battery = circuit_fed_battery("c-1", "battery-a")
+    snapshot = _snapshot(without_values(circuit, "connection/feeds-device-status"), battery)
+
+    assert snapshot.battery.connected is None
+    assert snapshot.battery.feed_circuit_id == "c-1"
+
+
+def test_an_unpublished_feed_names_no_circuit() -> None:
+    circuit, battery = circuit_fed_battery("c-1", "battery-a")
+    snapshot = _snapshot(without_values(circuit, "connection/feeds-device-id", "connection/feeds-device-status"), battery)
+
+    assert (snapshot.battery.connected, snapshot.battery.feed_circuit_id) == (None, None)
+
+
+def test_a_battery_behind_the_upstream_lugs_keeps_their_link() -> None:
+    battery = replay(parent_child_tree(), _REFERENCE_PANEL).build_snapshot().battery
+
+    assert battery.connected is True
+    assert battery.feed_circuit_id is None
+
+
+def test_a_flat_panel_names_no_feeding_circuit() -> None:
+    snapshot = SchemaZeroAdapter(serial_number="sim-40t-001", schema=flat_schema(40)).build_snapshot()
+
+    assert snapshot.battery.feed_circuit_id is None