Add switchbot vacuum support (#144550)

* add support for vacuum

* add vacuum unit test
This commit is contained in:
Retha Runolfsson 2025-05-10 15:34:51 +08:00 committed by GitHub
parent 626f8a9166
commit 977d2fe8b3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 360 additions and 0 deletions

View File

@ -73,6 +73,11 @@ PLATFORMS_BY_TYPE = {
],
SupportedModels.HUBMINI_MATTER.value: [Platform.SENSOR],
SupportedModels.CIRCULATOR_FAN.value: [Platform.FAN, Platform.SENSOR],
SupportedModels.K20_VACUUM.value: [Platform.VACUUM, Platform.SENSOR],
SupportedModels.S10_VACUUM.value: [Platform.VACUUM, Platform.SENSOR],
SupportedModels.K10_VACUUM.value: [Platform.VACUUM, Platform.SENSOR],
SupportedModels.K10_PRO_VACUUM.value: [Platform.VACUUM, Platform.SENSOR],
SupportedModels.K10_PRO_COMBO_VACUUM.value: [Platform.VACUUM, Platform.SENSOR],
}
CLASS_BY_DEVICE = {
SupportedModels.CEILING_LIGHT.value: switchbot.SwitchbotCeilingLight,
@ -89,6 +94,11 @@ CLASS_BY_DEVICE = {
SupportedModels.RELAY_SWITCH_1.value: switchbot.SwitchbotRelaySwitch,
SupportedModels.ROLLER_SHADE.value: switchbot.SwitchbotRollerShade,
SupportedModels.CIRCULATOR_FAN.value: switchbot.SwitchbotFan,
SupportedModels.K20_VACUUM.value: switchbot.SwitchbotVacuum,
SupportedModels.S10_VACUUM.value: switchbot.SwitchbotVacuum,
SupportedModels.K10_VACUUM.value: switchbot.SwitchbotVacuum,
SupportedModels.K10_PRO_VACUUM.value: switchbot.SwitchbotVacuum,
SupportedModels.K10_PRO_COMBO_VACUUM.value: switchbot.SwitchbotVacuum,
}

View File

@ -38,6 +38,11 @@ class SupportedModels(StrEnum):
ROLLER_SHADE = "roller_shade"
HUBMINI_MATTER = "hubmini_matter"
CIRCULATOR_FAN = "circulator_fan"
K20_VACUUM = "k20_vacuum"
S10_VACUUM = "s10_vacuum"
K10_VACUUM = "k10_vacuum"
K10_PRO_VACUUM = "k10_pro_vacuum"
K10_PRO_COMBO_VACUUM = "k10_pro_combo_vacumm"
CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@ -56,6 +61,11 @@ CONNECTABLE_SUPPORTED_MODEL_TYPES = {
SwitchbotModel.RELAY_SWITCH_1: SupportedModels.RELAY_SWITCH_1,
SwitchbotModel.ROLLER_SHADE: SupportedModels.ROLLER_SHADE,
SwitchbotModel.CIRCULATOR_FAN: SupportedModels.CIRCULATOR_FAN,
SwitchbotModel.K20_VACUUM: SupportedModels.K20_VACUUM,
SwitchbotModel.S10_VACUUM: SupportedModels.S10_VACUUM,
SwitchbotModel.K10_VACUUM: SupportedModels.K10_VACUUM,
SwitchbotModel.K10_PRO_VACUUM: SupportedModels.K10_PRO_VACUUM,
SwitchbotModel.K10_PRO_COMBO_VACUUM: SupportedModels.K10_PRO_COMBO_VACUUM,
}
NON_CONNECTABLE_SUPPORTED_MODEL_TYPES = {

View File

@ -180,6 +180,18 @@
}
}
}
},
"vacuum": {
"vacuum": {
"state_attributes": {
"last_run_success": {
"state": {
"true": "[%key:component::binary_sensor::entity_component::problem::state::off%]",
"false": "[%key:component::binary_sensor::entity_component::problem::state::on%]"
}
}
}
}
}
},
"exceptions": {

View File

@ -0,0 +1,126 @@
"""Support for switchbot vacuums."""
from __future__ import annotations
from typing import Any
import switchbot
from switchbot import SwitchbotModel
from homeassistant.components.vacuum import (
StateVacuumEntity,
VacuumActivity,
VacuumEntityFeature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
PARALLEL_UPDATES = 0
DEVICE_SUPPORT_PROTOCOL_VERSION_1 = [
SwitchbotModel.K10_VACUUM,
SwitchbotModel.K10_PRO_VACUUM,
]
PROTOCOL_VERSION_1_STATE_TO_HA_STATE: dict[int, VacuumActivity] = {
0: VacuumActivity.CLEANING,
1: VacuumActivity.DOCKED,
}
PROTOCOL_VERSION_2_STATE_TO_HA_STATE: dict[int, VacuumActivity] = {
1: VacuumActivity.IDLE, # idle
2: VacuumActivity.DOCKED, # charge
3: VacuumActivity.DOCKED, # charge complete
4: VacuumActivity.IDLE, # self-check
5: VacuumActivity.IDLE, # the drum is moist
6: VacuumActivity.CLEANING, # exploration
7: VacuumActivity.CLEANING, # re-location
8: VacuumActivity.CLEANING, # cleaning and sweeping
9: VacuumActivity.CLEANING, # cleaning
10: VacuumActivity.CLEANING, # sweeping
11: VacuumActivity.PAUSED, # pause
12: VacuumActivity.CLEANING, # getting out of trouble
13: VacuumActivity.ERROR, # trouble
14: VacuumActivity.CLEANING, # mpo cleaning
15: VacuumActivity.RETURNING, # returning
16: VacuumActivity.CLEANING, # deep cleaning
17: VacuumActivity.CLEANING, # Sewage extraction
18: VacuumActivity.CLEANING, # replenish water for mop
19: VacuumActivity.CLEANING, # dust collection
20: VacuumActivity.CLEANING, # dry
21: VacuumActivity.IDLE, # dormant
22: VacuumActivity.IDLE, # network configuration
23: VacuumActivity.CLEANING, # remote control
24: VacuumActivity.RETURNING, # return to base
25: VacuumActivity.IDLE, # shut down
26: VacuumActivity.IDLE, # mark water base station
27: VacuumActivity.IDLE, # rinse the filter screen
28: VacuumActivity.IDLE, # mark humidifier location
29: VacuumActivity.IDLE, # on the way to the humidifier
30: VacuumActivity.IDLE, # add water for humidifier
31: VacuumActivity.IDLE, # upgrading
32: VacuumActivity.PAUSED, # pause during recharging
33: VacuumActivity.IDLE, # integrated with the platform
34: VacuumActivity.CLEANING, # working for the platform
}
SWITCHBOT_VACUUM_STATE_MAP: dict[int, dict[int, VacuumActivity]] = {
1: PROTOCOL_VERSION_1_STATE_TO_HA_STATE,
2: PROTOCOL_VERSION_2_STATE_TO_HA_STATE,
}
async def async_setup_entry(
hass: HomeAssistant,
entry: SwitchbotConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switchbot vacuum."""
async_add_entities([SwitchbotVacuumEntity(entry.runtime_data)])
class SwitchbotVacuumEntity(SwitchbotEntity, StateVacuumEntity):
"""Representation of a SwitchBot vacuum."""
_device: switchbot.SwitchbotVacuum
_attr_supported_features = (
VacuumEntityFeature.BATTERY
| VacuumEntityFeature.RETURN_HOME
| VacuumEntityFeature.START
| VacuumEntityFeature.STATE
)
_attr_translation_key = "vacuum"
_attr_name = None
def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
"""Initialize the Switchbot."""
super().__init__(coordinator)
self.protocol_version = (
1 if coordinator.model in DEVICE_SUPPORT_PROTOCOL_VERSION_1 else 2
)
@property
def activity(self) -> VacuumActivity | None:
"""Return the status of the vacuum cleaner."""
status_code = self._device.get_work_status()
return SWITCHBOT_VACUUM_STATE_MAP[self.protocol_version].get(status_code)
@property
def battery_level(self) -> int:
"""Return the vacuum battery."""
return self._device.get_battery()
async def async_start(self) -> None:
"""Start or resume the cleaning task."""
self._last_run_success = bool(
await self._device.clean_up(self.protocol_version)
)
async def async_return_to_base(self, **kwargs: Any) -> None:
"""Return to dock."""
self._last_run_success = bool(
await self._device.return_to_dock(self.protocol_version)
)

View File

@ -555,3 +555,128 @@ CIRCULATOR_FAN_SERVICE_INFO = BluetoothServiceInfoBleak(
connectable=True,
tx_power=-127,
)
K20_VACUUM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="K20 Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfe\x01\xf3\x8f'\x01\x11S\x00\x10d\x0f",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b".\x00d"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="K20 Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfe\x01\xf3\x8f'\x01\x11S\x00\x10d\x0f",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b".\x00d"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "K20 Vacuum"),
time=0,
connectable=True,
tx_power=-127,
)
K10_PRO_VACUUM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="K10 Pro Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfeP\x8d\x8d\x02 d",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"(\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="K10 Pro Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfeP\x8d\x8d\x02 d",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"(\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "K10 Pro Vacuum"),
time=0,
connectable=True,
tx_power=-127,
)
K10_VACUUM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="K10 Vacuum",
manufacturer_data={
2409: b"\xca8\x06\xa9_\xf1\x02 d",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"}\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="K10 Vacuum",
manufacturer_data={
2409: b"\xca8\x06\xa9_\xf1\x02 d",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"}\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "K10 Vacuum"),
time=0,
connectable=True,
tx_power=-127,
)
K10_POR_COMBO_VACUUM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="K10 Pro Combo Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfe\x01\xf4\x1d\x0b\x01\x01\xb1\x03\x118\x01",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"3\x00\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="K10 Pro Combo Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfe\x01\xf4\x1d\x0b\x01\x01\xb1\x03\x118\x01",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"3\x00\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "K10 Pro Combo Vacuum"),
time=0,
connectable=True,
tx_power=-127,
)
S10_VACUUM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="S10 Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfe\x00\x08|\n\x01\x11\x05\x00\x10M\x02",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"z\x00\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="S10 Vacuum",
manufacturer_data={
2409: b"\xb0\xe9\xfe\x00\x08|\n\x01\x11\x05\x00\x10M\x02",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"z\x00\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "S10 Vacuum"),
time=0,
connectable=True,
tx_power=-127,
)

View File

@ -0,0 +1,77 @@
"""Tests for switchbot vacuum."""
from collections.abc import Callable
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.vacuum import (
DOMAIN as VACUUM_DOMAIN,
SERVICE_RETURN_TO_BASE,
SERVICE_START,
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from . import (
K10_POR_COMBO_VACUUM_SERVICE_INFO,
K10_PRO_VACUUM_SERVICE_INFO,
K10_VACUUM_SERVICE_INFO,
K20_VACUUM_SERVICE_INFO,
S10_VACUUM_SERVICE_INFO,
)
from tests.common import MockConfigEntry
from tests.components.bluetooth import inject_bluetooth_service_info
@pytest.mark.parametrize(
("sensor_type", "service_info"),
[
("k20_vacuum", K20_VACUUM_SERVICE_INFO),
("s10_vacuum", S10_VACUUM_SERVICE_INFO),
("k10_pro_combo_vacumm", K10_POR_COMBO_VACUUM_SERVICE_INFO),
("k10_vacuum", K10_VACUUM_SERVICE_INFO),
("k10_pro_vacuum", K10_PRO_VACUUM_SERVICE_INFO),
],
)
@pytest.mark.parametrize(
("service", "mock_method"),
[(SERVICE_START, "clean_up"), (SERVICE_RETURN_TO_BASE, "return_to_dock")],
)
async def test_vacuum_controlling(
hass: HomeAssistant,
mock_entry_factory: Callable[[str], MockConfigEntry],
sensor_type: str,
service: str,
mock_method: str,
service_info: BluetoothServiceInfoBleak,
) -> None:
"""Test switchbot vacuum controlling."""
inject_bluetooth_service_info(hass, service_info)
entry = mock_entry_factory(sensor_type)
entry.add_to_hass(hass)
mocked_instance = AsyncMock(return_value=True)
with patch.multiple(
"homeassistant.components.switchbot.vacuum.switchbot.SwitchbotVacuum",
update=MagicMock(return_value=None),
**{mock_method: mocked_instance},
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_id = "vacuum.test_name"
await hass.services.async_call(
VACUUM_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mocked_instance.assert_awaited_once()