aixan/midea-meiju-codec
1
0
Fork 0
forked from HomeAssistant/midea-meiju-codec
Code Pull Requests Activity

init commit

Browse Source
This commit is contained in:
unknown
2023-09-02 16:30:03 +08:00
parent 6554430ac4
commit 86cd072738
18 changed files with 1635 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -150,3 +150,8 @@ cython_debug/
# and can be added to the global gitignore or merged into this file. For a more nuclear # and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder. # option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/ #.idea/
.idea
test.py
*.lua

7
README.md
View File

@@ -1,2 +1,7 @@
# midea-meiju-codec # Midea Meiju Codec
通过网络获取你美居家庭中的设备，并且在本地配置这些设备，并通过本地更新状态及控制设备。
- 自动查找和发现设备
- 自动下载设备的协议文件
- 将设备状态更新为设备可见的属性

88
custom_components/midea_meiju_codec/__init__.py Normal file
View File

@@ -0,0 +1,88 @@
import logging
import os
import base64
from homeassistant.core import HomeAssistant
from homeassistant.const import (
Platform,
CONF_TYPE,
CONF_PORT,
CONF_MODEL,
CONF_IP_ADDRESS,
CONF_DEVICE_ID,
CONF_PROTOCOL,
CONF_TOKEN,
CONF_NAME
)
from .core.device import MiedaDevice
from .const import (
DOMAIN,
DEVICES,
CONF_KEY,
CONF_ACCOUNT,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, hass_config: dict):
hass.data.setdefault(DOMAIN, {})
cjson = os.getcwd() + "/cjson.lua"
bit = os.getcwd() + "/bit.lua"
if not os.path.exists(cjson):
from .const import CJSON_LUA
cjson_lua = base64.b64decode(CJSON_LUA.encode("utf-8")).decode("utf-8")
with open(cjson, "wt") as fp:
fp.write(cjson_lua)
if not os.path.exists(bit):
from .const import BIT_LUA
bit_lua = base64.b64decode(BIT_LUA.encode("utf-8")).decode("utf-8")
with open(bit, "wt") as fp:
fp.write(bit_lua)
return True
async def async_setup_entry(hass: HomeAssistant, config_entry):
device_type = config_entry.data.get(CONF_TYPE)
if device_type == CONF_ACCOUNT:
return True
name = config_entry.data.get(CONF_NAME)
device_id = config_entry.data.get(CONF_DEVICE_ID)
device_type = config_entry.data.get(CONF_TYPE)
token = config_entry.data.get(CONF_TOKEN)
key = config_entry.data.get(CONF_KEY)
ip_address = config_entry.options.get(CONF_IP_ADDRESS, None)
if not ip_address:
ip_address = config_entry.data.get(CONF_IP_ADDRESS)
port = config_entry.data.get(CONF_PORT)
model = config_entry.data.get(CONF_MODEL)
protocol = config_entry.data.get(CONF_PROTOCOL)
lua_file = config_entry.data.get("lua_file")
_LOGGER.error(f"lua_file = {lua_file}")
if protocol == 3 and (key is None or key is None):
_LOGGER.error("For V3 devices, the key and the token is required.")
return False
device = MiedaDevice(
name=name,
device_id=device_id,
device_type=device_type,
ip_address=ip_address,
port=port,
token=token,
key=key,
protocol=protocol,
model=model,
lua_file=lua_file,
)
if device:
device.open()
if DOMAIN not in hass.data:
hass.data[DOMAIN] = {}
if DEVICES not in hass.data[DOMAIN]:
hass.data[DOMAIN][DEVICES] = {}
hass.data[DOMAIN][DEVICES][device_id] = device
for platform in [Platform.BINARY_SENSOR]:
hass.async_create_task(hass.config_entries.async_forward_entry_setup(
config_entry, platform))
#config_entry.add_update_listener(update_listener)
return True
return False

53
custom_components/midea_meiju_codec/binary_sensor.py Normal file
View File

@@ -0,0 +1,53 @@
import logging
from homeassistant.const import (
CONF_DEVICE_ID,
STATE_ON,
STATE_OFF
)
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from .midea_entities import MideaEntity
from .const import (
DOMAIN,
DEVICES,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities):
device_id = config_entry.data.get(CONF_DEVICE_ID)
device = hass.data[DOMAIN][DEVICES].get(device_id)
binary_sensors = []
sensor = MideaDeviceStatusSensor(device, "online")
binary_sensors.append(sensor)
async_add_entities(binary_sensors)
class MideaDeviceStatusSensor(MideaEntity):
@property
def device_class(self):
return BinarySensorDeviceClass.CONNECTIVITY
@property
def state(self):
return STATE_ON if self._device.connected else STATE_OFF
@property
def is_on(self):
return self.state == STATE_ON
@property
def available(self):
return True
@property
def extra_state_attributes(self) -> dict:
return self._device.attributes
def update_state(self, status):
try:
_LOGGER.debug("=" * 50)
self.schedule_update_ha_state()
_LOGGER.debug("-" * 50)
except Exception as e:
pass

204
custom_components/midea_meiju_codec/config_flow.py Normal file
View File

