diff --git a/packages/schema-1/src/span_panel_api_schema_1/devices.py b/packages/schema-1/src/span_panel_api_schema_1/devices.py index 2b6645c..120b2bb 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/devices.py +++ b/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, ) diff --git a/packages/schema-1/src/span_panel_api_schema_1/snapshot.py b/packages/schema-1/src/span_panel_api_schema_1/snapshot.py index 4579528..87312f8 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/snapshot.py +++ b/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, ), diff --git a/src/span_panel_api/models.py b/src/span_panel_api/models.py index 7feb4db..317168a 100644 --- a/src/span_panel_api/models.py +++ b/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: diff --git a/tests/test_circuit_fed_battery_captures.py b/tests/test_circuit_fed_battery_captures.py new file mode 100644 index 0000000..b6b0f7a --- /dev/null +++ b/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) diff --git a/tests/test_schema_one_circuit_fed_battery.py b/tests/test_schema_one_circuit_fed_battery.py new file mode 100644 index 0000000..38438f4 --- /dev/null +++ b/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