From a72815f828049c213acfd0a8227e12fc2c452aa0 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 9 Oct 2026 16:52:25 +0200 Subject: [PATCH 1/6] fix(energy): align auto-update telemetry log wording and stream lifecycle properties --- OWNd/message/energy.py | 21 ++++++++++++++++++--- tests/test_who18_auto_update_reply.py | 11 ++++++++++- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 6a878c6..fffa79e 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -170,7 +170,10 @@ def __init__(self, data: str) -> None: self._type = MESSAGE_TYPE_AUTO_UPDATE_INTERVAL self._update_interval = _integer_value(self._dimension_value, 0) self._energy_type = _integer_value(self._dimension_param, 0, 0) if self._dimension_param else None - self._human_readable_log = f"Sensor {self._sensor} automatic updates every {self._update_interval} minutes (0 = stopped)." # pylint: disable=line-too-long + if self._update_interval == 0: + self._human_readable_log = f"Sensor {self._sensor} automatic updates stopped." + else: + self._human_readable_log = f"Sensor {self._sensor} automatic updates for {self._update_interval} minutes." elif self._dimension == 53: self._type = MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION self._current_month_partial_consumption = _integer_value( @@ -188,12 +191,24 @@ def sensor(self) -> str: @property def update_interval(self) -> int | None: - """Time field of a 1200 reply (minutes per Legrand WHO 18), 0 when stopped.""" + """Time field of a 1200 reply (stream duration in minutes per Legrand WHO 18), 0 when stopped.""" + return self._update_interval + + @property + def stream_duration(self) -> int | None: + """Time field of a 1200 reply (stream duration in minutes per Legrand WHO 18), 0 when stopped. Alias for update_interval.""" return self._update_interval + @property + def is_streaming_active(self) -> bool | None: + """True if automatic updates are streaming (> 0 minutes), False if stopped (0 minutes), None if not a 1200 event.""" + if self._update_interval is None: + return None + return self._update_interval > 0 + @property def energy_type(self) -> int | None: - """Energy type of a 1200 reply (1 electricity, 2 gas, 3 heat, 4 water).""" + """Energy type of a 1200 reply (1 electricity per Legrand WHO 18; 2 gas, 3 heat, 4 water per libqtdevices).""" return self._energy_type @property diff --git a/tests/test_who18_auto_update_reply.py b/tests/test_who18_auto_update_reply.py index 4ff9be0..83a6cdd 100644 --- a/tests/test_who18_auto_update_reply.py +++ b/tests/test_who18_auto_update_reply.py @@ -18,20 +18,29 @@ def test_auto_update_confirmation_is_parsed(): msg = OWNEvent.parse("*#18*51*1200#1*255##") assert isinstance(msg, OWNEnergyEvent) assert msg.message_type == MESSAGE_TYPE_AUTO_UPDATE_INTERVAL + assert msg.sensor == "1" assert msg.update_interval == 255 + assert msg.stream_duration == 255 + assert msg.is_streaming_active is True assert msg.energy_type == 1 - assert "255 minutes" in msg.human_readable_log + assert msg.human_readable_log == "Sensor 1 automatic updates for 255 minutes." def test_auto_update_stop_is_parsed(): msg = OWNEvent.parse("*#18*51*1200#1*0##") assert msg.message_type == MESSAGE_TYPE_AUTO_UPDATE_INTERVAL + assert msg.sensor == "1" assert msg.update_interval == 0 + assert msg.stream_duration == 0 + assert msg.is_streaming_active is False + assert msg.human_readable_log == "Sensor 1 automatic updates stopped." def test_other_dimensions_are_untouched(): msg = OWNEvent.parse("*#18*51*113*1234##") assert msg.update_interval is None + assert msg.stream_duration is None + assert msg.is_streaming_active is None assert msg.energy_type is None assert msg.message_type != MESSAGE_TYPE_AUTO_UPDATE_INTERVAL From 94979b21b27f0033e912ffbf2147638bd0bdd8ed Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:02:16 +0200 Subject: [PATCH 2/6] fix(energy): clamp interval boundaries and clarify stream_duration preference --- OWNd/message/energy.py | 23 +++++++++++++++++------ tests/test_who18_auto_update_reply.py | 19 +++++++++++++++++++ 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index fffa79e..181d64a 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -168,8 +168,13 @@ def __init__(self, data: str) -> None: # 719-726). type: 1 electricity, 2 gas, 3 heat, 4 water # (energy_device.cpp:50-56). self._type = MESSAGE_TYPE_AUTO_UPDATE_INTERVAL - self._update_interval = _integer_value(self._dimension_value, 0) - self._energy_type = _integer_value(self._dimension_param, 0, 0) if self._dimension_param else None + parsed_interval = _integer_value(self._dimension_value, 0) + self._update_interval = max(0, parsed_interval) + self._energy_type = ( + max(0, _integer_value(self._dimension_param, 0, 0)) + if self._dimension_param + else None + ) if self._update_interval == 0: self._human_readable_log = f"Sensor {self._sensor} automatic updates stopped." else: @@ -191,12 +196,15 @@ def sensor(self) -> str: @property def update_interval(self) -> int | None: - """Time field of a 1200 reply (stream duration in minutes per Legrand WHO 18), 0 when stopped.""" + """Time field of a 1200 reply (stream duration in minutes per Legrand WHO 18), 0 when stopped. + + Retained for backward compatibility; prefer :attr:`stream_duration`. + """ return self._update_interval @property def stream_duration(self) -> int | None: - """Time field of a 1200 reply (stream duration in minutes per Legrand WHO 18), 0 when stopped. Alias for update_interval.""" + """Time field of a 1200 reply (stream duration in minutes, 0..255 per Legrand WHO 18), 0 when stopped.""" return self._update_interval @property @@ -257,9 +265,12 @@ def start_sending_instant_power( electricity, 2 gas, 3 heat, 4 water (libqtdevices energy_device.cpp:50-56). """ where = f"{where}#0" if str(where).startswith("7") else str(where) - duration = 255 if duration > 255 else duration + duration = 255 if duration > 255 else max(0, int(duration)) message = cls(f"*#18*{where}*#1200#{energy_type}*{duration}##") - message._human_readable_log = f"Requesting instant power draw update from sensor {where} for {duration} minutes." # pylint: disable=line-too-long + if duration == 0: + message._human_readable_log = f"Stopping instant power draw updates from sensor {where}." # pylint: disable=line-too-long + else: + message._human_readable_log = f"Requesting instant power draw update from sensor {where} for {duration} minutes." # pylint: disable=line-too-long return message @classmethod diff --git a/tests/test_who18_auto_update_reply.py b/tests/test_who18_auto_update_reply.py index 83a6cdd..bf66a09 100644 --- a/tests/test_who18_auto_update_reply.py +++ b/tests/test_who18_auto_update_reply.py @@ -63,3 +63,22 @@ def test_builders_take_the_energy_type_and_the_7xx_address(): assert str(OWNEnergyCommand.start_sending_instant_power(51, 255, 2)) == "*#18*51*#1200#2*255##" assert str(OWNEnergyCommand.stop_sending_instant_power(51, 2)) == "*#18*51*#1200#2*0##" assert str(OWNEnergyCommand.stop_sending_instant_power("71")) == "*#18*71#0*#1200#1*0##" + + +def test_builder_duration_zero_or_negative(): + stop_via_start = OWNEnergyCommand.start_sending_instant_power(51, 0) + assert str(stop_via_start) == "*#18*51*#1200#1*0##" + assert "Stopping" in stop_via_start.human_readable_log + + negative_clamped = OWNEnergyCommand.start_sending_instant_power(51, -10) + assert str(negative_clamped) == "*#18*51*#1200#1*0##" + assert "Stopping" in negative_clamped.human_readable_log + + +def test_negative_interval_clamped_to_stopped(): + event = OWNEnergyEvent("*#18*51*1200#1*0##") + # Simulate a corrupted non-standard negative value in _dimension_value + event._dimension_value = ["-5"] + parsed_interval = max(0, int(event._dimension_value[0])) + assert parsed_interval == 0 + From 4c3f8aac67b1130cab4b75f494dc7ec33bfbfe61 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 9 Oct 2026 18:37:36 +0200 Subject: [PATCH 3/6] fix(energy): preserve #0 actuator suffix in where and sensor addresses --- OWNd/message/energy.py | 30 ++++++++++++++++------- tests/test_who18_auto_update_reply.py | 34 +++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 8 deletions(-) diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 181d64a..4852ba8 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -36,6 +36,8 @@ def __init__(self, data: str) -> None: super().__init__(data) self._type: str | None = None + if self._where and self._where_param: + self._where = f"{self._where}#{'#'.join(self._where_param)}" where = self._where or "" self._sensor = where[1:] if len(where) > 1 else where self._active_power = 0 @@ -247,11 +249,23 @@ def monthly_consumption(self) -> dict[str, Any]: def current_month_partial_consumption(self) -> int: return self._current_month_partial_consumption + @property + def is_actuator(self) -> bool: + """True if the event originates from an energy management actuator (7N#0).""" + return bool(self._where and (self._where.startswith("7") or self._where.endswith("#0"))) + @property def human_readable_log(self) -> str: return self._human_readable_log +def _normalize_actuator_where(where: str | int) -> str: + """Normalize actuator address by ensuring #0 suffix per WHO 18 specification.""" + target = str(where).strip() + if target.startswith("7") and not target.endswith("#0"): + return f"{target}#0" + return target + class OWNEnergyCommand(OWNCommand): @classmethod @@ -264,7 +278,7 @@ def start_sending_instant_power( the meter answers *#18*W*1200#type*time##). ``energy_type`` 1 is electricity, 2 gas, 3 heat, 4 water (libqtdevices energy_device.cpp:50-56). """ - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) duration = 255 if duration > 255 else max(0, int(duration)) message = cls(f"*#18*{where}*#1200#{energy_type}*{duration}##") if duration == 0: @@ -282,7 +296,7 @@ def stop_sending_instant_power( Firmware replay on bt_supervisione: *#18*51*#1200#1*0## ACK, bus D1 A1 02 32 00 02 1D 00; the meter then reports *#18*W*1200#type*0##. """ - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) message = cls(f"*#18*{where}*#1200#{energy_type}*0##") message._human_readable_log = f"Stopping instant power draw updates from sensor {where}." # pylint: disable=line-too-long return message @@ -291,7 +305,7 @@ def stop_sending_instant_power( def get_hourly_consumption( cls, where: str | int, date: datetime.date ) -> OWNEnergyCommand | None: - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) today = datetime.date.today() one_year_ago = today - relativedelta(years=1) if date < one_year_ago: @@ -304,7 +318,7 @@ def get_hourly_consumption( @classmethod def get_partial_daily_consumption(cls, where: str | int) -> OWNEnergyCommand: - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) message = cls(f"*#18*{where}*54##") message._human_readable_log = ( f"Requesting today's partial power consumption from sensor {where}." @@ -315,7 +329,7 @@ def get_partial_daily_consumption(cls, where: str | int) -> OWNEnergyCommand: def get_daily_consumption( cls, where: str | int, year: int, month: int ) -> OWNEnergyCommand | None: - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) today = datetime.date.today() one_year_ago = today - relativedelta(years=1) two_year_ago = today - relativedelta(years=2) @@ -333,7 +347,7 @@ def get_daily_consumption( @classmethod def get_partial_monthly_consumption(cls, where: str | int) -> OWNEnergyCommand: - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) message = cls(f"*#18*{where}*53##") message._human_readable_log = ( f"Requesting this month's partial power consumption from sensor {where}." @@ -344,14 +358,14 @@ def get_partial_monthly_consumption(cls, where: str | int) -> OWNEnergyCommand: def get_monthly_consumption( cls, where: str | int, year: int, month: int ) -> OWNEnergyCommand: - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) message = cls(f"*#18*{where}*52#{str(year)[2:]}#{month}##") message._human_readable_log = f"Requesting monthly power consumption for {year}-{month} from sensor {where}." # pylint: disable=line-too-long return message @classmethod def get_total_consumption(cls, where: str | int) -> OWNEnergyCommand: - where = f"{where}#0" if str(where).startswith("7") else str(where) + where = _normalize_actuator_where(where) message = cls(f"*#18*{where}*51##") message._human_readable_log = ( f"Requesting total power consumption from sensor {where}." diff --git a/tests/test_who18_auto_update_reply.py b/tests/test_who18_auto_update_reply.py index bf66a09..ada7469 100644 --- a/tests/test_who18_auto_update_reply.py +++ b/tests/test_who18_auto_update_reply.py @@ -82,3 +82,37 @@ def test_negative_interval_clamped_to_stopped(): parsed_interval = max(0, int(event._dimension_value[0])) assert parsed_interval == 0 + +def test_actuator_address_preserves_hash_zero(): + # Authentic actuator frames from issue #669 trace (*#18*76#0*113*0##, *#18*77#0*113*161##) + actuator_evt = OWNEnergyEvent("*#18*76#0*113*0##") + assert actuator_evt.where == "76#0" + assert actuator_evt.sensor == "6#0" + assert actuator_evt.is_actuator is True + assert actuator_evt.active_power == 0 + assert actuator_evt.human_readable_log == "Sensor 6#0 is reporting an active power draw of 0 W." + + actuator_evt2 = OWNEnergyEvent("*#18*77#0*113*161##") + assert actuator_evt2.where == "77#0" + assert actuator_evt2.sensor == "7#0" + assert actuator_evt2.is_actuator is True + assert actuator_evt2.active_power == 161 + + # Meter frame does not collide with actuator + meter_evt = OWNEnergyEvent("*#18*56*113*0##") + assert meter_evt.where == "56" + assert meter_evt.sensor == "6" + assert meter_evt.is_actuator is False + assert meter_evt.active_power == 0 + assert meter_evt.human_readable_log == "Sensor 6 is reporting an active power draw of 0 W." + + # Command builder preserves address whether #0 was already provided or inferred + cmd1 = OWNEnergyCommand.stop_sending_instant_power("76") + assert str(cmd1) == "*#18*76#0*#1200#1*0##" + cmd2 = OWNEnergyCommand.stop_sending_instant_power("76#0") + assert str(cmd2) == "*#18*76#0*#1200#1*0##" + cmd3 = OWNEnergyCommand.start_sending_instant_power("76#0", 60) + assert str(cmd3) == "*#18*76#0*#1200#1*60##" + cmd4 = OWNEnergyCommand.get_total_consumption("76#0") + assert str(cmd4) == "*#18*76#0*51##" + From d30129fd6e3638f714dd9f63a91189cf34c8d990 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 9 Oct 2026 18:57:12 +0200 Subject: [PATCH 4/6] fix(energy): preserve base where contract and expose actuator address via target_address --- OWNd/message/energy.py | 17 ++++++++++++++--- tests/test_who18_auto_update_reply.py | 12 +++++++----- 2 files changed, 21 insertions(+), 8 deletions(-) diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 4852ba8..1db724b 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -36,8 +36,6 @@ def __init__(self, data: str) -> None: super().__init__(data) self._type: str | None = None - if self._where and self._where_param: - self._where = f"{self._where}#{'#'.join(self._where_param)}" where = self._where or "" self._sensor = where[1:] if len(where) > 1 else where self._active_power = 0 @@ -249,10 +247,23 @@ def monthly_consumption(self) -> dict[str, Any]: def current_month_partial_consumption(self) -> int: return self._current_month_partial_consumption + @property + def target_address(self) -> str: + """The full target address including any actuator channel suffix (#0).""" + if self._where and self._where_param: + return f"{self._where}#{'#'.join(self._where_param)}" + return self._where or "" + @property def is_actuator(self) -> bool: """True if the event originates from an energy management actuator (7N#0).""" - return bool(self._where and (self._where.startswith("7") or self._where.endswith("#0"))) + return bool( + self._where + and ( + self._where.startswith("7") + or (self._where_param and self._where_param == ["0"]) + ) + ) @property def human_readable_log(self) -> str: diff --git a/tests/test_who18_auto_update_reply.py b/tests/test_who18_auto_update_reply.py index ada7469..f49b35e 100644 --- a/tests/test_who18_auto_update_reply.py +++ b/tests/test_who18_auto_update_reply.py @@ -86,15 +86,17 @@ def test_negative_interval_clamped_to_stopped(): def test_actuator_address_preserves_hash_zero(): # Authentic actuator frames from issue #669 trace (*#18*76#0*113*0##, *#18*77#0*113*161##) actuator_evt = OWNEnergyEvent("*#18*76#0*113*0##") - assert actuator_evt.where == "76#0" - assert actuator_evt.sensor == "6#0" + assert actuator_evt.where == "76" + assert actuator_evt.target_address == "76#0" + assert actuator_evt.sensor == "6" assert actuator_evt.is_actuator is True assert actuator_evt.active_power == 0 - assert actuator_evt.human_readable_log == "Sensor 6#0 is reporting an active power draw of 0 W." + assert actuator_evt.human_readable_log == "Sensor 6 is reporting an active power draw of 0 W." actuator_evt2 = OWNEnergyEvent("*#18*77#0*113*161##") - assert actuator_evt2.where == "77#0" - assert actuator_evt2.sensor == "7#0" + assert actuator_evt2.where == "77" + assert actuator_evt2.target_address == "77#0" + assert actuator_evt2.sensor == "7" assert actuator_evt2.is_actuator is True assert actuator_evt2.active_power == 161 From df6b0a02d0f8a8f0a6a2be16efb0d0f2aa1a67a8 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 9 Oct 2026 19:05:19 +0200 Subject: [PATCH 5/6] test(energy): cover target_address on meter frames and fallback to empty string --- tests/test_who18_auto_update_reply.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/test_who18_auto_update_reply.py b/tests/test_who18_auto_update_reply.py index f49b35e..9368dcb 100644 --- a/tests/test_who18_auto_update_reply.py +++ b/tests/test_who18_auto_update_reply.py @@ -103,11 +103,16 @@ def test_actuator_address_preserves_hash_zero(): # Meter frame does not collide with actuator meter_evt = OWNEnergyEvent("*#18*56*113*0##") assert meter_evt.where == "56" + assert meter_evt.target_address == "56" assert meter_evt.sensor == "6" assert meter_evt.is_actuator is False assert meter_evt.active_power == 0 assert meter_evt.human_readable_log == "Sensor 6 is reporting an active power draw of 0 W." + # Fallback to empty string if where is empty + meter_evt._where = None + assert meter_evt.target_address == "" + # Command builder preserves address whether #0 was already provided or inferred cmd1 = OWNEnergyCommand.stop_sending_instant_power("76") assert str(cmd1) == "*#18*76#0*#1200#1*0##" From 9b61ed9080b27edeca7bdd699420520ee8af911d Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Fri, 9 Oct 2026 21:51:14 +0200 Subject: [PATCH 6/6] feat(energy): add request_active_power and get_instant_power command builders --- OWNd/message/energy.py | 16 ++++++++++++++++ tests/test_who18_auto_update_reply.py | 18 ++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 1db724b..46a4175 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -312,6 +312,22 @@ def stop_sending_instant_power( message._human_readable_log = f"Stopping instant power draw updates from sensor {where}." # pylint: disable=line-too-long return message + @classmethod + def request_active_power(cls, where: str | int) -> OWNEnergyCommand: + """Request instantaneous active power: *#18*W*113##. + + Legrand WHO 18 section 5.2.18 Dimension 113. + """ + where = _normalize_actuator_where(where) + message = cls(f"*#18*{where}*113##") + message._human_readable_log = ( + f"Requesting active power from sensor {where}." + ) + return message + + # Alias for API symmetry with get_total_consumption / downstream integrations + get_instant_power = request_active_power + @classmethod def get_hourly_consumption( cls, where: str | int, date: datetime.date diff --git a/tests/test_who18_auto_update_reply.py b/tests/test_who18_auto_update_reply.py index 9368dcb..f61e211 100644 --- a/tests/test_who18_auto_update_reply.py +++ b/tests/test_who18_auto_update_reply.py @@ -123,3 +123,21 @@ def test_actuator_address_preserves_hash_zero(): cmd4 = OWNEnergyCommand.get_total_consumption("76#0") assert str(cmd4) == "*#18*76#0*51##" + +def test_request_active_power_builder(): + cmd = OWNEnergyCommand.request_active_power(51) + assert str(cmd) == "*#18*51*113##" + assert cmd.human_readable_log == "Requesting active power from sensor 51." + + cmd_alias = OWNEnergyCommand.get_instant_power("51") + assert str(cmd_alias) == "*#18*51*113##" + assert cmd_alias.human_readable_log == "Requesting active power from sensor 51." + + # 7x address receives #0 normalization + cmd_7x = OWNEnergyCommand.request_active_power("71") + assert str(cmd_7x) == "*#18*71#0*113##" + + cmd_7x_preset = OWNEnergyCommand.request_active_power("71#0") + assert str(cmd_7x_preset) == "*#18*71#0*113##" + +