Move xiaomi_miio base entity to separate module (#126198)
epenet
GitHub
parent
e9ac6b7482
commit
1893545705
|
|
@ -18,7 +18,7 @@ from .const import (
|
|||
MODEL_AIRQUALITYMONITOR_S1,
|
||||
MODEL_AIRQUALITYMONITOR_V1,
|
||||
)
|
||||
from .device import XiaomiMiioEntity
|
||||
from .entity import XiaomiMiioEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ from .const import (
|
|||
MODELS_VACUUM_WITH_MOP,
|
||||
MODELS_VACUUM_WITH_SEPARATE_MOP,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ from .const import (
|
|||
MODEL_AIRFRESH_T2017,
|
||||
MODELS_VACUUM,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
|
||||
# Fans
|
||||
ATTR_RESET_DUST_FILTER = "reset_dust_filter"
|
||||
|
|
|
|||
|
|
@ -1,24 +1,11 @@
|
|||
"""Code to handle a Xiaomi Device."""
|
||||
|
||||
import datetime
|
||||
from enum import Enum
|
||||
from functools import partial
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from construct.core import ChecksumError
|
||||
from miio import Device, DeviceException
|
||||
|
||||
from homeassistant.const import ATTR_CONNECTIONS, CONF_MAC, CONF_MODEL
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
)
|
||||
|
||||
from .const import DOMAIN, AuthException, SetupException
|
||||
from .const import AuthException, SetupException
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -66,131 +53,3 @@ class ConnectXiaomiDevice:
|
|||
self._device_info.firmware_version,
|
||||
self._device_info.hardware_version,
|
||||
)
|
||||
|
||||
|
||||
class XiaomiMiioEntity(Entity):
|
||||
"""Representation of a base Xiaomi Miio Entity."""
|
||||
|
||||
def __init__(self, name, device, entry, unique_id):
|
||||
"""Initialize the Xiaomi Miio Device."""
|
||||
self._device = device
|
||||
self._model = entry.data[CONF_MODEL]
|
||||
self._mac = entry.data[CONF_MAC]
|
||||
self._device_id = entry.unique_id
|
||||
self._unique_id = unique_id
|
||||
self._name = name
|
||||
self._available = None
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return an unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of this entity, if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device info."""
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self._device_id)},
|
||||
manufacturer="Xiaomi",
|
||||
model=self._model,
|
||||
name=self._name,
|
||||
)
|
||||
|
||||
if self._mac is not None:
|
||||
device_info[ATTR_CONNECTIONS] = {(dr.CONNECTION_NETWORK_MAC, self._mac)}
|
||||
|
||||
return device_info
|
||||
|
||||
|
||||
class XiaomiCoordinatedMiioEntity[_T: DataUpdateCoordinator[Any]](
|
||||
CoordinatorEntity[_T]
|
||||
):
|
||||
"""Representation of a base a coordinated Xiaomi Miio Entity."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(self, device, entry, unique_id, coordinator):
|
||||
"""Initialize the coordinated Xiaomi Miio Device."""
|
||||
super().__init__(coordinator)
|
||||
self._device = device
|
||||
self._model = entry.data[CONF_MODEL]
|
||||
self._mac = entry.data[CONF_MAC]
|
||||
self._device_id = entry.unique_id
|
||||
self._device_name = entry.title
|
||||
self._unique_id = unique_id
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return an unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device info."""
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self._device_id)},
|
||||
manufacturer="Xiaomi",
|
||||
model=self._model,
|
||||
name=self._device_name,
|
||||
)
|
||||
|
||||
if self._mac is not None:
|
||||
device_info[ATTR_CONNECTIONS] = {(dr.CONNECTION_NETWORK_MAC, self._mac)}
|
||||
|
||||
return device_info
|
||||
|
||||
async def _try_command(self, mask_error, func, *args, **kwargs):
|
||||
"""Call a miio device command handling error messages."""
|
||||
try:
|
||||
result = await self.hass.async_add_executor_job(
|
||||
partial(func, *args, **kwargs)
|
||||
)
|
||||
except DeviceException as exc:
|
||||
if self.available:
|
||||
_LOGGER.error(mask_error, exc)
|
||||
|
||||
return False
|
||||
|
||||
_LOGGER.debug("Response received from miio device: %s", result)
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def _extract_value_from_attribute(cls, state, attribute):
|
||||
value = getattr(state, attribute)
|
||||
if isinstance(value, Enum):
|
||||
return value.value
|
||||
if isinstance(value, datetime.timedelta):
|
||||
return cls._parse_time_delta(value)
|
||||
if isinstance(value, datetime.time):
|
||||
return cls._parse_datetime_time(value)
|
||||
if isinstance(value, datetime.datetime):
|
||||
return cls._parse_datetime_datetime(value)
|
||||
|
||||
if value is None:
|
||||
_LOGGER.debug("Attribute %s is None, this is unexpected", attribute)
|
||||
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _parse_time_delta(timedelta: datetime.timedelta) -> int:
|
||||
return int(timedelta.total_seconds())
|
||||
|
||||
@staticmethod
|
||||
def _parse_datetime_time(initial_time: datetime.time) -> str:
|
||||
time = datetime.datetime.now().replace(
|
||||
hour=initial_time.hour, minute=initial_time.minute, second=0, microsecond=0
|
||||
)
|
||||
|
||||
if time < datetime.datetime.now():
|
||||
time += datetime.timedelta(days=1)
|
||||
|
||||
return time.isoformat()
|
||||
|
||||
@staticmethod
|
||||
def _parse_datetime_datetime(time: datetime.datetime) -> str:
|
||||
return time.isoformat()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,193 @@
|
|||
"""Code to handle a Xiaomi Device."""
|
||||
|
||||
import datetime
|
||||
from enum import Enum
|
||||
from functools import partial
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from miio import DeviceException
|
||||
|
||||
from homeassistant.const import ATTR_CONNECTIONS, CONF_MAC, CONF_MODEL
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.update_coordinator import (
|
||||
CoordinatorEntity,
|
||||
DataUpdateCoordinator,
|
||||
)
|
||||
|
||||
from .const import ATTR_AVAILABLE, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class XiaomiMiioEntity(Entity):
|
||||
"""Representation of a base Xiaomi Miio Entity."""
|
||||
|
||||
def __init__(self, name, device, entry, unique_id):
|
||||
"""Initialize the Xiaomi Miio Device."""
|
||||
self._device = device
|
||||
self._model = entry.data[CONF_MODEL]
|
||||
self._mac = entry.data[CONF_MAC]
|
||||
self._device_id = entry.unique_id
|
||||
self._unique_id = unique_id
|
||||
self._name = name
|
||||
self._available = None
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return an unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of this entity, if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device info."""
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self._device_id)},
|
||||
manufacturer="Xiaomi",
|
||||
model=self._model,
|
||||
name=self._name,
|
||||
)
|
||||
|
||||
if self._mac is not None:
|
||||
device_info[ATTR_CONNECTIONS] = {(dr.CONNECTION_NETWORK_MAC, self._mac)}
|
||||
|
||||
return device_info
|
||||
|
||||
|
||||
class XiaomiCoordinatedMiioEntity[_T: DataUpdateCoordinator[Any]](
|
||||
CoordinatorEntity[_T]
|
||||
):
|
||||
"""Representation of a base a coordinated Xiaomi Miio Entity."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(self, device, entry, unique_id, coordinator):
|
||||
"""Initialize the coordinated Xiaomi Miio Device."""
|
||||
super().__init__(coordinator)
|
||||
self._device = device
|
||||
self._model = entry.data[CONF_MODEL]
|
||||
self._mac = entry.data[CONF_MAC]
|
||||
self._device_id = entry.unique_id
|
||||
self._device_name = entry.title
|
||||
self._unique_id = unique_id
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return an unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device info."""
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, self._device_id)},
|
||||
manufacturer="Xiaomi",
|
||||
model=self._model,
|
||||
name=self._device_name,
|
||||
)
|
||||
|
||||
if self._mac is not None:
|
||||
device_info[ATTR_CONNECTIONS] = {(dr.CONNECTION_NETWORK_MAC, self._mac)}
|
||||
|
||||
return device_info
|
||||
|
||||
async def _try_command(self, mask_error, func, *args, **kwargs):
|
||||
"""Call a miio device command handling error messages."""
|
||||
try:
|
||||
result = await self.hass.async_add_executor_job(
|
||||
partial(func, *args, **kwargs)
|
||||
)
|
||||
except DeviceException as exc:
|
||||
if self.available:
|
||||
_LOGGER.error(mask_error, exc)
|
||||
|
||||
return False
|
||||
|
||||
_LOGGER.debug("Response received from miio device: %s", result)
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def _extract_value_from_attribute(cls, state, attribute):
|
||||
value = getattr(state, attribute)
|
||||
if isinstance(value, Enum):
|
||||
return value.value
|
||||
if isinstance(value, datetime.timedelta):
|
||||
return cls._parse_time_delta(value)
|
||||
if isinstance(value, datetime.time):
|
||||
return cls._parse_datetime_time(value)
|
||||
if isinstance(value, datetime.datetime):
|
||||
return cls._parse_datetime_datetime(value)
|
||||
|
||||
if value is None:
|
||||
_LOGGER.debug("Attribute %s is None, this is unexpected", attribute)
|
||||
|
||||
return value
|
||||
|
||||
@staticmethod
|
||||
def _parse_time_delta(timedelta: datetime.timedelta) -> int:
|
||||
return int(timedelta.total_seconds())
|
||||
|
||||
@staticmethod
|
||||
def _parse_datetime_time(initial_time: datetime.time) -> str:
|
||||
time = datetime.datetime.now().replace(
|
||||
hour=initial_time.hour, minute=initial_time.minute, second=0, microsecond=0
|
||||
)
|
||||
|
||||
if time < datetime.datetime.now():
|
||||
time += datetime.timedelta(days=1)
|
||||
|
||||
return time.isoformat()
|
||||
|
||||
@staticmethod
|
||||
def _parse_datetime_datetime(time: datetime.datetime) -> str:
|
||||
return time.isoformat()
|
||||
|
||||
|
||||
class XiaomiGatewayDevice(CoordinatorEntity, Entity):
|
||||
"""Representation of a base Xiaomi Gateway Device."""
|
||||
|
||||
def __init__(self, coordinator, sub_device, entry):
|
||||
"""Initialize the Xiaomi Gateway Device."""
|
||||
super().__init__(coordinator)
|
||||
self._sub_device = sub_device
|
||||
self._entry = entry
|
||||
self._unique_id = sub_device.sid
|
||||
self._name = f"{sub_device.name} ({sub_device.sid})"
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return an unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of this entity, if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device info of the gateway."""
|
||||
return DeviceInfo(
|
||||
identifiers={(DOMAIN, self._sub_device.sid)},
|
||||
via_device=(DOMAIN, self._entry.unique_id),
|
||||
manufacturer="Xiaomi",
|
||||
name=self._sub_device.name,
|
||||
model=self._sub_device.model,
|
||||
sw_version=self._sub_device.firmware_version,
|
||||
hw_version=self._sub_device.zigbee_model,
|
||||
)
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if entity is available."""
|
||||
if self.coordinator.data is None:
|
||||
return False
|
||||
|
||||
return self.coordinator.data[ATTR_AVAILABLE]
|
||||
|
|
@ -91,7 +91,7 @@ from .const import (
|
|||
SERVICE_RESET_FILTER,
|
||||
SERVICE_SET_EXTRA_FEATURES,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
from .typing import ServiceMethodDetails
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
|
|
|||
|
|
@ -8,17 +8,11 @@ from micloud.micloudexception import MiCloudAccessDenied
|
|||
from miio import DeviceException, gateway
|
||||
from miio.gateway.gateway import GATEWAY_MODEL_EU
|
||||
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import (
|
||||
ATTR_AVAILABLE,
|
||||
CONF_CLOUD_COUNTRY,
|
||||
CONF_CLOUD_PASSWORD,
|
||||
CONF_CLOUD_SUBDEVICES,
|
||||
CONF_CLOUD_USERNAME,
|
||||
DOMAIN,
|
||||
AuthException,
|
||||
SetupException,
|
||||
)
|
||||
|
|
@ -134,46 +128,3 @@ class ConnectXiaomiGateway:
|
|||
"DeviceException during setup of xiaomi gateway with host"
|
||||
f" {self._host}"
|
||||
) from error
|
||||
|
||||
|
||||
class XiaomiGatewayDevice(CoordinatorEntity, Entity):
|
||||
"""Representation of a base Xiaomi Gateway Device."""
|
||||
|
||||
def __init__(self, coordinator, sub_device, entry):
|
||||
"""Initialize the Xiaomi Gateway Device."""
|
||||
super().__init__(coordinator)
|
||||
self._sub_device = sub_device
|
||||
self._entry = entry
|
||||
self._unique_id = sub_device.sid
|
||||
self._name = f"{sub_device.name} ({sub_device.sid})"
|
||||
|
||||
@property
|
||||
def unique_id(self):
|
||||
"""Return an unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of this entity, if any."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return the device info of the gateway."""
|
||||
return DeviceInfo(
|
||||
identifiers={(DOMAIN, self._sub_device.sid)},
|
||||
via_device=(DOMAIN, self._entry.unique_id),
|
||||
manufacturer="Xiaomi",
|
||||
name=self._sub_device.name,
|
||||
model=self._sub_device.model,
|
||||
sw_version=self._sub_device.firmware_version,
|
||||
hw_version=self._sub_device.zigbee_model,
|
||||
)
|
||||
|
||||
@property
|
||||
def available(self):
|
||||
"""Return if entity is available."""
|
||||
if self.coordinator.data is None:
|
||||
return False
|
||||
|
||||
return self.coordinator.data[ATTR_AVAILABLE]
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ from .const import (
|
|||
MODELS_HUMIDIFIER_MIOT,
|
||||
MODELS_HUMIDIFIER_MJJSQ,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -66,8 +66,7 @@ from .const import (
|
|||
SERVICE_SET_DELAYED_TURN_OFF,
|
||||
SERVICE_SET_SCENE,
|
||||
)
|
||||
from .device import XiaomiMiioEntity
|
||||
from .gateway import XiaomiGatewayDevice
|
||||
from .entity import XiaomiGatewayDevice, XiaomiMiioEntity
|
||||
from .typing import ServiceMethodDetails
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
|
|
|||
|
|
@ -96,7 +96,7 @@ from .const import (
|
|||
MODELS_PURIFIER_MIIO,
|
||||
MODELS_PURIFIER_MIOT,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
|
||||
ATTR_DELAY_OFF_COUNTDOWN = "delay_off_countdown"
|
||||
ATTR_FAN_LEVEL = "fan_level"
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ from .const import (
|
|||
MODEL_FAN_ZA3,
|
||||
MODEL_FAN_ZA4,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
|
||||
ATTR_DISPLAY_ORIENTATION = "display_orientation"
|
||||
ATTR_LED_BRIGHTNESS = "led_brightness"
|
||||
|
|
|
|||
|
|
@ -89,8 +89,7 @@ from .const import (
|
|||
ROBOROCK_GENERIC,
|
||||
ROCKROBO_GENERIC,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity, XiaomiMiioEntity
|
||||
from .gateway import XiaomiGatewayDevice
|
||||
from .entity import XiaomiCoordinatedMiioEntity, XiaomiGatewayDevice, XiaomiMiioEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
|
|
@ -113,8 +113,7 @@ from .const import (
|
|||
SERVICE_SET_WIFI_LED_ON,
|
||||
SUCCESS,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity, XiaomiMiioEntity
|
||||
from .gateway import XiaomiGatewayDevice
|
||||
from .entity import XiaomiCoordinatedMiioEntity, XiaomiGatewayDevice, XiaomiMiioEntity
|
||||
from .typing import ServiceMethodDetails
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
|
|
|||
|
|
@ -41,7 +41,7 @@ from .const import (
|
|||
SERVICE_START_REMOTE_CONTROL,
|
||||
SERVICE_STOP_REMOTE_CONTROL,
|
||||
)
|
||||
from .device import XiaomiCoordinatedMiioEntity
|
||||
from .entity import XiaomiCoordinatedMiioEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue