Airmx 秒新加湿器接入HA - 『瀚思彼岸』» 智能家居技术论坛

@app.route("/aw")
def aw() -> dict[str, Any] | str:
if (path := request.args.get("path")) != "aw/GET/genId":
app.logger.error(f"Unsupported path {path}")
raise HTTPException.NotImplemented()
params = json.loads(request.args.get("params", ""))
wifi_mac_b = bytearray.fromhex(params["mac"].rjust(12, "0"))
wifi_mac = int.from_bytes(wifi_mac_b, byteorder="little")
ble_mac = wifi_mac + 2
device = Device(
id=int.from_bytes(wifi_mac_b[:2], byteorder="little"),
wifi_mac=wifi_mac.to_bytes(6, byteorder="big").hex(),
ble_mac=ble_mac.to_bytes(6, byteorder="big").hex(),
key=params["key"],
type=int(params["type"]),
ts=int(datetime.now().timestamp()),
)
data: dict[str, Any] | None = None
match device.type:
case 11: # A5
pass
case 20: # A3S
pass
case 21: # A3S_V2 / Iris
data = {
"awId": device.id,
"electrolysisLevel4OffTime": 1800,
"electrolysisLevel2OpenTime": 600,
"electrolysisLevel3OpenTime": 1200,
"electrolysisLevel4OpenTime": 1800,
"electrolysisLevel2OffTime": 600,
"electrolysisLevel1OpenTime": 60,
"electrolysisLevel3OffTime": 1200,
"electrolysisLevel1OffTime": 120,
}
case 30: # A2
pass
case _:
app.logger.error(f"Unsupported device: {device}")
raise HTTPException.NotImplemented()
app.logger.info(f"New device registered: {device}")
devices[device.id] = device
with open(DEVICE_STORE_PATH, "w") as f:
json.dump(devices, f, cls=EnhancedJSONEncoder)
if not data:
return '{"status":200,"data":{"awId":%d}}' % device.id
return {"status": 200, "data": data}
@app.route("/gettime")
def gettime() -> dict[str, int]:
return {"time": int(datetime.now().timestamp())}

复制代码

import json
from typing import Any
import logging
_LOGGER = logging.getLogger(__name__)
_LOGGER.setLevel(logging.DEBUG)
from homeassistant.components.humidifier import (
MODE_AUTO,
HumidifierDeviceClass,
HumidifierEntity,
HumidifierEntityFeature,
HumidifierAction, # ***改为显示加湿、空闲、关闭等运行状态
)
from homeassistant.components.humidifier.const import MODE_SLEEP
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, device_registry as dr, entity_platform
from homeassistant.helpers.entity_platform import AddEntitiesCallback
import voluptuous as vol
from .airwater.const import AirWaterCommand, AirWaterMode
from .airwater.device import AirWaterDevice
from .const import ATTR_COMMAND_DATA, ATTR_COMMAND_ID, DEVICES, DOMAIN, MODE_MANUAL, SERVICE_SEND_COMMAND
from .entity import AirWaterEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
device: AirWaterDevice = hass.data[DOMAIN][DEVICES][entry.entry_id]
async_add_entities([AirWaterHumidifier(device, entry)])
platform = entity_platform.async_get_current_platform()
platform.async_register_entity_service(
SERVICE_SEND_COMMAND,
{
vol.Required(ATTR_ENTITY_ID): cv.entity_id,
vol.Required(ATTR_COMMAND_ID): vol.Coerce(int),
vol.Required(ATTR_COMMAND_DATA): cv.string,
},
"async_send_command",
)
class AirWaterHumidifier(AirWaterEntity, HumidifierEntity):
_attr_name = None
_attr_device_class = HumidifierDeviceClass.HUMIDIFIER
_attr_supported_features = HumidifierEntityFeature.MODES
_attr_available_modes = [MODE_AUTO, MODE_MANUAL, MODE_SLEEP]
_attr_translation_key = "humidifier"
@property
def is_on(self) -> bool | None:
return self._device.status.power
@property
def current_humidity(self) -> int | None:
# ***获取 sensor.temperature_humidity_sensor_6510_humidity 的湿度值
value = self.hass.states.get("sensor.temperature_humidity_sensor_6510_humidity").state # 改为传感器实体ID
#_LOGGER.debug("AirMX Room Current humidity value: %s", value)
# 确保湿度值有效，且转换为整数
if value:
try:
#return int(float(value)) # 确保转换为整数
return round(float(value))
except ValueError:
return None
return None
@property
def target_humidity(self) -> int | None:
#if self.mode == MODE_MANUAL: # ***改为手动模式也显示仪表盘运作状态
# return None
return self._device.status.target_humidity
@property
def mode(self) -> str | None:
return {
AirWaterMode.SLEEP: MODE_SLEEP,
AirWaterMode.MANUAL: MODE_MANUAL,
}.get(self._device.status.mode, MODE_AUTO)
@property # ***改为显示加湿、空闲、关闭等运行状态
def action(self) -> HumidifierAction | None:
if not self.is_on:
return HumidifierAction.OFF
if self.current_humidity is not None and self.target_humidity is not None:
if self.current_humidity < self.target_humidity:
return HumidifierAction.HUMIDIFYING
else:
return HumidifierAction.IDLE
return None
async def async_turn_on(self, **kwargs: Any) -> None:
await self._device.async_turn_on()
async def async_turn_off(self, **kwargs: Any) -> None:
await self._device.async_turn_off()
async def async_set_humidity(self, humidity: int) -> None:
await self._device.async_set_target_humidity(humidity)
async def async_set_mode(self, mode: str) -> None:
await self._device.async_set_mode(
{
MODE_SLEEP: AirWaterMode.SLEEP,
MODE_MANUAL: AirWaterMode.MANUAL,
}.get(mode, AirWaterMode.AUTO)
)
async def async_send_command(self, command_id: int, command_data: str) -> None:
await self._device.async_send_command(AirWaterCommand(command_id), json.loads(command_data))
@callback
def async_write_ha_state(self) -> None:
super().async_write_ha_state()
if self.device_info and self.device_info.get("sw_version") != self._device.status.firmware_version:
self._update_device_info()
def _update_device_info(self) -> None:
device_registry = dr.async_get(self.hass)
device = device_registry.async_get_device({(DOMAIN, self.unique_id)})
if device:
device_registry.async_update_device(device.id, sw_version=self._device.status.firmware_version)

复制代码