diff --git a/OWNd/message/__init__.py b/OWNd/message/__init__.py index 5d4aa9c..2029159 100644 --- a/OWNd/message/__init__.py +++ b/OWNd/message/__init__.py @@ -40,6 +40,7 @@ MESSAGE_TYPE_ACTIVE_POWER, MESSAGE_TYPE_CURRENT_DAY_CONSUMPTION, MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION, + MESSAGE_TYPE_AUTO_UPDATE_INTERVAL, MESSAGE_TYPE_DAILY_CONSUMPTION, MESSAGE_TYPE_ENERGY_TOTALIZER, MESSAGE_TYPE_HOURLY_CONSUMPTION, @@ -201,6 +202,7 @@ "MESSAGE_TYPE_MONTHLY_CONSUMPTION", "MESSAGE_TYPE_CURRENT_DAY_CONSUMPTION", "MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION", + "MESSAGE_TYPE_AUTO_UPDATE_INTERVAL", "_integer_value", "OWNEnergyEvent", "OWNEnergyCommand", diff --git a/OWNd/message/energy.py b/OWNd/message/energy.py index 5be4132..6a878c6 100644 --- a/OWNd/message/energy.py +++ b/OWNd/message/energy.py @@ -17,6 +17,7 @@ MESSAGE_TYPE_MONTHLY_CONSUMPTION = "monthly_consumption" MESSAGE_TYPE_CURRENT_DAY_CONSUMPTION = "current_day_partial_consumption" MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION = "current_month_partial_consumption" +MESSAGE_TYPE_AUTO_UPDATE_INTERVAL = "auto_update_interval" def _integer_value( @@ -45,6 +46,8 @@ def __init__(self, data: str) -> None: self._current_day_partial_consumption = 0 self._monthly_consumption: dict[str, Any] = {} self._current_month_partial_consumption = 0 + self._update_interval: int | None = None + self._energy_type: int | None = None if not where.startswith(("1", "5", "7")): return @@ -158,6 +161,16 @@ def __init__(self, data: str) -> None: self._dimension_value, 0 ) self._human_readable_log = f"Sensor {self._sensor} is reporting a power consumption of {self._monthly_consumption['value']} Wh for {self._monthly_consumption['date'].strftime('%B %Y')}." # pylint: disable=line-too-long + elif self._dimension == 1200 and self._dimension_value: + # *#18*W*1200#type*time##: the meter confirms (time > 0) or + # ends (time = 0) the automatic power updates; libqtdevices + # re-arms the stream on time = 0 (energy_device.cpp:289-322, + # 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 elif self._dimension == 53: self._type = MESSAGE_TYPE_CURRENT_MONTH_CONSUMPTION self._current_month_partial_consumption = _integer_value( @@ -173,6 +186,16 @@ def message_type(self) -> str | None: def sensor(self) -> str: return self._sensor + @property + def update_interval(self) -> int | None: + """Time field of a 1200 reply (minutes per Legrand WHO 18), 0 when stopped.""" + return self._update_interval + + @property + def energy_type(self) -> int | None: + """Energy type of a 1200 reply (1 electricity, 2 gas, 3 heat, 4 water).""" + return self._energy_type + @property def active_power(self) -> int: return self._active_power @@ -210,14 +233,34 @@ def human_readable_log(self) -> str: class OWNEnergyCommand(OWNCommand): @classmethod def start_sending_instant_power( - cls, where: str | int, duration: int = 65 + cls, where: str | int, duration: int = 65, energy_type: int = 1 ) -> OWNEnergyCommand: + """Start automatic updates: *#18*W*#1200#type*time##. + + ``duration`` is the Time field (1..255, minutes per Legrand WHO 18; + 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 - message = cls(f"*#18*{where}*#1200#1*{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 return message + @classmethod + def stop_sending_instant_power( + cls, where: str | int, energy_type: int = 1 + ) -> OWNEnergyCommand: + """Stop automatic updates: *#18*W*#1200#type*0## (Time = 0). + + 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) + 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 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 new file mode 100644 index 0000000..4ff9be0 --- /dev/null +++ b/tests/test_who18_auto_update_reply.py @@ -0,0 +1,56 @@ +"""WHO 18 dimension 1200 reply: *#18*W*1200#type*time## confirms (time > 0) +or ends (time = 0) the automatic power updates. libqtdevices TS10_1_0_23 +energy_device.cpp:111 (_DIM_STATE_UPDATE_INTERVAL = 1200), :289-322 +(handleAutomaticUpdate re-arms on time = 0), :719-726, :50-56 (energy types). +Firmware replay on bt_supervisione (MyHomeServer1, F454, F459, F461): +*#18*51*#1200#1*255## ACK bus D1 A1 02 32 00 02 1D FF, #1200#1*0 ... 1D 00, +#1200#2*255 ... 82 FF; the bare *#18*51*1200## is NACK with no bus frame.""" + +from OWNd.message import ( + MESSAGE_TYPE_AUTO_UPDATE_INTERVAL, + OWNEnergyCommand, + OWNEnergyEvent, + OWNEvent, +) + + +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.update_interval == 255 + assert msg.energy_type == 1 + assert "255 minutes" in msg.human_readable_log + + +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.update_interval == 0 + + +def test_other_dimensions_are_untouched(): + msg = OWNEvent.parse("*#18*51*113*1234##") + assert msg.update_interval is None + assert msg.energy_type is None + assert msg.message_type != MESSAGE_TYPE_AUTO_UPDATE_INTERVAL + + +def test_message_type_is_exported(): + import OWNd.message as message + + assert "MESSAGE_TYPE_AUTO_UPDATE_INTERVAL" in message.__all__ + + +def test_stop_builder_is_symmetrical_to_start(): + start = OWNEnergyCommand.start_sending_instant_power(51, 255) + stop = OWNEnergyCommand.stop_sending_instant_power(51) + assert str(start) == "*#18*51*#1200#1*255##" + assert str(stop) == "*#18*51*#1200#1*0##" + assert "Stopping" in stop.human_readable_log + + +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##"