Implement SystemRuntimeMixin

meross_iot/controller/mixins/runtime.py

@@ -0,0 +1,50 @@
+import logging
+from datetime import datetime
+from typing import Optional
+
+from meross_iot.model.enums import Namespace
+
+_LOGGER = logging.getLogger(__name__)
+
+_DATE_FORMAT = '%Y-%m-%d'
+
+
+class SystemRuntimeMixin(object):
+    _execute_command: callable
+
+    def __init__(self, device_uuid: str,
+                 manager,
+                 **kwargs):
+        super().__init__(device_uuid=device_uuid, manager=manager, **kwargs)
+        self._runtime_info = {}
+
+    async def async_update_runtime_info(self, timeout: Optional[float] = None, *args, **kwargs) -> dict:
+        """
+        Polls the device to gather the latest runtime information for this device.
+        Note that the returned value might vary with the time as Meross could add/remove/change runtime information
+        in the future.
+
+        :return: a `dict` object containing the runtime information provided by the Meross device
+        """
+        result = await self._execute_command(method="GET",
+                                             namespace=Namespace.SYSTEM_RUNTIME,
+                                             payload={},
+                                             timeout=timeout)
+        data = result.get('runtime')
+        self._runtime_info = data
+        return data
+
+    @property
+    def cached_system_runtime_info(self) -> Optional[dict]:
+        """
+        Returns the latest cached runtime info. If you want a fresh value, consider using the
+        `update_runtime_info` method instead.
+        """
+        return self._runtime_info
+
+    async def async_update(self,
+                           *args,
+                           **kwargs) -> None:
+        # call the superclass implementation first
+        await super().async_update(*args, **kwargs)
+        await self.async_get_runtime_info()

meross_iot/device_factory.py

@@ -11,6 +11,7 @@
 from meross_iot.controller.mixins.hub import HubMts100Mixin, HubMixn, HubMs100Mixin
 from meross_iot.controller.mixins.light import LightMixin
 from meross_iot.controller.mixins.roller_shutter import RollerShutterTimerMixin
+from meross_iot.controller.mixins.runtime import SystemRuntimeMixin
 from meross_iot.controller.mixins.spray import SprayMixin
 from meross_iot.controller.mixins.system import SystemAllMixin, SystemOnlineMixin
 from meross_iot.controller.mixins.thermostat import ThermostatModeMixin
@@ -55,6 +56,7 @@
     # System
     Namespace.SYSTEM_ALL.value: SystemAllMixin,
     Namespace.SYSTEM_ONLINE.value: SystemOnlineMixin,
+    Namespace.SYSTEM_RUNTIME.value: SystemRuntimeMixin,
 
     # Hub
     Namespace.HUB_ONLINE.value: HubMixn,

meross_iot/model/enums.py

@@ -74,6 +74,7 @@ class Namespace(Enum):
     SYSTEM_ONLINE = 'Appliance.System.Online'
     SYSTEM_REPORT = 'Appliance.System.Report'
     SYSTEM_DEBUG = 'Appliance.System.Debug'
+    SYSTEM_RUNTIME = 'Appliance.System.Runtime'
 
     CONTROL_BIND = 'Appliance.Control.Bind'
     CONTROL_UNBIND = 'Appliance.Control.Unbind'

tests/test_runtime.py

@@ -0,0 +1,51 @@
+import os
+from typing import List, Union
+
+from aiohttp import web
+from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
+
+from meross_iot.controller.device import BaseDevice
+from meross_iot.controller.mixins.runtime import SystemRuntimeMixin
+from meross_iot.manager import MerossManager
+from meross_iot.model.enums import OnlineStatus
+from tests import async_get_client
+
+if os.name == 'nt':
+    import asyncio
+
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+else:
+    import asyncio
+
+
+class TestSystemRuntime(AioHTTPTestCase):
+    async def get_application(self):
+        return web.Application()
+
+    async def setUpAsync(self):
+        # Wait some time before next test-burst
+        await asyncio.sleep(10)
+        self.meross_client, self.requires_logout = await async_get_client()
+
+        # Look for a device to be used for this test
+        self.meross_manager = MerossManager(http_client=self.meross_client)
+        await self.meross_manager.async_init()
+        await self.meross_manager.async_device_discovery()
+        self.test_devices: List[Union[BaseDevice, SystemRuntimeMixin]] = self.meross_manager.find_devices(device_class=SystemRuntimeMixin, online_status=OnlineStatus.ONLINE)
+
+    @unittest_run_loop
+    async def test_runtime_manual_update(self):
+        if len(self.test_devices) < 1:
+            self.skipTest("No device has been found to run this test.")
+        for d in self.test_devices:
+            info = await d.async_update_runtime_info()
+            print(f"Wifi signal for device {d.name} is {info.get('signal')}%")
+            self.assertEqual(d.cached_system_runtime_info, info)
+
+    async def tearDownAsync(self):
+        if self.requires_logout:
+            await self.meross_client.async_logout()
+        self.meross_manager.close()
+
+        # Give a change to asyncio clean everything up
+        await asyncio.sleep(1)