r202639: circuit feeds role

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,

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 -----------------------

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

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:

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()

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

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."
+    ),
 }
 
 

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.
 

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())