feat(sensors): calculate Victron load power from voltage and current

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.

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

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