Add BLE emulation for dev purposes

Author: AlexxIT

custom_components/xiaomi_gateway3/gateway3.py

@@ -6,7 +6,7 @@
 import time
 from telnetlib import Telnet
 from threading import Thread
-from typing import Optional
+from typing import Optional, Union
 
 from paho.mqtt.client import Client, MQTTMessage
 from . import utils
@@ -240,6 +240,8 @@ def _get_devices_v3(self):
                     continue
 
                 retain = json.loads(data[did + '.prop'])['props']
+                _LOGGER.debug(f"{self.host} | {model} retain: {retain}")
+
                 params = {
                     p[2]: retain.get(p[1])
                     for p in desc['params']
@@ -422,8 +424,9 @@ def process_message(self, data: dict):
             device['mac'] = '0x' + device['mac']
             self.setup_devices([device])
 
-    def process_ble_event(self, raw: bytes):
-        data = json.loads(raw[10:])['params']
+    def process_ble_event(self, raw: Union[bytes, str]):
+        data = json.loads(raw[10:])['params'] \
+            if isinstance(raw, bytes) else json.loads(raw)
 
         _LOGGER.debug(f"{self.host} | Process BLE {data}")
 

custom_components/xiaomi_gateway3/remote.py

@@ -1,5 +1,6 @@
 import logging
 
+from homeassistant.components.remote import ATTR_DEVICE
 from homeassistant.helpers.entity import ToggleEntity
 
 from . import DOMAIN, Gateway3Device
@@ -40,3 +41,9 @@ def turn_on(self):
 
     def turn_off(self):
         self.gw.send(self.device, {'pairing_stop': 0})
+
+    async def async_send_command(self, command, **kwargs):
+        for cmd in command:
+            # for testing purposes
+            if cmd == 'ble':
+                self.gw.process_ble_event(kwargs[ATTR_DEVICE])