Add sensors to Ecovacs (#108686)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
pull/108739/head
Robert Resch 2024-01-23 21:17:18 +01:00 committed by GitHub
parent bfd9bd3ff2
commit 37f5c75752
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1099 additions and 56 deletions

View File

@ -26,6 +26,7 @@ CONFIG_SCHEMA = vol.Schema(
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.SENSOR,
Platform.VACUUM,
]

View File

@ -17,13 +17,14 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .controller import EcovacsController
from .entity import EcovacsDescriptionEntity, EcovacsEntityDescription, EventT
from .entity import EcovacsCapabilityEntityDescription, EcovacsDescriptionEntity, EventT
from .util import get_supported_entitites
@dataclass(kw_only=True, frozen=True)
class EcovacsBinarySensorEntityDescription(
BinarySensorEntityDescription,
EcovacsEntityDescription,
EcovacsCapabilityEntityDescription,
Generic[EventT],
):
"""Class describing Deebot binary sensor entity."""
@ -49,15 +50,13 @@ async def async_setup_entry(
) -> None:
"""Add entities for passed config_entry in HA."""
controller: EcovacsController = hass.data[DOMAIN][config_entry.entry_id]
controller.register_platform_add_entities(
EcovacsBinarySensor, ENTITY_DESCRIPTIONS, async_add_entities
async_add_entities(
get_supported_entitites(controller, EcovacsBinarySensor, ENTITY_DESCRIPTIONS)
)
class EcovacsBinarySensor(
EcovacsDescriptionEntity[
CapabilityEvent[EventT], EcovacsBinarySensorEntityDescription
],
EcovacsDescriptionEntity[CapabilityEvent[EventT]],
BinarySensorEntity,
):
"""Ecovacs binary sensor."""

View File

@ -23,9 +23,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import EcovacsDescriptionEntity, EcovacsEntityDescription
from .util import get_client_device_id
_LOGGER = logging.getLogger(__name__)
@ -88,23 +86,6 @@ class EcovacsController:
_LOGGER.debug("Controller initialize complete")
def register_platform_add_entities(
self,
entity_class: type[EcovacsDescriptionEntity],
descriptions: tuple[EcovacsEntityDescription, ...],
async_add_entities: AddEntitiesCallback,
) -> None:
"""Create entities from descriptions and add them."""
new_entites: list[EcovacsDescriptionEntity] = []
for device in self.devices:
for description in descriptions:
if capability := description.capability_fn(device.capabilities):
new_entites.append(entity_class(device, capability, description))
if new_entites:
async_add_entities(new_entites)
async def teardown(self) -> None:
"""Disconnect controller."""
for device in self.devices:

View File

@ -16,26 +16,13 @@ from homeassistant.helpers.entity import Entity, EntityDescription
from .const import DOMAIN
_EntityDescriptionT = TypeVar("_EntityDescriptionT", bound=EntityDescription)
CapabilityT = TypeVar("CapabilityT")
EventT = TypeVar("EventT", bound=Event)
@dataclass(kw_only=True, frozen=True)
class EcovacsEntityDescription(
EntityDescription,
Generic[CapabilityT],
):
"""Ecovacs entity description."""
capability_fn: Callable[[Capabilities], CapabilityT | None]
class EcovacsEntity(Entity, Generic[CapabilityT, _EntityDescriptionT]):
class EcovacsEntity(Entity, Generic[CapabilityT]):
"""Ecovacs entity."""
entity_description: _EntityDescriptionT
_attr_should_poll = False
_attr_has_entity_name = True
_always_available: bool = False
@ -106,16 +93,26 @@ class EcovacsEntity(Entity, Generic[CapabilityT, _EntityDescriptionT]):
self._device.events.request_refresh(event_type)
class EcovacsDescriptionEntity(EcovacsEntity[CapabilityT, _EntityDescriptionT]):
class EcovacsDescriptionEntity(EcovacsEntity[CapabilityT]):
"""Ecovacs entity."""
def __init__(
self,
device: Device,
capability: CapabilityT,
entity_description: _EntityDescriptionT,
entity_description: EntityDescription,
**kwargs: Any,
) -> None:
"""Initialize entity."""
self.entity_description = entity_description
super().__init__(device, capability, **kwargs)
@dataclass(kw_only=True, frozen=True)
class EcovacsCapabilityEntityDescription(
EntityDescription,
Generic[CapabilityT],
):
"""Ecovacs entity description."""
capability_fn: Callable[[Capabilities], CapabilityT | None]

View File

@ -7,6 +7,44 @@
"on": "mdi:water"
}
}
},
"sensor": {
"error": {
"default": "mdi:alert-circle"
},
"lifespan_brush": {
"default": "mdi:broom"
},
"lifespan_filter": {
"default": "mdi:air-filter"
},
"lifespan_side_brush": {
"default": "mdi:broom"
},
"network_ip": {
"default": "mdi:ip-network-outline"
},
"network_rssi": {
"default": "mdi:signal-variant"
},
"network_ssid": {
"default": "mdi:wifi"
},
"stats_area": {
"default": "mdi:floor-plan"
},
"stats_time": {
"default": "mdi:timer-outline"
},
"total_stats_area": {
"default": "mdi:floor-plan"
},
"total_stats_time": {
"default": "mdi:timer-outline"
},
"total_stats_cleanings": {
"default": "mdi:counter"
}
}
}
}