@@ -0,0 +1,204 @@
import voluptuous as vol
import logging
import os
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from homeassistant import config_entries
from homeassistant.const import (
CONF_TYPE,
CONF_USERNAME,
CONF_PASSWORD,
CONF_DEVICE,
CONF_PORT,
CONF_MODEL,
CONF_IP_ADDRESS,
CONF_DEVICE_ID,
CONF_PROTOCOL,
CONF_TOKEN,
CONF_NAME
)
from .core.cloud import MeijuCloudExtend
from .core.discover import discover
from .core.device import MiedaDevice
from .const import (
DOMAIN,
STORAGE_PATH,
CONF_ACCOUNT,
CONF_HOME,
CONF_KEY
)
_LOGGER = logging.getLogger(__name__)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_session = None
_cloud = None
_current_home = None
_device_list = {}
def _get_configured_account(self):
for entry in self._async_current_entries():
if entry.data.get(CONF_TYPE) == CONF_ACCOUNT:
return entry.data.get(CONF_USERNAME), entry.data.get(CONF_PASSWORD)
return None, None
def _device_configured(self, device_id):
for entry in self._async_current_entries():
if device_id == entry.data.get(CONF_DEVICE_ID):
return True
return False
async def async_step_user(self, user_input=None, error=None):
username, password = self._get_configured_account()
if username is not None and password is not None:
if self._session is None:
self._session = async_create_clientsession(self.hass)
if self._cloud is None:
self._cloud = MeijuCloudExtend(self._session, username, password)
if await self._cloud.login():
return await self.async_step_home()
if user_input is not None:
return self.async_create_entry(
title=f"{user_input[CONF_USERNAME]}",
data={
CONF_TYPE: CONF_ACCOUNT,
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD]
})
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str
}),
errors={"base": error} if error else None
)
async def async_step_home(self, user_input=None, error=None):
if user_input is not None:
self._current_home = user_input[CONF_HOME]
return await self.async_step_device()
homes = await self._cloud.get_homegroups()
home_list = {}
for home in homes:
home_list[int(home.get("homegroupId"))] = home.get("name")
return self.async_show_form(
step_id="home",
data_schema=vol.Schema({
vol.Required(CONF_HOME, default=list(home_list.keys())[0]):
vol.In(home_list),
}),
errors={"base": error} if error else None
)
async def async_step_device(self, user_input=None, error=None):
if user_input is not None:
# 下载lua
# 本地尝试连接设备
device = self._device_list[user_input[CONF_DEVICE]]
if not device.get("online"):
return await self.async_step_device(error="offline_error")
discover_devices = discover([device["type"]])
_LOGGER.debug(discover_devices)
if discover_devices is None or len(discover_devices) == 0:
return await self.async_step_device(error="discover_failed")
current_device = discover_devices.get(user_input[CONF_DEVICE])
if current_device is None:
return await self.async_step_device(error="discover_failed")
os.makedirs(self.hass.config.path(STORAGE_PATH), exist_ok=True)
path = self.hass.config.path(STORAGE_PATH)
file = await self._cloud.get_lua(device["sn"], device["type"], path, device["enterprise_code"])
if file is None:
return await self.async_step_device(error="download_lua_failed")
use_token = None
use_key = None
connected = False
if current_device.get("protocol") == 3:
for byte_order_big in [False, True]:
token, key = await self._cloud.get_token(user_input[CONF_DEVICE], byte_order_big=byte_order_big)
if token and key:
dm = MiedaDevice(
name=device.get("name"),
device_id=user_input[CONF_DEVICE],
device_type=current_device.get(CONF_TYPE),
ip_address=current_device.get(CONF_IP_ADDRESS),
port=current_device.get(CONF_PORT),
token=token,
key=key,
protocol=3,
model=device.get(CONF_MODEL),
lua_file=None
)
if dm.connect():
use_token = token
use_key = key
connected = True
else:
return await self.async_step_device(error="cant_get_token")
else:
dm = MiedaDevice(
name=device.get("name"),
device_id=user_input[CONF_DEVICE],
device_type=current_device.get(CONF_TYPE),
ip_address=current_device.get(CONF_IP_ADDRESS),
port=current_device.get(CONF_PORT),
token=use_token,
key=use_key,
protocol=2,
model=device.get(CONF_MODEL),
lua_file=None
)
if dm.connect():
connected = True
if not connected:
return await self.async_step_device(error="connect_error")
return self.async_create_entry(
title=device.get("name"),
data={
CONF_NAME: device.get("name"),
CONF_DEVICE_ID: user_input[CONF_DEVICE],
CONF_TYPE: current_device.get("type"),
CONF_PROTOCOL: current_device.get("protocol"),
CONF_IP_ADDRESS: current_device.get("ip_address"),
CONF_PORT: current_device.get("port"),
CONF_MODEL: device.get("model"),
CONF_TOKEN: use_token,
CONF_KEY: use_key,
"lua_file": file,
"sn": device.get("sn"),
"sn8": device.get("sn8"),
})
devices = await self._cloud.get_devices(self._current_home)
self._device_list = {}
device_list = {}
for device in devices:
if not self._device_configured(int(device.get("applianceCode"))):
self._device_list[int(device.get("applianceCode"))] = {
"name": device.get("name"),
"type": int(device.get("type"), 16),
"sn8": device.get("sn8"),
"sn": device.get("sn"),
"model": device.get("productModel"),
"enterprise_code": device.get("enterpriseCode"),
"online": device.get("onlineStatus") == "1"
}
device_list[int(device.get("applianceCode"))] = \
f"{device.get('name')} ({'在线' if device.get('onlineStatus') == '1' else '离线'})"
if len(self._device_list) == 0:
return await self.async_step_device(error="no_new_devices")
return self.async_show_form(
step_id="device",
data_schema=vol.Schema({
vol.Required(CONF_DEVICE, default=list(device_list.keys())[0]):
vol.In(device_list),
}),
errors={"base": error} if error else None
)
class OptionsFlowHandler(config_entries.OptionsFlow):
def __init__(self, config_entry: config_entries.ConfigEntry):
pass

8
custom_components/midea_meiju_codec/const.py Normal file
View File

File diff suppressed because one or more lines are too long

203
custom_components/midea_meiju_codec/core/cloud.py Normal file
View File

@@ -0,0 +1,203 @@
from aiohttp import ClientSession
from secrets import token_hex, token_urlsafe
from .security import CloudSecurity
from threading import Lock
import datetime
import logging
import time
import json
_LOGGER = logging.getLogger(__name__)
CLIENT_TYPE = 1 # Android
FORMAT = 2 # JSON
APP_KEY = "4675636b"
class MideaCloudBase:
LANGUAGE = "en_US"
APP_ID = "1010"
SRC = "1010"
LOGIN_KEY = None
IOT_KEY = None
DEVICE_ID = int(time.time() * 100000)
def __init__(self, session: ClientSession, security, username: str, password: str, server: str = None):
self.session = session
self.username = username
self.password = password
self.server = None
self.login_id = None
self.access_token = ""
self.key = None
self._api_lock = Lock()
self.login_session = None
self.security = security
self.server = server
async def api_request(self, endpoint, args=None, data=None):
args = args or {}
headers = {}
if data is None:
data = {
"appId": self.APP_ID,
"format": FORMAT,
"clientType": CLIENT_TYPE,
"language": self.LANGUAGE,
"src": self.SRC,
"stamp": datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
"deviceId": self.DEVICE_ID,
}
data.update(args)
if not data.get("reqId"):
data.update({
"reqId": token_hex(16),
})
url = self.server + endpoint
random = str(int(time.time()))
sign = self.security.sign(json.dumps(data), random)
headers.update({
"Content-Type": "application/json",
"secretVersion": "1",
"sign": sign,
"random": random,
"accessToken": self.access_token
})
response = {"code": -1}
for i in range(0, 3):
try:
with self._api_lock:
r = await self.session.request("POST", url, headers=headers, data=json.dumps(data), timeout=10)
raw = await r.read()
_LOGGER.debug(f"Endpoint: {endpoint}, Response: {str(raw)}")
response = json.loads(raw)
break
except Exception as e:
_LOGGER.debug(f"Cloud error: {repr(e)}")
if int(response["code"]) == 0 and "data" in response:
return response["data"]
return None
async def get_login_id(self):
response = await self.api_request(
"/v1/user/login/id/get",
args={"loginAccount": self.username}
)
if response:
self.login_id = response["loginId"]
return True
return False
async def login(self):
result = await self.get_login_id()
if result:
response = await self.api_request(
"/mj/user/login",
data={
"data": {
"appKey": APP_KEY,
"platform": FORMAT,
"deviceId": self.DEVICE_ID
},
"iotData": {
"appId": self.APP_ID,
"clientType": CLIENT_TYPE,
"iampwd": self.security.encrypt_iam_password(self.login_id, self.password),
"loginAccount": self.username,
"password": self.security.encrypt_password(self.login_id, self.password),
"pushToken": token_urlsafe(120),
"reqId": token_hex(16),
"src": self.SRC,
"stamp": datetime.time().strftime("%Y%m%d%H%M%S"),
},
}
)
if response:
self.access_token = response["mdata"]["accessToken"]
if "key" in response:
self.key = CloudSecurity.decrypt(bytes.fromhex(response["key"]))
return True
return False
async def get_token(self, device_id: int, byte_order_big=False):
if byte_order_big:
udpid = CloudSecurity.get_udpid(device_id.to_bytes(6, "big"))
else:
udpid = CloudSecurity.get_udpid(device_id.to_bytes(6, "little"))
_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",
args={"udpid": udpid}
)
if response and "tokenlist" in response:
for token in response["tokenlist"]:
if token["udpId"] == udpid:
return token["token"].upper(), token["key"].upper()
return None, None
class MeijuCloudExtend(MideaCloudBase):
LANGUAGE = "zh_CN"
LOGIN_KEY = "ad0ee21d48a64bf49f4fb583ab76e799"
IOT_KEY = "prod_secret123@muc"
SERVER = "https://mp-prod.smartmidea.net/mas/v5/app/proxy?alias="
def __init__(self, session: ClientSession, username: str, password: str):
super().__init__(session=session,
security=CloudSecurity(self.IOT_KEY, self.LOGIN_KEY),
username=username,
password=password,
server=self.SERVER)
async def get_homegroups(self):
response = await self.api_request("/v1/homegroup/list/get", args={})
return response.get("homeList")
async def get_devices(self, homegroupID=None):
if homegroupID is None:
homes = []
homegroups = await self.get_homegroups()
if homegroups:
for home in homegroups:
homes.append(home["homegroupId"])
else:
homes = [homegroupID]
devices = []
for home in homes:
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)
return devices
async def get_lua(self, sn, device_type, path, enterprise_code=None):
response = await self.api_request(
"/v1/appliance/protocol/lua/luaGet",
data={
"applianceSn": sn,
"applianceType": f"0x{'%02X' % device_type}",
"applianceMFCode": enterprise_code if enterprise_code else "0000",
'version': "0",
"iotAppId": "900",
"stamp": datetime.datetime.now().strftime("%Y%m%d%H%M%S")
}
)
fnm = None
if response:
res = await self.session.get(response["url"])
if res.status == 200:
lua = await res.text()
if lua:
stream = 'local bit = require "bit" ' + CloudSecurity.decrypt(bytes.fromhex(lua)).decode()
fnm = f"{path}/{response['fileName']}"
with open(fnm, "w") as fp:
fp.write(stream)
return fnm

46
custom_components/midea_meiju_codec/core/crc8.py Normal file
View File

@@ -0,0 +1,46 @@
crc8_854_table = [
0x00, 0x5E, 0xBC, 0xE2, 0x61, 0x3F, 0xDD, 0x83,
0xC2, 0x9C, 0x7E, 0x20, 0xA3, 0xFD, 0x1F, 0x41,
0x9D, 0xC3, 0x21, 0x7F, 0xFC, 0xA2, 0x40, 0x1E,
0x5F, 0x01, 0xE3, 0xBD, 0x3E, 0x60, 0x82, 0xDC,
0x23, 0x7D, 0x9F, 0xC1, 0x42, 0x1C, 0xFE, 0xA0,
0xE1, 0xBF, 0x5D, 0x03, 0x80, 0xDE, 0x3C, 0x62,
0xBE, 0xE0, 0x02, 0x5C, 0xDF, 0x81, 0x63, 0x3D,
0x7C, 0x22, 0xC0, 0x9E, 0x1D, 0x43, 0xA1, 0xFF,
0x46, 0x18, 0xFA, 0xA4, 0x27, 0x79, 0x9B, 0xC5,
0x84, 0xDA, 0x38, 0x66, 0xE5, 0xBB, 0x59, 0x07,
0xDB, 0x85, 0x67, 0x39, 0xBA, 0xE4, 0x06, 0x58,
0x19, 0x47, 0xA5, 0xFB, 0x78, 0x26, 0xC4, 0x9A,
0x65, 0x3B, 0xD9, 0x87, 0x04, 0x5A, 0xB8, 0xE6,
0xA7, 0xF9, 0x1B, 0x45, 0xC6, 0x98, 0x7A, 0x24,
0xF8, 0xA6, 0x44, 0x1A, 0x99, 0xC7, 0x25, 0x7B,
0x3A, 0x64, 0x86, 0xD8, 0x5B, 0x05, 0xE7, 0xB9,
0x8C, 0xD2, 0x30, 0x6E, 0xED, 0xB3, 0x51, 0x0F,
0x4E, 0x10, 0xF2, 0xAC, 0x2F, 0x71, 0x93, 0xCD,
0x11, 0x4F, 0xAD, 0xF3, 0x70, 0x2E, 0xCC, 0x92,
0xD3, 0x8D, 0x6F, 0x31, 0xB2, 0xEC, 0x0E, 0x50,
0xAF, 0xF1, 0x13, 0x4D, 0xCE, 0x90, 0x72, 0x2C,
0x6D, 0x33, 0xD1, 0x8F, 0x0C, 0x52, 0xB0, 0xEE,
0x32, 0x6C, 0x8E, 0xD0, 0x53, 0x0D, 0xEF, 0xB1,
0xF0, 0xAE, 0x4C, 0x12, 0x91, 0xCF, 0x2D, 0x73,
0xCA, 0x94, 0x76, 0x28, 0xAB, 0xF5, 0x17, 0x49,
0x08, 0x56, 0xB4, 0xEA, 0x69, 0x37, 0xD5, 0x8B,
0x57, 0x09, 0xEB, 0xB5, 0x36, 0x68, 0x8A, 0xD4,
0x95, 0xCB, 0x29, 0x77, 0xF4, 0xAA, 0x48, 0x16,
0xE9, 0xB7, 0x55, 0x0B, 0x88, 0xD6, 0x34, 0x6A,
0x2B, 0x75, 0x97, 0xC9, 0x4A, 0x14, 0xF6, 0xA8,
0x74, 0x2A, 0xC8, 0x96, 0x15, 0x4B, 0xA9, 0xF7,
0xB6, 0xE8, 0x0A, 0x54, 0xD7, 0x89, 0x6B, 0x35
]
def calculate(data):
crc_value = 0
for m in data:
k = crc_value ^ m
if k > 256:
k -= 256
if k < 0:
k += 256
crc_value = crc8_854_table[k]
return crc_value

297
custom_components/midea_meiju_codec/core/device.py Normal file
View File

@@ -0,0 +1,297 @@
import threading
from enum import IntEnum
from .security import LocalSecurity, MSGTYPE_HANDSHAKE_REQUEST, MSGTYPE_ENCRYPTED_REQUEST
from .packet_builder import PacketBuilder
from .lua_runtime import MideaCodec
import socket
import logging
import json
import time
_LOGGER = logging.getLogger(__name__)
class AuthException(Exception):
pass
class ResponseException(Exception):
pass
class RefreshFailed(Exception):
pass
class ParseMessageResult(IntEnum):
SUCCESS = 0
PADDING = 1
ERROR = 99
class MiedaDevice(threading.Thread):
def __init__(self,
name: str,
device_id: int,
device_type: int,
ip_address: str,
port: int,
token: str | None,
key: str | None,
protocol: int,
model: str,
lua_file: str | None):
threading.Thread.__init__(self)
self._socket = None
self._ip_address = ip_address
self._port = port
self._security = LocalSecurity()
self._token = bytes.fromhex(token) if token else None
self._key = bytes.fromhex(key) if key else None
self._buffer = b""
self._device_name = name
self._device_id = device_id
self._device_type = device_type
self._protocol = protocol
self._model = model
self._updates = []
self._unsupported_protocol = []
self._is_run = False
self._device_protocol_version = 0
self._sub_type = None
self._sn = None
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
@property
def device_name(self):
return self._device_name
@property
def device_id(self):
return self._device_id
@property
def device_type(self):
return self._device_type
@property
def model(self):
return self._model
@property
def attributes(self):
return self._attributes
@property
def connected(self):
return self._connected
@staticmethod
def fetch_v2_message(msg):
result = []
while len(msg) > 0:
factual_msg_len = len(msg)
if factual_msg_len < 6:
break
alleged_msg_len = msg[4] + (msg[5] << 8)
if factual_msg_len >= alleged_msg_len:
result.append(msg[:alleged_msg_len])
msg = msg[alleged_msg_len:]
else:
break
return result, msg
def connect(self, refresh=False):
try:
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._socket.settimeout(10)
_LOGGER.debug(f"[{self._device_id}] Connecting to {self._ip_address}:{self._port}")
self._socket.connect((self._ip_address, self._port))
_LOGGER.debug(f"[{self._device_id}] Connected")
if self._protocol == 3:
self.authenticate()
_LOGGER.debug(f"[{self._device_id}] Authentication success")
self.device_connected(True)
if refresh:
self.refresh_status()
return True
except socket.timeout:
_LOGGER.debug(f"[{self._device_id}] Connection timed out")
except socket.error:
_LOGGER.debug(f"[{self._device_id}] Connection error")
except AuthException:
_LOGGER.debug(f"[{self._device_id}] Authentication failed")
except ResponseException:
_LOGGER.debug(f"[{self._device_id}] Unexpected response received")
except RefreshFailed:
_LOGGER.debug(f"[{self._device_id}] Refresh status is timed out")
except Exception as e:
_LOGGER.error(f"[{self._device_id}] Unknown error: {e.__traceback__.tb_frame.f_globals['__file__']}, "
f"{e.__traceback__.tb_lineno}, {repr(e)}")
self.device_connected(False)
return False
def authenticate(self):
request = self._security.encode_8370(
self._token, MSGTYPE_HANDSHAKE_REQUEST)
_LOGGER.debug(f"[{self._device_id}] Handshaking")
self._socket.send(request)
response = self._socket.recv(512)
if len(response) < 20:
raise AuthException()
response = response[8: 72]
self._security.tcp_key(response, self._key)
def send_message(self, data):
if self._protocol == 3:
self.send_message_v3(data, msg_type=MSGTYPE_ENCRYPTED_REQUEST)
else:
self.send_message_v2(data)
def send_message_v2(self, data):
if self._socket is not None:
self._socket.send(data)
else:
_LOGGER.debug(f"[{self._device_id}] Send failure, device disconnected, data: {data.hex()}")
def send_message_v3(self, data, msg_type=MSGTYPE_ENCRYPTED_REQUEST):
data = self._security.encode_8370(data, msg_type)
self.send_message_v2(data)
def build_send(self, cmd):
_LOGGER.debug(f"[{self._device_id}] Sending: {cmd}")
bytes_cmd = bytes.fromhex(cmd)
msg = PacketBuilder(self._device_id, bytes_cmd).finalize()
self.send_message(msg)
def refresh_status(self):
query_cmd = self._lua_runtime.build_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)
_LOGGER.debug(f"[{self._device_id}] Received: {decrypted.hex()}")
# 这就是最终消息
status = self._lua_runtime.decode_status(decrypted.hex())
_LOGGER.debug(f"[{self._device_id}] 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):
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}
self.update_all(status)
def register_update(self, update):
self._updates.append(update)
def update_all(self, status):
_LOGGER.debug(f"[{self._device_id}] Status update: {status}")
for update in self._updates:
update(status)
def open(self):
if not self._is_run:
self._is_run = True
threading.Thread.start(self)
def close(self):
if self._is_run:
self._is_run = False
self.close_socket()
def close_socket(self):
self._unsupported_protocol = []
self._buffer = b""
if self._socket:
self._socket.close()
self._socket = None
def set_ip_address(self, ip_address):
_LOGGER.debug(f"[{self._device_id}] Update IP address to {ip_address}")
self._ip_address = ip_address
self.close_socket()
def run(self):
while self._is_run:
while self._socket is None:
if self.connect(refresh=True) is False:
if not self._is_run:
return
self.close_socket()
time.sleep(5)
timeout_counter = 0
start = time.time()
previous_refresh = start
previous_heartbeat = start
self._socket.settimeout(1)
while True:
try:
now = time.time()
if now - previous_refresh >= self._refresh_interval:
self.refresh_status()
previous_refresh = now
if now - previous_heartbeat >= self._heartbeat_interval:
self.send_heartbeat()
previous_heartbeat = now
msg = self._socket.recv(512)
msg_len = len(msg)
if msg_len == 0:
raise socket.error("Connection closed by peer")
result = self.parse_message(msg)
if result == ParseMessageResult.ERROR:
_LOGGER.debug(f"[{self._device_id}] Message 'ERROR' received")
self.close_socket()
break
elif result == ParseMessageResult.SUCCESS:
timeout_counter = 0
except socket.timeout:
timeout_counter = timeout_counter + 1
if timeout_counter >= 120:
_LOGGER.debug(f"[{self._device_id}] Heartbeat timed out")
self.close_socket()
break
except socket.error as e:
_LOGGER.debug(f"[{self._device_id}] Socket error {repr(e)}")
self.close_socket()
break
except Exception as e:
_LOGGER.error(f"[{self._device_id}] Unknown error :{e.__traceback__.tb_frame.f_globals['__file__']}, "
f"{e.__traceback__.tb_lineno}, {repr(e)}")
self.close_socket()
break

171
custom_components/midea_meiju_codec/core/discover.py Normal file
View File

@@ -0,0 +1,171 @@
import logging
import socket
import ifaddr
from ipaddress import IPv4Network
from .security import LocalSecurity
try:
import xml.etree.cElementTree as ET
except ImportError:
import xml.etree.ElementTree as ET
_LOGGER = logging.getLogger(__name__)
BROADCAST_MSG = bytearray([
0x5a, 0x5a, 0x01, 0x11, 0x48, 0x00, 0x92, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x7f, 0x75, 0xbd, 0x6b, 0x3e, 0x4f, 0x8b, 0x76,
0x2e, 0x84, 0x9c, 0x6e, 0x57, 0x8d, 0x65, 0x90,
0x03, 0x6e, 0x9d, 0x43, 0x42, 0xa5, 0x0f, 0x1f,
0x56, 0x9e, 0xb8, 0xec, 0x91, 0x8e, 0x92, 0xe5
])
DEVICE_INFO_MSG = bytearray([
0x5a, 0x5a, 0x15, 0x00, 0x00, 0x38, 0x00, 0x04,
0x00, 0x00, 0x00, 0x00, 0x00, 0x27, 0x33, 0x05,
0x13, 0x06, 0x14, 0x14, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x03, 0xe8, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0xca, 0x8d, 0x9b, 0xf9, 0xa0, 0x30, 0x1a, 0xe3,
0xb7, 0xe4, 0x2d, 0x53, 0x49, 0x47, 0x62, 0xbe
])
def discover(discover_type=None, ip_address=None):
_LOGGER.debug(f"Begin discover, type: {discover_type}, ip_address: {ip_address}")
if discover_type is None:
discover_type = []
security = LocalSecurity()
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.settimeout(5)
found_devices = {}
if ip_address is None:
addrs = enum_all_broadcast()
else:
addrs = [ip_address]
for v in range(0, 3):
for addr in addrs:
sock.sendto(BROADCAST_MSG, (addr, 6445))
sock.sendto(BROADCAST_MSG, (addr, 20086))
while True:
try:
data, addr = sock.recvfrom(512)
ip = addr[0]
_LOGGER.debug(f"Received broadcast from {addr}: {data.hex()}")
if len(data) >= 104 and (data[:2].hex() == "5a5a" or data[8:10].hex() == "5a5a"):
if data[:2].hex() == "5a5a":
protocol = 2
elif data[:2].hex() == "8370":
protocol = 3
if data[8:10].hex() == "5a5a":
data = data[8:-16]
else:
continue
device_id = int.from_bytes(bytearray.fromhex(data[20:26].hex()), "little")
if device_id in found_devices:
continue
encrypt_data = data[40:-16]
reply = security.aes_decrypt(encrypt_data)
_LOGGER.debug(f"Declassified reply: {reply.hex()}")
ssid = reply[41:41 + reply[40]].decode("utf-8")
device_type = ssid.split("_")[1]
port = bytes2port(reply[4:8])
model = reply[17:25].decode("utf-8")
sn = reply[8:40].decode("utf-8")
elif data[:6].hex() == "3c3f786d6c20":
protocol = 1
root = ET.fromstring(data.decode(
encoding="utf-8", errors="replace"))
child = root.find("body/device")
m = child.attrib
port, sn, device_type = int(m["port"]), m["apc_sn"], str(
hex(int(m["apc_type"])))[2:]
response = get_device_info(ip, int(port))
device_id = get_id_from_response(response)
if len(sn) == 32:
model = sn[9:17]
elif len(sn) == 22:
model = sn[3:11]
else:
model = ""
else:
continue
device = {
"device_id": device_id,
"type": int(device_type, 16),
"ip_address": ip,
"port": port,
"model": model,
"sn": sn,
"protocol": protocol
}
if len(discover_type) == 0 or device.get("type") in discover_type:
found_devices[device_id] = device
_LOGGER.debug(f"Found a supported device: {device}")
else:
_LOGGER.debug(f"Found a unsupported device: {device}")
except socket.timeout:
break
except socket.error as e:
_LOGGER.debug(f"Socket error: {repr(e)}")
return found_devices
def get_id_from_response(response):
if response[64:-16][:6].hex() == "3c3f786d6c20":
xml = response[64:-16]
root = ET.fromstring(xml.decode(encoding="utf-8", errors="replace"))
child = root.find("smartDevice")
m = child.attrib
return int.from_bytes(bytearray.fromhex(m["devId"]), "little")
else:
return 0
def bytes2port(paramArrayOfbyte):
if paramArrayOfbyte is None:
return 0
b, i = 0, 0
while b < 4:
if b < len(paramArrayOfbyte):
b1 = paramArrayOfbyte[b] & 0xFF
else:
b1 = 0
i |= b1 << b * 8
b += 1
return i
def get_device_info(device_ip, device_port: int):
response = bytearray(0)
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(8)
device_address = (device_ip, device_port)
sock.connect(device_address)
_LOGGER.debug(f"Sending to {device_ip}:{device_port} {DEVICE_INFO_MSG.hex()}")
sock.sendall(DEVICE_INFO_MSG)
response = sock.recv(512)
except socket.timeout:
_LOGGER.warning(f"Connect the device {device_ip}:{device_port} timed out for 8s. "
f"Don't care about a small amount of this. if many maybe not support."
)
except socket.error:
_LOGGER.warning(f"Can't connect to Device {device_ip}:{device_port}")
return response
def enum_all_broadcast():
nets = []
adapters = ifaddr.get_adapters()
for adapter in adapters:
for ip in adapter.ips:
if ip.is_IPv4 and ip.network_prefix < 32:
localNet = IPv4Network(f"{ip.ip}/{ip.network_prefix}", strict=False)
if localNet.is_private and not localNet.is_loopback and not localNet.is_link_local:
nets.append(str(localNet.broadcast_address))
return nets

