Remove file sync read warning

custom_components/midea_auto_codec/core/cloud.py

@@ -96,7 +96,7 @@ class MideaCloud:
                     break
             except Exception as e:
                 pass
-        print(response)
+
         if int(response["code"]) == 0 and "data" in response:
             return response["data"]
 

custom_components/midea_auto_codec/core/device.py

@@ -1,10 +1,8 @@
 import threading
 import socket
-import time
 from enum import IntEnum
 from .security import LocalSecurity, MSGTYPE_HANDSHAKE_REQUEST, MSGTYPE_ENCRYPTED_REQUEST
 from .packet_builder import PacketBuilder
-from .lua_runtime import MideaCodec
 from .message import MessageQuestCustom
 from .logger import MideaLogger
 
@@ -41,8 +39,7 @@ class MiedaDevice(threading.Thread):
                  subtype: int | None,
                  connected: bool,
                  sn: str | None,
-                 sn8: str | None,
-                 lua_file: str | None):
+                 sn8: str | None):
         threading.Thread.__init__(self)
         self._socket = None
         self._ip_address = ip_address
@@ -74,7 +71,6 @@ class MiedaDevice(threading.Thread):
         self._centralized = []
         self._calculate_get = []
         self._calculate_set = []
-        self._lua_runtime = MideaCodec(lua_file, sn=sn, subtype=subtype) if lua_file is not None else None
 
     @property
     def device_name(self):
@@ -136,8 +132,6 @@ class MiedaDevice(threading.Thread):
             for attr in self._centralized:
                 new_status[attr] = self._attributes.get(attr)
             new_status[attribute] = value
-            if set_cmd := self._lua_runtime.build_control(new_status):
-                self._build_send(set_cmd)
 
     def set_attributes(self, attributes):
         new_status = {}
@@ -148,9 +142,6 @@ class MiedaDevice(threading.Thread):
             if attribute in self._attributes.keys():
                 has_new = True
                 new_status[attribute] = value
-        if has_new:
-            if set_cmd := self._lua_runtime.build_control(new_status):
-                self._build_send(set_cmd)
 
     def set_ip_address(self, ip_address):
         MideaLogger.debug(f"Update IP address to {ip_address}")
@@ -219,72 +210,6 @@ class MiedaDevice(threading.Thread):
         msg = PacketBuilder(self._device_id, bytes_cmd).finalize()
         self._send_message(msg)
 
-    def _refresh_status(self):
-        for query in self._queries:
-            if query_cmd := self._lua_runtime.build_query(query):
-                self._build_send(query_cmd)
-
-    def _parse_message(self, msg):
-        if self._protocol == 3:
-            messages, self._buffer = self._security.decode_8370(self._buffer + msg)
-        else:
-            messages, self._buffer = self.fetch_v2_message(self._buffer + msg)
-        if len(messages) == 0:
-            return ParseMessageResult.PADDING
-        for message in messages:
-            if message == b"ERROR":
-                return ParseMessageResult.ERROR
-            payload_len = message[4] + (message[5] << 8) - 56
-            payload_type = message[2] + (message[3] << 8)
-            if payload_type in [0x1001, 0x0001]:
-                # Heartbeat detected
-                pass
-            elif len(message) > 56:
-                cryptographic = message[40:-16]
-                if payload_len % 16 == 0:
-                    decrypted = self._security.aes_decrypt(cryptographic)
-                    MideaLogger.debug(f"Received: {decrypted.hex().lower()}")
-                    if status := self._lua_runtime.decode_status(decrypted.hex()):
-                        MideaLogger.debug(f"Decoded: {status}")
-                        new_status = {}
-                        for single in status.keys():
-                            value = status.get(single)
-                            if single not in self._attributes or self._attributes[single] != value:
-                                self._attributes[single] = value
-                                new_status[single] = value
-                        if len(new_status) > 0:
-                            for c in self._calculate_get:
-                                lvalue = c.get("lvalue")
-                                rvalue = c.get("rvalue")
-                                if lvalue and rvalue:
-                                    calculate = False
-                                    for s, v in new_status.items():
-                                        if rvalue.find(f"[{s}]") >= 0:
-                                            calculate = True
-                                            break
-                                    if calculate:
-                                        calculate_str1 = \
-                                            (f"{lvalue.replace('[', 'self._attributes[')} = "
-                                             f"{rvalue.replace('[', 'self._attributes[')}") \
-                                            .replace("[","[\"").replace("]","\"]")
-                                        calculate_str2 = \
-                                            (f"{lvalue.replace('[', 'new_status[')} = "
-                                             f"{rvalue.replace('[', 'self._attributes[')}") \
-                                            .replace("[","[\"").replace("]","\"]")
-                                        try:
-                                            exec(calculate_str1)
-                                            exec(calculate_str2)
-                                        except Exception:
-                                            MideaLogger.warning(
-                                                f"Calculation Error: {lvalue} = {rvalue}", self._device_id
-                                            )
-                            self._update_all(new_status)
-        return ParseMessageResult.SUCCESS
-
-    def _send_heartbeat(self):
-        msg = PacketBuilder(self._device_id, bytearray([0x00])).finalize(msg_type=0)
-        self._send_message(msg)
-
     def _device_connected(self, connected=True):
         self._connected = connected
         status = {"connected": connected}