View File

@ -0,0 +1,256 @@
"""Ecovacs sensor module."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Generic
from deebot_client.capabilities import CapabilityEvent, CapabilityLifeSpan
from deebot_client.events import (
BatteryEvent,
ErrorEvent,
Event,
LifeSpan,
LifeSpanEvent,
NetworkInfoEvent,
StatsEvent,
TotalStatsEvent,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
AREA_SQUARE_METERS,
ATTR_BATTERY_LEVEL,
CONF_DESCRIPTION,
PERCENTAGE,
EntityCategory,
UnitOfTime,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import DOMAIN
from .controller import EcovacsController
from .entity import (
EcovacsCapabilityEntityDescription,
EcovacsDescriptionEntity,
EcovacsEntity,
EventT,
)
from .util import get_supported_entitites
@dataclass(kw_only=True, frozen=True)
class EcovacsSensorEntityDescription(
EcovacsCapabilityEntityDescription,
SensorEntityDescription,
Generic[EventT],
):
"""Ecovacs sensor entity description."""
value_fn: Callable[[EventT], StateType]
ENTITY_DESCRIPTIONS: tuple[EcovacsSensorEntityDescription, ...] = (
# Stats
EcovacsSensorEntityDescription[StatsEvent](
key="stats_area",
capability_fn=lambda caps: caps.stats.clean,
value_fn=lambda e: e.area,
translation_key="stats_area",
native_unit_of_measurement=AREA_SQUARE_METERS,
),
EcovacsSensorEntityDescription[StatsEvent](
key="stats_time",
capability_fn=lambda caps: caps.stats.clean,
value_fn=lambda e: e.time,
translation_key="stats_time",
native_unit_of_measurement=UnitOfTime.SECONDS,
suggested_unit_of_measurement=UnitOfTime.MINUTES,
),
# TotalStats
EcovacsSensorEntityDescription[TotalStatsEvent](
capability_fn=lambda caps: caps.stats.total,
value_fn=lambda e: e.area,
key="total_stats_area",
translation_key="total_stats_area",
native_unit_of_measurement=AREA_SQUARE_METERS,
state_class=SensorStateClass.TOTAL_INCREASING,
),
EcovacsSensorEntityDescription[TotalStatsEvent](
capability_fn=lambda caps: caps.stats.total,
value_fn=lambda e: e.time,
key="total_stats_time",
translation_key="total_stats_time",
native_unit_of_measurement=UnitOfTime.MINUTES,
suggested_unit_of_measurement=UnitOfTime.HOURS,
state_class=SensorStateClass.TOTAL_INCREASING,
),
EcovacsSensorEntityDescription[TotalStatsEvent](
capability_fn=lambda caps: caps.stats.total,
value_fn=lambda e: e.cleanings,
key="total_stats_cleanings",
translation_key="total_stats_cleanings",
state_class=SensorStateClass.TOTAL_INCREASING,
),
EcovacsSensorEntityDescription[BatteryEvent](
capability_fn=lambda caps: caps.battery,
value_fn=lambda e: e.value,
key=ATTR_BATTERY_LEVEL,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
),
EcovacsSensorEntityDescription[NetworkInfoEvent](
capability_fn=lambda caps: caps.network,
value_fn=lambda e: e.ip,
key="network_ip",
translation_key="network_ip",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
),
EcovacsSensorEntityDescription[NetworkInfoEvent](
capability_fn=lambda caps: caps.network,
value_fn=lambda e: e.rssi,
key="network_rssi",
translation_key="network_rssi",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
),
EcovacsSensorEntityDescription[NetworkInfoEvent](
capability_fn=lambda caps: caps.network,
value_fn=lambda e: e.ssid,
key="network_ssid",
translation_key="network_ssid",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
),
)
@dataclass(kw_only=True, frozen=True)
class EcovacsLifespanSensorEntityDescription(SensorEntityDescription):
"""Ecovacs lifespan sensor entity description."""
component: LifeSpan
value_fn: Callable[[LifeSpanEvent], int | float]
LIFESPAN_ENTITY_DESCRIPTIONS = tuple(
EcovacsLifespanSensorEntityDescription(
component=component,
value_fn=lambda e: e.percent,
key=f"lifespan_{component.name.lower()}",
translation_key=f"lifespan_{component.name.lower()}",
native_unit_of_measurement=PERCENTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
)
for component in (
LifeSpan.BRUSH,
LifeSpan.FILTER,
LifeSpan.SIDE_BRUSH,
)
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add entities for passed config_entry in HA."""
controller: EcovacsController = hass.data[DOMAIN][config_entry.entry_id]
entities: list[EcovacsEntity] = get_supported_entitites(
controller, EcovacsSensor, ENTITY_DESCRIPTIONS
)
for device in controller.devices:
lifespan_capability = device.capabilities.life_span
for description in LIFESPAN_ENTITY_DESCRIPTIONS:
if description.component in lifespan_capability.types:
entities.append(
EcovacsLifespanSensor(device, lifespan_capability, description)
)
if capability := device.capabilities.error:
entities.append(EcovacsErrorSensor(device, capability))
async_add_entities(entities)
class EcovacsSensor(
EcovacsDescriptionEntity[CapabilityEvent],
SensorEntity,
):
"""Ecovacs sensor."""
entity_description: EcovacsSensorEntityDescription
async def async_added_to_hass(self) -> None:
"""Set up the event listeners now that hass is ready."""
await super().async_added_to_hass()
async def on_event(event: Event) -> None:
value = self.entity_description.value_fn(event)
if value is None:
return
self._attr_native_value = value
self.async_write_ha_state()
self._subscribe(self._capability.event, on_event)
class EcovacsLifespanSensor(
EcovacsDescriptionEntity[CapabilityLifeSpan],
SensorEntity,
):
"""Lifespan sensor."""
entity_description: EcovacsLifespanSensorEntityDescription
async def async_added_to_hass(self) -> None:
"""Set up the event listeners now that hass is ready."""
await super().async_added_to_hass()
async def on_event(event: LifeSpanEvent) -> None:
if event.type == self.entity_description.component:
self._attr_native_value = self.entity_description.value_fn(event)
self.async_write_ha_state()
self._subscribe(self._capability.event, on_event)
class EcovacsErrorSensor(
EcovacsEntity[CapabilityEvent[ErrorEvent]],
SensorEntity,
):
"""Error sensor."""
_always_available = True
_unrecorded_attributes = frozenset({CONF_DESCRIPTION})
entity_description: SensorEntityDescription = SensorEntityDescription(
key="error",
translation_key="error",
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
)
async def async_added_to_hass(self) -> None:
"""Set up the event listeners now that hass is ready."""
await super().async_added_to_hass()
async def on_event(event: ErrorEvent) -> None:
self._attr_native_value = event.code
self._attr_extra_state_attributes = {CONF_DESCRIPTION: event.description}
self.async_write_ha_state()
self._subscribe(self._capability.event, on_event)

