fix(energy): parse the WHO 18 dimension 1200 automatic-update reply

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",

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

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##"