added interface for MideaCodec

custom_components/midea_meiju_codec/core/device.py

@@ -61,14 +61,13 @@ class MiedaDevice(threading.Thread):
         self._is_run = False
         self._device_protocol_version = 0
         self._sub_type = None
-        self._sn = sn
         self._sn8 = sn8
         self._attributes = {}
         self._refresh_interval = 30
         self._heartbeat_interval = 10
         self._default_refresh_interval = 30
         self._connected = False
-        self._lua_runtime = MideaCodec(lua_file) if lua_file is not None else None
+        self._lua_runtime = MideaCodec(lua_file, sn=sn) if lua_file is not None else None
 
     @property
     def device_name(self):

custom_components/midea_meiju_codec/core/lua_runtime.py

@@ -27,16 +27,49 @@ class LuaRuntime:
 
 
 class MideaCodec(LuaRuntime):
-    def __init__(self, file):
+    def __init__(self, file, sn=None, sub_type=None):
         super().__init__(file)
+        self._sn = sn
+        self._sub_type = sub_type
+
+    def _build_base_dict(self):
+        device_info ={}
+        if self._sn is not None:
+            device_info["deviceSN"] = self._sn
+        if self._sub_type is not None:
+            device_info["deviceSN"] = self._sub_type
+        base_dict = {
+            "deviceinfo": device_info
+        }
+        return base_dict
 
     def build_query(self, append=None):
-        json_str = '{"deviceinfo":{"deviceSubType":1},"query":{}}'
+        query_dict = self._build_base_dict()
+        query_dict["query"] = {} if append is None else append
+        json_str = json.dumps(query_dict)
         result = self.json_to_data(json_str)
         return result
 
-    def decode_status(self, data):
+    def build_control(self, append=None):
-        data = '{"deviceinfo": {"deviceSubType":1},"msg":{"data": "' + data + '"}}'
+        query_dict = self._build_base_dict()
-        result = self.data_to_json(data)
+        query_dict["control"] = {} if append is None else append
+        json_str = json.dumps(query_dict)
+        result = self.json_to_data(json_str)
+        return result
+
+    def build_status(self, append=None):
+        query_dict = self._build_base_dict()
+        query_dict["status"] = {} if append is None else append
+        json_str = json.dumps(query_dict)
+        result = self.json_to_data(json_str)
+        return result
+
+    def decode_status(self, data: str):
+        data_dict = self._build_base_dict()
+        data_dict["msg"] = {
+            "data": data
+        }
+        json_str = json.dumps(data_dict)
+        result = self.data_to_json(json_str)
         status = json.loads(result)
         return status.get("status")