diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 6a878c6..46a4175 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -168,9 +168,17 @@ 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 - self._human_readable_log = f"Sensor {self._sensor} automatic updates every {self._update_interval} minutes (0 = stopped)." # pylint: disable=line-too-long + 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: + 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 +196,27 @@ 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. + + 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, 0..255 per Legrand WHO 18), 0 when stopped.""" 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 @@ -224,11 +247,36 @@ 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_param and self._where_param == ["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 @@ -241,10 +289,13 @@ 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) - duration = 255 if duration > 255 else duration + where = _normalize_actuator_where(where) + 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 @@ -256,16 +307,32 @@ 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 + @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 ) -> 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: @@ -278,7 +345,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}." @@ -289,7 +356,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) @@ -307,7 +374,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}." @@ -318,14 +385,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 4ff9be0..f61e211 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 @@ -54,3 +63,81 @@ 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 + + +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" + 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 is reporting an active power draw of 0 W." + + actuator_evt2 = OWNEnergyEvent("*#18*77#0*113*161##") + 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 + + # 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##" + 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##" + + +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##" + +