From bcaee1f9e299eae766671d45d828456d1faafd25 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:11:58 +0200 Subject: [PATCH 1/5] feat(load): add WHO 3 load control parser and builders from firmware evidence OWNLoadEvent / OWNLoadCommand replace the base-class pass-through for WHO 3. Builders exist only for frames the firmware oracle (energy-ts10, F459 020105 and MH202 010024) shows reaching the bus: force *3*2*#N## (bus $04B70N1300), priority status *#3*#N## and control-unit measurements *#3*10*D##. WHAT 0/1/3, general WHERE 0 and *#3*0## are refused by the gateways and get no builder. State replies (WHAT 0-3) follow the WHO 3 v1.0.0 specification. Measurement replies stay raw: no device capture confirms their scaling or units yet. Co-Authored-By: Claude Opus 5.5 --- OWNd/message/__init__.py | 21 +++ OWNd/message/base.py | 3 +- OWNd/message/load.py | 173 +++++++++++++++++++++++ README.md | 2 +- tests/test_who3_load_control.py | 236 ++++++++++++++++++++++++++++++++ 5 files changed, 433 insertions(+), 2 deletions(-) create mode 100644 OWNd/message/load.py create mode 100644 tests/test_who3_load_control.py diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index b8c503f..b976d76 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -147,6 +147,17 @@ OWNLightingCommand, OWNLightingEvent, ) +from .load import ( + LOAD_CONTROL_UNIT, + LOAD_MEASUREMENT_DIMENSIONS, + LOAD_PRIORITIES, + LOAD_STATE_DISABLED, + LOAD_STATE_ENABLED, + LOAD_STATE_FORCED, + LOAD_STATE_FORCING_REMOVED, + OWNLoadCommand, + OWNLoadEvent, +) from .scenario import ( OWNAuxEvent, OWNScenarioEvent, @@ -182,6 +193,16 @@ # Automation (WHO 2) "OWNAutomationEvent", "OWNAutomationCommand", + # Load control (WHO 3) + "LOAD_STATE_DISABLED", + "LOAD_STATE_ENABLED", + "LOAD_STATE_FORCED", + "LOAD_STATE_FORCING_REMOVED", + "LOAD_CONTROL_UNIT", + "LOAD_PRIORITIES", + "LOAD_MEASUREMENT_DIMENSIONS", + "OWNLoadEvent", + "OWNLoadCommand", # Heating (WHO 4) "MESSAGE_TYPE_MAIN_TEMPERATURE", "MESSAGE_TYPE_MAIN_HUMIDITY", diff --git a/OWNd/message/base.py b/OWNd/message/base.py index 3f8decf..eba9c0a 100644 --- a/OWNd/message/base.py +++ b/OWNd/message/base.py @@ -20,6 +20,7 @@ "gateway", "heating", "lighting", + "load", "scenario", "sound", ) @@ -406,7 +407,7 @@ def parse(cls, data: str) -> OWNCommand | None: parser = _COMMAND_DISPATCH.get(_who) if parser is not None: return parser(data) - if _who in (0, 3, 14, 22, 24) or _who > 1000: + if _who in (0, 14, 22, 24) or _who > 1000: return cls(data) if _who in (7, 9): return ( diff --git a/OWNd/message/load.py b/OWNd/message/load.py new file mode 100644 index 0000000..b7f90a9 --- /dev/null +++ b/OWNd/message/load.py @@ -0,0 +1,173 @@ +"""WHO 3: Load control (load shedding) events and commands. + +Evidence, in order of weight: + +- Firmware: own-firmware-oracle suite ``energy-ts10`` (vendored in + tests/golden/firmware_oracle.json). On F459 020105 and MH202 010024, whose + rows are aligned, only ``*3*2*#N##`` (force) is accepted as a command. It is + ACKed and put on the bus as ``$04B70N1300``, with the priority in the + telegram. ``*3*0*``, ``*3*1*`` and ``*3*3*`` are NACKed with nothing on the + bus, for a priority and for the general WHERE 0 alike. ``*#3*#N##`` and + ``*#3*10*D##`` (D 0-4) are forwarded to the bus; ``*#3*0##`` is NACKed. + F453AV, F454, H4684 and MH200N agree once their one-row capture lag is + allowed for, and three of them echo ``*3*2*#N##`` on the event session. +- Specification: WHO 3 Load Control v1.0.0 (2006), as summarised by the + OpenWebNet Encyclopedia (functional/who-3-load-management). WHAT 0 load + disabled, 1 enabled, 2 forced, 3 forcing removed; WHERE 0 general, 10 control + unit, #1-#8 priority; DIMENSION 0-4 at WHERE 10 are measurements. + +No capture from a plant with a load-control central unit exists yet, so the +state and measurement replies (``*3*STATE*#N##``, ``*#3*10*D*VAL##``) are +parsed from the specification only. Measurement values are kept raw: the +specification names the dimensions but defines no scaling, sign or unit. +""" + +from __future__ import annotations + +import re + +from .base import OWNCommand, OWNEvent, register_command_parser, register_event_parser + +LOAD_STATE_DISABLED = 0 +LOAD_STATE_ENABLED = 1 +LOAD_STATE_FORCED = 2 +LOAD_STATE_FORCING_REMOVED = 3 + +LOAD_CONTROL_UNIT = "10" +LOAD_PRIORITIES = range(1, 9) +LOAD_MEASUREMENT_DIMENSIONS = range(0, 5) + +_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", +} + +_PRIORITY_WHERE = re.compile(r"^#([1-8])$") + + +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 "") + return int(match.group(1)) if match else None + + +def _priority_where(priority: int) -> str: + if priority not in LOAD_PRIORITIES: + raise ValueError("Load priority must be between 1 and 8") + return f"#{priority}" + + +class OWNLoadEvent(OWNEvent): + """A load priority state or a control-unit measurement reply.""" + + def __init__(self, data: str) -> None: + super().__init__(data) + + self._priority = _priority_of(self._where) + self._state: int | None = None + self._measurement_values: list[str] = [] + + if self._priority is not None: + _target = f"Load priority {self._priority} is" + elif self._where == "0": + _target = "All load priorities are" + else: + _target = f"Load {self._where} is" + + if self._message_type == "STATUS" and self._what in _STATE_TEXT: + self._state = self._what + self._human_readable_log = f"{_target} {_STATE_TEXT[self._what]}." + elif ( + self._message_type == "DIMENSION_REQUEST_REPLY" + and self._where == LOAD_CONTROL_UNIT + 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}." + ) + + @property + def priority(self) -> int | None: + """Load priority 1-8 from a ``#N`` WHERE, None for any other target.""" + return self._priority + + @property + def state(self) -> int | None: + """WHAT 0-3 of a state frame (see ``LOAD_STATE_*``), None otherwise.""" + return self._state + + @property + def is_forced(self) -> bool | None: + """Whether the load is forced, None when the frame carries no state.""" + return None if self._state is None else self._state == LOAD_STATE_FORCED + + @property + def is_control_unit(self) -> bool: + return self._where == LOAD_CONTROL_UNIT + + @property + def measurement_values(self) -> list[str]: + """Raw values of a control-unit measurement reply, unscaled and unitless.""" + return self._measurement_values + + +class OWNLoadCommand(OWNCommand): + """WHO 3 commands and requests. + + Only frames that the firmware oracle shows reaching the bus get a builder. + WHAT 0, 1 and 3, the general WHERE 0 and the general status request + ``*#3*0##`` are refused by the gateways, so there is no builder for them. + """ + + def __init__(self, data: str) -> None: + super().__init__(data) + _priority = _priority_of(self._where) + if self._message_type == "STATUS" and self._what == LOAD_STATE_FORCED: + if _priority is not None: + self._human_readable_log = f"Forcing load priority {_priority}." + elif self._message_type == "STATUS_REQUEST" and _priority is not None: + self._human_readable_log = f"Requesting load priority {_priority} status." + elif ( + self._message_type == "DIMENSION_REQUEST" + and self._where == LOAD_CONTROL_UNIT + and self._dimension is not None + ): + self._human_readable_log = ( + f"Requesting load control unit dimension {self._dimension}." + ) + + @classmethod + def force(cls, priority: int) -> OWNLoadCommand: + """Force the load on ``priority`` 1-8 (``*3*2*#N##``).""" + return cls(f"*3*{LOAD_STATE_FORCED}*{_priority_where(priority)}##") + + @classmethod + def status(cls, priority: int) -> OWNLoadCommand: + """Request the state of load ``priority`` 1-8 (``*#3*#N##``).""" + return cls(f"*#3*{_priority_where(priority)}##") + + @classmethod + def request_measurement(cls, dimension: int = 0) -> OWNLoadCommand: + """Request control-unit measurement ``dimension`` 0-4 (``*#3*10*D##``).""" + if dimension not in LOAD_MEASUREMENT_DIMENSIONS: + raise ValueError("Load measurement dimension must be between 0 and 4") + return cls(f"*#3*{LOAD_CONTROL_UNIT}*{dimension}##") + + +register_event_parser(3, OWNLoadEvent) +register_command_parser(3, OWNLoadCommand) diff --git a/README.md b/README.md index 49b9506..cc9f26b 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 | Load shedding status, circuit priority management | `OWNCommand`, `OWNEvent` | +| **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` | | **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` | diff --git a/tests/test_who3_load_control.py b/tests/test_who3_load_control.py new file mode 100644 index 0000000..b5c3781 --- /dev/null +++ b/tests/test_who3_load_control.py @@ -0,0 +1,236 @@ +"""WHO 3 load control, pinned to the firmware oracle first and the specification second. + +Firmware: tests/golden/firmware_oracle.json, suite energy-ts10. F459 020105 and +MH202 010024 have aligned rows and are the reference; F453AV, F454, H4684 and +MH200N lag by one row in that suite, so their per-row replies are not used. + +Specification: WHO 3 Load Control v1.0.0 (2006) via the OpenWebNet Encyclopedia +(functional/who-3-load-management). No real-plant capture exists yet, so the +state and measurement replies below come from the specification. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from OWNd.message import ( + LOAD_STATE_DISABLED, + LOAD_STATE_ENABLED, + LOAD_STATE_FORCED, + LOAD_STATE_FORCING_REMOVED, + OWNCommand, + OWNEvent, + OWNLoadCommand, + OWNLoadEvent, + OWNMessage, +) + +ORACLE = json.loads( + (Path(__file__).parent / "golden" / "firmware_oracle.json").read_text(encoding="utf-8") +)["verdicts"] +ALIGNED_GATEWAYS = ("F459", "MH202") +WHO3_INPUTS = sorted(f for f in ORACLE if f.startswith(("*3*", "*#3*"))) + + +def _aligned(frame: str) -> list[dict[str, Any]]: + rows = [r for r in ORACLE[frame] if r["product"] in ALIGNED_GATEWAYS] + assert {r["product"] for r in rows} == set(ALIGNED_GATEWAYS) + return rows + + +def _forwarded(frame: str) -> bool: + """Not NACKed and put on the bus by every aligned gateway.""" + return all(r["reply"] != "nack" and r["verdict"] == "out" for r in _aligned(frame)) + + +def _refused(frame: str) -> bool: + return all(r["reply"] == "nack" and r["verdict"] == "silent" for r in _aligned(frame)) + + +def _bus_text(row: dict[str, Any]) -> list[str]: + return [bytes.fromhex(b.replace(" ", "")).decode().strip() for b in row["bus_frames"]] + + +def _all_builder_frames() -> set[str]: + frames = {str(OWNLoadCommand.force(p)) for p in range(1, 9)} + frames |= {str(OWNLoadCommand.status(p)) for p in range(1, 9)} + frames |= {str(OWNLoadCommand.request_measurement(d)) for d in range(5)} + return frames + + +def test_oracle_covers_who3(): + assert len(WHO3_INPUTS) == 15 + + +@pytest.mark.parametrize("frame", WHO3_INPUTS) +def test_every_oracle_input_dispatches_to_load_classes(frame: str): + parsed = OWNMessage.parse(frame) + expected = OWNLoadEvent if frame.startswith("*3*") else OWNLoadCommand + assert isinstance(parsed, expected) + assert parsed.is_valid + assert isinstance(OWNCommand.parse(frame), OWNLoadCommand) + + +@pytest.mark.parametrize("priority", [1, 2]) +def test_force_is_forwarded_with_priority_in_the_telegram(priority: int): + frame = str(OWNLoadCommand.force(priority)) + assert frame == f"*3*2*#{priority}##" + for row in _aligned(frame): + assert row["reply"] == "ack" + assert _bus_text(row) == [f"$04B70{priority}1300"] + + +def test_force_echo_on_event_session_parses_as_forced(): + echoed = {own for rows in ORACLE.values() for r in rows for own in r["emitted_own"]} + load_echoes = sorted(e for e in echoed if e.startswith("*3*")) + assert load_echoes == ["*3*2*#1##", "*3*2*#2##"] + for frame in load_echoes: + event = OWNEvent.parse(frame) + assert isinstance(event, OWNLoadEvent) + assert event.is_forced is True + assert event.priority == int(frame[-3]) + + +def test_every_oracle_backed_builder_frame_reaches_the_bus(): + in_oracle = _all_builder_frames() & set(ORACLE) + assert in_oracle == { + "*3*2*#1##", + "*3*2*#2##", + "*#3*#1##", + "*#3*#2##", + "*#3*10*0##", + "*#3*10*1##", + "*#3*10*2##", + "*#3*10*3##", + "*#3*10*4##", + } + assert all(_forwarded(f) for f in in_oracle) + + +def test_no_builder_emits_a_frame_the_gateways_refuse(): + refused = {f for f in WHO3_INPUTS if _refused(f)} + assert refused == { + "*#3*0##", + "*3*0*#1##", + "*3*0*0##", + "*3*1*#1##", + "*3*1*0##", + "*3*3*#1##", + } + assert not refused & _all_builder_frames() + + +def test_measurement_requests_read_distinct_registers(): + """Each dimension maps to its own register run; dimension 0 reads the others. + + Run widths 1/1/2/4 fit the specification's voltage/current/power/energy + order, but the bus telegrams name no quantity, so values stay raw. + """ + for gateway in ALIGNED_GATEWAYS: + registers = { + d: next( + _bus_text(r) for r in _aligned(f"*#3*10*{d}##") if r["product"] == gateway + ) + for d in range(5) + } + assert registers[1] == ["$03AA001613"] + assert registers[2] == ["$03AA001612"] + assert registers[3] == ["$03AA001610", "$03AA001611"] + assert registers[4] == ["$03AA001606", "$03AA001607", "$03AA001608", "$03AA001609"] + others = registers[1] + registers[2] + registers[3] + registers[4] + assert registers[0][:4] == others[:4] + assert set(registers[0]) <= set(others) + + +@pytest.mark.parametrize( + ("frame", "state", "text"), + [ + ("*3*0*#3##", LOAD_STATE_DISABLED, "Load priority 3 is disabled."), + ("*3*1*#8##", LOAD_STATE_ENABLED, "Load priority 8 is enabled."), + ("*3*2*#1##", LOAD_STATE_FORCED, "Load priority 1 is forced."), + ("*3*3*#5##", LOAD_STATE_FORCING_REMOVED, "Load priority 5 is no longer forced."), + ("*3*1*0##", LOAD_STATE_ENABLED, "All load priorities are enabled."), + ], +) +def test_specified_state_frames(frame: str, state: int, text: str): + event = OWNEvent.parse(frame) + assert isinstance(event, OWNLoadEvent) + assert event.state == state + assert event.is_forced is (state == LOAD_STATE_FORCED) + assert event.human_readable_log == text + + +def test_priority_keeps_its_hash_and_rejects_out_of_range(): + assert OWNLoadEvent("*3*2*#4##").where == "#4" + assert OWNLoadEvent("*3*2*#4##").priority == 4 + assert OWNLoadEvent("*3*2*#9##").priority is None + assert OWNLoadEvent("*3*2*4##").priority is None + assert OWNLoadEvent("*3*2*4##").human_readable_log == "Load 4 is forced." + + +def test_unknown_what_keeps_the_raw_frame(): + event = OWNLoadEvent("*3*7*#1##") + assert event.state is None + assert event.is_forced is None + assert event.priority == 1 + assert event.human_readable_log == "*3*7*#1##" + + +def test_measurement_reply_stays_raw(): + 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.state is None + assert event.human_readable_log == ( + "Load control unit dimension 0 (all measurements per specification): " + "raw values ['230', '12', '2760', '1520']." + ) + + +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']." + ) + other = OWNLoadEvent("*#3*#1*1*5##") + assert other.measurement_values == [] + assert not other.is_control_unit + + +@pytest.mark.parametrize( + ("command", "frame", "text"), + [ + (OWNLoadCommand.force(3), "*3*2*#3##", "Forcing load priority 3."), + (OWNLoadCommand.status(8), "*#3*#8##", "Requesting load priority 8 status."), + (OWNLoadCommand.request_measurement(), "*#3*10*0##", "Requesting load control unit dimension 0."), + (OWNLoadCommand.request_measurement(4), "*#3*10*4##", "Requesting load control unit dimension 4."), + ], +) +def test_builders(command: OWNLoadCommand, frame: str, text: str): + assert str(command) == frame + assert command.is_valid + assert command.human_readable_log == text + + +@pytest.mark.parametrize("frame", ["*3*2*0##", "*#3*0##", "*#3*#1*1##", "*3*1*#1##"]) +def test_parsed_commands_without_a_specific_log_keep_the_raw_frame(frame: str): + assert OWNLoadCommand(frame).human_readable_log == frame + + +@pytest.mark.parametrize("priority", [0, 9, -1]) +def test_priority_validation(priority: int): + with pytest.raises(ValueError, match="between 1 and 8"): + OWNLoadCommand.force(priority) + with pytest.raises(ValueError, match="between 1 and 8"): + OWNLoadCommand.status(priority) + + +@pytest.mark.parametrize("dimension", [-1, 5]) +def test_dimension_validation(dimension: int): + with pytest.raises(ValueError, match="between 0 and 4"): + OWNLoadCommand.request_measurement(dimension) From e743c2a996c2c2ab68e49fddf471d72236930dcb Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:04:39 +0200 Subject: [PATCH 2/5] feat(profiles): exclude WHO 3 load control from MyHomeServer1Profile based on firmware oracle --- OWNd/profiles.py | 13 +++++++++++++ scripts/sync_documentation.py | 6 +++--- tests/test_profiles.py | 10 ++++++++++ 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/OWNd/profiles.py b/OWNd/profiles.py index dbda8d4..defa463 100644 --- a/OWNd/profiles.py +++ b/OWNd/profiles.py @@ -519,6 +519,16 @@ def __init__(self) -> None: class MyHomeServer1Profile(GatewayProfile): + """The MyHomeServer1 (Eliot / Legrand Living Now gateway). + + Subsystems: + - WHO 3 (Load Control): MyHomeServer1 firmware does not package the + ``bt_energia`` translator (unlike MH202, F454, and F459), so it strictly + NACKs all WHO 3 requests without bus activity (see own-firmware-oracle + findings/MyHomeServer1/energy-ts10.md). WHO 3 is therefore excluded from + ``supported_who``. + """ + def __init__(self) -> None: super().__init__( model_name="MyHomeServer1", @@ -530,6 +540,9 @@ def __init__(self) -> None: supports_hmac=True, supports_native_transitions=True, supports_extended_frames=True, + supported_who=tuple( + w for w in DEFAULT_SUPPORTED_WHO if w != WHO_LOAD_CONTROL + ), ) diff --git a/scripts/sync_documentation.py b/scripts/sync_documentation.py index b127e9a..9af4c91 100644 --- a/scripts/sync_documentation.py +++ b/scripts/sync_documentation.py @@ -65,6 +65,7 @@ "OWNd/message/gateway.py": "WHO 13: Gateway management, clock/date configuration, and model telemetry", "OWNd/message/heating.py": "WHO 4: Thermoregulation, heating/cooling zone control, and temperature decoders", "OWNd/message/lighting.py": "WHO 1: Lighting controls, dimming, RGB/HSV color, and motion/PIR sensors", + "OWNd/message/load.py": "WHO 3: Load control, priority forcing, and central-unit measurement events/commands", "OWNd/message/scenario.py": "WHO 0, WHO 9, WHO 17: Scenario activation, auxiliary commands, and MH200/MH202 scenes", "OWNd/message/sound.py": "WHO 16: Sound system and audio/video matrix routing (WHO 22 is not implemented)", "OWNd/profiles.py": "Declarative hardware gateway models across 14 canonical hardware profiles", @@ -107,9 +108,8 @@ class WhoCatalogDefinition: WhoCatalogDefinition( who="**3**", name="Load Control", - description="Load shedding status, circuit priority management", - dispatch_who_keys=(), - explicit_classes=("OWNCommand", "OWNEvent"), + 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", + dispatch_who_keys=(3,), ), WhoCatalogDefinition( who="**4**", diff --git a/tests/test_profiles.py b/tests/test_profiles.py index d92846a..af6cb65 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -97,6 +97,16 @@ def test_profile_lookup_accepts_common_name_variants() -> None: assert profile.model_name == "MyHomeServer1" assert profile.supports_session_count(4) assert profile.supports_who(WHO_SOUND) + # Firmware oracle proves MyHomeServer1 omits bt_energia and rejects WHO 3 + assert not profile.supports_who(WHO_LOAD_CONTROL) + + +def test_myhomeserver1_lacks_who3_load_control() -> None: + """MyHomeServer1 firmware does not package bt_energia and rejects WHO 3.""" + profile = get_gateway_profile("MyHomeServer1") + assert WHO_LOAD_CONTROL not in profile.supported_who + assert not profile.supports_who(WHO_LOAD_CONTROL) + def test_f461_profile_lookup() -> None: From 5f43444be5cffdd02ffe22c46be14a1c258039d6 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 3/5] feat(load): add typed measurement properties for WHO 3 dimensions 1-4 --- OWNd/message/__init__.py | 10 +++ OWNd/message/load.py | 117 ++++++++++++++++++++++++++++---- README.md | 3 +- scripts/sync_documentation.py | 2 +- tests/test_who3_load_control.py | 57 +++++++++++++++- 5 files changed, 171 insertions(+), 18 deletions(-) diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index b976d76..db6639d 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -149,6 +149,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, @@ -201,6 +206,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 cc9f26b..7aa618d 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` | @@ -342,6 +342,7 @@ OWNd maintains an automated test suite with strict **100.0% line coverage** veri | [`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 9af4c91..df17452 100644 --- a/scripts/sync_documentation.py +++ b/scripts/sync_documentation.py @@ -108,7 +108,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']." From 3582df68f64909cdb16b13eb66274d20f40c85a8 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 16 Oct 2026 20:52:45 +0200 Subject: [PATCH 4/5] feat(profiles,load): align WHO 3 profile gating and oracle tests across fleet --- OWNd/profiles.py | 22 ++++++++++++++ tests/test_profiles.py | 14 +++++---- tests/test_who3_load_control.py | 51 +++++++++++++++++++++++++++++---- 3 files changed, 76 insertions(+), 11 deletions(-) diff --git a/OWNd/profiles.py b/OWNd/profiles.py index defa463..3f2d736 100644 --- a/OWNd/profiles.py +++ b/OWNd/profiles.py @@ -366,6 +366,15 @@ def __init__(self) -> None: class F461Profile(GatewayProfile): + """The F461 server gateway stack (AArch64 / ARM64). + + Subsystems: + - WHO 3 (Load Control): Like MyHomeServer1 and F460, F461 firmware + (`020011`) does not package the ``bt_energia`` translator and strictly + NACKs all WHO 3 requests without bus activity (`results/F461/020011/oracle/full/energy-ts10.tsv`). + WHO 3 is therefore excluded from ``supported_who``. + """ + def __init__(self) -> None: super().__init__( model_name="F461", @@ -377,6 +386,9 @@ def __init__(self) -> None: supports_hmac=True, supports_native_transitions=True, supports_extended_frames=True, + supported_who=tuple( + w for w in DEFAULT_SUPPORTED_WHO if w != WHO_LOAD_CONTROL + ), ) @@ -390,6 +402,10 @@ class MH200Profile(GatewayProfile): subsystems are copied from the MH200N profile and have not been measured on an MH200. + WHO 3 (Load Control) is verified by the firmware oracle: MH200 firmware + 2.1.2 (``020102``) ships ``bt_energia`` and translates all 9 WHO 3 force + and measurement requests onto the SCS bus (``results/MH200/020102/oracle/full/energy-ts10.tsv``). + WHO 25 covers both CEN+ and dry contact / IR state functions (WHAT 31/32). As documented in Legrand's WHO 25 specification ("Dry contact and IR state functions", v1.0.0, 2010, https://developer.legrand.com/uploads/2019/12/WHO_25.pdf, @@ -414,6 +430,7 @@ def __init__(self) -> None: supported_who=( WHO_LIGHTING, WHO_AUTOMATION, + WHO_LOAD_CONTROL, WHO_HEATING, WHO_DOOR_ENTRY, WHO_LOCK, @@ -431,6 +448,10 @@ class MH200NProfile(GatewayProfile): ``*#16*0*5##`` without NACK and returning full source and amplifier inventory (MyHOME#427 / comment 5848181845). + WHO 3 (Load Control) is verified by the firmware oracle: MH200N firmware + 1.1.8 (``010108``) ships ``bt_energia`` and translates all 9 WHO 3 force + and measurement requests onto the SCS bus (``results/MH200N/010108/oracle/full/energy-ts10.tsv``). + WHO 6 and 8 are listed by family with the MH200, whose capture proves the relay (see ``MH200Profile``). Nothing has been captured on an MH200N itself; its 1.1.8 image ships a ``bt_vct`` translator, which the firmware oracle @@ -448,6 +469,7 @@ def __init__(self) -> None: supported_who=( WHO_LIGHTING, WHO_AUTOMATION, + WHO_LOAD_CONTROL, WHO_HEATING, WHO_DOOR_ENTRY, WHO_LOCK, diff --git a/tests/test_profiles.py b/tests/test_profiles.py index af6cb65..b93baff 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -102,11 +102,14 @@ def test_profile_lookup_accepts_common_name_variants() -> None: def test_myhomeserver1_lacks_who3_load_control() -> None: - """MyHomeServer1 firmware does not package bt_energia and rejects WHO 3.""" - profile = get_gateway_profile("MyHomeServer1") - assert WHO_LOAD_CONTROL not in profile.supported_who - assert not profile.supports_who(WHO_LOAD_CONTROL) - + """MyHomeServer1 and F461 firmwares do not package bt_energia and reject WHO 3.""" + for model in ("MyHomeServer1", "F461"): + profile = get_gateway_profile(model) + assert WHO_LOAD_CONTROL not in profile.supported_who + assert not profile.supports_who(WHO_LOAD_CONTROL) + for model in ("MH200", "MH200N", "F454", "MH202"): + profile = get_gateway_profile(model) + assert profile.supports_who(WHO_LOAD_CONTROL) def test_f461_profile_lookup() -> None: @@ -117,6 +120,7 @@ def test_f461_profile_lookup() -> None: assert profile.supports_hmac is True assert profile.command_queue_delay == 0.05 assert profile.supports_native_transitions is True + assert not profile.supports_who(WHO_LOAD_CONTROL) def test_f455_basic_gateway_has_no_audio() -> None: diff --git a/tests/test_who3_load_control.py b/tests/test_who3_load_control.py index 636cef2..7ea7c85 100644 --- a/tests/test_who3_load_control.py +++ b/tests/test_who3_load_control.py @@ -1,8 +1,9 @@ """WHO 3 load control, pinned to the firmware oracle first and the specification second. -Firmware: tests/golden/firmware_oracle.json, suite energy-ts10. F459 020105 and -MH202 010024 have aligned rows and are the reference; F453AV, F454, H4684 and -MH200N lag by one row in that suite, so their per-row replies are not used. +Firmware: tests/golden/firmware_oracle.json, suite energy-ts10. Gateways +packaging ``bt_energia`` (F453AV, F454, F459, H4684, MH200, MH200N, MH202) +translate WHO 3 commands onto the SCS bus; gateways without ``bt_energia`` +(MyHomeServer1, F461, F460, F450) silently NACK all 15 WHO 3 inputs. Specification: WHO 3 Load Control v1.0.0 (2006) via the OpenWebNet Encyclopedia (functional/who-3-load-management). No real-plant capture exists yet, so the @@ -33,7 +34,33 @@ ORACLE = json.loads( (Path(__file__).parent / "golden" / "firmware_oracle.json").read_text(encoding="utf-8") )["verdicts"] -ALIGNED_GATEWAYS = ("F459", "MH202") +BT_ENERGIA_CANDIDATES = ( + "F453AV", + "F454", + "F459", + "H4684", + "MH200", + "MH200N", + "MH202", +) +ALIGNED_GATEWAYS = tuple( + p + for p in BT_ENERGIA_CANDIDATES + if any( + r["product"] == p + and r["reply"] == "ack" + and [bytes.fromhex(b.replace(" ", "")).decode().strip() for b in r["bus_frames"]] + == ["$04B7011300"] + for r in ORACLE["*3*2*#1##"] + ) + and any( + r["product"] == p + and [bytes.fromhex(b.replace(" ", "")).decode().strip() for b in r["bus_frames"]] + == ["$03AA001614"] + for r in ORACLE["*#3*#1##"] + ) +) +WITHOUT_BT_ENERGIA_GATEWAYS = ("F450", "F460", "F461", "MyHomeServer1") WHO3_INPUTS = sorted(f for f in ORACLE if f.startswith(("*3*", "*#3*"))) @@ -44,8 +71,10 @@ def _aligned(frame: str) -> list[dict[str, Any]]: def _forwarded(frame: str) -> bool: - """Not NACKed and put on the bus by every aligned gateway.""" - return all(r["reply"] != "nack" and r["verdict"] == "out" for r in _aligned(frame)) + """Put on the bus with at least one SCS telegram by every aligned bt_energia gateway.""" + return all( + r["verdict"] == "out" and bool(r["bus_frames"]) for r in _aligned(frame) + ) def _refused(frame: str) -> bool: @@ -63,6 +92,16 @@ def _all_builder_frames() -> set[str]: return frames +def test_gateways_without_bt_energia_refuse_all_who3_frames(): + assert len(ALIGNED_GATEWAYS) >= 2 + for frame in WHO3_INPUTS: + rows = {r["product"]: r for r in ORACLE[frame]} + for gw in WITHOUT_BT_ENERGIA_GATEWAYS: + assert rows[gw]["reply"] == "nack" + assert rows[gw]["verdict"] == "silent" + assert rows[gw]["bus_frames"] == [] + + def test_oracle_covers_who3(): assert len(WHO3_INPUTS) == 15 From 739d551eb701125dae627d57ec7de546ee880b47 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:46:43 +0200 Subject: [PATCH 5/5] fix(load): align WHO 3 dimension order (1=voltage, 2=current, 3=power, 4=energy) with WHO_3.pdf and bt_energia binary --- OWNd/message/__init__.py | 6 +- OWNd/message/load.py | 157 +++++++++++++++++--------------- README.md | 2 +- scripts/sync_documentation.py | 2 +- tests/test_who3_load_control.py | 105 ++++++++++++--------- 5 files changed, 152 insertions(+), 120 deletions(-) diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index db6639d..fe7f4cc 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -151,8 +151,8 @@ LOAD_CONTROL_UNIT, LOAD_DIMENSION_ALL, LOAD_DIMENSION_CURRENT, + LOAD_DIMENSION_ENERGY, LOAD_DIMENSION_POWER, - LOAD_DIMENSION_POWER_FACTOR, LOAD_DIMENSION_VOLTAGE, LOAD_MEASUREMENT_DIMENSIONS, LOAD_PRIORITIES, @@ -207,10 +207,10 @@ "LOAD_PRIORITIES", "LOAD_MEASUREMENT_DIMENSIONS", "LOAD_DIMENSION_ALL", - "LOAD_DIMENSION_POWER", "LOAD_DIMENSION_VOLTAGE", "LOAD_DIMENSION_CURRENT", - "LOAD_DIMENSION_POWER_FACTOR", + "LOAD_DIMENSION_POWER", + "LOAD_DIMENSION_ENERGY", "OWNLoadEvent", "OWNLoadCommand", # Heating (WHO 4) diff --git a/OWNd/message/load.py b/OWNd/message/load.py index 13bc8cc..05dabdb 100644 --- a/OWNd/message/load.py +++ b/OWNd/message/load.py @@ -2,24 +2,45 @@ Evidence, in order of weight: -- Firmware: own-firmware-oracle suite ``energy-ts10`` (vendored in - tests/golden/firmware_oracle.json). On F459 020105 and MH202 010024, whose - rows are aligned, only ``*3*2*#N##`` (force) is accepted as a command. It is - ACKed and put on the bus as ``$04B70N1300``, with the priority in the - telegram. ``*3*0*``, ``*3*1*`` and ``*3*3*`` are NACKed with nothing on the - bus, for a priority and for the general WHERE 0 alike. ``*#3*#N##`` and - ``*#3*10*D##`` (D 0-4) are forwarded to the bus; ``*#3*0##`` is NACKed. - F453AV, F454, H4684 and MH200N agree once their one-row capture lag is - allowed for, and three of them echo ``*3*2*#N##`` on the event session. -- Specification: WHO 3 Load Control v1.0.0 (2006), as summarised by the - OpenWebNet Encyclopedia (functional/who-3-load-management). WHAT 0 load - disabled, 1 enabled, 2 forced, 3 forcing removed; WHERE 0 general, 10 control - unit, #1-#8 priority; DIMENSION 0-4 at WHERE 10 are measurements. - -No capture from a plant with a load-control central unit exists yet, so the -state and measurement replies (``*3*STATE*#N##``, ``*#3*10*D*VAL##``) are -parsed from the specification only. Measurement values are kept raw: the -specification names the dimensions but defines no scaling, sign or unit. +- Firmware: ``bt_energia`` v2.0.2 (2003-05-08) ARM binary and + ``own-firmware-oracle`` suite ``energy-ts10`` (vendored in + ``tests/golden/firmware_oracle.json``; see + https://github.com/OpenWebNet-HA/own-firmware-oracle/blob/main/oracle/cases/energy-ts10.cases#L56-L65 + and + https://github.com/OpenWebNet-HA/own-firmware-oracle/blob/main/findings/MyHomeServer1/energy-ts10.md#35-who-3-load-management--f421). + On F459 020105 and MH202 010024, whose rows are aligned, only ``*3*2*#N##`` + (force) is accepted as a command. It is ACKed and put on the bus as + ``$04B70N1300``, with the priority in the telegram. ``*3*0*``, ``*3*1*`` and + ``*3*3*`` are NACKed with nothing on the bus, for a priority and for the + general WHERE 0 alike. ``*#3*#N##`` and ``*#3*10*D##`` (D 0-4) are forwarded + to the bus; ``*#3*0##`` is NACKed (``bt_energia`` at ``0x00b91c``/``0x00b998`` + only accepts ``*#3##`` and ``*#3*##`` for general status). F453AV, F454, + H4684 and MH200N agree once their one-row capture lag is allowed for, and + three of them echo ``*3*2*#N##`` on the event session. + In ``bt_energia``, ``0x00b140`` (jump table ``0x00b3e0``) polls the F421 + central unit registers and ``0x00a63c`` decodes the ``$031900`` + + ``$031A00`` replies via ``htod`` into unsigned integers formatted with + ``%d``: + - Dimension 1 (Voltage, V): register ``0x13`` (1 byte, 8-bit int ``0..255``, + ``0x00b674`` / ``0x00af2c`` -> ``*#3*10*1*%d##``) + - Dimension 2 (Current, A): register ``0x12`` (1 byte, 8-bit int ``0..255``, + ``0x00b6f4`` / ``0x00ae7c`` -> ``*#3*10*2*%d##``) + - Dimension 3 (Power, W): registers ``0x10..0x11`` (2 bytes, 16-bit int + ``0..65535``, ``0x00b774`` / ``0x00ad88`` -> ``*#3*10*3*%d##``) + - Dimension 4 (Energy): registers ``0x06..0x09`` (4 bytes, 32-bit int, + ``0x00b830`` / ``0x00ac34`` -> ``*#3*10*4*%d##``) + - Dimension 0 (All measurements): polls all 8 registers ``13, 12, 10..11, + 06..09`` (``0x00b3f4``) and joins them as ``*#3*10*0*%s*%s*%s*%s##`` + (``0x00aa24``) in ``T*C*P*E`` (Voltage, Current, Power, Energy) order. +- Specification: WHO 3 Load Control v1.0.0 (2006-08-30, + ``OpenWebNet_Community_6_powermanagement_v1_0_0_EN.doc``, archived at + https://archive.openwebnet-ha.org/sha256/4f/68/4f685a6ee3b16e6a7994a338186687bcea1dabba74dc5d24d22a07becb24dea1.pdf, + pp. 4, 6-7), as summarised by the OpenWebNet Encyclopedia + (https://github.com/OpenWebNet-HA/OpenWebNet-Encyclopedia/blob/main/functional/who-3-load-management/README.md#measurement-dimension-values). + WHAT 0 load disabled, 1 enabled, 2 forced, 3 forcing removed; WHERE 0 + general, 10 control unit, #1-#8 priority; DIMENSION 0-4 at WHERE 10 are + All dimensions (``T*C*P*E``), Voltage (``T``, Volt), Current (``C``, Ampere), + Power (``P``, Watt), and Energy (``E``, unit unspecified in the spec). """ from __future__ import annotations @@ -38,10 +59,10 @@ 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 +LOAD_DIMENSION_VOLTAGE = 1 +LOAD_DIMENSION_CURRENT = 2 +LOAD_DIMENSION_POWER = 3 +LOAD_DIMENSION_ENERGY = 4 _STATE_TEXT = { LOAD_STATE_DISABLED: "disabled", @@ -50,18 +71,19 @@ LOAD_STATE_FORCING_REMOVED: "no longer forced", } -# 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 mappings verified against WHO_3.pdf v1.0.0 (pp. 4, 6-7) and +# central unit register polling/decoding in bt_energia v2.0.2 (0x00b140 / 0x00a63c): +# - Dimension 0: All measurements (T*C*P*E) -> 8 registers $03AA001613, 12, 10..11, 06..09 +# - Dimension 1: Voltage (T, V) -> 1 register $03AA001613 (8-bit int) +# - Dimension 2: Current (C, A) -> 1 register $03AA001612 (8-bit int) +# - Dimension 3: Power (P, W) -> 2 registers $03AA001610..$03AA001611 (16-bit int) +# - Dimension 4: Energy (E) -> 4 registers $03AA001606..$03AA001609 (32-bit int) _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", + LOAD_DIMENSION_POWER: "power", + LOAD_DIMENSION_ENERGY: "energy", } _PRIORITY_WHERE = re.compile(r"^#([1-8])$") @@ -76,15 +98,6 @@ def _safe_int(value: str | None) -> int | None: 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 "") @@ -106,10 +119,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 + self._current: int | None = None + self._power: int | None = None + self._energy: int | None = None if self._priority is not None: _target = f"Load priority {self._priority} is" @@ -127,42 +140,38 @@ def __init__(self, data: str) -> None: and self._dimension is not None ): self._measurement_values = list(self._dimension_value) - 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: + if 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._current = _safe_int(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]) + elif 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: {self._power} W." + ) + elif self._dimension == LOAD_DIMENSION_ENERGY and self._dimension_value: + self._energy = _safe_int(self._dimension_value[0]) self._human_readable_log = ( - f"Load control unit power factor: {self._power_factor}." + f"Load control unit energy: {self._energy}." ) - elif self._dimension == LOAD_DIMENSION_ALL: - if len(self._dimension_value) >= 1: - self._power = _safe_int(self._dimension_value[0]) + elif self._dimension == LOAD_DIMENSION_ALL and self._dimension_value: + self._voltage = _safe_int(self._dimension_value[0]) if len(self._dimension_value) >= 2: - self._voltage = _safe_int(self._dimension_value[1]) + self._current = _safe_int(self._dimension_value[1]) if len(self._dimension_value) >= 3: - self._current = _safe_float(self._dimension_value[2]) + self._power = _safe_int(self._dimension_value[2]) if len(self._dimension_value) >= 4: - self._power_factor = _safe_float(self._dimension_value[3]) + self._energy = _safe_int(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}." + f"voltage={self._voltage} V, current={self._current} A, " + f"power={self._power} W, energy={self._energy}." ) else: _name = _DIMENSION_NAMES.get(self._dimension) @@ -196,25 +205,25 @@ 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.""" + """Measured voltage in V (dimension 1, or index 0 of dimension 0), None if not reported.""" return self._voltage @property - def current(self) -> float | None: - """Current in A (dimension 3), None if not reported.""" + def current(self) -> int | None: + """Measured current in A (dimension 2, or index 1 of dimension 0), 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 + def power(self) -> int | None: + """Measured power in W (dimension 3, or index 2 of dimension 0), None if not reported.""" + return self._power + + @property + def energy(self) -> int | None: + """Measured energy counter (dimension 4, or index 3 of dimension 0), None if not reported.""" + return self._energy class OWNLoadCommand(OWNCommand): diff --git a/README.md b/README.md index 7aa618d..97e2c2d 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 with typed power, voltage, current, and power factor properties from firmware register mappings | `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 voltage, current, power, and energy properties (`WHO_3.pdf` v1.0.0 & `bt_energia` 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` | diff --git a/scripts/sync_documentation.py b/scripts/sync_documentation.py index df17452..279bd1a 100644 --- a/scripts/sync_documentation.py +++ b/scripts/sync_documentation.py @@ -108,7 +108,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 with typed power, voltage, current, and power factor properties from firmware register mappings", + description="Priority forcing (`*3*2*#N##`, firmware-verified), priority states 0–3, priority status and central-unit measurement requests with typed voltage, current, power, and energy properties (`WHO_3.pdf` v1.0.0 & `bt_energia` register mappings)", dispatch_who_keys=(3,), ), WhoCatalogDefinition( diff --git a/tests/test_who3_load_control.py b/tests/test_who3_load_control.py index 7ea7c85..6fe1ac6 100644 --- a/tests/test_who3_load_control.py +++ b/tests/test_who3_load_control.py @@ -1,13 +1,17 @@ """WHO 3 load control, pinned to the firmware oracle first and the specification second. -Firmware: tests/golden/firmware_oracle.json, suite energy-ts10. Gateways -packaging ``bt_energia`` (F453AV, F454, F459, H4684, MH200, MH200N, MH202) -translate WHO 3 commands onto the SCS bus; gateways without ``bt_energia`` -(MyHomeServer1, F461, F460, F450) silently NACK all 15 WHO 3 inputs. - -Specification: WHO 3 Load Control v1.0.0 (2006) via the OpenWebNet Encyclopedia -(functional/who-3-load-management). No real-plant capture exists yet, so the -state and measurement replies below come from the specification. +Firmware: ``bt_energia`` v2.0.2 (2003-05-08) ARM binary and +``tests/golden/firmware_oracle.json``, suite ``energy-ts10`` +(https://github.com/OpenWebNet-HA/own-firmware-oracle/blob/main/oracle/cases/energy-ts10.cases#L56-L65). +Gateways packaging ``bt_energia`` (F453AV, F454, F459, H4684, MH200, MH200N, +MH202) translate WHO 3 commands onto the SCS bus; gateways without +``bt_energia`` (MyHomeServer1, F461, F460, F450) silently NACK all 15 WHO 3 +inputs. + +Specification: WHO 3 Load Control v1.0.0 (2006-08-30, archived at +https://archive.openwebnet-ha.org/sha256/4f/68/4f685a6ee3b16e6a7994a338186687bcea1dabba74dc5d24d22a07becb24dea1.pdf, +pp. 4, 6-7) via the OpenWebNet Encyclopedia +(https://github.com/OpenWebNet-HA/OpenWebNet-Encyclopedia/blob/main/functional/who-3-load-management/README.md#measurement-dimension-values). """ from __future__ import annotations @@ -19,6 +23,11 @@ import pytest from OWNd.message import ( + LOAD_DIMENSION_ALL, + LOAD_DIMENSION_CURRENT, + LOAD_DIMENSION_ENERGY, + LOAD_DIMENSION_POWER, + LOAD_DIMENSION_VOLTAGE, LOAD_STATE_DISABLED, LOAD_STATE_ENABLED, LOAD_STATE_FORCED, @@ -29,7 +38,7 @@ OWNLoadEvent, OWNMessage, ) -from OWNd.message.load import _safe_float, _safe_int +from OWNd.message.load import _safe_int ORACLE = json.loads( (Path(__file__).parent / "golden" / "firmware_oracle.json").read_text(encoding="utf-8") @@ -167,9 +176,16 @@ def test_no_builder_emits_a_frame_the_gateways_refuse(): def test_measurement_requests_read_distinct_registers(): """Each dimension maps to its own register run; dimension 0 reads the others. - Run widths 1/1/2/4 fit the specification's voltage/current/power/energy - order, but the bus telegrams name no quantity, so values stay raw. + In ``bt_energia`` v2.0.2 (``0x00b140`` / ``0x00a63c``), register widths + 1/1/2/4 bytes are decoded as 8/8/16/32-bit integers formatted with ``%d``, + matching the specification's Voltage (1) / Current (2) / Power (3) / + Energy (4) order. """ + assert LOAD_DIMENSION_ALL == 0 + assert LOAD_DIMENSION_VOLTAGE == 1 + assert LOAD_DIMENSION_CURRENT == 2 + assert LOAD_DIMENSION_POWER == 3 + assert LOAD_DIMENSION_ENERGY == 4 for gateway in ALIGNED_GATEWAYS: registers = { d: next( @@ -177,13 +193,23 @@ def test_measurement_requests_read_distinct_registers(): ) for d in range(5) } - assert registers[1] == ["$03AA001613"] - assert registers[2] == ["$03AA001612"] - assert registers[3] == ["$03AA001610", "$03AA001611"] - assert registers[4] == ["$03AA001606", "$03AA001607", "$03AA001608", "$03AA001609"] - others = registers[1] + registers[2] + registers[3] + registers[4] - assert registers[0][:4] == others[:4] - assert set(registers[0]) <= set(others) + assert registers[LOAD_DIMENSION_VOLTAGE] == ["$03AA001613"] + assert registers[LOAD_DIMENSION_CURRENT] == ["$03AA001612"] + assert registers[LOAD_DIMENSION_POWER] == ["$03AA001610", "$03AA001611"] + assert registers[LOAD_DIMENSION_ENERGY] == [ + "$03AA001606", + "$03AA001607", + "$03AA001608", + "$03AA001609", + ] + others = ( + registers[LOAD_DIMENSION_VOLTAGE] + + registers[LOAD_DIMENSION_CURRENT] + + registers[LOAD_DIMENSION_POWER] + + registers[LOAD_DIMENSION_ENERGY] + ) + assert registers[LOAD_DIMENSION_ALL][:4] == others[:4] + assert set(registers[LOAD_DIMENSION_ALL]) <= set(others) @pytest.mark.parametrize( @@ -226,28 +252,28 @@ def test_measurement_reply_all_measurements(): 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.voltage == 230 + assert event.current == 12 + assert event.power == 2760 + assert event.energy == 1520 assert event.state is None assert event.human_readable_log == ( - "Load control unit all measurements: power=230 W, voltage=12 V, " - "current=2760.0 A, power factor=1520.0." + "Load control unit all measurements: voltage=230 V, current=12 A, " + "power=2760 W, energy=1520." ) @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"), + ("*#3*10*1*230##", LOAD_DIMENSION_VOLTAGE, "voltage", 230, "voltage: 230 V."), + ("*#3*10*2*12##", LOAD_DIMENSION_CURRENT, "current", 12, "current: 12 A."), + ("*#3*10*3*2760##", LOAD_DIMENSION_POWER, "power", 2760, "power: 2760 W."), + ("*#3*10*4*1520##", LOAD_DIMENSION_ENERGY, "energy", 1520, "energy: 1520."), ], ) def test_measurement_reply_individual_dimensions( - frame: str, dim: int, attr: str, expected_value: int | float, log_phrase: str + frame: str, dim: int, attr: str, expected_value: int, log_phrase: str ): event = OWNEvent.parse(frame) assert isinstance(event, OWNLoadEvent) @@ -257,30 +283,27 @@ def test_measurement_reply_individual_dimensions( assert log_phrase in event.human_readable_log -def test_safe_helpers(): +def test_safe_int(): 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*##") + empty_voltage = OWNEvent.parse("*#3*10*1*##") + assert isinstance(empty_voltage, OWNLoadEvent) + assert empty_voltage.voltage is None + + empty_power = OWNEvent.parse("*#3*10*3*##") 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.voltage == 230 assert partial.current is None - assert partial.power_factor is None + assert partial.power is None + assert partial.energy is None def test_unnamed_dimension_and_other_where_are_not_measurements():