mirror of
https://github.com/sususweet/midea-meiju-codec.git
synced 2025-10-15 18:58:29 +00:00
226 lines
8.4 KiB
Python
226 lines
8.4 KiB
Python
"""Base entity class for Midea Auto Cloud integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from enum import IntEnum
|
|
from typing import Any
|
|
|
|
from homeassistant.helpers.debounce import Debouncer
|
|
from homeassistant.helpers.device_registry import DeviceInfo
|
|
from homeassistant.helpers.entity import Entity
|
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
|
|
from .const import DOMAIN
|
|
from .core.logger import MideaLogger
|
|
from .data_coordinator import MideaDataUpdateCoordinator
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
class Rationale(IntEnum):
|
|
EQUALLY = 0
|
|
GREATER = 1
|
|
LESS = 2
|
|
|
|
class MideaEntity(CoordinatorEntity[MideaDataUpdateCoordinator], Entity):
|
|
"""Base class for Midea entities."""
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: MideaDataUpdateCoordinator,
|
|
device_id: int,
|
|
device_name: str,
|
|
device_type: str,
|
|
sn: str,
|
|
sn8: str,
|
|
model: str,
|
|
entity_key: str,
|
|
*,
|
|
device: Any | None = None,
|
|
manufacturer: str | None = None,
|
|
rationale: list | None = None,
|
|
config: dict | None = None,
|
|
) -> None:
|
|
"""Initialize the entity."""
|
|
super().__init__(coordinator)
|
|
self._device_id = device_id
|
|
self._device_name = device_name
|
|
self._device_type = device_type
|
|
self._entity_key = entity_key
|
|
self._sn = sn
|
|
self._sn8 = sn8
|
|
self._model = model
|
|
# Legacy/extended fields
|
|
self._device = device
|
|
self._config = config or {}
|
|
self._rationale = rationale
|
|
if (self._config.get("rationale")) is not None:
|
|
self._rationale = self._config.get("rationale")
|
|
if self._rationale is None:
|
|
self._rationale = ["off", "on"]
|
|
|
|
# Display and identification
|
|
self._attr_has_entity_name = True
|
|
# Prefer legacy unique_id scheme if device object is available (device_id based)
|
|
if self._device is not None:
|
|
self._attr_unique_id = f"{DOMAIN}.{self._device_id}_{self._entity_key}"
|
|
self.entity_id_base = f"midea_{self._device_id}"
|
|
manu = "Midea" if manufacturer is None else manufacturer
|
|
self.manufacturer = manu
|
|
self._attr_device_info = DeviceInfo(
|
|
identifiers={(DOMAIN, str(self._device_id))},
|
|
model=self._model,
|
|
serial_number=sn,
|
|
manufacturer=manu,
|
|
name=device_name,
|
|
)
|
|
# Presentation attributes from config
|
|
self._attr_native_unit_of_measurement = self._config.get("unit_of_measurement")
|
|
self._attr_device_class = self._config.get("device_class")
|
|
self._attr_state_class = self._config.get("state_class")
|
|
self._attr_icon = self._config.get("icon")
|
|
# Prefer translated name; allow explicit override via config.name
|
|
self._attr_translation_key = self._config.get("translation_key") or self._entity_key
|
|
name_cfg = self._config.get("name")
|
|
if name_cfg is not None:
|
|
self._attr_name = f"{name_cfg}"
|
|
self.entity_id = self._attr_unique_id
|
|
# Register device updates for HA state refresh
|
|
try:
|
|
self._device.register_update(self.update_state) # type: ignore[attr-defined]
|
|
except Exception:
|
|
pass
|
|
else:
|
|
# Fallback to sn8-based unique id/device info
|
|
self._attr_unique_id = f"{sn8}_{self.entity_id_suffix}"
|
|
self.entity_id_base = f"midea_{sn8.lower()}"
|
|
self._attr_device_info = DeviceInfo(
|
|
identifiers={(DOMAIN, sn8)},
|
|
model=model,
|
|
serial_number=sn,
|
|
manufacturer="Midea",
|
|
name=device_name,
|
|
)
|
|
|
|
# Debounced command publishing
|
|
self._debounced_publish_command = Debouncer(
|
|
hass=self.coordinator.hass,
|
|
logger=_LOGGER,
|
|
cooldown=2,
|
|
immediate=True,
|
|
background=True,
|
|
function=self._publish_command,
|
|
)
|
|
|
|
if self.coordinator.config_entry:
|
|
self.coordinator.config_entry.async_on_unload(
|
|
self._debounced_publish_command.async_shutdown
|
|
)
|
|
|
|
@property
|
|
def entity_id_suffix(self) -> str:
|
|
"""Return the suffix for entity ID."""
|
|
return "base"
|
|
|
|
@property
|
|
def device_attributes(self) -> dict:
|
|
"""Return device attributes."""
|
|
return self.coordinator.data.attributes
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
"""Return if entity is available."""
|
|
if self.coordinator.data:
|
|
return self.coordinator.data.available
|
|
else:
|
|
return False
|
|
|
|
async def _publish_command(self) -> None:
|
|
"""Publish commands to the device."""
|
|
# This will be implemented by subclasses
|
|
pass
|
|
|
|
# ===== Unified helpers migrated from legacy entity base =====
|
|
def _get_status_on_off(self, attribute_key: str | None) -> bool:
|
|
"""Return boolean value from device attributes for given key.
|
|
|
|
Accepts common truthy representations: True/1/"on"/"true".
|
|
"""
|
|
result = False
|
|
if attribute_key is None:
|
|
return result
|
|
status = self.device_attributes.get(attribute_key)
|
|
if status is not None:
|
|
try:
|
|
result = bool(self._rationale.index(status))
|
|
except ValueError:
|
|
MideaLogger.error(f"The value of attribute {attribute_key} ('{status}') "
|
|
f"is not in rationale {self._rationale}")
|
|
return result
|
|
|
|
async def _async_set_status_on_off(self, attribute_key: str | None, turn_on: bool) -> None:
|
|
"""Set boolean attribute via coordinator, no-op if key is None."""
|
|
if attribute_key is None:
|
|
return
|
|
await self.async_set_attribute(attribute_key, self._rationale[int(turn_on)])
|
|
|
|
def _list_get_selected(self, key_of_list: list, rationale: Rationale = Rationale.EQUALLY):
|
|
for index in range(0, len(key_of_list)):
|
|
match = True
|
|
for attr, value in key_of_list[index].items():
|
|
state_value = self.device_attributes.get(attr)
|
|
if state_value is None:
|
|
match = False
|
|
break
|
|
if rationale is Rationale.EQUALLY and state_value != value:
|
|
match = False
|
|
break
|
|
if rationale is Rationale.GREATER and state_value < value:
|
|
match = False
|
|
break
|
|
if rationale is Rationale.LESS and state_value > value:
|
|
match = False
|
|
break
|
|
if match:
|
|
return index
|
|
return None
|
|
|
|
def _dict_get_selected(self, key_of_dict: dict, rationale: Rationale = Rationale.EQUALLY):
|
|
for mode, status in key_of_dict.items():
|
|
match = True
|
|
for attr, value in status.items():
|
|
state_value = self.device_attributes.get(attr)
|
|
if state_value is None:
|
|
match = False
|
|
break
|
|
if rationale is Rationale.EQUALLY and state_value != value:
|
|
match = False
|
|
break
|
|
if rationale is Rationale.GREATER and state_value < value:
|
|
match = False
|
|
break
|
|
if rationale is Rationale.LESS and state_value > value:
|
|
match = False
|
|
break
|
|
if match:
|
|
return mode
|
|
return None
|
|
|
|
async def publish_command_from_current_state(self) -> None:
|
|
"""Publish commands to the device from current state."""
|
|
self.coordinator.mute_state_update_for_a_while()
|
|
self.coordinator.async_update_listeners()
|
|
await self._debounced_publish_command.async_call()
|
|
|
|
async def async_set_attribute(self, attribute: str, value: Any) -> None:
|
|
"""Set a device attribute."""
|
|
await self.coordinator.async_set_attribute(attribute, value)
|
|
|
|
async def async_set_attributes(self, attributes: dict) -> None:
|
|
"""Set multiple device attributes."""
|
|
await self.coordinator.async_set_attributes(attributes)
|
|
|
|
async def async_send_command(self, cmd_type: int, cmd_body: str) -> None:
|
|
"""Send a command to the device."""
|
|
await self.coordinator.async_send_command(cmd_type, cmd_body)
|