custom_components/midea_auto_codec/core/lua_runtime.py

@@ -1,91 +0,0 @@
-import lupa
-import threading
-import json
-from .logger import MideaLogger
-
-
-class LuaRuntime:
-    def __init__(self, file):
-        self._runtimes = lupa.LuaRuntime()
-        string = f'dofile("{file}")'
-        self._runtimes.execute(string)
-        self._lock = threading.Lock()
-        self._json_to_data = self._runtimes.eval("function(param) return jsonToData(param) end")
-        self._data_to_json = self._runtimes.eval("function(param) return dataToJson(param) end")
-
-    def json_to_data(self, json_value):
-        with self._lock:
-            result = self._json_to_data(json_value)
-
-        return result
-
-    def data_to_json(self, data_value):
-        with self._lock:
-            result = self._data_to_json(data_value)
-        return result
-
-
-class MideaCodec(LuaRuntime):
-    def __init__(self, file, sn=None, subtype=None):
-        super().__init__(file)
-        self._sn = sn
-        self._subtype = subtype
-
-    def _build_base_dict(self):
-        device_info ={}
-        if self._sn is not None:
-            device_info["deviceSN"] = self._sn
-        if self._subtype is not None:
-            device_info["deviceSubType"] = self._subtype
-        base_dict = {
-            "deviceinfo": device_info
-        }
-        return base_dict
-
-    def build_query(self, append=None):
-        query_dict = self._build_base_dict()
-        query_dict["query"] = {} if append is None else append
-        json_str = json.dumps(query_dict)
-        try:
-            result = self.json_to_data(json_str)
-            return result
-        except lupa.LuaError as e:
-            MideaLogger.error(f"LuaRuntimeError in build_query {json_str}: {repr(e)}")
-        return None
-
-    def build_control(self, append=None):
-        query_dict = self._build_base_dict()
-        query_dict["control"] = {} if append is None else append
-        json_str = json.dumps(query_dict)
-        try:
-            result = self.json_to_data(json_str)
-            return result
-        except lupa.LuaError as e:
-            MideaLogger.error(f"LuaRuntimeError in build_control {json_str}: {repr(e)}")
-            return None
-
-    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)
-        try:
-            result = self.json_to_data(json_str)
-            return result
-        except lupa.LuaError as e:
-            MideaLogger.error(f"LuaRuntimeError in build_status {json_str}: {repr(e)}")
-            return None
-
-    def decode_status(self, data: str):
-        data_dict = self._build_base_dict()
-        data_dict["msg"] = {
-            "data": data
-        }
-        json_str = json.dumps(data_dict)
-        try:
-            result = self.data_to_json(json_str)
-            status = json.loads(result)
-            return status.get("status")
-        except lupa.LuaError as e:
-            MideaLogger.error(f"LuaRuntimeError in decode_status {data}: {repr(e)}")
-        return None
-