From 0e925f2c61dfe4a21f680b84f2f2eb7d1566ca40 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Wed, 14 Oct 2026 21:32:54 +0200 Subject: [PATCH] feat(load): add typed measurement properties for WHO 3 dimensions 1-4 --- OWNd/message/__init__.py | 10 +++ OWNd/message/load.py | 117 ++++++++++++++++++++++++++++---- README.md | 4 +- scripts/sync_documentation.py | 2 +- tests/test_who3_load_control.py | 57 +++++++++++++++- 5 files changed, 172 insertions(+), 18 deletions(-) diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index 98a17b1..c52f6d9 100644 --- a/OWNd/message/__init__.py +++ b/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) diff --git a/OWNd/message/load.py b/OWNd/message/load.py index b7f90a9..13bc8cc 100644 --- a/OWNd/message/load.py +++ b/OWNd/message/load.py @@ -37,6 +37,12 @@ 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", @@ -44,18 +50,41 @@ 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. diff --git a/README.md b/README.md index 1fb9167..76e7530 100755 --- a/README.md +++ b/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 | diff --git a/scripts/sync_documentation.py b/scripts/sync_documentation.py index 3f963e9..5c3c3b6 100644 --- a/scripts/sync_documentation.py +++ b/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( diff --git a/tests/test_who3_load_control.py b/tests/test_who3_load_control.py index b5c3781..636cef2 100644 --- a/tests/test_who3_load_control.py +++ b/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']."