WIP: improve miotdevice API, add gosund plug support (cuco.plug.cp1)

miio/__init__.py

@@ -22,6 +22,7 @@
 from miio.exceptions import DeviceError, DeviceException
 from miio.fan import Fan, FanP5, FanSA1, FanV2, FanZA1, FanZA4
 from miio.gateway import Gateway
+from miio.gosund_plug import GosundPlug
 from miio.heater import Heater
 from miio.philips_bulb import PhilipsBulb, PhilipsWhiteBulb
 from miio.philips_eyecare import PhilipsEyecare

miio/airpurifier_miot.py

@@ -255,16 +255,6 @@ def __json__(self):
 class AirPurifierMiot(MiotDevice):
     """Main class representing the air purifier which uses MIoT protocol."""
 
-    def __init__(
-        self,
-        ip: str = None,
-        token: str = None,
-        start_id: int = 0,
-        debug: int = 0,
-        lazy_discover: bool = True,
-    ) -> None:
-        super().__init__(_MAPPING, ip, token, start_id, debug, lazy_discover)
-
     @command(
         default_output=format_output(
             "",
@@ -297,19 +287,19 @@ def status(self) -> AirPurifierMiotStatus:
         return AirPurifierMiotStatus(
             {
                 prop["did"]: prop["value"] if prop["code"] == 0 else None
-                for prop in self.get_properties_for_mapping()
+                for prop in self.get_properties_for_mapping(_MAPPING)
             }
         )
 
     @command(default_output=format_output("Powering on"))
     def on(self):
         """Power on."""
-        return self.set_property("power", True)
+        return self.set_property_from_mapping(_MAPPING, "power", True)
 
     @command(default_output=format_output("Powering off"))
     def off(self):
         """Power off."""
-        return self.set_property("power", False)
+        return self.set_property_from_mapping(_MAPPING, "power", False)
 
     @command(
         click.argument("level", type=int),
@@ -319,7 +309,7 @@ def set_fan_level(self, level: int):
         """Set fan level."""
         if level < 1 or level > 3:
             raise AirPurifierMiotException("Invalid fan level: %s" % level)
-        return self.set_property("fan_level", level)
+        return self.set_property_from_mapping(_MAPPING, "fan_level", level)
 
     @command(
         click.argument("rpm", type=int),
@@ -333,7 +323,7 @@ def set_favorite_rpm(self, rpm: int):
                 "Invalid favorite motor speed: %s. Must be between 300 and 2300 and divisible by 10"
                 % rpm
             )
-        return self.set_property("favorite_rpm", rpm)
+        return self.set_property_from_mapping(_MAPPING, "favorite_rpm", rpm)
 
     @command(
         click.argument("volume", type=int),
@@ -345,15 +335,15 @@ def set_volume(self, volume: int):
             raise AirPurifierMiotException(
                 "Invalid volume: %s. Must be between 0 and 100" % volume
             )
-        return self.set_property("buzzer_volume", volume)
+        return self.set_property_from_mapping(_MAPPING, "buzzer_volume", volume)
 
     @command(
         click.argument("mode", type=EnumType(OperationMode, False)),
         default_output=format_output("Setting mode to '{mode.value}'"),
     )
     def set_mode(self, mode: OperationMode):
         """Set mode."""
-        return self.set_property("mode", mode.value)
+        return self.set_property_from_mapping(_MAPPING, "mode", mode.value)
 
     @command(
         click.argument("level", type=int),
@@ -366,15 +356,17 @@ def set_favorite_level(self, level: int):
         if level < 0 or level > 14:
             raise AirPurifierMiotException("Invalid favorite level: %s" % level)
 
-        return self.set_property("favorite_level", level)
+        return self.set_property_from_mapping(_MAPPING, "favorite_level", level)
 
     @command(
         click.argument("brightness", type=EnumType(LedBrightness, False)),
         default_output=format_output("Setting LED brightness to {brightness}"),
     )
     def set_led_brightness(self, brightness: LedBrightness):
         """Set led brightness."""
-        return self.set_property("led_brightness", brightness.value)
+        return self.set_property_from_mapping(
+            _MAPPING, "led_brightness", brightness.value
+        )
 
     @command(
         click.argument("led", type=bool),
@@ -384,7 +376,7 @@ def set_led_brightness(self, brightness: LedBrightness):
     )
     def set_led(self, led: bool):
         """Turn led on/off."""
-        return self.set_property("led", led)
+        return self.set_property_from_mapping(_MAPPING, "led", led)
 
     @command(
         click.argument("buzzer", type=bool),
@@ -394,7 +386,7 @@ def set_led(self, led: bool):
     )
     def set_buzzer(self, buzzer: bool):
         """Set buzzer on/off."""
-        return self.set_property("buzzer", buzzer)
+        return self.set_property_from_mapping(_MAPPING, "buzzer", buzzer)
 
     @command(
         click.argument("lock", type=bool),
@@ -404,4 +396,4 @@ def set_buzzer(self, buzzer: bool):
     )
     def set_child_lock(self, lock: bool):
         """Set child lock on/off."""
-        return self.set_property("child_lock", lock)
+        return self.set_property_from_mapping(_MAPPING, "child_lock", lock)

miio/gosund_plug.py

@@ -0,0 +1,35 @@
+import logging
+from dataclasses import dataclass, field
+
+from .click_common import command, format_output
+from .miot_device import MiotDevice
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class GosundPlugStatus:
+    _max_properties = 1
+    state: bool = field(metadata={"siid": 2, "piid": 1})
+
+
+class GosundPlug(MiotDevice):
+    """Support for gosund miot plug (cuco.plug.cp1)."""
+
+    _MAPPING = GosundPlugStatus
+
+    @command()
+    def status(self) -> GosundPlugStatus:
+        """Return current state of the plug."""
+
+        return self.get_properties_for_dataclass(GosundPlugStatus)
+
+    @command(default_output=format_output("Powering on"))
+    def on(self):
+        """Power on."""
+        return self.set_property(state=True)
+
+    @command(default_output=format_output("Powering off"))
+    def off(self):
+        """Power off."""
+        return self.set_property(state=False)

miio/miot_device.py

@@ -1,37 +1,135 @@
 import logging
+from dataclasses import dataclass, field
 
+from .click_common import command
 from .device import Device
+from .exceptions import DeviceException
 
 _LOGGER = logging.getLogger(__name__)
 
 
+@dataclass
+class MiotInfo:
+    """Container for common MiotInfo service."""
+
+    _siid = 1
+    _max_properties = 1  # some devices respond with broken json otherwise
+
+    manufacturer: str = field(metadata={"piid": 1})
+    model: str = field(metadata={"piid": 2})
+    serial_number: str = field(metadata={"piid": 3})
+    firmware_version: str = field(metadata={"piid": 4})
+
+
 class MiotDevice(Device):
     """Main class representing a MIoT device."""
 
     def __init__(
         self,
-        mapping: dict,
         ip: str = None,
         token: str = None,
         start_id: int = 0,
         debug: int = 0,
         lazy_discover: bool = True,
     ) -> None:
-        self.mapping = mapping
         super().__init__(ip, token, start_id, debug, lazy_discover)
 
-    def get_properties_for_mapping(self) -> list:
+    @command()
+    def miot_info(self) -> MiotInfo:
+        """Return common miot information."""
+        return self.get_properties_for_dataclass(MiotInfo)
+
+    def get_properties_for_dataclass(self, cls):
+        """Run a query to fill property container."""
+        fields = cls.__dataclass_fields__
+        property_mapping = {}
+
+        for field_name in fields:
+            field_meta = fields[field_name].metadata
+
+            if "piid" not in field_meta:
+                continue
+            piid = field_meta["piid"]
+
+            siid = field_meta.get("siid", getattr(cls, "_siid", None))
+            if siid is None:
+                raise DeviceException(
+                    f"no siid defined for {field_name} or for class {cls}"
+                )
+
+            property_mapping[field_name] = {"siid": siid, "piid": piid}
+
+        response = {
+            prop["did"]: prop["value"] if prop["code"] == 0 else None
+            for prop in self.get_properties_for_mapping(
+                property_mapping, max_properties=cls._max_properties
+            )
+        }
+
+        return cls(**response)
+
+    def set_property(self, **kwargs):
+        """Helper to set properties using the device specific mapping."""
+        if getattr(self, "_MAPPING") is None:
+            raise DeviceException("Device class does not have _MAPPING")
+
+        return self.set_properties_from_dataclass(self._MAPPING(**kwargs))
+
+    def set_properties_from_dataclass(self, obj):
+        """Set properties as defined in the given dataclass object."""
+        # TODO: this needs cleanup.
+        fields = obj.__dataclass_fields__
+        properties_to_set = []
+
+        for field_name in fields:
+            field_meta = fields[field_name].metadata
+
+            if "piid" not in field_meta:
+                continue
+
+            piid = field_meta["piid"]
+
+            siid = field_meta.get("siid", getattr(obj, "_siid", None))
+            if siid is None:
+                raise DeviceException(
+                    f"no siid defined for {field_name} or for class {obj.__class__}"
+                )
+
+            value = obj.__getattribute__(field_name)
+            if value is None:
+                continue
+
+            property_set = {
+                "siid": siid,
+                "piid": piid,
+                "did": field_name,
+                "value": value,
+            }
+
+            properties_to_set.append(property_set)
+
+        if not properties_to_set:
+            raise DeviceException("No values to set!")
+
+        # TODO: handle splitting based on _max_properties
+
+        _LOGGER.debug("Going to set %s" % properties_to_set)
+        return self.send("set_properties", properties_to_set)
+
+    def get_properties_for_mapping(
+        self, property_mapping, *, max_properties=15
+    ) -> list:
         """Retrieve raw properties based on mapping."""
 
         # We send property key in "did" because it's sent back via response and we can identify the property.
-        properties = [{"did": k, **v} for k, v in self.mapping.items()]
+        properties = [{"did": k, **v} for k, v in property_mapping.items()]
 
-        return self.get_properties(properties, max_properties=15)
+        return self.get_properties(properties, max_properties=max_properties)
 
-    def set_property(self, property_key: str, value):
+    def set_property_from_mapping(self, property_mapping, property_key: str, value):
         """Sets property value."""
 
         return self.send(
             "set_properties",
-            [{"did": property_key, **self.mapping[property_key], "value": value}],
+            [{"did": property_key, **property_mapping[property_key], "value": value}],
         )

miio/tests/dummies.py

@@ -63,7 +63,7 @@ def __init__(self, *args, **kwargs):
         self.state = [{"did": k, "value": v, "code": 0} for k, v in self.state.items()]
         super().__init__(*args, **kwargs)
 
-    def get_properties_for_mapping(self):
+    def get_properties_for_mapping(self, mapping=None):
         return self.state
 
     def set_property(self, property_key: str, value):