diff --git a/CHANGELOG.md b/CHANGELOG.md index 57e8e48..28a12fb 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/packages/schema-1/CHANGELOG.md b/packages/schema-1/CHANGELOG.md index 7663f8f..c00c216 100644 --- a/packages/schema-1/CHANGELOG.md +++ b/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 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 092b09b..edf357c 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 @@ -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: diff --git a/src/span_panel_api/models.py b/src/span_panel_api/models.py index 9510064..b544d4c 100644 --- a/src/span_panel_api/models.py +++ b/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 `:` range the limit's declaration states in + `$format`, from the same resolution as `charge_current_limit_a`; firmware + r202639 declares `6:`. 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 `:`, + 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. diff --git a/tests/test_mqtt_homie.py b/tests/test_mqtt_homie.py index 3fe7452..0b0a19c 100644 --- a/tests/test_mqtt_homie.py +++ b/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" diff --git a/tests/test_schema_one_charge_limit.py b/tests/test_schema_one_charge_limit.py index d4c8f28..8a64c89 100644 --- a/tests/test_schema_one_charge_limit.py +++ b/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,9 +510,12 @@ 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: @@ -518,7 +523,7 @@ def _with_limit_format(fmt: str | None, device_id: str = EVSE) -> dict[str, dict 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 `:` 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 `:` 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 # ---------------------------------------------------------------------------