starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-25 17:04:04 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add clean_area support for Ecovacs mqtt vacuums (#163580)

Browse Source
This commit is contained in:
Robert Resch
2026-02-23 22:42:25 +01:00
committed by GitHub
parent fc9bdb3cb1
commit 977d29956b
2 changed files with 636 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/ecovacs/vacuum.py
+145 -9
View File
@@ -8,17 +8,24 @@ from typing import TYPE_CHECKING, Any
from deebot_client.capabilities import Capabilities, DeviceType
from deebot_client.device import Device
from deebot_client.events import FanSpeedEvent, RoomsEvent, StateEvent
from deebot_client.models import CleanAction, CleanMode, Room, State
from deebot_client.events import (
CachedMapInfoEvent,
FanSpeedEvent,
RoomsEvent,
StateEvent,
)
from deebot_client.events.map import Map
from deebot_client.models import CleanAction, CleanMode, State
import sucks
from homeassistant.components.vacuum import (
Segment,
StateVacuumEntity,
StateVacuumEntityDescription,
VacuumActivity,
VacuumEntityFeature,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import slugify
@@ -29,6 +36,7 @@ from .entity import EcovacsEntity, EcovacsLegacyEntity
from .util import get_name_key
_LOGGER = logging.getLogger(__name__)
_SEGMENTS_SEPARATOR = "_"
ATTR_ERROR = "error"
@@ -218,7 +226,8 @@ class EcovacsVacuum(
"""Initialize the vacuum."""
super().__init__(device, device.capabilities)
self._rooms: list[Room] = []
self._room_event: RoomsEvent | None = None
self._maps: dict[str, Map] = {}
if fan_speed := self._capability.fan_speed:
self._attr_supported_features |= VacuumEntityFeature.FAN_SPEED
@@ -226,14 +235,13 @@ class EcovacsVacuum(
get_name_key(level) for level in fan_speed.types
]
if self._capability.map and self._capability.clean.action.area:
self._attr_supported_features |= VacuumEntityFeature.CLEAN_AREA
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_rooms(event: RoomsEvent) -> None:
self._rooms = event.rooms
self.async_write_ha_state()
async def on_status(event: StateEvent) -> None:
self._attr_activity = _STATE_TO_VACUUM_STATE[event.state]
self.async_write_ha_state()
@@ -249,8 +257,20 @@ class EcovacsVacuum(
self._subscribe(self._capability.fan_speed.event, on_fan_speed)
if map_caps := self._capability.map:
async def on_rooms(event: RoomsEvent) -> None:
self._room_event = event
self._check_segments_changed()
self.async_write_ha_state()
self._subscribe(map_caps.rooms.event, on_rooms)
async def on_map_info(event: CachedMapInfoEvent) -> None:
self._maps = {map_obj.id: map_obj for map_obj in event.maps}
self._check_segments_changed()
self._subscribe(map_caps.cached_info.event, on_map_info)
@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return entity specific state attributes.
@@ -259,7 +279,10 @@ class EcovacsVacuum(
is lowercase snake_case.
"""
rooms: dict[str, Any] = {}
for room in self._rooms:
if self._room_event is None:
return rooms
for room in self._room_event.rooms:
# convert room name to snake_case to meet the convention
room_name = slugify(room.name)
room_values = rooms.get(room_name)
@@ -374,3 +397,116 @@ class EcovacsVacuum(
)
return await self._device.execute_command(position_commands[0])
@callback
def _check_segments_changed(self) -> None:
"""Check if segments have changed and create repair issue."""
last_seen = self.last_seen_segments
if last_seen is None:
return
last_seen_ids = {seg.id for seg in last_seen}
current_ids = {seg.id for seg in self._get_segments()}
if current_ids != last_seen_ids:
self.async_create_segments_issue()
def _get_segments(self) -> list[Segment]:
"""Get the segments that can be cleaned."""
last_seen = self.last_seen_segments or []
if self._room_event is None or not self._maps:
# If we don't have the necessary information to determine segments, return the last
# seen segments to avoid temporarily losing all segments until we get the necessary
# information, which could cause unnecessary issues to be created
return last_seen
map_id = self._room_event.map_id
if (map_obj := self._maps.get(map_id)) is None:
_LOGGER.warning("Map ID %s not found in available maps", map_id)
return []
id_prefix = f"{map_id}{_SEGMENTS_SEPARATOR}"
other_map_ids = {
map_obj.id
for map_obj in self._maps.values()
if map_obj.id != self._room_event.map_id
}
# Include segments from the current map and any segments from other maps that were
# previously seen, as we want to continue showing segments from other maps for
# mapping purposes
segments = [
seg for seg in last_seen if _split_composite_id(seg.id)[0] in other_map_ids
]
segments.extend(
Segment(
id=f"{id_prefix}{room.id}",
name=room.name,
group=map_obj.name,
)
for room in self._room_event.rooms
)
return segments
async def async_get_segments(self) -> list[Segment]:
"""Get the segments that can be cleaned."""
return self._get_segments()
async def async_clean_segments(self, segment_ids: list[str], **kwargs: Any) -> None:
"""Perform an area clean.
Only cleans segments from the currently selected map.
"""
if not self._maps:
_LOGGER.warning("No map information available, cannot clean segments")
return
valid_room_ids: list[int | float] = []
for composite_id in segment_ids:
map_id, segment_id = _split_composite_id(composite_id)
if (map_obj := self._maps.get(map_id)) is None:
_LOGGER.warning("Map ID %s not found in available maps", map_id)
continue
if not map_obj.using:
room_name = next(
(
segment.name
for segment in self.last_seen_segments or []
if segment.id == composite_id
),
"",
)
_LOGGER.warning(
'Map "%s" is not currently selected, skipping segment "%s" (%s)',
map_obj.name,
room_name,
segment_id,
)
continue
valid_room_ids.append(int(segment_id))
if not valid_room_ids:
_LOGGER.warning(
"No valid segments to clean after validation, skipping clean segments command"
)
return
if TYPE_CHECKING:
# Supported feature is only added if clean.action.area is not None
assert self._capability.clean.action.area is not None
await self._device.execute_command(
self._capability.clean.action.area(
CleanMode.SPOT_AREA,
valid_room_ids,
1,
)
)
@callback
def _split_composite_id(composite_id: str) -> tuple[str, str]:
"""Split a composite ID into its components."""
map_id, _, segment_id = composite_id.partition(_SEGMENTS_SEPARATOR)
return map_id, segment_id
tests/components/ecovacs/test_vacuum.py
+491
View File
@@ -0,0 +1,491 @@
"""Tests for Ecovacs vacuum entities."""
from dataclasses import asdict
import logging
from deebot_client.events import CachedMapInfoEvent, Event, RoomsEvent
from deebot_client.events.map import Map
from deebot_client.models import CleanMode, Room
from deebot_client.rs.map import RotationAngle # pylint: disable=no-name-in-module
import pytest
from homeassistant.components import vacuum
from homeassistant.components.ecovacs.controller import EcovacsController
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er, issue_registry as ir
from .util import notify_and_wait
from tests.typing import WebSocketGenerator
pytestmark = [pytest.mark.usefixtures("init_integration")]
@pytest.fixture
def platforms() -> Platform | list[Platform]:
"""Platforms, which should be loaded during the test."""
return Platform.VACUUM
def _prepare_test(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
entity_id: str,
) -> None:
entity_registry.async_update_entity_options(
entity_id,
vacuum.DOMAIN,
{
"area_mapping": {
"area_kitchen": ["1_1"],
"area_living_room": ["1_2"],
"area_bedroom": ["2_1"],
},
"last_seen_segments": [
{"id": "1_1", "name": "Kitchen", "group": "Main map"},
{"id": "1_2", "name": "Living room", "group": "Main map"},
{"id": "2_1", "name": "Bedroom", "group": "Second map"},
],
},
)
vacuum_obj = hass.data[vacuum.DATA_COMPONENT].get_entity(entity_id)
assert vacuum_obj.last_seen_segments == [
vacuum.Segment(id="1_1", name="Kitchen", group="Main map"),
vacuum.Segment(id="1_2", name="Living room", group="Main map"),
vacuum.Segment(id="2_1", name="Bedroom", group="Second map"),
]
assert vacuum_obj.supported_features & vacuum.VacuumEntityFeature.CLEAN_AREA
@pytest.mark.parametrize(("device_fixture", "entity_id"), [("qhe2o2", "vacuum.dusty")])
async def test_clean_area(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
controller: EcovacsController,
entity_id: str,
) -> None:
"""Test clean_area service call."""
_prepare_test(hass, entity_registry, entity_id)
device = controller.devices[0]
event_bus = device.events
await notify_and_wait(
hass,
event_bus,
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
Map(
id="2",
name="Second map",
using=False,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
)
device._execute_command.reset_mock()
await hass.services.async_call(
vacuum.DOMAIN,
vacuum.SERVICE_CLEAN_AREA,
{
ATTR_ENTITY_ID: entity_id,
"cleaning_area_id": ["area_living_room", "area_kitchen"],
},
blocking=True,
)
assert device._execute_command.call_count == 1
command = device._execute_command.call_args.args[0]
expected_command = device.capabilities.clean.action.area(
CleanMode.SPOT_AREA, [2, 1], 1
)
assert command == expected_command
@pytest.mark.parametrize(("device_fixture", "entity_id"), [("qhe2o2", "vacuum.dusty")])
async def test_clean_area_no_map(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
caplog: pytest.LogCaptureFixture,
controller: EcovacsController,
entity_id: str,
) -> None:
"""Test clean_area service call logs warning when no map info is available."""
_prepare_test(hass, entity_registry, entity_id)
device = controller.devices[0]
device._execute_command.reset_mock()
await hass.services.async_call(
vacuum.DOMAIN,
vacuum.SERVICE_CLEAN_AREA,
{
ATTR_ENTITY_ID: entity_id,
"cleaning_area_id": ["area_living_room", "area_kitchen"],
},
blocking=True,
)
assert caplog.record_tuples == [
(
"homeassistant.components.ecovacs.vacuum",
logging.WARNING,
"No map information available, cannot clean segments",
),
]
assert device._execute_command.call_count == 0
@pytest.mark.parametrize(("device_fixture", "entity_id"), [("qhe2o2", "vacuum.dusty")])
async def test_clean_area_invalid_map_id(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
caplog: pytest.LogCaptureFixture,
controller: EcovacsController,
entity_id: str,
) -> None:
"""Test clean_area service call logs warning when invalid map ID is provided."""
_prepare_test(hass, entity_registry, entity_id)
device = controller.devices[0]
event_bus = device.events
await notify_and_wait(
hass,
event_bus,
CachedMapInfoEvent(
{
Map(
id="2",
name="Second map",
using=False,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
)
device._execute_command.reset_mock()
caplog.clear()
await hass.services.async_call(
vacuum.DOMAIN,
vacuum.SERVICE_CLEAN_AREA,
{
ATTR_ENTITY_ID: entity_id,
"cleaning_area_id": ["area_living_room", "area_kitchen"],
},
blocking=True,
)
assert caplog.record_tuples == [
(
"homeassistant.components.ecovacs.vacuum",
logging.WARNING,
"Map ID 1 not found in available maps",
),
# twice as both areas reference the same missing map ID
(
"homeassistant.components.ecovacs.vacuum",
logging.WARNING,
"Map ID 1 not found in available maps",
),
(
"homeassistant.components.ecovacs.vacuum",
logging.WARNING,
"No valid segments to clean after validation, skipping clean segments command",
),
]
assert device._execute_command.call_count == 0
@pytest.mark.parametrize(("device_fixture", "entity_id"), [("qhe2o2", "vacuum.dusty")])
async def test_clean_area_room_from_not_current_map(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
caplog: pytest.LogCaptureFixture,
controller: EcovacsController,
entity_id: str,
) -> None:
"""Test clean_area service call logs warning when room is from a not current map."""
_prepare_test(hass, entity_registry, entity_id)
device = controller.devices[0]
event_bus = device.events
await notify_and_wait(
hass,
event_bus,
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
Map(
id="2",
name="Second map",
using=False,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
)
await notify_and_wait(
hass,
event_bus,
RoomsEvent(
map_id="1",
rooms=[
Room(name="Kitchen", id=1, coordinates=""),
Room(name="Living room", id=2, coordinates=""),
],
),
)
device._execute_command.reset_mock()
caplog.clear()
await hass.services.async_call(
vacuum.DOMAIN,
vacuum.SERVICE_CLEAN_AREA,
{
ATTR_ENTITY_ID: entity_id,
"cleaning_area_id": ["area_bedroom", "area_kitchen"],
},
blocking=True,
)
assert caplog.record_tuples == [
(
"homeassistant.components.ecovacs.vacuum",
logging.WARNING,
'Map "Second map" is not currently selected, skipping segment "Bedroom" (1)',
),
]
assert device._execute_command.call_count == 1
command = device._execute_command.call_args.args[0]
expected_command = device.capabilities.clean.action.area(
CleanMode.SPOT_AREA, [1], 1
)
assert command == expected_command
@pytest.mark.parametrize(
"events",
[
(
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
Map(
id="2",
name="Second map",
using=False,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
RoomsEvent(
map_id="1",
rooms=[
Room(name="Kitchen", id=1, coordinates=""),
],
),
),
(
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
RoomsEvent(
map_id="1",
rooms=[
Room(name="Kitchen", id=1, coordinates=""),
Room(name="Living room", id=2, coordinates=""),
],
),
),
(
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
Map(
id="2",
name="Second map",
using=False,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
RoomsEvent(
map_id="1",
rooms=[
Room(name="Kitchen", id=1, coordinates=""),
Room(name="Living room", id=2, coordinates=""),
Room(name="Bedroom", id=3, coordinates=""),
],
),
),
],
ids=[
"room removed",
"map removed",
"room added",
],
)
@pytest.mark.parametrize(("device_fixture", "entity_id"), [("qhe2o2", "vacuum.dusty")])
async def test_raise_segment_changed_issue(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
controller: EcovacsController,
entity_id: str,
events: tuple[Event, ...],
) -> None:
"""Test that the issue is raised on segment changes."""
_prepare_test(hass, entity_registry, entity_id)
device = controller.devices[0]
event_bus = device.events
for event in events:
await notify_and_wait(hass, event_bus, event)
entity_entry = entity_registry.async_get(entity_id)
issue_id = f"{vacuum.ISSUE_SEGMENTS_CHANGED}_{entity_entry.id}"
issue = ir.async_get(hass).async_get_issue(vacuum.DOMAIN, issue_id)
assert issue is not None
@pytest.mark.parametrize(
("events", "expected_segments", "expected_log_messages"),
[
((), [], []),
(
(
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
RoomsEvent(
map_id="2",
rooms=[
Room(name="Kitchen", id=1, coordinates=""),
Room(name="Living room", id=2, coordinates=""),
],
),
),
[],
[
(
"homeassistant.components.ecovacs.vacuum",
logging.WARNING,
"Map ID 2 not found in available maps",
)
],
),
(
(
CachedMapInfoEvent(
{
Map(
id="1",
name="Main map",
using=True,
built=True,
angle=RotationAngle.DEG_0,
),
}
),
RoomsEvent(
map_id="1",
rooms=[
Room(name="Kitchen", id=1, coordinates=""),
Room(name="Living room", id=2, coordinates=""),
],
),
),
[
vacuum.Segment(id="1_1", name="Kitchen", group="Main map"),
vacuum.Segment(id="1_2", name="Living room", group="Main map"),
],
[],
),
],
ids=[
"no room event available",
"invalid map ID in room event",
"room added",
],
)
@pytest.mark.parametrize(("device_fixture", "entity_id"), [("qhe2o2", "vacuum.dusty")])
async def test_get_segments(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
caplog: pytest.LogCaptureFixture,
controller: EcovacsController,
entity_id: str,
events: tuple[Event, ...],
expected_segments: list[vacuum.Segment],
expected_log_messages: list[tuple[str, int, str]],
) -> None:
"""Test vacuum/get_segments websocket command."""
device = controller.devices[0]
event_bus = device.events
for event in events:
await notify_and_wait(hass, event_bus, event)
caplog.clear()
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "vacuum/get_segments", "entity_id": entity_id}
)
msg = await client.receive_json()
assert msg["success"]
assert msg["result"] == {"segments": [asdict(seg) for seg in expected_segments]}
for log_message in expected_log_messages:
assert log_message in caplog.record_tuples