From 1f2fb86c78ff348669c2e4af09e62ea400c1ed8f Mon Sep 17 00:00:00 2001 From: yellowcooln <12516003+yellowcooln@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:33:23 -0400 Subject: [PATCH] feat(sensors): calculate Victron load power from voltage and current --- docs/victron_vedirect.md | 5 ++ repeater/sensors/victron_vedirect.py | 30 +++++++ tests/test_victron_load_power.py | 116 +++++++++++++++++++++++++++ 3 files changed, 151 insertions(+) create mode 100644 tests/test_victron_load_power.py diff --git a/docs/victron_vedirect.md b/docs/victron_vedirect.md index a48908b..d108997 100644 --- a/docs/victron_vedirect.md +++ b/docs/victron_vedirect.md @@ -82,6 +82,7 @@ Fields depend on model, firmware and configured auxiliary input. This table desc |---|---|---| | `V`, `I` | `battery_voltage_v`, `battery_current_a` | mV→V, mA→A; current sign preserved | | `VPV`, `PPV`, `IL` | `pv_voltage_v`, `pv_power_w`, `load_current_a` | mV→V, W, mA→A | +| Calculated from `V` and `IL` | `load_power_w` | Battery voltage (V) × load current (A) → W; **Load power (calculated)** in Measurements | | `CS`, `ERR`, `MPPT` | `charge_state_code`, `error_code`, `tracker_state_code` | Numeric codes; official descriptions in the corresponding card labels | | `LOAD`, `Relay`, `Alarm` | `load_output`, `relay`, `alarm` | Case-insensitive ON/OFF→boolean; other values unavailable | | `AR`, `OR`, `MON` | `alarm_reason`, `off_reason`, `monitor_mode_code` | Decimal alarm mask, hexadecimal off mask, signed monitor mode; decoded wording in card labels | @@ -97,6 +98,10 @@ Fields depend on model, firmware and configured auxiliary input. This table desc `H17/H18` mean discharged/charged energy for a BMV, but produced/consumed energy in DC monitor mode; their keys intentionally avoid assigning a meaning that requires knowing the mode. `MON` is available only on supporting firmware. `SOC` outside 0–1000 per-mille and `HSDS` outside 0–364 are treated as invalid. Negative unsigned status codes/masks are invalid; `MON` remains signed. +### Calculated load power + +VE.Direct does not provide a direct load-power field here. When both battery voltage and load current are valid in the current poll, the sensor calculates `load_power_w = battery_voltage_v * load_current_a` after merging that poll's validated blocks. This is an estimate using battery voltage, not a separately measured power value. Zero is a valid result and the current's sign is preserved. Missing or invalid inputs never reuse a previous value; a previously discovered calculated card becomes unavailable with the corresponding reason. Its descriptor uses `/derived/load_power_w` to identify a calculated local field, not a wire-protocol label. No UI or configuration change is required, and the existing V1 grouping remains intact. + ### Readable status values — sensor only Keep V1's numeric codes, measurement units and Measurements/Status grouping. Decode statuses inside the sensor and append the wording to the existing descriptor's `label`: for example **Charge state — Float** above the unchanged value **5**. This is a card-label change, not a new string value type or `display_value` contract. There are no extra flat label fields, duplicate cards, **Legacy values** section, UI dependency, or compatibility/version switch. Numeric measurements, codes and counts remain numbers; LOAD/Relay/Alarm remain booleans. diff --git a/repeater/sensors/victron_vedirect.py b/repeater/sensors/victron_vedirect.py index e13fb53..fae9646 100644 --- a/repeater/sensors/victron_vedirect.py +++ b/repeater/sensors/victron_vedirect.py @@ -303,6 +303,7 @@ class VictronVEDirectSensor(SensorBase): self._seen = set() self._data = {} + self._load_power_seen = False def _reading_metrics(self, ok): metrics = [] @@ -334,6 +335,27 @@ class VictronVEDirectSensor(SensorBase): "read_failed" if not ok else "invalid" if key in self._data else "missing" ) metrics.append(descriptor) + if self._load_power_seen: + available = ok and self._data.get("load_power_w") is not None + descriptor = { + "id": "/derived/load_power_w", + "source_path": "/derived/load_power_w", + "data_key": "load_power_w", + "label": "Load power (calculated)", + "unit": "W", + "kind": "number", + "category": "measurement", + "available": available, + } + if not available: + descriptor["reason"] = ( + "read_failed" + if not ok + else "invalid" + if "load_power_w" in self._data + else "missing" + ) + metrics.append(descriptor) return metrics def _read(self): @@ -348,6 +370,14 @@ class VictronVEDirectSensor(SensorBase): raise RuntimeError( "VE.Direct serial transport unavailable; check device and permissions" ) from None + # Derived after merging this poll's validated blocks, never across polls. + if "battery_voltage_v" in self._data and "load_current_a" in self._data: + self._load_power_seen = True + voltage = self._data["battery_voltage_v"] + current = self._data["load_current_a"] + self._data["load_power_w"] = ( + voltage * current if voltage is not None and current is not None else None + ) return self._data def _read_serial(self): diff --git a/tests/test_victron_load_power.py b/tests/test_victron_load_power.py new file mode 100644 index 0000000..7ca040c --- /dev/null +++ b/tests/test_victron_load_power.py @@ -0,0 +1,116 @@ +"""Synthetic load-power tests through the real sensor manager; no live hardware.""" + +from unittest.mock import MagicMock + +import pytest + +from tests.test_victron_vedirect import frame, manager + + +def serial_port(monkeypatch, payload): + import serial + + port = MagicMock() + port.__enter__.return_value = port + port.read.return_value = payload + monkeypatch.setattr(serial, "Serial", MagicMock(return_value=port)) + return port + + +@pytest.mark.parametrize( + "voltage,current", [(12800, 1500), (12800, 0), (0, 1500), (0, 0), (12800, -500)] +) +def test_load_power_is_calculated_in_watts_and_grouped_as_measurement( + monkeypatch, voltage, current +): + port = serial_port(monkeypatch, frame({"V": voltage, "IL": current})) + result = manager().read_all()[0] + assert result["ok"] + assert result["data"]["load_power_w"] == pytest.approx((voltage / 1000) * (current / 1000)) + descriptor = next( + metric for metric in result["metrics"] if metric["data_key"] == "load_power_w" + ) + assert descriptor == { + "id": "/derived/load_power_w", + "source_path": "/derived/load_power_w", + "data_key": "load_power_w", + "label": "Load power (calculated)", + "unit": "W", + "kind": "number", + "category": "measurement", + "available": True, + } + port.write.assert_not_called() + + +@pytest.mark.parametrize( + "fields", + [ + {}, + {"V": 12800}, + {"IL": 1500}, + {"V": "---", "IL": 1500}, + {"V": 12800, "IL": "---"}, + {"V": "NaN", "IL": 1500}, + {"V": 12800, "IL": "inf"}, + ], +) +def test_load_power_is_not_fabricated_without_valid_inputs(monkeypatch, fields): + serial_port(monkeypatch, frame(fields)) + result = manager().read_all()[0] + assert result["data"].get("load_power_w") is None + assert all( + not metric["available"] + for metric in result["metrics"] + if metric["data_key"] == "load_power_w" + ) + + +@pytest.mark.parametrize( + "fields,reason", + [ + ({"V": 12800}, "missing"), + ({"IL": 1500}, "missing"), + ({"V": 12800, "IL": "---"}, "invalid"), + ({"V": "---", "IL": 1500}, "invalid"), + (None, "read_failed"), + ], +) +def test_derived_card_becomes_unavailable_without_stale_power(monkeypatch, fields, reason): + port = serial_port(monkeypatch, frame({"V": 12800, "IL": 1500})) + sensor_manager = manager() + first = sensor_manager.read_all()[0] + assert first["data"]["load_power_w"] == pytest.approx(12.8 * 1.5) + if fields is None: + port.read.side_effect = OSError("synthetic disconnect") + else: + port.read.return_value = frame(fields) + second = sensor_manager.read_all()[0] + assert second["data"].get("load_power_w") is None + descriptor = next( + metric for metric in second["metrics"] if metric["data_key"] == "load_power_w" + ) + assert not descriptor["available"] + assert descriptor["reason"] == reason + + +def test_load_power_uses_separate_validated_blocks_of_same_poll(monkeypatch): + serial_port(monkeypatch, frame({"V": 12800}) + frame({"IL": 1500})) + result = manager().read_all()[0] + assert result["data"]["load_power_w"] == pytest.approx(12.8 * 1.5) + + +def test_load_power_never_combines_inputs_from_different_polls(monkeypatch): + port = serial_port(monkeypatch, frame({"V": 12800})) + sensor_manager = manager() + for fields in ({"V": 12800}, {"IL": 1500}, {"V": 12800}): + port.read.return_value = frame(fields) + result = sensor_manager.read_all()[0] + assert "load_power_w" not in result["data"] + assert all(metric["data_key"] != "load_power_w" for metric in result["metrics"]) + + +def test_load_power_does_not_use_value_invalidated_later_in_same_poll(monkeypatch): + serial_port(monkeypatch, frame({"V": 12800, "IL": 1500}) + frame({"IL": "---"})) + result = manager().read_all()[0] + assert result["data"]["load_power_w"] is None