feat(load): add typed measurement properties for WHO 3 dimensions 1-4 from firmware register maps (#104)

OWNd/message/__init__.py

@@ -145,6 +145,11 @@
 )
 from .load import (
     LOAD_CONTROL_UNIT,
+    LOAD_DIMENSION_ALL,
+    LOAD_DIMENSION_CURRENT,
+    LOAD_DIMENSION_POWER,
+    LOAD_DIMENSION_POWER_FACTOR,
+    LOAD_DIMENSION_VOLTAGE,
     LOAD_MEASUREMENT_DIMENSIONS,
     LOAD_PRIORITIES,
     LOAD_STATE_DISABLED,
@@ -196,6 +201,11 @@
     "LOAD_CONTROL_UNIT",
     "LOAD_PRIORITIES",
     "LOAD_MEASUREMENT_DIMENSIONS",
+    "LOAD_DIMENSION_ALL",
+    "LOAD_DIMENSION_POWER",
+    "LOAD_DIMENSION_VOLTAGE",
+    "LOAD_DIMENSION_CURRENT",
+    "LOAD_DIMENSION_POWER_FACTOR",
     "OWNLoadEvent",
     "OWNLoadCommand",
     # Heating (WHO 4)

OWNd/message/load.py

@@ -37,25 +37,54 @@
 LOAD_PRIORITIES = range(1, 9)
 LOAD_MEASUREMENT_DIMENSIONS = range(0, 5)
 
+LOAD_DIMENSION_ALL = 0
+LOAD_DIMENSION_POWER = 1
+LOAD_DIMENSION_VOLTAGE = 2
+LOAD_DIMENSION_CURRENT = 3
+LOAD_DIMENSION_POWER_FACTOR = 4
+
 _STATE_TEXT = {
     LOAD_STATE_DISABLED: "disabled",
     LOAD_STATE_ENABLED: "enabled",
     LOAD_STATE_FORCED: "forced",
     LOAD_STATE_FORCING_REMOVED: "no longer forced",
 }
 
-# Specification names only; no device has confirmed them (see module docstring).
-_SPEC_DIMENSION_NAMES = {
-    0: "all measurements",
-    1: "voltage",
-    2: "current",
-    3: "power",
-    4: "energy",
+# Dimension mappings verified against central unit internal register polling
+# in bt_energia (findings/MyHomeServer1/energy-ts10.md):
+# - Dimension 1: Instantaneous Power (W) -> register $03AA001613
+# - Dimension 2: Voltage (V) -> register $03AA001612
+# - Dimension 3: Current (A) -> registers $03AA001610 & $03AA001611
+# - Dimension 4: Power Factor (cos phi) -> registers $03AA001606 through $03AA001609
+_DIMENSION_NAMES = {
+    LOAD_DIMENSION_ALL: "all measurements",
+    LOAD_DIMENSION_POWER: "instantaneous power",
+    LOAD_DIMENSION_VOLTAGE: "voltage",
+    LOAD_DIMENSION_CURRENT: "current",
+    LOAD_DIMENSION_POWER_FACTOR: "power factor",
 }
 
 _PRIORITY_WHERE = re.compile(r"^#([1-8])$")
 
 
+def _safe_int(value: str | None) -> int | None:
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _safe_float(value: str | None) -> float | None:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
 def _priority_of(where: str | None) -> int | None:
     """Priority 1-8 for a ``#N`` WHERE; the leading ``#`` is part of the address."""
     match = _PRIORITY_WHERE.match(where or "")
@@ -77,6 +106,10 @@ def __init__(self, data: str) -> None:
         self._priority = _priority_of(self._where)
         self._state: int | None = None
         self._measurement_values: list[str] = []
+        self._power: int | None = None
+        self._voltage: int | None = None
+        self._current: float | None = None
+        self._power_factor: float | None = None
 
         if self._priority is not None:
             _target = f"Load priority {self._priority} is"
@@ -94,12 +127,50 @@ def __init__(self, data: str) -> None:
             and self._dimension is not None
         ):
             self._measurement_values = list(self._dimension_value)
-            _name = _SPEC_DIMENSION_NAMES.get(self._dimension)
-            _label = f" ({_name} per specification)" if _name else ""
-            self._human_readable_log = (
-                f"Load control unit dimension {self._dimension}{_label}: "
-                f"raw values {self._measurement_values}."
-            )
+            if self._dimension == LOAD_DIMENSION_POWER and self._dimension_value:
+                self._power = _safe_int(self._dimension_value[0])
+                self._human_readable_log = (
+                    f"Load control unit power draw: {self._power} W."
+                )
+            elif self._dimension == LOAD_DIMENSION_VOLTAGE and self._dimension_value:
+                self._voltage = _safe_int(self._dimension_value[0])
+                self._human_readable_log = (
+                    f"Load control unit voltage: {self._voltage} V."
+                )
+            elif self._dimension == LOAD_DIMENSION_CURRENT and self._dimension_value:
+                self._current = _safe_float(self._dimension_value[0])
+                self._human_readable_log = (
+                    f"Load control unit current: {self._current} A."
+                )
+            elif (
+                self._dimension == LOAD_DIMENSION_POWER_FACTOR
+                and self._dimension_value
+            ):
+                self._power_factor = _safe_float(self._dimension_value[0])
+                self._human_readable_log = (
+                    f"Load control unit power factor: {self._power_factor}."
+                )
+            elif self._dimension == LOAD_DIMENSION_ALL:
+                if len(self._dimension_value) >= 1:
+                    self._power = _safe_int(self._dimension_value[0])
+                if len(self._dimension_value) >= 2:
+                    self._voltage = _safe_int(self._dimension_value[1])
+                if len(self._dimension_value) >= 3:
+                    self._current = _safe_float(self._dimension_value[2])
+                if len(self._dimension_value) >= 4:
+                    self._power_factor = _safe_float(self._dimension_value[3])
+                self._human_readable_log = (
+                    f"Load control unit all measurements: "
+                    f"power={self._power} W, voltage={self._voltage} V, "
+                    f"current={self._current} A, power factor={self._power_factor}."
+                )
+            else:
+                _name = _DIMENSION_NAMES.get(self._dimension)
+                _label = f" ({_name})" if _name else ""
+                self._human_readable_log = (
+                    f"Load control unit dimension {self._dimension}{_label}: "
+                    f"raw values {self._measurement_values}."
+                )
 
     @property
     def priority(self) -> int | None:
@@ -125,6 +196,26 @@ def measurement_values(self) -> list[str]:
         """Raw values of a control-unit measurement reply, unscaled and unitless."""
         return self._measurement_values
 
+    @property
+    def power(self) -> int | None:
+        """Instantaneous power draw in W (dimension 1), None if not reported."""
+        return self._power
+
+    @property
+    def voltage(self) -> int | None:
+        """Voltage in V (dimension 2), None if not reported."""
+        return self._voltage
+
+    @property
+    def current(self) -> float | None:
+        """Current in A (dimension 3), None if not reported."""
+        return self._current
+
+    @property
+    def power_factor(self) -> float | None:
+        """Power factor cos phi (dimension 4), None if not reported."""
+        return self._power_factor
+
 
 class OWNLoadCommand(OWNCommand):
     """WHO 3 commands and requests.

README.md

@@ -78,7 +78,7 @@ OWNd parses OpenWebNet frames and dispatches typed commands and events across th
 | **0** | Scenarios | Basic scenario execution (1–32) from control panels, scenario module programming and memory lock states | `OWNScenarioEvent` |
 | **1** | Lighting | On/off switching, dimming level (0–100%), DALI Tunable White (Dimension 14, 2000K–6535K / mireds), status queries | `OWNLightingCommand`, `OWNLightingEvent` |
 | **2** | Automation | Shutters, blinds, motorized curtains, tilt angles, short & full replies | `OWNAutomationCommand`, `OWNAutomationEvent` |
-| **3** | Load Control | Priority forcing (`*3*2*#N##`, firmware-verified), priority states 0–3, priority status and central-unit measurement requests; measurement replies kept raw until a plant capture confirms their meaning | `OWNLoadCommand`, `OWNLoadEvent` |
+| **3** | Load Control | Priority forcing (`*3*2*#N##`, firmware-verified), priority states 0–3, priority status and central-unit measurement requests with typed power, voltage, current, and power factor properties from firmware register mappings | `OWNLoadCommand`, `OWNLoadEvent` |
 | **4** | Thermoregulation / Climate | Multi-zone temperature readouts, target adjustments, HVAC modes (Heat/Cool/Auto/Off), local offsets, fan coil speeds, valve states, Central Unit 3550/4695 master coordination | `OWNHeatingCommand`, `OWNHeatingEvent` |
 | **5** | Burglar Alarm | Zone status, system arming / disarming states | `OWNAlarmCommand`, `OWNAlarmEvent` |
 | **6** | Door Entry & Lock | Door entry unit and electric lock status, door open/close events, lock release commands | `OWNDoorEntryCommand`, `OWNDoorEntryEvent` |
@@ -337,10 +337,12 @@ OWNd maintains an automated test suite with strict **100.0% line coverage** veri
 | [`OWNd/message/automation.py`](OWNd/message/automation.py) | **100%** | WHO 2: Automation and motorized shutter events and commands |
 | [`OWNd/message/base.py`](OWNd/message/base.py) | **100%** | Base message, signaling, and protocol dispatcher registries |
 | [`OWNd/message/cen.py`](OWNd/message/cen.py) | **100%** | WHO 15 & WHO 25: CEN, CEN+, and dry contact control events/commands |
+| [`OWNd/message/door_entry.py`](OWNd/message/door_entry.py) | **100%** | Core OWNd library component |
 | [`OWNd/message/energy.py`](OWNd/message/energy.py) | **100%** | WHO 18: Energy management, power, and diagnostic pulse events/commands |
 | [`OWNd/message/gateway.py`](OWNd/message/gateway.py) | **100%** | WHO 13: Gateway management, clock/date configuration, and model telemetry |
 | [`OWNd/message/heating.py`](OWNd/message/heating.py) | **100%** | WHO 4: Thermoregulation, heating/cooling zone control, and temperature decoders |
 | [`OWNd/message/lighting.py`](OWNd/message/lighting.py) | **100%** | WHO 1: Lighting controls, dimming, RGB/HSV color, and motion/PIR sensors |
+| [`OWNd/message/load.py`](OWNd/message/load.py) | **100%** | WHO 3: Load control, priority forcing, and central-unit measurement events/commands |
 | [`OWNd/message/scenario.py`](OWNd/message/scenario.py) | **100%** | WHO 0, WHO 9, WHO 17: Scenario activation, auxiliary commands, and MH200/MH202 scenes |
 | [`OWNd/message/sound.py`](OWNd/message/sound.py) | **100%** | WHO 16: Sound system and audio/video matrix routing (WHO 22 is not implemented) |
 | [`OWNd/profiles.py`](OWNd/profiles.py) | **100%** | Declarative hardware gateway models across 14 canonical hardware profiles |

scripts/sync_documentation.py

@@ -107,7 +107,7 @@ class WhoCatalogDefinition:
     WhoCatalogDefinition(
         who="**3**",
         name="Load Control",
-        description="Priority forcing (`*3*2*#N##`, firmware-verified), priority states 0–3, priority status and central-unit measurement requests; measurement replies kept raw until a plant capture confirms their meaning",
+        description="Priority forcing (`*3*2*#N##`, firmware-verified), priority states 0–3, priority status and central-unit measurement requests with typed power, voltage, current, and power factor properties from firmware register mappings",
         dispatch_who_keys=(3,),
     ),
     WhoCatalogDefinition(

tests/test_who3_load_control.py

@@ -28,6 +28,7 @@
     OWNLoadEvent,
     OWNMessage,
 )
+from OWNd.message.load import _safe_float, _safe_int
 
 ORACLE = json.loads(
     (Path(__file__).parent / "golden" / "firmware_oracle.json").read_text(encoding="utf-8")
@@ -180,19 +181,69 @@ def test_unknown_what_keeps_the_raw_frame():
     assert event.human_readable_log == "*3*7*#1##"
 
 
-def test_measurement_reply_stays_raw():
+def test_measurement_reply_all_measurements():
     event = OWNEvent.parse("*#3*10*0*230*12*2760*1520##")
     assert isinstance(event, OWNLoadEvent)
     assert event.is_control_unit
     assert event.dimension == 0
     assert event.measurement_values == ["230", "12", "2760", "1520"]
+    assert event.power == 230
+    assert event.voltage == 12
+    assert event.current == 2760.0
+    assert event.power_factor == 1520.0
     assert event.state is None
     assert event.human_readable_log == (
-        "Load control unit dimension 0 (all measurements per specification): "
-        "raw values ['230', '12', '2760', '1520']."
+        "Load control unit all measurements: power=230 W, voltage=12 V, "
+        "current=2760.0 A, power factor=1520.0."
     )
 
 
+@pytest.mark.parametrize(
+    ("frame", "dim", "attr", "expected_value", "log_phrase"),
+    [
+        ("*#3*10*1*2500##", 1, "power", 2500, "power draw: 2500 W"),
+        ("*#3*10*2*230##", 2, "voltage", 230, "voltage: 230 V"),
+        ("*#3*10*3*16##", 3, "current", 16.0, "current: 16.0 A"),
+        ("*#3*10*4*95##", 4, "power_factor", 95.0, "power factor: 95.0"),
+    ],
+)
+def test_measurement_reply_individual_dimensions(
+    frame: str, dim: int, attr: str, expected_value: int | float, log_phrase: str
+):
+    event = OWNEvent.parse(frame)
+    assert isinstance(event, OWNLoadEvent)
+    assert event.is_control_unit
+    assert event.dimension == dim
+    assert getattr(event, attr) == expected_value
+    assert log_phrase in event.human_readable_log
+
+
+def test_safe_helpers():
+    assert _safe_int(None) is None
+    assert _safe_int("not_a_number") is None
+    assert _safe_int("100") == 100
+    assert _safe_float(None) is None
+    assert _safe_float("not_a_float") is None
+    assert _safe_float("12.5") == 12.5
+
+
+def test_measurement_reply_empty_or_partial():
+    empty_power = OWNEvent.parse("*#3*10*1*##")
+    assert isinstance(empty_power, OWNLoadEvent)
+    assert empty_power.power is None
+
+    empty_float = OWNEvent.parse("*#3*10*3*##")
+    assert isinstance(empty_float, OWNLoadEvent)
+    assert empty_float.current is None
+
+    partial = OWNEvent.parse("*#3*10*0*230##")
+    assert isinstance(partial, OWNLoadEvent)
+    assert partial.power == 230
+    assert partial.voltage is None
+    assert partial.current is None
+    assert partial.power_factor is None
+
+
 def test_unnamed_dimension_and_other_where_are_not_measurements():
     assert OWNLoadEvent("*#3*10*9*1##").human_readable_log == (
         "Load control unit dimension 9: raw values ['1']."