Report the charge-current limit range a SPAN Drive declares

CHANGELOG.md

@@ -16,6 +16,7 @@ Snapshot fields that held a fabricated constant can now be unknown, and with `sp
 - **spanos3/r202639/03 batch 2: reads more of the data this firmware publishes.**
 - **`SpanMqttClient.masked_tree()` returns the panel's device tree with identifiers and site details masked**, for a consumer's diagnostics; `None` on a flat panel.
 - **`V2StatusInfo.hardware_version` reports the hardware version string from `GET /api/v2/status`.**
+- **`SpanEvseSnapshot.charge_current_limit_minimum_a` and `charge_current_limit_maximum_a` report the range a SPAN Drive declares for its charge-current limit**, `None` where undeclared.
 
 ### Changed
 

packages/schema-1/CHANGELOG.md

@@ -17,6 +17,7 @@ Requires `span-panel-api` **3.8.0 or newer**, whose snapshot fields accept the u
 
 - **spanos3/r202639/03 batch 2: reads more of the data this firmware publishes.**
 - **`SchemaOneAdapter` exports the tree it holds, masked**, for `SpanMqttClient.masked_tree()`.
+- **Each SPAN Drive's declared charge-current limit range is reported.**
 
 ### Changed
 

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

@@ -54,7 +54,7 @@
 from typing import TYPE_CHECKING
 
 from span_panel_api.models import SpanBatterySnapshot, SpanEvseSnapshot, SpanMidSnapshot, SpanPVSnapshot
-from span_panel_api_schema_1.charge_limit import ChargeLimitProperty, ChargeLimitSurface, resolve_charge_limit
+from span_panel_api_schema_1.charge_limit import ChargeLimitProperty, ChargeLimitSurface, IntegerBounds, resolve_charge_limit
 from span_panel_api_schema_1.const import (
     NODE_CONNECTION,
     NODE_GRID,
@@ -369,9 +369,12 @@ def build_evse(
     The charge-current pair is read through `resolve_charge_limit` rather than
     from named constants, because which node and properties carry it is a
     question only this charger's `$description` can answer. See
-    `span_panel_api_schema_1.charge_limit`.
+    `span_panel_api_schema_1.charge_limit`. The limit's declared range comes from
+    the same resolution, so the range a control offers and the range
+    `SchemaOneAdapter.evse_charge_limit_payload` enforces cannot disagree.
     """
     limit = resolve_charge_limit(evse)
+    minimum, maximum = _limit_bounds(limit)
     return SpanEvseSnapshot(
         node_id=node_id,
         feed_circuit_id=feeds.get(evse.device_id, ""),
@@ -394,9 +397,23 @@ def build_evse(
         charge_current_ceiling_a=_limit_value(evse, limit, limit.ceiling) if limit else None,
         charge_current_limit_target_a=_limit_target(evse, limit),
         charge_current_limit_settable=limit is not None and limit.limit is not None and limit.limit.settable,
+        charge_current_limit_minimum_a=minimum,
+        charge_current_limit_maximum_a=maximum,
     )
 
 
+def _limit_bounds(surface: ChargeLimitSurface | None) -> IntegerBounds:
+    """The range the resolved limit declares, `None` on a side it leaves open or cannot be read.
+
+    `ChargeLimitProperty.bounds` is already None for a format
+    `declared_integer_bounds` will not interpret, so a declaration that bounds
+    nothing on the write path reports no range here either.
+    """
+    if surface is None or surface.limit is None or surface.limit.bounds is None:
+        return None, None
+    return surface.limit.bounds
+
+
 def _limit_value(evse: DiscoveredDevice, surface: ChargeLimitSurface, declaration: ChargeLimitProperty | None) -> int | None:
     """One half of the resolved charge-limit pair, or None where it is not declared."""
     if declaration is None:

src/span_panel_api/models.py

@@ -465,6 +465,36 @@ class SpanEvseSnapshot:
     set topic when it is not.
     """
 
+    charge_current_limit_minimum_a: int | None = None
+    """The lowest charge-current limit the charger declares a user may set, in amps. v1.0 only.
+
+    The minimum of the `<min>:<max>` range the limit's declaration states in
+    `$format`, from the same resolution as `charge_current_limit_a`; firmware
+    r202639 declares `6:<commissioned maximum>`. A write below it is refused
+    rather than published, so a control offers nothing under it.
+
+    A declaration, not a reading: it is reported as soon as the charger
+    describes itself, and whether or not the limit is settable, which
+    `charge_current_limit_settable` answers. `None` where the charger declares
+    no limit, no `$format`, or the format leaves this side open, and where the
+    format cannot be read as an integer range (not `integer`, not `<min>:<max>`,
+    a bound that is not a plain integer, or a minimum above the maximum). `None`
+    on the flat schema, which declares no range.
+    """
+
+    charge_current_limit_maximum_a: int | None = None
+    """The highest charge-current limit the charger declares a user may set, in amps. v1.0 only.
+
+    The maximum of the same declared range, `None` on the same terms. A write
+    above it is refused rather than published.
+
+    **Not `charge_current_ceiling_a`.** That is the commissioned ceiling, a
+    published value; this is the declared range, part of the limit's
+    description. Every r202639 capture declares a maximum equal to the ceiling
+    it publishes, but they are separate facts, and a write must satisfy both: a
+    control's upper bound is the lower of the two where both are present.
+    """
+
     lock_control: ControlTarget | None = None
     """Where the charger's connector lock is commanded, and which property reports it. v1.0 only.
 

tests/test_mqtt_homie.py

@@ -1434,6 +1434,23 @@ def test_evse_metadata_parsed(self):
         assert evse.serial_number == "SN12345"
         assert evse.software_version == "2.1.0"
 
+    def test_evse_reports_no_charge_limit_range(self):
+        """Flat EVSE types declare no charge-current limit, so no range: None, never a guessed one."""
+        circuit_uuid = "aabbccdd-1122-3344-5566-778899001122"
+        acc, consumer = _build_ready_consumer(
+            {
+                "core": {"type": TYPE_CORE},
+                circuit_uuid: {"type": TYPE_CIRCUIT},
+                "evse-0": {"type": TYPE_EVSE},
+            }
+        )
+        acc.handle_message(f"{PREFIX}/evse-0/feed", circuit_uuid)
+        acc.handle_message(f"{PREFIX}/evse-0/advertised-current", "32.0")
+
+        evse = consumer.build_snapshot().evse["evse-0"]
+        assert evse.charge_current_limit_minimum_a is None
+        assert evse.charge_current_limit_maximum_a is None
+
     def test_evse_multiple_devices(self):
         """Two EVSE nodes produce two snapshot entries."""
         circ_a = "aaaaaaaa-1111-2222-3333-444444444444"

tests/test_schema_one_charge_limit.py

@@ -29,6 +29,7 @@
 
 from ebus_sdk.homie import DiscoveredDevice
 
+from reference_payloads.captures import CAPTURES, capture_snapshot, capture_tree
 from reference_payloads.schema_one import device_from_topics, parent_child_tree
 from span_panel_api.exceptions import SpanPanelServerError
 from span_panel_api.models import V2HomieSchema
@@ -42,6 +43,7 @@
 PANEL = "example-40t-001"
 EVSE = "evse"
 EVSE_2 = "evse-2"
+_EVSE_TYPE = "energy.ebus.device.evse"
 
 CEILING_TOPIC = "config/max-charge-current"
 LIMIT_TOPIC = "config/user-max-charge-current"
@@ -508,17 +510,20 @@ def _declared_minimum(device_id: str) -> int:
     return int(declared["user-max-charge-current"]["format"].split(":")[0])
 
 
-def _with_limit_format(fmt: str | None, device_id: str = EVSE) -> dict[str, dict[str, str]]:
-    """The capture with the limit's declared `$format` replaced, or removed where `None`."""
-    topics = dict(_TREE[device_id])
+def _with_limit_format(
+    fmt: str | None, device_id: str = EVSE, base: dict[str, dict[str, str]] | None = None
+) -> dict[str, dict[str, str]]:
+    """The capture (or `base`, a tree built from it) with the limit's declared `$format` replaced, or removed where `None`."""
+    source = _TREE if base is None else base
+    topics = dict(source[device_id])
     description = json.loads(topics["$description"])
     declaration = description["nodes"]["config"]["properties"]["user-max-charge-current"]
     if fmt is None:
         declaration.pop("format")
     else:
         declaration["format"] = fmt
     topics["$description"] = json.dumps(description)
-    tree = {other: dict(values) for other, values in _TREE.items()}
+    tree = {other: dict(values) for other, values in source.items()}
     tree[device_id] = topics
     return tree
 
@@ -655,6 +660,144 @@ def test_a_declared_maximum_bounds_a_charger_that_publishes_no_ceiling() -> None
     assert adapter.evse_charge_limit_payload(key, maximum + 1) is None
 
 
+# ---------------------------------------------------------------------------
+# The declared range — what a control may offer, reported before any write
+# ---------------------------------------------------------------------------
+
+
+def _range(evse: object) -> tuple[int | None, int | None]:
+    return evse.charge_current_limit_minimum_a, evse.charge_current_limit_maximum_a
+
+
+def _declared_range(device_id: str) -> tuple[int, int]:
+    """The `<min>:<max>` the capture's limit declares, read from its `$description`."""
+    declared = json.loads(_TREE[device_id]["$description"])["nodes"]["config"]["properties"]
+    low, high = declared["user-max-charge-current"]["format"].split(":")
+    return int(low), int(high)
+
+
+def test_each_charger_reports_the_range_its_limit_declares() -> None:
+    """The capture's own declarations, read the way the broker replays them.
+
+    Each charger declares `6:32`; a control bounded by anything else offers a
+    value the write path then refuses.
+    """
+    adapter = _adapter()
+    snapshot = adapter.build_snapshot()
+
+    for device_id in (EVSE, EVSE_2):
+        assert _declared_range(device_id) == (6, 32)
+        assert _range(snapshot.evse[_key(adapter, device_id)]) == _declared_range(device_id)
+
+
+def test_the_range_is_not_the_ceiling() -> None:
+    """The declared maximum and the commissioned ceiling are separate readings.
+
+    The ceiling is lowered below the declared maximum so a mapper that reported
+    the ceiling as the range, or the range as the ceiling, would show here.
+    """
+    lowered = _declared_range(EVSE)[1] - 8
+    evse = _snapshot_evse(_tree(**{EVSE: {CEILING_TOPIC: str(lowered)}}), EVSE)
+
+    assert evse.charge_current_ceiling_a == lowered
+    assert _range(evse) == _declared_range(EVSE)
+
+
+def test_each_charger_reports_its_own_range() -> None:
+    """Both chargers declare the same range in the capture, so they are made to differ first."""
+    tree = _with_limit_format("8:16", EVSE_2, base=_with_limit_format("6:24", EVSE))
+    adapter = _adapter(tree)
+    snapshot = adapter.build_snapshot()
+
+    assert _range(snapshot.evse[_key(adapter, EVSE)]) == (6, 24)
+    assert _range(snapshot.evse[_key(adapter, EVSE_2)]) == (8, 16)
+
+
+def test_no_declared_format_reports_no_range() -> None:
+    assert _range(_snapshot_evse(_with_limit_format(None), EVSE)) == (None, None)
+
+
+@pytest.mark.parametrize(
+    "fmt", ["garbage", "6", "a:b", "6:32:1", "6.5:32", "UNLOCKED,LOCKED", "1_0:32", "+6:32", "6:+32", "٦:32", "32:6"]
+)
+def test_an_uninterpretable_or_inverted_format_reports_no_range(fmt: str) -> None:
+    """The same formats the write path refuses to read bound nothing here either."""
+    assert _range(_snapshot_evse(_with_limit_format(fmt), EVSE)) == (None, None)
+
+
+def test_an_open_side_is_none_on_that_side_only() -> None:
+    assert _range(_snapshot_evse(_with_limit_format(":20"), EVSE)) == (None, 20)
+    assert _range(_snapshot_evse(_with_limit_format("6:"), EVSE)) == (6, None)
+
+
+def test_a_float_limit_reports_no_range() -> None:
+    """Only an `integer` declaration's range is read, as on the write path."""
+    topics = dict(_TREE[EVSE])
+    description = json.loads(topics["$description"])
+    description["nodes"]["config"]["properties"]["user-max-charge-current"]["datatype"] = "float"
+    topics["$description"] = json.dumps(description)
+    tree = {other: dict(values) for other, values in _TREE.items()}
+    tree[EVSE] = topics
+
+    assert _range(_snapshot_evse(tree, EVSE)) == (None, None)
+
+
+def test_a_charger_with_no_charge_limit_node_reports_no_range() -> None:
+    assert _range(_snapshot_evse(_without_node(EVSE), EVSE)) == (None, None)
+
+
+def test_the_range_is_reported_whether_or_not_the_limit_is_settable() -> None:
+    """A declaration, reported as declared; `charge_current_limit_settable` is the separate answer to whether to offer it.
+
+    The same rule `lock_state_options` follows for a lock the charger does not
+    accept writes to.
+    """
+    evse = _snapshot_evse(_without_settable(EVSE), EVSE)
+
+    assert evse.charge_current_limit_settable is False
+    assert _range(evse) == _declared_range(EVSE)
+
+
+def test_the_catalogued_spelling_reports_the_same_range() -> None:
+    catalogued = _snapshot_evse(_renamed_to_catalog(EVSE), EVSE)
+
+    assert _range(catalogued) == _declared_range(EVSE)
+
+
+def test_the_range_does_not_wait_for_a_value() -> None:
+    """The range is in the declaration, so it is reported before the charger publishes either half."""
+    evse = _snapshot_evse(_tree(**{EVSE: {LIMIT_TOPIC: None, CEILING_TOPIC: None}}), EVSE)
+
+    assert evse.charge_current_ceiling_a is None
+    assert _range(evse) == _declared_range(EVSE)
+
+
+def _captured_ranges(stem: str) -> list[tuple[int, int]]:
+    """The `<min>:<max>` each charger in a reference capture declares for its limit."""
+    tree = capture_tree(stem)
+    declared = []
+    for device_id in tree.devices:
+        if tree.device_type(device_id) != _EVSE_TYPE:
+            continue
+        properties = json.loads(tree.devices[device_id].description)["nodes"]["config"]["properties"]
+        low, high = properties["user-max-charge-current"]["format"].split(":")
+        declared.append((int(low), int(high)))
+    return sorted(declared)
+
+
+@pytest.mark.parametrize("stem", CAPTURES)
+def test_each_captured_charger_reports_the_range_it_declares(stem: str) -> None:
+    """Every reference capture, against its own declarations rather than the reference tree's."""
+    snapshot = capture_snapshot(stem)
+
+    assert sorted(_range(charger) for charger in snapshot.evse.values()) == _captured_ranges(stem)
+
+
+def test_the_captures_hold_a_declared_range() -> None:
+    """Without one, the test above compares two empty lists and proves nothing."""
+    assert any(_captured_ranges(stem) for stem in CAPTURES)
+
+
 # ---------------------------------------------------------------------------
 # The transport — what actually reaches the wire
 # ---------------------------------------------------------------------------