Improved water leak sensor device, added sample and version bump

Author: albertogeniola

.version

@@ -1 +1 @@
-0.4.10.3alpha3
+0.4.10.3

README.md

@@ -32,7 +32,7 @@ Due to the popularity of the library, I've decided to list it publicly on the Pi
 So, the installation is as simple as typing the following command:
 
 ```bash
-pip install meross_iot==0.4.10.2
+pip install meross_iot==0.4.10.3
 ```
 
 ## Usage & Full Documentation
@@ -124,6 +124,7 @@ The list of tested devices is the following:
 - MS100F (temperature/humidity sensor)
 - MSS710
 - MSXH0 (Smart Humidifier)
+- MS400 and MS405 (Water leak sensors)
 
 I'd like to thank all the people who contributed to the early stage of library development,
 who stimulated me to continue the development and making this library support more devices.
@@ -199,14 +200,17 @@ Anyway, feel free to contribute via donations!
 </p>
 
 ## Changelog
-### 0.4.10.2
-- Allows to pass a specific ssl context to the manager to perform MQTT ssl connections using that SSL context. 
-This is a breaking change: signature of MerossManager() has changed.
-
+### 0.4.10.3
+- Add support for water leak sensors such as MS400 and MS405
+- Add water-leak sensor examples
 
 <details>
     <summary>Older</summary>
 
+### 0.4.10.2
+- Allows to pass a specific ssl context to the manager to perform MQTT ssl connections using that SSL context. 
+This is a breaking change: signature of MerossManager() has changed.
+
 ### 0.4.9.2
 - Partially Address #569, fix error when dealing with unknown sub-devices. Water leak sensor not supported yet.
 

examples/water_leak_sensor.py

@@ -1,13 +1,26 @@
 import asyncio
 import os
+from typing import List
 
+from meross_iot.controller.subdevice import Ms405Sensor
 from meross_iot.http_api import MerossHttpClient
 from meross_iot.manager import MerossManager
+from meross_iot.model.enums import Namespace, OnlineStatus
 
 EMAIL = os.environ.get('MEROSS_EMAIL') or "YOUR_MEROSS_CLOUD_EMAIL"
 PASSWORD = os.environ.get('MEROSS_PASSWORD') or "YOUR_MEROSS_CLOUD_PASSWORD"
 
 
+async def water_leak_event(namespace: Namespace, data: dict, device_internal_id: str, *args, **kwargs):
+    print("An event has occurred!")
+    if namespace == Namespace.CONTROL_ALARM:
+        print(f"Alarm occurred! Event data: {data}")
+    elif namespace == Namespace.HUB_SENSOR_WATERLEAK:
+        print(f"Water leak occurred! Event data: {data}")
+    else:
+        print(f"Another event occurred: {namespace.value}, Event data: {data}")
+
+
 async def main():
     # Setup the HTTP client API from user-password
     http_api_client = await MerossHttpClient.async_from_user_password(email=EMAIL, password=PASSWORD, api_base_url="https://iot.meross.com")
@@ -19,26 +32,29 @@ async def main():
     # Retrieve all the MS100 devices that are registered on this account
     await manager.async_device_discovery()
 
-    msh400 = manager.find_devices(device_type="msh400")
-    hub = msh400[0]
-
-
-    water_leak_sensors = manager.find_devices(device_type="ms405")
+    # Retrieve water leak sensors. Can either be ms400 or ms405
+    water_leak_sensors: List[Ms405Sensor] = manager.find_devices(device_class=Ms405Sensor, online_status=OnlineStatus.ONLINE)
 
     if len(water_leak_sensors) < 1:
-        print("No MSH405 sensor found...")
+        print("No online water leak sensors found!")
     else:
-        dev = water_leak_sensors[0]
+        # Let's register an event handle to quickly react in case of water leaks
+        for sensor in water_leak_sensors:
+            sensor.register_push_notification_handler_coroutine(water_leak_event)
 
         # Manually force and update to retrieve the latest temperature sensed from
         # the device. This ensures we get the most recent data and not a cached value
-        await dev.async_update()
-
-        # Access read cached data
-        print(f"IS_LEAKING={dev.is_leaking}")
-
-        # Let's wait a bit for some events to occur
-        await asyncio.sleep(3600)
+        while True:
+            try:
+                for sensor in water_leak_sensors:
+                    print(f"Sensor {sensor.name} - Current leak status = {sensor.is_leaking}. "
+                          f"Is currently leaking? {sensor.is_leaking}. "
+                          f"Last timestamp of leak = {sensor.latest_detected_water_leak_ts if sensor.latest_detected_water_leak_ts is not None else 'NEVER'}")
+                    print("Press CTRL+C to terminate.")
+                    # Let's wait a bit for some events to occur
+                    await asyncio.sleep(10)
+            except InterruptedError as e:
+                print("Execution terminated by the user")
 
     # Close the manager and logout from http_api
     manager.close()

meross_iot/controller/subdevice.py

@@ -1,6 +1,7 @@
 import logging
+from collections import deque
 from datetime import datetime
-from typing import Optional, Iterable
+from typing import Optional, Iterable, List, Dict
 
 from meross_iot.controller.device import GenericSubDevice
 from meross_iot.model.enums import OnlineStatus, ThermostatV3Mode