View File

@ -24,6 +24,49 @@
"name": "Mop attached"
}
},
"sensor": {
"error": {
"name": "Error",
"state_attributes": {
"description": {
"name": "Description"
}
}
},
"lifespan_brush": {
"name": "Brush lifespan"
},
"lifespan_filter": {
"name": "Filter lifespan"
},
"lifespan_side_brush": {
"name": "Side brush lifespan"
},
"network_ip": {
"name": "IP address"
},
"network_rssi": {
"name": "Wi-Fi RSSI"
},
"network_ssid": {
"name": "Wi-Fi SSID"
},
"stats_area": {
"name": "Area cleaned"
},
"stats_time": {
"name": "Time cleaned"
},
"total_stats_area": {
"name": "Total area cleaned"
},
"total_stats_cleanings": {
"name": "Total cleanings"
},
"total_stats_time": {
"name": "Total time cleaned"
}
},
"vacuum": {
"vacuum": {
"state_attributes": {

View File

@ -1,7 +1,18 @@
"""Ecovacs util functions."""
from __future__ import annotations
import random
import string
from typing import TYPE_CHECKING
from .entity import (
EcovacsCapabilityEntityDescription,
EcovacsDescriptionEntity,
EcovacsEntity,
)
if TYPE_CHECKING:
from .controller import EcovacsController
def get_client_device_id() -> str:
@ -9,3 +20,19 @@ def get_client_device_id() -> str:
return "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(8)
)
def get_supported_entitites(
controller: EcovacsController,
entity_class: type[EcovacsDescriptionEntity],
descriptions: tuple[EcovacsCapabilityEntityDescription, ...],
) -> list[EcovacsEntity]:
"""Return all supported entities for all devices."""
entities: list[EcovacsEntity] = []
for device in controller.devices:
for description in descriptions:
if capability := description.capability_fn(device.capabilities):
entities.append(entity_class(device, capability, description))
return entities

View File

@ -210,7 +210,7 @@ _ATTR_ROOMS = "rooms"
class EcovacsVacuum(
EcovacsEntity[Capabilities, StateVacuumEntityDescription],
EcovacsEntity[Capabilities],
StateVacuumEntity,
):
"""Ecovacs vacuum."""

View File

@ -38,13 +38,13 @@ def mock_config_entry() -> MockConfigEntry:
@pytest.fixture
def device_classes() -> list[str]:
"""Device classes, which should be returned by the get_devices api call."""
return ["yna5x1"]
def device_fixture() -> str:
"""Device class, which should be returned by the get_devices api call."""
return "yna5x1"
@pytest.fixture
def mock_authenticator(device_classes: list[str]) -> Generator[Mock, None, None]:
def mock_authenticator(device_fixture: str) -> Generator[Mock, None, None]:
"""Mock the authenticator."""
with patch(
"homeassistant.components.ecovacs.controller.Authenticator",
@ -56,11 +56,9 @@ def mock_authenticator(device_classes: list[str]) -> Generator[Mock, None, None]
authenticator = mock.return_value
authenticator.authenticate.return_value = Credentials("token", "user_id", 0)
devices = []
for device_class in device_classes:
devices.append(
load_json_object_fixture(f"devices/{device_class}/device.json", DOMAIN)
)
devices = [
load_json_object_fixture(f"devices/{device_fixture}/device.json", DOMAIN)
]
def post_authenticated(
path: str,

View File

@ -0,0 +1,585 @@
# serializer version: 1
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_area_cleaned:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ozmo_950_area_cleaned',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Area cleaned',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'stats_area',
'unique_id': 'E1234567890000000001_stats_area',
'unit_of_measurement': 'm²',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_area_cleaned:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Area cleaned',
'unit_of_measurement': 'm²',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_area_cleaned',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '10',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_battery:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_battery',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
'original_name': 'Battery',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': 'E1234567890000000001_battery_level',
'unit_of_measurement': '%',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_battery:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'battery',
'friendly_name': 'Ozmo 950 Battery',
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_battery',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '100',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_brush_lifespan:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_brush_lifespan',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Brush lifespan',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'lifespan_brush',
'unique_id': 'E1234567890000000001_lifespan_brush',
'unit_of_measurement': '%',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_brush_lifespan:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Brush lifespan',
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_brush_lifespan',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '80',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_error:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_error',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Error',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'error',
'unique_id': 'E1234567890000000001_error',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_error:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'description': 'NoError: Robot is operational',
'friendly_name': 'Ozmo 950 Error',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_error',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '0',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_filter_lifespan:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_filter_lifespan',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Filter lifespan',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'lifespan_filter',
'unique_id': 'E1234567890000000001_lifespan_filter',
'unit_of_measurement': '%',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_filter_lifespan:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Filter lifespan',
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_filter_lifespan',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '56',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_ip_address:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_ip_address',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'IP address',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'network_ip',
'unique_id': 'E1234567890000000001_network_ip',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_ip_address:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 IP address',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_ip_address',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '192.168.0.10',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_side_brush_lifespan:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_side_brush_lifespan',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Side brush lifespan',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'lifespan_side_brush',
'unique_id': 'E1234567890000000001_lifespan_side_brush',
'unit_of_measurement': '%',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_side_brush_lifespan:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Side brush lifespan',
'unit_of_measurement': '%',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_side_brush_lifespan',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '40',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_time_cleaned:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ozmo_950_time_cleaned',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Time cleaned',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'stats_time',
'unique_id': 'E1234567890000000001_stats_time',
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_time_cleaned:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Time cleaned',
'unit_of_measurement': <UnitOfTime.SECONDS: 's'>,
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_time_cleaned',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '300',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_total_area_cleaned:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ozmo_950_total_area_cleaned',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Total area cleaned',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'total_stats_area',
'unique_id': 'E1234567890000000001_total_stats_area',
'unit_of_measurement': 'm²',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_total_area_cleaned:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Total area cleaned',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': 'm²',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_total_area_cleaned',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '60',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_total_cleanings:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ozmo_950_total_cleanings',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Total cleanings',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'total_stats_cleanings',
'unique_id': 'E1234567890000000001_total_stats_cleanings',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_total_cleanings:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Total cleanings',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_total_cleanings',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '123',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_total_time_cleaned:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.ozmo_950_total_time_cleaned',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
'sensor.private': dict({
'suggested_unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
}),
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Total time cleaned',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'total_stats_time',
'unique_id': 'E1234567890000000001_total_stats_time',
'unit_of_measurement': <UnitOfTime.HOURS: 'h'>,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_total_time_cleaned:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Total time cleaned',
'state_class': <SensorStateClass.TOTAL_INCREASING: 'total_increasing'>,
'unit_of_measurement': <UnitOfTime.MINUTES: 'min'>,
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_total_time_cleaned',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '144000',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_wi_fi_rssi:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_wi_fi_rssi',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Wi-Fi RSSI',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'network_rssi',
'unique_id': 'E1234567890000000001_network_rssi',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_wi_fi_rssi:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Wi-Fi RSSI',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_wi_fi_rssi',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': '-62',
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_wi_fi_ssid:entity-registry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.ozmo_950_wi_fi_ssid',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Wi-Fi SSID',
'platform': 'ecovacs',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'network_ssid',
'unique_id': 'E1234567890000000001_network_ssid',
'unit_of_measurement': None,
})
# ---
# name: test_sensors[yna5x1-entity_ids0][sensor.ozmo_950_wi_fi_ssid:state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Ozmo 950 Wi-Fi SSID',
}),
'context': <ANY>,
'entity_id': 'sensor.ozmo_950_wi_fi_ssid',
'last_changed': <ANY>,
'last_updated': <ANY>,
'state': 'Testnetwork',
})
# ---