42
custom_components/midea_meiju_codec/core/lua_runtime.py Normal file
View File

@@ -0,0 +1,42 @@
import lupa
import logging
import threading
import json
_LOGGER = logging.getLogger(__name__)
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):
with self._lock:
result = self._json_to_data(json)
return result
def data_to_json(self, data):
with self._lock:
result = self._data_to_json(data)
return result
class MideaCodec(LuaRuntime):
def __init__(self, file):
super().__init__(file)
def build_query(self, append=None):
json_str = '{"deviceinfo":{"deviceSubType":1},"query":{}}'
result = self.json_to_data(json_str)
return result
def decode_status(self, data):
data = '{"deviceinfo": {"deviceSubType":1},"msg":{"data": "' + data + '"}}'
result = self.data_to_json(data)
status = json.loads(result)
return status.get("status")

158
custom_components/midea_meiju_codec/core/message.py Normal file
View File

@@ -0,0 +1,158 @@
from abc import ABC
from enum import IntEnum
class MessageLenError(Exception):
pass
class MessageBodyError(Exception):
pass
class MessageCheckSumError(Exception):
pass
class MessageType(IntEnum):
set = 0x02,
query = 0x03,
notify1 = 0x04,
notify2 = 0x05,
exception = 0x06,
querySN = 0x07,
exception2 = 0x0A,
querySubtype = 0xA0
class MessageBase(ABC):
HEADER_LENGTH = 10
def __init__(self):
self._device_type = 0x00
self._message_type = 0x00
self._body_type = 0x00
self._device_protocol_version = 0
@staticmethod
def checksum(data):
return (~ sum(data) + 1) & 0xff
@property
def header(self):
raise NotImplementedError
@property
def body(self):
raise NotImplementedError
@property
def message_type(self):
return self._message_type
@message_type.setter
def message_type(self, value):
self._message_type = value
@property
def device_type(self):
return self._device_type
@device_type.setter
def device_type(self, value):
self._device_type = value
@property
def body_type(self):
return self._body_type
@body_type.setter
def body_type(self, value):
self._body_type = value
@property
def device_protocol_version(self):
return self._device_protocol_version
@device_protocol_version.setter
def device_protocol_version(self, value):
self._device_protocol_version = value
def __str__(self) -> str:
output = {
"header": self.header.hex(),
"body": self.body.hex(),
"message type": "%02x" % self._message_type,
"body type": ("%02x" % self._body_type) if self._body_type is not None else "None"
}
return str(output)
class MessageRequest(MessageBase):
def __init__(self, device_protocol_version, device_type, message_type, body_type):
super().__init__()
self.device_protocol_version = device_protocol_version
self.device_type = device_type
self.message_type = message_type
self.body_type = body_type
@property
def header(self):
length = self.HEADER_LENGTH + len(self.body)
return bytearray([
# flag
0xAA,
# length
length,
# device type
self._device_type,
# frame checksum
0x00, # self._device_type ^ length,
# unused
0x00, 0x00,
# frame ID
0x00,
# frame protocol version
0x00,
# device protocol version
self._device_protocol_version,
# frame type
self._message_type
])
@property
def _body(self):
raise NotImplementedError
@property
def body(self):
body = bytearray([])
if self.body_type is not None:
body.append(self.body_type)
if self._body is not None:
body.extend(self._body)
return body
def serialize(self):
stream = self.header + self.body
stream.append(MessageBase.checksum(stream[1:]))
return stream
class MessageQuestCustom(MessageRequest):
def __init__(self, device_type, cmd_type, cmd_body):
super().__init__(
device_protocol_version=0,
device_type=device_type,
message_type=cmd_type,
body_type=None)
self._cmd_body = cmd_body
@property
def _body(self):
return bytearray([])
@property
def body(self):
return self._cmd_body