@@ -463,22 +464,51 @@ class Ms405Sensor(GenericSubDevice):
     Class that represents a Meross MS400 Smart Water Leak Sensor.
     """
 
-    def __init__(self, hubdevice_uuid: str, subdevice_id: str, manager, **kwargs):
+    def __init__(self, hubdevice_uuid: str, subdevice_id: str, manager, max_events_queue_len=30, **kwargs):
         super().__init__(hubdevice_uuid, subdevice_id, manager, **kwargs)
-        self.__water_leak = {}
-        self._last_active_time = None
+
+        self._last_active_time: Optional[int] = None
+        # Represents the last time we contacted the device
+
+        self.__water_leak_state: Optional[bool] = None
+        # Represents the current state
+
+        self.__last_event_ts: Optional[int] = None
+        # Represents the timestamp of the last sample (current state sampling)
+
+        self.__cached_events: deque = deque(maxlen=max_events_queue_len)
+        # Last N samples we collected
+
+        self.__last_waterleak_event_ts: Optional[int] = None
+        # Last timestamp we've seen a leak
 
     @property
     def is_leaking(self) -> Optional[bool]:
         """
-        Returns True if a water leak is detected (1), False otherwise (0).
-        Returns None if the state has not been fetched yet.
+        Returns the latest updated state available for the water leak sensor, if available.
         """
-        cur_val = self.__water_leak.get("latestWaterLeak")
-        if cur_val is None:
-            return None
-        else:
-            return cur_val == 1
+        return self.__water_leak_state
+
+    @property
+    def latest_sample_time(self) -> Optional[int]:
+        """
+        Returns the timestamp (GMT) of the latest available sampling.
+        """
+        return self.__last_event_ts
+
+    @property
+    def latest_detected_water_leak_ts(self) -> Optional[int]:
+        """
+        Return the timestamp (GMT) of the latest time the sensor sampled a water leak.
+        """
+        return self.__last_waterleak_event_ts
+
+    @property
+    def get_last_events(self) -> List[Dict]:
+        """
+        Returns the last cached items
+        """
+        return [x for x in self.__cached_events]
 
     async def async_update(self,
                            timeout: Optional[float] = None,
@@ -503,6 +533,26 @@ async def async_update(self,
             await self.async_handle_subdevice_notification(namespace=Namespace.HUB_SENSOR_ALL, data=subdev_state)
             break
 
+    def _handle_water_leak_fresh_data(self, leaking: bool, timestamp: int):
+        # If handling an event with an older timestamp than the one we have, just discard it.
+        if self.latest_sample_time is not None and timestamp <= self.latest_sample_time:
+            return
+
+        # If this is the first update or if it's more recent than the last we have, update the current state.
+        if self.__last_event_ts is None or timestamp >= self.__last_event_ts:
+            self.__last_event_ts = timestamp
+            self.__water_leak_state = leaking
+
+        # If the event is a leak and is more recent than the latest leak event, update it.
+        if leaking and (self.__last_waterleak_event_ts is None or timestamp >= self.__last_waterleak_event_ts):
+            self.__last_waterleak_event_ts = timestamp
+
+        # In any case, register the event in the queue
+        self.__cached_events.append({
+            "leaking": leaking,
+            "timestamp": timestamp
+        })
+
     async def async_handle_push_notification(self, namespace: Namespace, data: dict) -> bool:
         locally_handled = False
         if namespace == Namespace.HUB_ONLINE:
@@ -511,7 +561,10 @@ async def async_handle_push_notification(self, namespace: Namespace, data: dict)
                 self._online = OnlineStatus(update_element.get('status', -1))
                 locally_handled = True
         elif namespace == Namespace.HUB_SENSOR_WATERLEAK:
-            self.__water_leak = data.get('waterLeak')
+            water_leak_state = data.get('waterLeak')
+            latestWaterLeak = water_leak_state.get('latestWaterLeak')
+            latestSampleTime = water_leak_state.get('latestSampleTime')
+            self._handle_water_leak_fresh_data(leaking=latestWaterLeak==1, timestamp=latestSampleTime)
             locally_handled = True
 
         return locally_handled
@@ -523,15 +576,17 @@ async def async_handle_subdevice_notification(self, namespace: Namespace, data:
             self._last_active_time = data.get('online', {}).get('lastActiveTime')
         elif namespace == Namespace.HUB_SENSOR_WATERLEAK:
             latestWaterLeak = data.get('latestWaterLeak')
-            latestSaampleTime = data.get('latestSampleTime')
-            if latestWaterLeak is not None:
-                self.__water_leak["latestWaterLeak"]=latestWaterLeak
-            if latestSaampleTime is not None:
-                self.__water_leak["latestSaampleTime"] = latestSaampleTime
+            latestSampleTime = data.get('latestSampleTime')
+            self._handle_water_leak_fresh_data(leaking=latestWaterLeak==1, timestamp=latestSampleTime)
             locally_handled = True
         elif namespace == Namespace.HUB_SENSOR_ALL:
             self._online = OnlineStatus(data.get('online', {}).get('status', -1))
-            self.__water_leak.update(data.get('waterLeak', {}))
+            water_leak_state = data.get('waterLeak')
+            if water_leak_state is not None:
+                latestWaterLeak = water_leak_state.get('latestWaterLeak')
+                latestSampleTime = water_leak_state.get('latestSampleTime')
+                self._handle_water_leak_fresh_data(leaking=latestWaterLeak == 1, timestamp=latestSampleTime)
+
             locally_handled = True
         else:
             _LOGGER.warning(f"Could not handle event %s in subdevice %s handler", namespace, self.name)