View File

@ -0,0 +1,115 @@
"""Tests for Ecovacs sensors."""
from deebot_client.event_bus import EventBus
from deebot_client.events import (
BatteryEvent,
ErrorEvent,
LifeSpan,
LifeSpanEvent,
NetworkInfoEvent,
StatsEvent,
TotalStatsEvent,
)
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.ecovacs.controller import EcovacsController
from homeassistant.const import STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .util import block_till_done
pytestmark = [pytest.mark.usefixtures("init_integration")]
async def notify_events(hass: HomeAssistant, event_bus: EventBus):
"""Notify events."""
event_bus.notify(StatsEvent(10, 300, "spotArea"))
event_bus.notify(TotalStatsEvent(60, 144000, 123))
event_bus.notify(BatteryEvent(100))
event_bus.notify(BatteryEvent(100))
event_bus.notify(
NetworkInfoEvent("192.168.0.10", "Testnetwork", -62, "AA:BB:CC:DD:EE:FF")
)
event_bus.notify(LifeSpanEvent(LifeSpan.BRUSH, 80, 60 * 60))
event_bus.notify(LifeSpanEvent(LifeSpan.FILTER, 56, 40 * 60))
event_bus.notify(LifeSpanEvent(LifeSpan.SIDE_BRUSH, 40, 20 * 60))
event_bus.notify(ErrorEvent(0, "NoError: Robot is operational"))
await block_till_done(hass, event_bus)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("device_fixture", "entity_ids"),
[
(
"yna5x1",
[
"sensor.ozmo_950_area_cleaned",
"sensor.ozmo_950_battery",
"sensor.ozmo_950_brush_lifespan",
"sensor.ozmo_950_error",
"sensor.ozmo_950_filter_lifespan",
"sensor.ozmo_950_ip_address",
"sensor.ozmo_950_side_brush_lifespan",
"sensor.ozmo_950_time_cleaned",
"sensor.ozmo_950_total_area_cleaned",
"sensor.ozmo_950_total_cleanings",
"sensor.ozmo_950_total_time_cleaned",
"sensor.ozmo_950_wi_fi_rssi",
"sensor.ozmo_950_wi_fi_ssid",
],
),
],
)
async def test_sensors(
hass: HomeAssistant,
controller: EcovacsController,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
entity_ids: list[str],
) -> None:
"""Test that sensor entity snapshots match."""
assert entity_ids == sorted(hass.states.async_entity_ids(Platform.SENSOR))
for entity_id in entity_ids:
assert (state := hass.states.get(entity_id)), f"State of {entity_id} is missing"
assert state.state == STATE_UNKNOWN
await notify_events(hass, controller.devices[0].events)
for entity_id in entity_ids:
assert (state := hass.states.get(entity_id)), f"State of {entity_id} is missing"
assert snapshot(name=f"{entity_id}:state") == state
assert (entity_entry := entity_registry.async_get(state.entity_id))
assert snapshot(name=f"{entity_id}:entity-registry") == entity_entry
assert entity_entry.device_id
@pytest.mark.parametrize(
("device_fixture", "entity_ids"),
[
(
"yna5x1",
[
"sensor.ozmo_950_error",
"sensor.ozmo_950_ip_address",
"sensor.ozmo_950_wi_fi_rssi",
"sensor.ozmo_950_wi_fi_ssid",
],
),
],
)
async def test_disabled_by_default_sensors(
hass: HomeAssistant, entity_registry: er.EntityRegistry, entity_ids: list[str]
) -> None:
"""Test the disabled by default sensors."""
for entity_id in entity_ids:
assert not hass.states.get(entity_id)
assert (
entry := entity_registry.async_get(entity_id)
), f"Entity registry entry for {entity_id} is missing"
assert entry.disabled
assert entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION

View File

@ -1,6 +1,4 @@
"""Ecovacs test util."""
import asyncio
from deebot_client.event_bus import EventBus
@ -9,10 +7,15 @@ from deebot_client.events import Event
from homeassistant.core import HomeAssistant
async def block_till_done(hass: HomeAssistant, event_bus: EventBus) -> None:
"""Block till done."""
await asyncio.gather(*event_bus._tasks)
await hass.async_block_till_done()
async def notify_and_wait(
hass: HomeAssistant, event_bus: EventBus, event: Event
) -> None:
"""Block till done."""
event_bus.notify(event)
await asyncio.gather(*event_bus._tasks)
await hass.async_block_till_done()
await block_till_done(hass, event_bus)