60
custom_components/midea_meiju_codec/core/packet_builder.py Normal file
View File

@@ -0,0 +1,60 @@
from .security import LocalSecurity
import datetime
class PacketBuilder:
def __init__(self, device_id: int, command):
self.command = None
self.security = LocalSecurity()
# aa20ac00000000000003418100ff03ff000200000000000000000000000006f274
# Init the packet with the header data.
self.packet = bytearray([
# 2 bytes - StaicHeader
0x5a, 0x5a,
# 2 bytes - mMessageType
0x01, 0x11,
# 2 bytes - PacketLenght
0x00, 0x00,
# 2 bytes
0x20, 0x00,
# 4 bytes - MessageId
0x00, 0x00, 0x00, 0x00,
# 8 bytes - Date&Time
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
# 6 bytes - mDeviceID
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
# 12 bytes
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00
])
self.packet[12:20] = self.packet_time()
self.packet[20:28] = device_id.to_bytes(8, "little")
self.command = command
def finalize(self, msg_type=1):
if msg_type != 1:
self.packet[3] = 0x10
self.packet[6] = 0x7b
else:
self.packet.extend(self.security.aes_encrypt(self.command))
# PacketLenght
self.packet[4:6] = (len(self.packet) + 16).to_bytes(2, "little")
# Append a basic checksum data(16 bytes) to the packet
self.packet.extend(self.encode32(self.packet))
return self.packet
def encode32(self, data: bytearray):
return self.security.encode32_data(data)
@staticmethod
def checksum(data):
return (~ sum(data) + 1) & 0xff
@staticmethod
def packet_time():
t = datetime.datetime.now().strftime("%Y%m%d%H%M%S%f")[
:16]
b = bytearray()
for i in range(0, len(t), 2):
d = int(t[i:i+2])
b.insert(0, d)
return b

160
custom_components/midea_meiju_codec/core/security.py Normal file
View File

@@ -0,0 +1,160 @@
import logging
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Util.strxor import strxor
from Crypto.Random import get_random_bytes
from hashlib import md5, sha256
from urllib.parse import urlparse
import hmac
import urllib
_LOGGER = logging.getLogger(__name__)
MSGTYPE_HANDSHAKE_REQUEST = 0x0
MSGTYPE_HANDSHAKE_RESPONSE = 0x1
MSGTYPE_ENCRYPTED_RESPONSE = 0x3
MSGTYPE_ENCRYPTED_REQUEST = 0x6
class CloudSecurity:
def __init__(self, iotKey, loginKey):
self._hmackey = "PROD_VnoClJI9aikS8dyy"
self._iotkey = iotKey
self._loginKey = loginKey
def sign(self, data: str, random: str) -> str:
msg = self._iotkey
if data:
msg += data
msg += random
sign = hmac.new(self._hmackey.encode("ascii"), msg.encode("ascii"), sha256)
return sign.hexdigest()
def encrypt_password(self, loginId, data):
m = sha256()
m.update(data.encode("ascii"))
login_hash = loginId + m.hexdigest() + self._loginKey
m = sha256()
m.update(login_hash.encode("ascii"))
return m.hexdigest()
def encrypt_iam_password(self, loginId, data) -> str:
md = md5()
md.update(data.encode("ascii"))
md_second = md5()
md_second.update(md.hexdigest().encode("ascii"))
return md_second.hexdigest()
@staticmethod
def get_udpid(data):
data = bytearray(sha256(data).digest())
for i in range(0, 16):
data[i] ^= data[i + 16]
return data[0: 16].hex()
@staticmethod
def decrypt(data:bytes, key:bytes="96c7acdfdb8af79a".encode()):
return unpad(AES.new(key, AES.MODE_ECB).decrypt(data), 16)
@staticmethod
def encrypt(data:bytes, key:bytes="96c7acdfdb8af79a".encode()):
return AES.new(key, AES.MODE_ECB).encrypt(pad(data, 16))
class LocalSecurity:
def __init__(self):
self.blockSize = 16
self.iv = b"\0" * 16
self.aes_key = bytes.fromhex("6a92ef406bad2f0359baad994171ea6d")
self.salt = bytes.fromhex("78686469776a6e6368656b6434643531326368646a783564386534633339344432443753")
self._tcp_key = None
self._request_count = 0
self._response_count = 0
def aes_decrypt(self, raw):
try:
return unpad(AES.new(self.aes_key, AES.MODE_ECB).decrypt(bytearray(raw)), self.blockSize)
except ValueError as e:
_LOGGER.error(f"Error in aes_decrypt: {repr(e)} - data: {raw.hex()}")
return bytearray(0)
def aes_encrypt(self, raw):
return AES.new(self.aes_key, AES.MODE_ECB).encrypt(bytearray(pad(raw, self.blockSize)))
def aes_cbc_decrypt(self, raw, key):
return AES.new(key=key, mode=AES.MODE_CBC, iv=self.iv).decrypt(raw)
def aes_cbc_encrypt(self, raw, key):
return AES.new(key=key, mode=AES.MODE_CBC, iv=self.iv).encrypt(raw)
def encode32_data(self, raw):
return md5(raw + self.salt).digest()
def tcp_key(self, response, key):
if response == b"ERROR":
raise Exception("authentication failed")
if len(response) != 64:
raise Exception("unexpected data length")
payload = response[:32]
sign = response[32:]
plain = self.aes_cbc_decrypt(payload, key)
if sha256(plain).digest() != sign:
raise Exception("sign does not match")
self._tcp_key = strxor(plain, key)
self._request_count = 0
self._response_count = 0
return self._tcp_key
def encode_8370(self, data, msgtype):
header = bytearray([0x83, 0x70])
size, padding = len(data), 0
if msgtype in (MSGTYPE_ENCRYPTED_RESPONSE, MSGTYPE_ENCRYPTED_REQUEST):
if (size + 2) % 16 != 0:
padding = 16 - (size + 2 & 0xf)
size += padding + 32
data += get_random_bytes(padding)
header += size.to_bytes(2, "big")
header += bytearray([0x20, padding << 4 | msgtype])
data = self._request_count.to_bytes(2, "big") + data
self._request_count += 1
if self._request_count >= 0xFFFF:
self._request_count = 0
if msgtype in (MSGTYPE_ENCRYPTED_RESPONSE, MSGTYPE_ENCRYPTED_REQUEST):
sign = sha256(header + data).digest()
data = self.aes_cbc_encrypt(raw=data, key=self._tcp_key) + sign
return header + data
def decode_8370(self, data):
if len(data) < 6:
return [], data
header = data[:6]
if header[0] != 0x83 or header[1] != 0x70:
raise Exception("not an 8370 message")
size = int.from_bytes(header[2:4], "big") + 8
leftover = None
if len(data) < size:
return [], data
elif len(data) > size:
leftover = data[size:]
data = data[:size]
if header[4] != 0x20:
raise Exception("missing byte 4")
padding = header[5] >> 4
msgtype = header[5] & 0xf
data = data[6:]
if msgtype in (MSGTYPE_ENCRYPTED_RESPONSE, MSGTYPE_ENCRYPTED_REQUEST):
sign = data[-32:]
data = data[:-32]
data = self.aes_cbc_decrypt(raw=data, key=self._tcp_key)
if sha256(header + data).digest() != sign:
raise Exception("sign does not match")
if padding:
data = data[:-padding]
self._response_count = int.from_bytes(data[:2], "big")
data = data[2:]
if leftover:
packets, incomplete = self.decode_8370(leftover)
return [data] + packets, incomplete
return [data], b""

