diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index b8c503f..fe7f4cc 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -147,6 +147,22 @@ OWNLightingCommand, OWNLightingEvent, ) +from .load import ( + LOAD_CONTROL_UNIT, + LOAD_DIMENSION_ALL, + LOAD_DIMENSION_CURRENT, + LOAD_DIMENSION_ENERGY, + LOAD_DIMENSION_POWER, + LOAD_DIMENSION_VOLTAGE, + 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 +198,21 @@ # 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", + "LOAD_DIMENSION_ALL", + "LOAD_DIMENSION_VOLTAGE", + "LOAD_DIMENSION_CURRENT", + "LOAD_DIMENSION_POWER", + "LOAD_DIMENSION_ENERGY", + "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..05dabdb --- /dev/null +++ b/OWNd/message/load.py @@ -0,0 +1,273 @@ +"""WHO 3: Load control (load shedding) events and commands. + +Evidence, in order of weight: + +- 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 + +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) + +LOAD_DIMENSION_ALL = 0 +LOAD_DIMENSION_VOLTAGE = 1 +LOAD_DIMENSION_CURRENT = 2 +LOAD_DIMENSION_POWER = 3 +LOAD_DIMENSION_ENERGY = 4 + +_STATE_TEXT = { + LOAD_STATE_DISABLED: "disabled", + LOAD_STATE_ENABLED: "enabled", + LOAD_STATE_FORCED: "forced", + LOAD_STATE_FORCING_REMOVED: "no longer forced", +} + +# 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_VOLTAGE: "voltage", + LOAD_DIMENSION_CURRENT: "current", + LOAD_DIMENSION_POWER: "power", + LOAD_DIMENSION_ENERGY: "energy", +} + +_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 _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] = [] + self._voltage: int | 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" + 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) + 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_int(self._dimension_value[0]) + self._human_readable_log = ( + f"Load control unit current: {self._current} A." + ) + 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 energy: {self._energy}." + ) + 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._current = _safe_int(self._dimension_value[1]) + if len(self._dimension_value) >= 3: + self._power = _safe_int(self._dimension_value[2]) + if len(self._dimension_value) >= 4: + self._energy = _safe_int(self._dimension_value[3]) + self._human_readable_log = ( + f"Load control unit all measurements: " + f"voltage={self._voltage} V, current={self._current} A, " + f"power={self._power} W, energy={self._energy}." + ) + 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: + """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 + + @property + def voltage(self) -> int | None: + """Measured voltage in V (dimension 1, or index 0 of dimension 0), None if not reported.""" + return self._voltage + + @property + 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(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): + """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/OWNd/profiles.py b/OWNd/profiles.py index dbda8d4..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, @@ -519,6 +541,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 +562,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/README.md b/README.md index 49b9506..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 | 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 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` | @@ -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 b127e9a..279bd1a 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 with typed voltage, current, power, and energy properties (`WHO_3.pdf` v1.0.0 & `bt_energia` register mappings)", + dispatch_who_keys=(3,), ), WhoCatalogDefinition( who="**4**", diff --git a/tests/test_profiles.py b/tests/test_profiles.py index d92846a..b93baff 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -97,6 +97,19 @@ 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 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: @@ -107,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 new file mode 100644 index 0000000..6fe1ac6 --- /dev/null +++ b/tests/test_who3_load_control.py @@ -0,0 +1,349 @@ +"""WHO 3 load control, pinned to the firmware oracle first and the specification second. + +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 + +import json +from pathlib import Path +from typing import Any + +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, + LOAD_STATE_FORCING_REMOVED, + OWNCommand, + OWNEvent, + OWNLoadCommand, + OWNLoadEvent, + OWNMessage, +) +from OWNd.message.load import _safe_int + +ORACLE = json.loads( + (Path(__file__).parent / "golden" / "firmware_oracle.json").read_text(encoding="utf-8") +)["verdicts"] +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*"))) + + +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: + """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: + 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_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 + + +@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. + + 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( + _bus_text(r) for r in _aligned(f"*#3*10*{d}##") if r["product"] == gateway + ) + for d in range(5) + } + 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( + ("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_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.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: voltage=230 V, current=12 A, " + "power=2760 W, energy=1520." + ) + + +@pytest.mark.parametrize( + ("frame", "dim", "attr", "expected_value", "log_phrase"), + [ + ("*#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, 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_int(): + assert _safe_int(None) is None + assert _safe_int("not_a_number") is None + assert _safe_int("100") == 100 + + +def test_measurement_reply_empty_or_partial(): + 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 + + partial = OWNEvent.parse("*#3*10*0*230##") + assert isinstance(partial, OWNLoadEvent) + assert partial.voltage == 230 + assert partial.current is None + assert partial.power is None + assert partial.energy 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']." + ) + 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)