Zigbee suqare button and cube support via fake_device

miio/fake_device.py

@@ -0,0 +1,110 @@
+import calendar
+import datetime
+import logging
+import socket
+import struct
+from functools import reduce
+
+import ifaddr
+from miio.protocol import Message
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def ipv4_nonloop_ips():
+    def flatten(a, b):
+        a.extend(b)
+        return a
+
+    return list(
+        filter(
+            lambda ip: isinstance(ip, str) and not ip.startswith("127"),
+            map(
+                lambda ip: ip.ip,
+                reduce(
+                    flatten, map(lambda adapter: adapter.ips, ifaddr.get_adapters()), []
+                ),
+            ),
+        )
+    )
+
+
+class FakeDevice:
+    _device_id = None
+    _address = None
+    _token = None
+
+    def __init__(self, device_id, token, address="0.0.0.0"):
+        self._device_id = device_id
+        self._token = token
+        self._address = address
+
+    def run(self, callback):
+        def build_ack(device: int):
+            # Original devices are using year 1970, but it seems current datetime is fine
+            timestamp = calendar.timegm(datetime.datetime.now().timetuple())
+            # ACK packet not signed, 16 bytes header + 16 bytes of zeroes
+            return struct.pack(">HHIII16s", 0x2131, 32, 0, device, timestamp, bytes(16))
+
+        helobytes = bytes.fromhex(
+            "21310020ffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+        )
+
+        sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
+        # Gateway interacts only with port 54321
+        sock.bind((self._address, 54321))
+        _LOGGER.info(
+            "fake miio device started with  address=%s device_id=%s callback=%s token=****",
+            self._address,
+            self._device_id,
+            callback,
+        )
+
+        while True:
+            data, [host, port] = sock.recvfrom(1024)
+            if data == helobytes:
+                _LOGGER.debug("%s:%s=>PING", host, port)
+                m = build_ack(self._device_id)
+                sock.sendto(m, (host, port))
+                _LOGGER.debug("%s:%s<=ACK(device_id=%s)", host, port, self._device_id)
+            else:
+                request = Message.parse(data, token=self._token)
+                value = request.data.value
+                _LOGGER.debug("%s:%s=>%s", host, port, value)
+                action, device_call_id = value["method"].split("_")
+                source_device_id = (
+                    f"lumi.{device_call_id}"  # All known devices use lumi. prefix
+                )
+                callback(source_device_id, action, value["params"])
+                # This result means OK, but some methods return ['ok'] instead of 0
+                # might be necessary to use different results for different methods
+                result = {"result": 0, "id": value["id"]}
+                header = {
+                    "length": 0,
+                    "unknown": 0,
+                    "device_id": self._device_id,
+                    "ts": datetime.datetime.now(),
+                }
+                msg = {
+                    "data": {"value": result},
+                    "header": {"value": header},
+                    "checksum": 0,
+                }
+                response = Message.build(msg, token=self._token)
+                _LOGGER.debug("%s:%s<=%s", host, port, result)
+                sock.sendto(response, (host, port))
+
+
+if __name__ == "__main__":
+
+    def callback(source_device, action, params):
+        _LOGGER.debug(f"CALLBACK {source_device}=>{action}({params})")
+
+    from gateway_scripts import tokens, fake_device_id
+
+    logging.basicConfig(level="DEBUG")
+
+    device_id = int(fake_device_id)
+    device_token = bytes.fromhex(tokens["real"])
+    fake_device = FakeDevice(device_id, device_token)
+    fake_device.run(callback)

miio/gateway.py

@@ -11,6 +11,21 @@
 from .click_common import EnumType, command, format_output
 from .device import Device
 from .exceptions import DeviceException
+from .fake_device import ipv4_nonloop_ips
+from .gateway_scripts import (
+    action_id,
+    build_doublepress,
+    build_flip90,
+    build_flip180,
+    build_longpress,
+    build_move,
+    build_rotate,
+    build_shake,
+    build_shakeair,
+    build_singlepress,
+    build_taptap,
+    tokens,
+)
 from .utils import brightness_and_color_to_int, int_to_brightness, int_to_rgb
 
 _LOGGER = logging.getLogger(__name__)
@@ -161,6 +176,8 @@ def discover_devices(self):
             DeviceType.Plug: AqaraPlug,
             DeviceType.SensorHT: SensorHT,
             DeviceType.AqaraHT: AqaraHT,
+            DeviceType.Cube: Cube,
+            DeviceType.AqaraSquareButton: AqaraSquareButton,
             DeviceType.AqaraMagnet: AqaraMagnet,
             DeviceType.AqaraSwitchOneChannel: AqaraSwitchOneChannel,
             DeviceType.AqaraSwitchTwoChannels: AqaraSwitchTwoChannels,
@@ -201,6 +218,38 @@ def discover_devices(self):
 
         return self._devices
 
+    def x_del(self, script_id):
+        """Delete script by id."""
+        return self.send("miIO.xdel", [script_id])
+
+    @command(click.argument("sid"), click.argument("command"))
+    def subdevice_command(self, sid, command):
+        self.discover_devices()
+        target = list(filter(lambda subdevice: subdevice.sid == sid, self.devices))
+        if len(target) < 1:
+            return f"Device with sid={sid} not found"
+        elif not hasattr(target[0], command):
+            return f"Device with sid={sid} has no method {command}"
+        else:
+            return getattr(target[0], command)()
+
+    def install_script(self, sid, action, builder):
+        addresses = ipv4_nonloop_ips()
+        my_ip = addresses[0]  # Taking first public IP ;(
+        _LOGGER.info("Using address %s for action %s of %s", my_ip, action, sid)
+        data_tkn = tokens["data_tkn"]
+        source = builder(sid, my_ip)
+        return self.send(
+            "send_data_frame",
+            {
+                "cur": 0,
+                "data": source,
+                "data_tkn": data_tkn,
+                "total": 1,
+                "type": "scene",
+            },
+        )
+
     @command(click.argument("property"))
     def get_prop(self, property):
         """Get the value of a property for given sid."""
@@ -738,6 +787,21 @@ def get_firmware_version(self) -> Optional[int]:
             )
         return self._fw_ver
 