12
custom_components/midea_meiju_codec/manifest.json Normal file
View File

@@ -0,0 +1,12 @@
{
"domain": "midea_meiju_codec",
"name": "Midea Meiju Codec",
"version": "v0.0.1",
"config_flow": true,
"documentation": "https://github.com/georgezhao2010/midea-meiju-codec",
"issue_tracker": "https://github.com/georgezhao2010/midea-meiju-codec/issues",
"requirements": ["lupa>=2.0"],
"dependencies": [],
"iot_class": "local_polling",
"codeowners": ["@georgezhao2010"]
}

48
custom_components/midea_meiju_codec/midea_entities.py Normal file
View File

@@ -0,0 +1,48 @@
from homeassistant.helpers.entity import Entity
from .const import DOMAIN
class MideaEntity(Entity):
def __init__(self, device, entity_key: str):
self._device = device
self._device.register_update(self.update_state)
self._entity_key = entity_key
self._unique_id = f"{DOMAIN}.{self._device.device_id}_{entity_key}"
self.entity_id = self._unique_id
self._device_name = self._device.name
@property
def device(self):
return self._device
@property
def device_info(self):
return {
"manufacturer": "Midea",
"model": self._device.model,
"identifiers": {(DOMAIN, self._device.device_id)},
"name": self._device_name
}
@property
def unique_id(self):
return self._unique_id
@property
def should_poll(self):
return False
@property
def state(self):
return self._device.get_attribute(self._entity_key)
@property
def available(self):
return self._device.connected
def update_state(self, status):
if self._entity_key in status or "connected" in status:
try:
self.schedule_update_ha_state()
except Exception as e:
pass

37
custom_components/midea_meiju_codec/translations/en.json Normal file
View File

@@ -0,0 +1,37 @@
{
"config": {
"error": {
"offline_error": "只能配置在线设备",
"download_lua_failed": "下载设备协议脚本失败",
"discover_failed": "无法在本地搜索到该设备",
"no_new_devices": "没有可用的设备",
"cant_get_token": "无法连接美的云获取设备关键信息(Token和Key)",
"config_incorrect": "配置信息不正确, 请检查后重新输入",
"connect_error": "无法连接到指定设备"
},
"step": {
"user": {
"data": {
"username": "用户名（手机号）",
"password": "密码"
},
"description": "登录并保存你的美居账号及密码",
"title": "登录"
},
"home": {
"description": "选择设备所在家庭",
"title": "家庭",
"data": {
"home": "设备所在家庭"
}
},
"device": {
"description": "选择要添加的设备",
"title": "设备",
"data": {
"device": "要添加的设备"
}
}
}
}
}

37
custom_components/midea_meiju_codec/translations/zh-Hans.json Normal file
View File

@@ -0,0 +1,37 @@
{
"config": {
"error": {
"offline_error": "只能配置在线设备",
"download_lua_failed": "下载设备协议脚本失败",
"discover_failed": "无法在本地搜索到该设备",
"no_new_devices": "没有可用的设备",
"cant_get_token": "无法连接美的云获取设备关键信息(Token和Key)",
"config_incorrect": "配置信息不正确, 请检查后重新输入",
"connect_error": "无法连接到指定设备"
},
"step": {
"user": {
"data": {
"username": "用户名（手机号）",
"password": "密码"
},
"description": "登录并保存你的美居账号及密码",
"title": "登录"
},
"home": {
"description": "选择设备所在家庭",
"title": "家庭",
"data": {
"home": "设备所在家庭"
}
},
"device": {
"description": "选择要添加的设备",
"title": "设备",
"data": {
"device": "要添加的设备"
}
}
}
}
}