mirror of
https://github.com/sususweet/midea-meiju-codec.git
synced 2025-09-28 02:32:40 +00:00
275 lines
10 KiB
Python
275 lines
10 KiB
Python
import os
|
|
import base64
|
|
import voluptuous as vol
|
|
from importlib import import_module
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.util.json import load_json
|
|
try:
|
|
from homeassistant.helpers.json import save_json
|
|
except ImportError:
|
|
from homeassistant.util.json import save_json
|
|
from homeassistant.helpers.typing import ConfigType
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.core import (
|
|
HomeAssistant,
|
|
ServiceCall
|
|
)
|
|
from homeassistant.const import (
|
|
Platform,
|
|
CONF_TYPE,
|
|
CONF_PORT,
|
|
CONF_MODEL,
|
|
CONF_IP_ADDRESS,
|
|
CONF_DEVICE_ID,
|
|
CONF_PROTOCOL,
|
|
CONF_TOKEN,
|
|
CONF_NAME,
|
|
CONF_DEVICE,
|
|
CONF_ENTITIES
|
|
)
|
|
|
|
from .core.logger import MideaLogger
|
|
from .core.device import MiedaDevice
|
|
from .data_coordinator import MideaDataUpdateCoordinator
|
|
from .core.cloud import get_midea_cloud
|
|
from .const import (
|
|
DOMAIN,
|
|
DEVICES,
|
|
CONF_REFRESH_INTERVAL,
|
|
CONFIG_PATH,
|
|
CONF_KEY,
|
|
CONF_ACCOUNT,
|
|
CONF_SN8,
|
|
CONF_SN,
|
|
CONF_MODEL_NUMBER,
|
|
CONF_LUA_FILE, CONF_SERVERS
|
|
)
|
|
# 账号型：登录云端、获取设备列表，并为每台设备建立协调器（无本地控制）
|
|
from .const import CONF_PASSWORD as CONF_PASSWORD_KEY, CONF_SERVER as CONF_SERVER_KEY
|
|
|
|
PLATFORMS: list[Platform] = [
|
|
Platform.BINARY_SENSOR,
|
|
# Platform.SENSOR,
|
|
# Platform.SWITCH,
|
|
Platform.CLIMATE,
|
|
Platform.SELECT,
|
|
Platform.WATER_HEATER,
|
|
Platform.FAN
|
|
]
|
|
|
|
def get_sn8_used(hass: HomeAssistant, sn8):
|
|
entries = hass.config_entries.async_entries(DOMAIN)
|
|
count = 0
|
|
for entry in entries:
|
|
if sn8 == entry.data.get("sn8"):
|
|
count += 1
|
|
return count
|
|
|
|
|
|
def remove_device_config(hass: HomeAssistant, sn8):
|
|
config_file = hass.config.path(f"{CONFIG_PATH}/{sn8}.json")
|
|
try:
|
|
os.remove(config_file)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def load_device_config(hass: HomeAssistant, device_type, sn8):
|
|
os.makedirs(hass.config.path(CONFIG_PATH), exist_ok=True)
|
|
config_file = hass.config.path(f"{CONFIG_PATH}/{sn8}.json")
|
|
json_data = load_json(config_file, default={})
|
|
if len(json_data) > 0:
|
|
json_data = json_data.get(sn8)
|
|
else:
|
|
device_path = f".device_mapping.{'T0x%02X' % device_type}"
|
|
try:
|
|
mapping_module = import_module(device_path, __package__)
|
|
if sn8 in mapping_module.DEVICE_MAPPING.keys():
|
|
json_data = mapping_module.DEVICE_MAPPING[sn8]
|
|
elif "default" in mapping_module.DEVICE_MAPPING:
|
|
json_data = mapping_module.DEVICE_MAPPING["default"]
|
|
if len(json_data) > 0:
|
|
save_data = {sn8: json_data}
|
|
save_json(config_file, save_data)
|
|
except ModuleNotFoundError:
|
|
MideaLogger.warning(f"Can't load mapping file for type {'T0x%02X' % device_type}")
|
|
return json_data
|
|
|
|
|
|
def register_services(hass: HomeAssistant):
|
|
|
|
async def async_set_attributes(service: ServiceCall):
|
|
device_id = service.data.get("device_id")
|
|
attributes = service.data.get("attributes")
|
|
MideaLogger.debug(f"Service called: set_attributes, device_id: {device_id}, attributes: {attributes}")
|
|
try:
|
|
coordinator: MideaDataUpdateCoordinator = hass.data[DOMAIN][DEVICES][device_id].get("coordinator")
|
|
except KeyError:
|
|
MideaLogger.error(f"Failed to call service set_attributes: the device {device_id} isn't exist.")
|
|
return
|
|
if coordinator:
|
|
await coordinator.async_set_attributes(attributes)
|
|
|
|
async def async_send_command(service: ServiceCall):
|
|
device_id = service.data.get("device_id")
|
|
cmd_type = service.data.get("cmd_type")
|
|
cmd_body = service.data.get("cmd_body")
|
|
try:
|
|
coordinator: MideaDataUpdateCoordinator = hass.data[DOMAIN][DEVICES][device_id].get("coordinator")
|
|
except KeyError:
|
|
MideaLogger.error(f"Failed to call service send_command: the device {device_id} isn't exist.")
|
|
return
|
|
if coordinator:
|
|
await coordinator.async_send_command(cmd_type, cmd_body)
|
|
|
|
hass.services.async_register(
|
|
DOMAIN,
|
|
"set_attributes",
|
|
async_set_attributes,
|
|
schema=vol.Schema({
|
|
vol.Required("device_id"): vol.Coerce(int),
|
|
vol.Required("attributes"): vol.Any(dict)
|
|
})
|
|
)
|
|
hass.services.async_register(
|
|
DOMAIN, "send_command", async_send_command,
|
|
schema=vol.Schema({
|
|
vol.Required("device_id"): vol.Coerce(int),
|
|
vol.Required("cmd_type"): vol.In([2, 3]),
|
|
vol.Required("cmd_body"): str
|
|
})
|
|
)
|
|
|
|
|
|
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry):
|
|
device_id = config_entry.data.get(CONF_DEVICE_ID)
|
|
if device_id is not None:
|
|
ip_address = config_entry.options.get(
|
|
CONF_IP_ADDRESS, None
|
|
)
|
|
refresh_interval = config_entry.options.get(
|
|
CONF_REFRESH_INTERVAL, None
|
|
)
|
|
device: MiedaDevice = hass.data[DOMAIN][DEVICES][device_id][CONF_DEVICE]
|
|
if device:
|
|
if ip_address is not None:
|
|
device.set_ip_address(ip_address)
|
|
if refresh_interval is not None:
|
|
device.set_refresh_interval(refresh_interval)
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType):
|
|
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)
|
|
|
|
register_services(hass)
|
|
return True
|
|
|
|
|
|
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
|
|
device_type = config_entry.data.get(CONF_TYPE)
|
|
MideaLogger.debug(f"async_setup_entry type={device_type} data={config_entry.data}")
|
|
if device_type == CONF_ACCOUNT:
|
|
account = config_entry.data.get(CONF_ACCOUNT)
|
|
password = config_entry.data.get(CONF_PASSWORD_KEY)
|
|
server = config_entry.data.get(CONF_SERVER_KEY)
|
|
cloud_name = CONF_SERVERS.get(server)
|
|
cloud = get_midea_cloud(
|
|
cloud_name=cloud_name,
|
|
session=async_get_clientsession(hass),
|
|
account=account,
|
|
password=password,
|
|
)
|
|
if not cloud or not await cloud.login():
|
|
MideaLogger.error("Midea cloud login failed")
|
|
return False
|
|
|
|
# 拉取家庭与设备列表
|
|
appliances = None
|
|
try:
|
|
homes = await cloud.list_home()
|
|
if homes and len(homes) > 0:
|
|
first_home_id = list(homes.keys())[0]
|
|
appliances = await cloud.list_appliances(first_home_id)
|
|
else:
|
|
appliances = await cloud.list_appliances(None)
|
|
except Exception as e:
|
|
MideaLogger.error(f"Fetch appliances failed: {e}")
|
|
appliances = None
|
|
|
|
if appliances is None:
|
|
appliances = {}
|
|
|
|
hass.data.setdefault(DOMAIN, {})
|
|
hass.data[DOMAIN].setdefault("accounts", {})
|
|
bucket = {"device_list": {}, "coordinator_map": {}}
|
|
|
|
# 为每台设备构建占位设备与协调器（不连接本地）
|
|
for appliance_code, info in appliances.items():
|
|
MideaLogger.debug(f"info={info} ")
|
|
try:
|
|
device = MiedaDevice(
|
|
name=info.get(CONF_NAME) or info.get("name"),
|
|
device_id=appliance_code,
|
|
device_type=info.get(CONF_TYPE) or info.get("type"),
|
|
ip_address=None,
|
|
port=None,
|
|
token=None,
|
|
key=None,
|
|
connected=info.get("online"),
|
|
protocol=info.get(CONF_PROTOCOL) or 2,
|
|
model=info.get(CONF_MODEL),
|
|
subtype=info.get(CONF_MODEL_NUMBER),
|
|
sn=info.get(CONF_SN) or info.get("sn"),
|
|
sn8=info.get(CONF_SN8) or info.get("sn8"),
|
|
lua_file=None,
|
|
)
|
|
coordinator = MideaDataUpdateCoordinator(hass, config_entry, device)
|
|
await coordinator.async_config_entry_first_refresh()
|
|
bucket["device_list"][appliance_code] = info
|
|
bucket["coordinator_map"][appliance_code] = coordinator
|
|
except Exception as e:
|
|
MideaLogger.error(f"Init device failed: {appliance_code}, error: {e}")
|
|
|
|
hass.data[DOMAIN]["accounts"][config_entry.entry_id] = bucket
|
|
|
|
hass.async_create_task(hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS))
|
|
return True
|
|
|
|
|
|
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry):
|
|
device_id = config_entry.data.get(CONF_DEVICE_ID)
|
|
device_type = config_entry.data.get(CONF_TYPE)
|
|
if device_type == CONF_ACCOUNT:
|
|
# 卸载平台并清理账号桶
|
|
unload_ok = await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS)
|
|
if unload_ok:
|
|
try:
|
|
hass.data.get(DOMAIN, {}).get("accounts", {}).pop(config_entry.entry_id, None)
|
|
except Exception:
|
|
pass
|
|
return unload_ok
|
|
if device_id is not None:
|
|
device: MiedaDevice = hass.data[DOMAIN][DEVICES][device_id][CONF_DEVICE]
|
|
if device is not None:
|
|
if get_sn8_used(hass, device.sn8) == 1:
|
|
lua_file = config_entry.data.get("lua_file")
|
|
os.remove(lua_file)
|
|
remove_device_config(hass, device.sn8)
|
|
# device.close()
|
|
hass.data[DOMAIN][DEVICES].pop(device_id)
|
|
for platform in ALL_PLATFORM:
|
|
await hass.config_entries.async_forward_entry_unload(config_entry, platform)
|
|
return True
|