+    @command()
+    def uninstall_scripts(self):
+        return dict(
+            map(
+                lambda action: (
+                    action,
+                    (
+                        action_id[action](self.sid),
+                        self._gw.x_del(action_id[action](self.sid)),
+                    ),
+                ),
+                action_id.keys(),
+            )
+        )
+
 
 class AqaraHT(SubDevice):
     """Subdevice AqaraHT specific properties and methods"""
@@ -921,3 +985,55 @@ def update(self):
         self._props.status_ch0 = values[0]
         self._props.status_ch1 = values[1]
         self._props.load_power = values[2]
+
+
+class Cube(SubDevice):
+    """Subdevice Cube specific properties and methods"""
+
+    properties = []
+
+    @command()
+    def install_move_script(self):
+        return self._gw.install_script(self.sid, "move", build_move)
+
+    @command()
+    def install_rotate_script(self):
+        return self._gw.install_script(self.sid, "rotate", build_rotate)
+
+    @command()
+    def install_shake_script(self):
+        return self._gw.install_script(self.sid, "shakeair", build_shakeair)
+
+    @command()
+    def install_flip90_script(self):
+        return self._gw.install_script(self.sid, "flip90", build_flip90)
+
+    @command()
+    def install_taptap_script(self):
+        return self._gw.install_script(self.sid, "taptap", build_taptap)
+
+    @command()
+    def install_flip180_script(self):
+        return self._gw.install_script(self.sid, "flip180", build_flip180)
+
+
+class AqaraSquareButton(SubDevice):
+    """Subdevice AqaraSquareButton specific properties and methods"""
+
+    properties = []
+
+    @command()
+    def install_singlepress_script(self):
+        return self._gw.install_script(self.sid, "singlepress", build_singlepress)
+
+    @command()
+    def install_doublepress_script(self):
+        return self._gw.install_script(self.sid, "doublepress", build_doublepress)
+
+    @command()
+    def install_longpress_script(self):
+        return self._gw.install_script(self.sid, "longpress", build_longpress)
+
+    @command()
+    def install_shake_script(self):
+        return self._gw.install_script(self.sid, "shake", build_shake)