HomeAssistant/midea-meiju-codec
1
0
Fork 1
mirror of https://github.com/sususweet/midea-meiju-codec.git synced 2025-09-28 02:32:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

v0.0.4

Browse Source
This commit is contained in:
unknown
2023-09-10 12:08:27 +08:00
parent 216bfc53db
commit 2246948ddf
15 changed files with 636 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
custom_components/midea_meiju_codec/core/cloud.py
View File

@@ -35,7 +35,7 @@ class MideaCloudBase:
self.security = security
self.server = server
async def api_request(self, endpoint, args=None, data=None):
async def api_request(self, endpoint, args=None, data=None) -> dict | None:
args = args or {}
headers = {}
if data is None:
@@ -127,7 +127,7 @@ class MideaCloudBase:
udpid = CloudSecurity.get_udpid(device_id.to_bytes(6, "big"))
else:
udpid = CloudSecurity.get_udpid(device_id.to_bytes(6, "little"))
_LOGGER.error(f"The udpid of deivce [{device_id}] generated "
_LOGGER.debug(f"The udpid of deivce [{device_id}] generated "
f"with byte order '{'big' if byte_order_big else 'little'}': {udpid}")
response = await self.api_request(
"/v1/iot/secure/getToken",
@@ -171,11 +171,12 @@ class MeijuCloudExtend(MideaCloudBase):
response = await self.api_request("/v1/appliance/home/list/get", args={
'homegroupId': home
})
for h in response.get("homeList") or []:
for r in h.get("roomList") or []:
for a in r.get("applianceList"):
a["sn"] = CloudSecurity.decrypt(bytes.fromhex(a["sn"]), self.key).decode()
devices.append(a)
if response:
for h in response.get("homeList") or []:
for r in h.get("roomList") or []:
for a in r.get("applianceList"):
a["sn"] = CloudSecurity.decrypt(bytes.fromhex(a["sn"]), self.key).decode()
devices.append(a)
return devices
async def get_lua(self, sn, device_type, path, enterprise_code=None):
@@ -183,7 +184,7 @@ class MeijuCloudExtend(MideaCloudBase):
"/v1/appliance/protocol/lua/luaGet",
data={
"applianceSn": sn,
"applianceType": f"0x{'%02X' % device_type}",
"applianceType": "0x%02X" % device_type,
"applianceMFCode": enterprise_code if enterprise_code else "0000",
'version': "0",
"iotAppId": "900",

33
custom_components/midea_meiju_codec/core/device.py
View File

@@ -60,6 +60,7 @@ class MiedaDevice(threading.Thread):
self._sn = sn
self._sn8 = sn8
self._attributes = {
"device_type": "T0x%02X" % device_type,
"sn": sn,
"sn8": sn8,
"subtype": subtype
@@ -135,8 +136,8 @@ class MiedaDevice(threading.Thread):
for attr in self._centralized:
new_status[attr] = self._attributes.get(attr)
new_status[attribute] = value
set_cmd = self._lua_runtime.build_control(new_status)
self.build_send(set_cmd)
if set_cmd := self._lua_runtime.build_control(new_status):
self.build_send(set_cmd)
def set_attributes(self, attributes):
new_status = {}
@@ -148,8 +149,8 @@ class MiedaDevice(threading.Thread):
has_new = True
new_status[attribute] = value
if has_new:
set_cmd = self._lua_runtime.build_control(new_status)
self.build_send(set_cmd)
if set_cmd := self._lua_runtime.build_control(new_status):
self.build_send(set_cmd)
@staticmethod
def fetch_v2_message(msg):
@@ -231,8 +232,8 @@ class MiedaDevice(threading.Thread):
def refresh_status(self):
for query in self._queries:
query_cmd = self._lua_runtime.build_query(query)
self.build_send(query_cmd)
if query_cmd := self._lua_runtime.build_query(query):
self.build_send(query_cmd)
def parse_message(self, msg):
if self._protocol == 3:
@@ -255,16 +256,16 @@ class MiedaDevice(threading.Thread):
decrypted = self._security.aes_decrypt(cryptographic)
MideaLogger.debug(f"Received: {decrypted.hex()}")
# 这就是最终消息
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:
self.update_all(new_status)
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:
self.update_all(new_status)
return ParseMessageResult.SUCCESS
def send_heartbeat(self):

38
custom_components/midea_meiju_codec/core/lua_runtime.py
View File

@@ -1,7 +1,8 @@
import lupa
import threading
import json
from .logger import MideaLogger
lupa.LuaMemoryError
class LuaRuntime:
def __init__(self, file):
@@ -44,22 +45,34 @@ class MideaCodec(LuaRuntime):
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
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)
result = self.json_to_data(json_str)
return result
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)
result = self.json_to_data(json_str)
return result
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()
@@ -67,6 +80,11 @@ class MideaCodec(LuaRuntime):
"data": data
}
json_str = json.dumps(data_dict)
result = self.data_to_json(json_str)
status = json.loads(result)
return status.get("status")
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