HomeAssistant/midea-meiju-codec
1
0
Fork 1
mirror of https://github.com/sususweet/midea-meiju-codec.git synced 2025-11-12 07:51:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add device support for T0x21 switch.

Browse Source
This commit is contained in:
sususweet
2025-10-27 22:38:12 +08:00
parent 087c5db497
commit 171d76ee3e
4 changed files with 196 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
custom_components/midea_auto_cloud/core/cloud.py
View File

@@ -63,7 +63,7 @@ class MideaCloud:
def _make_general_data(self):
return {}
async def _api_request(self, endpoint: str, data: dict, header=None) -> dict | None:
async def _api_request(self, endpoint: str, data: dict, header=None, method="POST") -> dict | None:
header = header or {}
if not data.get("reqId"):
data.update({
@@ -91,15 +91,18 @@ class MideaCloud:
_LOGGER.debug(f"Midea cloud API header: {header}")
_LOGGER.debug(f"Midea cloud API dump_data: {dump_data}")
try:
r = await self._session.request("POST", url, headers=header, data=dump_data, timeout=5)
r = await self._session.request(method, url, headers=header, data=dump_data, timeout=5)
raw = await r.read()
_LOGGER.debug(f"Midea cloud API url: {url}, data: {data}, response: {raw}")
response = json.loads(raw)
except Exception as e:
_LOGGER.debug(f"API request attempt failed: {e}")
if int(response["code"]) == 0 and "data" in response:
return response["data"]
if int(response["code"]) == 0:
if "data" in response:
return response["data"]
else:
return {"message": "ok"}
return None
@@ -207,6 +210,10 @@ class MideaCloud:
"""Get status of central AC devices. Subclasses should implement if supported."""
raise NotImplementedError()
async def send_switch_control(self, device_id: str, nodeid: str, switch_control: dict) -> bool:
"""Send control to switch device. Subclasses should implement if supported."""
raise NotImplementedError()
class MeijuCloud(MideaCloud):
APP_ID = "900"
@@ -405,6 +412,39 @@ class MeijuCloud(MideaCloud):
)
return response
async def send_switch_control(self, device_id: str, nodeid: str, switch_control: dict) -> bool:
"""Send control to switch device using the controlPanelFour API with PUT method."""
import uuid
# switch_control 格式: {"endPoint": 1, "attribute": 0}
end_point = switch_control.get("endPoint", 1)
attribute = switch_control.get("attribute", 0)
# 构建请求数据
request_data = {
"msgId": str(uuid.uuid4()).replace("-", ""),
"deviceControlList": [{
"endPoint": end_point,
"attribute": attribute
}],
"deviceId": device_id,
"nodeId": nodeid
}
MideaLogger.debug(f"Sending switch control to device {device_id}: {request_data}")
# 使用PUT方法发送到开关控制API
if response := await self._api_request(
endpoint="/v1/appliance/operation/controlPanelFour/" + device_id,
data=request_data,
method="PUT"
):
MideaLogger.debug(f"[{device_id}] Switch control response: {response}")
return True
else:
MideaLogger.warning(f"[{device_id}] Switch control failed: {response}")
return False
async def download_lua(
self, path: str,
device_type: int,

105
custom_components/midea_auto_cloud/data_coordinator.py
View File

@@ -123,9 +123,7 @@ class MideaDataUpdateCoordinator(DataUpdateCoordinator[MideaDeviceData]):
for appliance in status_data["appliances"]:
if appliance.get("type") == "0x21" and "extraData" in appliance:
extra_data = appliance["extraData"]
if "attr" in extra_data and "state" in extra_data["attr"]:
state = extra_data["attr"]["state"]
if "attr" in extra_data:
if "nodeid" in extra_data["attr"]:
self.device._attributes["nodeid"] = extra_data["attr"]["nodeid"]
if "masterId" in extra_data["attr"]:
@@ -135,7 +133,8 @@ class MideaDataUpdateCoordinator(DataUpdateCoordinator[MideaDeviceData]):
if "idType" in extra_data["attr"]:
self.device._attributes["idType"] = extra_data["attr"]["idType"]
if "condition_attribute" in state:
if "state" in extra_data["attr"] and "condition_attribute" in extra_data["attr"]["state"]:
state = extra_data["attr"]["state"]
condition = state["condition_attribute"]
# 将状态数据更新到设备属性中
for key, value in condition.items():
@@ -153,6 +152,32 @@ class MideaDataUpdateCoordinator(DataUpdateCoordinator[MideaDeviceData]):
self.device._attributes[key] = value
else:
self.device._attributes[key] = value
if "endlist" in extra_data["attr"]:
endlist = extra_data["attr"]["endlist"]
# endlist是一个数组，包含多个endpoint对象
if isinstance(endlist, list):
for endpoint in endlist:
if "event" in endpoint:
event = endpoint["event"]
endpoint_id = endpoint.get("endpoint", 1)
endpoint_name = endpoint.get("name", f"按键{endpoint_id}")
# 为每个endpoint创建独立的状态属性
for key, value in event.items():
# 创建带endpoint标识的属性名
attr_key = f"endpoint_{endpoint_id}_{key}"
attr_name_key = f"endpoint_{endpoint_id}_name"
# 保存endpoint名称
self.device._attributes[attr_name_key] = endpoint_name
self.device._attributes[attr_key] = value
# 同时保持原有的属性名（用于兼容性）
for key, value in event.items():
# 尝试将数字字符串转换为数字
self.device._attributes[key] = value
break
except Exception as e:
MideaLogger.debug(f"Error polling central AC state: {e}")
@@ -223,6 +248,78 @@ class MideaDataUpdateCoordinator(DataUpdateCoordinator[MideaDeviceData]):
MideaLogger.debug(f"Error sending control to {self.device.device_name}: {e}")
return False
async def async_send_switch_control(self, control: dict) -> bool:
"""发送开关控制命令（subtype为00000000的设备）"""
try:
cloud = self._cloud
if cloud and hasattr(cloud, "send_switch_control"):
# 获取设备ID和nodeId
masterid = str(self.device.attributes.get("masterId"))
nodeid = str(self.device.attributes.get("nodeid"))
if not nodeid:
MideaLogger.warning(f"No nodeid found for switch device {self._device_id}")
return False
# 根据控制命令确定endPoint和attribute值
end_point = control.get("endpoint", 1) # 从control中获取endpoint，默认1
attribute = 0 # 默认attribute
# 根据control内容设置attribute值
if "run_mode" in control:
if control["run_mode"] == "1":
attribute = 1 # 开启
else:
attribute = 0 # 关闭
# 构建控制数据
switch_control = {
"endPoint": end_point,
"attribute": attribute
}
MideaLogger.debug(f"Sending switch control to {self.device.device_name}: {switch_control}")
success = await cloud.send_switch_control(masterid, nodeid, switch_control)
if success:
# 更新本地状态 - 使用类似poll_central的解析方法
await self._update_switch_status_from_control(control)
self.mute_state_update_for_a_while()
self.async_update_listeners()
return True
else:
MideaLogger.debug(f"Failed to send switch control to {self.device.device_name}")
return False
else:
MideaLogger.debug("Cloud service not available for switch control")
return False
except Exception as e:
MideaLogger.debug(f"Error sending switch control to {self.device.device_name}: {e}")
return False
async def _update_switch_status_from_control(self, control: dict) -> None:
"""根据控制命令更新开关状态，参照poll_central的解析方法"""
try:
# 获取endpoint ID
endpoint_id = control.get("endpoint", 1)
run_mode = control.get("run_mode", "0")
# 模拟endlist数据结构来更新状态
# 根据run_mode设置OnOff状态
onoff_value = "1" if run_mode == "1" else "0"
# 更新endpoint特定的状态属性
attr_key = f"endpoint_{endpoint_id}_OnOff"
self.device._attributes[attr_key] = onoff_value
# 同时更新兼容性属性
self.device._attributes["OnOff"] = onoff_value
MideaLogger.debug(f"Updated switch status for endpoint {endpoint_id}: OnOff={onoff_value}")
except Exception as e:
MideaLogger.debug(f"Error updating switch status from control: {e}")
def _build_full_central_ac_control(self, new_control: dict) -> dict:
"""构建完整控制命令"""
full_control = {}

19
custom_components/midea_auto_cloud/device_mapping/T0x21.py
View File

@@ -3,7 +3,24 @@ from homeassistant.components.sensor import SensorStateClass, SensorDeviceClass
from homeassistant.components.switch import SwitchDeviceClass
DEVICE_MAPPING = {
"default": {
"00000000": {
"rationale": ["0", "1"],
"queries": [{}],
"centralized": [],
"entities": {
Platform.SWITCH: {
"endpoint_1_OnOff": {
"device_class": SwitchDeviceClass.SWITCH,
"rationale": ['0', '1']
},
"endpoint_2_OnOff": {
"device_class": SwitchDeviceClass.SWITCH,
"rationale": ['0', '1']
}
},
}
},
"b17-0000": {
"rationale": ["off", "on"],
"queries": [{}],
"centralized": ["run_mode", "fan_speed", "cooling_temp", "heating_temp", "extflag"],

37
custom_components/midea_auto_cloud/switch.py
View File

@@ -5,6 +5,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .core.logger import MideaLogger
from .midea_entity import MideaEntity
from . import load_device_config
@@ -43,6 +44,9 @@ class MideaSwitchEntity(MideaEntity, SwitchEntity):
"""Midea switch entity."""
def __init__(self, coordinator, device, manufacturer, rationale, entity_key, config):
# 自动判断是否为中央空调设备（T0x21）
self._is_central_ac = device.device_type == 0x21
super().__init__(
coordinator,
device.device_id,
@@ -67,12 +71,37 @@ class MideaSwitchEntity(MideaEntity, SwitchEntity):
async def async_turn_on(self):
"""Turn the switch on."""
# Use attribute from config if available, otherwise fall back to entity_key
attribute = self._config.get("attribute", self._entity_key)
await self._async_set_status_on_off(attribute, True)
if self._is_central_ac:
await self._async_set_central_ac_switch_status(True)
else:
await self._async_set_status_on_off(attribute, True)
async def async_turn_off(self):
"""Turn the switch off."""
# Use attribute from config if available, otherwise fall back to entity_key
attribute = self._config.get("attribute", self._entity_key)
await self._async_set_status_on_off(attribute, False)
if self._is_central_ac:
await self._async_set_central_ac_switch_status(False)
else:
await self._async_set_status_on_off(attribute, False)
async def _async_set_central_ac_switch_status(self, is_on: bool):
"""设置中央空调开关设备的状态"""
# 从entity_key中提取endpoint ID
# entity_key格式: endpoint_1_OnOff -> 提取出 1
endpoint_id = 1 # 默认值
if self._entity_key.startswith("endpoint_"):
try:
# 提取endpoint_后面的数字
parts = self._entity_key.split("_")
if len(parts) >= 2:
endpoint_id = int(parts[1])
except (ValueError, IndexError):
MideaLogger.warning(f"Failed to extract endpoint ID from {self._entity_key}, using default 1")
# 构建控制命令
control = {
"run_mode": "1" if is_on else "0",
"endpoint": endpoint_id
}
await self.coordinator.async_send_switch_control(control)