diff --git a/packages/schema-1/src/span_panel_api_schema_1/circuits.py b/packages/schema-1/src/span_panel_api_schema_1/circuits.py index 73b6b88..3b857e4 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/circuits.py +++ b/packages/schema-1/src/span_panel_api_schema_1/circuits.py @@ -33,9 +33,10 @@ from typing import TYPE_CHECKING -from span_panel_api.models import SpanCircuitSnapshot +from span_panel_api.models import FEEDS_ROLES, FeedsRole, SpanCircuitSnapshot from span_panel_api_schema_1.const import ( NODE_BREAKER, + NODE_CONNECTION, NODE_INFO, NODE_LOAD_SHED, NODE_METER, @@ -45,6 +46,7 @@ PROP_ACTIVE_POWER, PROP_CURRENT, PROP_EXPORTED_ENERGY, + PROP_FEEDS_ROLE, PROP_IMPORTED_ENERGY, PROP_MANAGED, PROP_NAME, @@ -261,6 +263,12 @@ def relay_is_settable(device: DiscoveredDevice) -> bool: return _settable(device, NODE_SWITCH, PROP_RELAY) and _relay_controllable(device) +def _feeds_role(device: DiscoveredDevice) -> FeedsRole | None: + """The declared `connection/feeds-role`, or None when unpublished or outside the role set.""" + published = _text(device, NODE_CONNECTION, PROP_FEEDS_ROLE) + return next((role for role in FEEDS_ROLES if role == published), None) + + def build_circuit( device: DiscoveredDevice, device_type: str = "circuit", relative_position: str = "" ) -> SpanCircuitSnapshot: @@ -304,6 +312,7 @@ def build_circuit( is_never_backup=not priority_settable, device_type=device_type, relative_position=relative_position, + feeds_role=_feeds_role(device), # Unknown without a pole count: a meter outside the panel has no # breaker, and a breaker whose count has not arrived has not said. is_240v=None if poles is None else poles >= 2, diff --git a/packages/schema-1/src/span_panel_api_schema_1/field_metadata.py b/packages/schema-1/src/span_panel_api_schema_1/field_metadata.py index a0ebd9e..7e2105d 100644 --- a/packages/schema-1/src/span_panel_api_schema_1/field_metadata.py +++ b/packages/schema-1/src/span_panel_api_schema_1/field_metadata.py @@ -432,6 +432,7 @@ def _lookup( (TYPE_PANEL, NODE_SHED_FORECAST, "full-charge-total-time-remaining"), # --- Circuit topology and PCS membership -------------------------------- (TYPE_CIRCUIT, NODE_CONNECTION, "feeds-device-id"), + (TYPE_CIRCUIT, NODE_CONNECTION, "feeds-role"), (TYPE_CIRCUIT, NODE_PCS, "managed"), (TYPE_CIRCUIT, NODE_PCS, "priority"), # --- Lugs direction and the upstream device's link ----------------------- diff --git a/src/span_panel_api/__init__.py b/src/span_panel_api/__init__.py index c068239..b1ab08c 100644 --- a/src/span_panel_api/__init__.py +++ b/src/span_panel_api/__init__.py @@ -41,12 +41,14 @@ ADOPTION_IDENTITY_NODE, ADOPTION_TOPOLOGY_NODE, DISCOVERY_NAMESPACE, + FEEDS_ROLES, AdoptedDevice, AdoptedProperty, ControlTarget, DiscoveredMetadata, ExtensionProperty, ExtensionSubject, + FeedsRole, FieldMetadata, HomieSchemaTypes, PassphraseRotation, @@ -149,6 +151,10 @@ "SpanPVSnapshot", "SpanPanelSnapshot", "SpanPcsSnapshot", + # What a circuit's `connection/feeds-role` declares, read into + # `SpanCircuitSnapshot.feeds_role`. + "FEEDS_ROLES", + "FeedsRole", # Factory "create_span_client", # Detection diff --git a/src/span_panel_api/models.py b/src/span_panel_api/models.py index 332c2a7..cc72d60 100644 --- a/src/span_panel_api/models.py +++ b/src/span_panel_api/models.py @@ -11,11 +11,18 @@ from collections.abc import Mapping from dataclasses import dataclass, field +from typing import Final, Literal # Homie schema type: {type_name: {property_name: {attribute: value}}} # Values are heterogeneous JSON (str, int, bool, nested dicts). type HomieSchemaTypes = dict[str, dict[str, object]] +type FeedsRole = Literal["LOADS", "SUBPANEL", "SOLAR", "STORAGE", "GENERATOR", "MIXED", "UNUSED"] +"""What a circuit feeds, as its `connection/feeds-role` declares it.""" + +FEEDS_ROLES: Final[tuple[FeedsRole, ...]] = ("LOADS", "SUBPANEL", "SOLAR", "STORAGE", "GENERATOR", "MIXED", "UNUSED") +"""Every `FeedsRole`, for reading a wire value into one.""" + @dataclass(frozen=True, slots=True) class SpanCircuitSnapshot: @@ -88,6 +95,15 @@ class SpanCircuitSnapshot: has no relay, no breaker and no shed priority, and occupies no position. """ + feeds_role: FeedsRole | None = None + """What the circuit feeds, from its `connection/feeds-role`. + + `None` where the circuit declares no role, has not published it, or + publishes a value outside `FEEDS_ROLES`. Reported as declared and nothing + more: `device_type` stays as published, so a circuit whose role is solar is + still a circuit. The flat schema declares no roles. + """ + @dataclass(frozen=True, slots=True) class SpanPVSnapshot: diff --git a/tests/test_feeds_role_captures.py b/tests/test_feeds_role_captures.py new file mode 100644 index 0000000..10cebeb --- /dev/null +++ b/tests/test_feeds_role_captures.py @@ -0,0 +1,30 @@ +"""Every circuit's feeds role equals what its capture declares, over every reference capture.""" + +from __future__ import annotations + +import pytest + +from reference_payloads.captures import CAPTURES, capture_adapter, capture_snapshot, capture_tree +from span_panel_api import FEEDS_ROLES + + +@pytest.mark.parametrize("stem", CAPTURES) +def test_each_circuit_reports_its_declared_feeds_role(stem: str) -> None: + tree, snapshot = capture_tree(stem), capture_snapshot(stem) + for circuit_id in tree.circuits(): + published = tree.value(circuit_id, "connection/feeds-role") + expected = published if published in FEEDS_ROLES else None + assert snapshot.circuits[circuit_id].feeds_role == expected, circuit_id + + +def test_the_captures_declare_feeds_roles() -> None: + assert any( + capture_tree(stem).declares(circuit_id, "connection/feeds-role") + for stem in CAPTURES + for circuit_id in capture_tree(stem).circuits() + ) + + +@pytest.mark.parametrize("stem", CAPTURES) +def test_the_feeds_role_a_circuit_declares_is_not_reported_as_discovered(stem: str) -> None: + assert "discovered.circuit/connection/feeds-role" not in capture_adapter(stem).build_field_metadata() diff --git a/tests/test_public_api_unchanged.py b/tests/test_public_api_unchanged.py index e5c53af..3e527bc 100644 --- a/tests/test_public_api_unchanged.py +++ b/tests/test_public_api_unchanged.py @@ -83,6 +83,11 @@ # `None` on every existing consumer's data and nothing that reads the # snapshot today changes. "SpanPcsSnapshot", + # Added 2026-10-09: the closed set a circuit's `connection/feeds-role` + # declares, read into `SpanCircuitSnapshot.feeds_role`. Purely additive -- + # the field defaults to `None`, and the flat schema declares no roles. + "FEEDS_ROLES", + "FeedsRole", # Factory "create_span_client", # Detection diff --git a/tests/test_schema_one_conformance.py b/tests/test_schema_one_conformance.py index f11fd78..d2de2b6 100644 --- a/tests/test_schema_one_conformance.py +++ b/tests/test_schema_one_conformance.py @@ -282,6 +282,12 @@ def _emitter_declared() -> set[tuple[str, str]]: "in test_schema_one_charge_limit.py drives it from a synthetic description, which is " "evidence of a parser and not of a producer -- which is what this entry records." ), + ("connection", "feeds-role"): ( + "what a circuit feeds when that is not itself a device on the tree. The emitter's " + "reference tree declares no role on any circuit; the reference captures declare it on " + "their circuits and leave it unvalued there, so the values are driven by synthetic trees in " + "test_schema_one_feeds_role.py." + ), } diff --git a/tests/test_schema_one_discovery.py b/tests/test_schema_one_discovery.py index d4a76c0..2d6b754 100644 --- a/tests/test_schema_one_discovery.py +++ b/tests/test_schema_one_discovery.py @@ -266,13 +266,50 @@ def test_an_off_snapshot_route_that_became_observable_must_be_retired() -> None: ) +_FEEDS_ROLE: Declaration = (TYPE_CIRCUIT, "connection", "feeds-role") +_ABSENT_FROM_THE_CAPTURE: frozenset[Declaration] = frozenset({_FEEDS_ROLE}) +"""Addressed entries the capture never declares, each proved on a synthetic declaration below.""" + +_A_LOAD_CIRCUIT = "0ab966b95f92a6a51ec548485aa85f54" +"""A circuit that feeds no device, so nothing else labels it.""" + + def test_no_addressed_entry_has_gone_stale() -> None: """A table entry outlives its declaration silently; the file only ever grows.""" declared = _declared() - stale = [entry for entry in (*_CONSUMED_WITHOUT_A_ROW, *_CONSUMED_OFF_SNAPSHOT) if entry not in declared] + stale = [ + entry + for entry in (*_CONSUMED_WITHOUT_A_ROW, *_CONSUMED_OFF_SNAPSHOT) + if entry not in declared and entry not in _ABSENT_FROM_THE_CAPTURE + ] assert not stale, f"addressed-property entries the reference tree no longer declares:\n{_rendered(stale)}" +def test_an_entry_excused_from_the_capture_is_still_absent_from_it() -> None: + """Once the capture declares it, the experiment above can prove it directly.""" + assert not (_ABSENT_FROM_THE_CAPTURE & _declared()) + + +def test_feeds_role_moves_the_snapshot_on_a_circuit_that_declares_it() -> None: + """The experiment above, run on a circuit given the declaration the capture lacks.""" + _type, node, prop = _FEEDS_ROLE + tree = _tree() + description = json.loads(tree[_A_LOAD_CIRCUIT]["$description"]) + description["nodes"][node]["properties"][prop] = { + "datatype": "enum", + "format": "LOADS,SUBPANEL,SOLAR,STORAGE,GENERATOR,MIXED,UNUSED", + } + tree[_A_LOAD_CIRCUIT]["$description"] = json.dumps(description) + tree[_A_LOAD_CIRCUIT][f"{node}/{prop}"] = "LOADS" + baseline = _snapshot_fields(_snapshot(tree)) + tree[_A_LOAD_CIRCUIT][f"{node}/{prop}"] = "SOLAR" + after = _snapshot_fields(_snapshot(tree)) + + moved = {path for path, value in after.items() if baseline.get(path) != value} + assert moved == {f"circuits@{_A_LOAD_CIRCUIT}.feeds_role"} + assert _FEEDS_ROLE in _CONSUMED_WITHOUT_A_ROW + + def test_no_addressed_entry_duplicates_a_metadata_row() -> None: """The four tables partition the addressed set; they do not overlap. diff --git a/tests/test_schema_one_feeds_role.py b/tests/test_schema_one_feeds_role.py new file mode 100644 index 0000000..d226df2 --- /dev/null +++ b/tests/test_schema_one_feeds_role.py @@ -0,0 +1,65 @@ +"""Each circuit's declared feeds role, as `SpanCircuitSnapshot.feeds_role`. + +`connection/feeds-role` says what a circuit feeds when that is not itself a +device on the tree. The role is reported as declared and changes nothing else: +a circuit whose role is solar is still a circuit, and no inverter is made up +for it. A value outside the role set, or none at all, reads as `None`. +""" + +from __future__ import annotations + +from typing import get_args + +import pytest + +from conftest import flat_schema +from reference_payloads.schema_one import replay +from reference_payloads.synthetic_trees import PANEL_ID, hosted_circuit, panel, tree, without_values +from span_panel_api import FEEDS_ROLES, FeedsRole, SpanPanelSnapshot +from span_panel_api_schema_0 import SchemaZeroAdapter + + +def _snapshot(role: str | None) -> SpanPanelSnapshot: + return replay(tree(panel(), hosted_circuit("c-1", (1,), feeds_role=role)), PANEL_ID).build_snapshot() + + +def test_the_role_set_is_the_literal() -> None: + assert set(FEEDS_ROLES) == set(get_args(FeedsRole.__value__)) + assert len(FEEDS_ROLES) == len(set(FEEDS_ROLES)) + + +@pytest.mark.parametrize("role", FEEDS_ROLES) +def test_a_declared_role_is_reported_as_published(role: FeedsRole) -> None: + assert _snapshot(role).circuits["c-1"].feeds_role == role + + +@pytest.mark.parametrize("role", ["NOT_A_ROLE", "solar", ""]) +def test_a_value_outside_the_role_set_reads_none(role: str) -> None: + assert _snapshot(role).circuits["c-1"].feeds_role is None + + +def test_an_undeclared_role_reads_none() -> None: + assert _snapshot(None).circuits["c-1"].feeds_role is None + + +def test_a_declared_role_not_yet_published_reads_none() -> None: + circuit = without_values(hosted_circuit("c-1", (1,), feeds_role="LOADS"), "connection/feeds-role") + snapshot = replay(tree(panel(), circuit), PANEL_ID).build_snapshot() + + assert snapshot.circuits["c-1"].feeds_role is None + + +@pytest.mark.spec_only +def test_a_solar_role_keeps_the_circuit_a_circuit() -> None: + """The role travels in `feeds_role`; `device_type` stays as published and no inverter appears.""" + snapshot = _snapshot("SOLAR") + + assert snapshot.circuits["c-1"].feeds_role == "SOLAR" + assert snapshot.circuits["c-1"].device_type == "circuit" + assert snapshot.pv_inverters == {} + + +def test_a_flat_panel_reports_no_role() -> None: + snapshot = SchemaZeroAdapter(serial_number="sim-40t-001", schema=flat_schema(40)).build_snapshot() + + assert all(circuit.feeds_role is None for circuit in snapshot.circuits.values())