Add switchbot roller shade and hubmini matter support (#142168)

* Add roller shade and hubmini matter support

* add unit tests

* fix adv data
This commit is contained in:
Retha Runolfsson 2025-04-07 22:25:00 +08:00 committed by GitHub
parent bf003d643c
commit 79b984d612
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 358 additions and 1 deletions

View File

@ -66,6 +66,12 @@ PLATFORMS_BY_TYPE = {
SupportedModels.RELAY_SWITCH_1.value: [Platform.SWITCH],
SupportedModels.LEAK.value: [Platform.BINARY_SENSOR, Platform.SENSOR],
SupportedModels.REMOTE.value: [Platform.SENSOR],
SupportedModels.ROLLER_SHADE.value: [
Platform.COVER,
Platform.BINARY_SENSOR,
Platform.SENSOR,
],
SupportedModels.HUBMINI_MATTER.value: [Platform.SENSOR],
}
CLASS_BY_DEVICE = {
SupportedModels.CEILING_LIGHT.value: switchbot.SwitchbotCeilingLight,
@ -80,6 +86,7 @@ CLASS_BY_DEVICE = {
SupportedModels.BLIND_TILT.value: switchbot.SwitchbotBlindTilt,
SupportedModels.RELAY_SWITCH_1PM.value: switchbot.SwitchbotRelaySwitch,
SupportedModels.RELAY_SWITCH_1.value: switchbot.SwitchbotRelaySwitch,
SupportedModels.ROLLER_SHADE.value: switchbot.SwitchbotRollerShade,
}

View File

@ -35,6 +35,8 @@ class SupportedModels(StrEnum):
RELAY_SWITCH_1 = "relay_switch_1"
LEAK = "leak"
REMOTE = "remote"
ROLLER_SHADE = "roller_shade"
HUBMINI_MATTER = "hubmini_matter"
CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@ -51,6 +53,7 @@ CONNECTABLE_SUPPORTED_MODEL_TYPES = {
SwitchbotModel.HUB2: SupportedModels.HUB2,
SwitchbotModel.RELAY_SWITCH_1PM: SupportedModels.RELAY_SWITCH_1PM,
SwitchbotModel.RELAY_SWITCH_1: SupportedModels.RELAY_SWITCH_1,
SwitchbotModel.ROLLER_SHADE: SupportedModels.ROLLER_SHADE,
}
NON_CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@ -62,6 +65,7 @@ NON_CONNECTABLE_SUPPORTED_MODEL_TYPES = {
SwitchbotModel.MOTION_SENSOR: SupportedModels.MOTION,
SwitchbotModel.LEAK: SupportedModels.LEAK,
SwitchbotModel.REMOTE: SupportedModels.REMOTE,
SwitchbotModel.HUBMINI_MATTER: SupportedModels.HUBMINI_MATTER,
}
SUPPORTED_MODEL_TYPES = (

View File

@ -37,6 +37,8 @@ async def async_setup_entry(
coordinator = entry.runtime_data
if isinstance(coordinator.device, switchbot.SwitchbotBlindTilt):
async_add_entities([SwitchBotBlindTiltEntity(coordinator)])
elif isinstance(coordinator.device, switchbot.SwitchbotRollerShade):
async_add_entities([SwitchBotRollerShadeEntity(coordinator)])
else:
async_add_entities([SwitchBotCurtainEntity(coordinator)])
@ -199,3 +201,85 @@ class SwitchBotBlindTiltEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
self._attr_is_opening = self.parsed_data["motionDirection"]["opening"]
self._attr_is_closing = self.parsed_data["motionDirection"]["closing"]
self.async_write_ha_state()
class SwitchBotRollerShadeEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
"""Representation of a Switchbot."""
_device: switchbot.SwitchbotRollerShade
_attr_device_class = CoverDeviceClass.SHADE
_attr_supported_features = (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.STOP
| CoverEntityFeature.SET_POSITION
)
_attr_translation_key = "cover"
_attr_name = None
def __init__(self, coordinator: SwitchbotDataUpdateCoordinator) -> None:
"""Initialize the switchbot."""
super().__init__(coordinator)
self._attr_is_closed = None
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added."""
await super().async_added_to_hass()
last_state = await self.async_get_last_state()
if not last_state or ATTR_CURRENT_POSITION not in last_state.attributes:
return
self._attr_current_cover_position = last_state.attributes.get(
ATTR_CURRENT_POSITION
)
self._last_run_success = last_state.attributes.get("last_run_success")
if self._attr_current_cover_position is not None:
self._attr_is_closed = self._attr_current_cover_position <= 20
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the roller shade."""
_LOGGER.debug("Switchbot to open roller shade %s", self._address)
self._last_run_success = bool(await self._device.open())
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the roller shade."""
_LOGGER.debug("Switchbot to close roller shade %s", self._address)
self._last_run_success = bool(await self._device.close())
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the moving of roller shade."""
_LOGGER.debug("Switchbot to stop roller shade %s", self._address)
self._last_run_success = bool(await self._device.stop())
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
position = kwargs.get(ATTR_POSITION)
_LOGGER.debug("Switchbot to move at %d %s", position, self._address)
self._last_run_success = bool(await self._device.set_position(position))
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._attr_is_closing = self._device.is_closing()
self._attr_is_opening = self._device.is_opening()
self._attr_current_cover_position = self.parsed_data["position"]
self._attr_is_closed = self.parsed_data["position"] <= 20
self.async_write_ha_state()

View File

@ -386,3 +386,53 @@ def make_advertisement(
connectable=True,
tx_power=-127,
)
HUBMINI_MATTER_SERVICE_INFO = BluetoothServiceInfoBleak(
name="HubMini Matter",
manufacturer_data={
2409: b"\xe6\xa1\xcd\x1f[e\x00\x00\x00\x00\x00\x00\x14\x01\x985\x00",
},
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="HubMini Matter",
manufacturer_data={
2409: b"\xe6\xa1\xcd\x1f[e\x00\x00\x00\x00\x00\x00\x14\x01\x985\x00",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"v\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "HubMini Matter"),
time=0,
connectable=True,
tx_power=-127,
)
ROLLER_SHADE_SERVICE_INFO = BluetoothServiceInfoBleak(
name="RollerShade",
manufacturer_data={
2409: b"\xb0\xe9\xfeT\x90\x1b,\x08\x9f\x11\x04'\x00",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b",\x00'\x9f\x11\x04"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=generate_advertisement_data(
local_name="RollerShade",
manufacturer_data={
2409: b"\xb0\xe9\xfeT\x90\x1b,\x08\x9f\x11\x04'\x00",
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b",\x00'\x9f\x11\x04"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "RollerShade"),
time=0,
connectable=True,
tx_power=-127,
)

View File

@ -24,7 +24,12 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant, State
from . import WOBLINDTILT_SERVICE_INFO, WOCURTAIN3_SERVICE_INFO, make_advertisement
from . import (
ROLLER_SHADE_SERVICE_INFO,
WOBLINDTILT_SERVICE_INFO,
WOCURTAIN3_SERVICE_INFO,
make_advertisement,
)
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.bluetooth import inject_bluetooth_service_info
@ -325,3 +330,163 @@ async def test_blindtilt_controlling(
state = hass.states.get(entity_id)
assert state.state == CoverState.OPEN
assert state.attributes[ATTR_CURRENT_TILT_POSITION] == 50
async def test_roller_shade_setup(
hass: HomeAssistant, mock_entry_factory: Callable[[str], MockConfigEntry]
) -> None:
"""Test setting up the RollerShade."""
inject_bluetooth_service_info(hass, WOCURTAIN3_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="roller_shade")
entity_id = "cover.test_name"
mock_restore_cache(
hass,
[
State(
entity_id,
CoverState.OPEN,
{ATTR_CURRENT_POSITION: 60},
)
],
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.update",
new=AsyncMock(return_value=True),
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state.state == CoverState.OPEN
assert state.attributes[ATTR_CURRENT_POSITION] == 60
async def test_roller_shade_controlling(
hass: HomeAssistant, mock_entry_factory: Callable[[str], MockConfigEntry]
) -> None:
"""Test Roller Shade controlling."""
inject_bluetooth_service_info(hass, ROLLER_SHADE_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="roller_shade")
entry.add_to_hass(hass)
info = {"battery": 39}
with (
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.get_basic_info",
new=AsyncMock(return_value=info),
),
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.open",
new=AsyncMock(return_value=True),
) as mock_open,
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.close",
new=AsyncMock(return_value=True),
) as mock_close,
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.stop",
new=AsyncMock(return_value=True),
) as mock_stop,
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.set_position",
new=AsyncMock(return_value=True),
) as mock_set_position,
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_id = "cover.test_name"
address = "AA:BB:CC:DD:EE:FF"
service_data = b",\x00'\x9f\x11\x04"
# Test open
manufacturer_data = b"\xb0\xe9\xfeT\x90\x1b,\x08\xa0\x11\x04'\x00"
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
with patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.get_basic_info",
new=AsyncMock(return_value=info),
):
inject_bluetooth_service_info(
hass, make_advertisement(address, manufacturer_data, service_data)
)
await hass.async_block_till_done()
mock_open.assert_awaited_once()
state = hass.states.get(entity_id)
assert state.state == CoverState.OPEN
assert state.attributes[ATTR_CURRENT_POSITION] == 68
# Test close
manufacturer_data = b"\xb0\xe9\xfeT\x90\x1b,\x08\x5a\x11\x04'\x00"
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
with patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.get_basic_info",
return_value=info,
):
inject_bluetooth_service_info(
hass, make_advertisement(address, manufacturer_data, service_data)
)
await hass.async_block_till_done()
mock_close.assert_awaited_once()
state = hass.states.get(entity_id)
assert state.state == CoverState.CLOSED
assert state.attributes[ATTR_CURRENT_POSITION] == 10
# Test stop
manufacturer_data = b"\xb0\xe9\xfeT\x90\x1b,\x08\x5f\x11\x04'\x00"
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
with patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.get_basic_info",
return_value=info,
):
inject_bluetooth_service_info(
hass, make_advertisement(address, manufacturer_data, service_data)
)
await hass.async_block_till_done()
mock_stop.assert_awaited_once()
state = hass.states.get(entity_id)
assert state.state == CoverState.CLOSED
assert state.attributes[ATTR_CURRENT_POSITION] == 5
# Test set position
manufacturer_data = b"\xb0\xe9\xfeT\x90\x1b,\x08\x32\x11\x04'\x00"
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: entity_id, ATTR_POSITION: 50},
blocking=True,
)
with patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotRollerShade.get_basic_info",
return_value=info,
):
inject_bluetooth_service_info(
hass, make_advertisement(address, manufacturer_data, service_data)
)
await hass.async_block_till_done()
mock_set_position.assert_awaited_once()
state = hass.states.get(entity_id)
assert state.state == CoverState.OPEN
assert state.attributes[ATTR_CURRENT_POSITION] == 50

View File

@ -22,6 +22,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from . import (
HUBMINI_MATTER_SERVICE_INFO,
LEAK_SERVICE_INFO,
REMOTE_SERVICE_INFO,
WOHAND_SERVICE_INFO,
@ -293,3 +294,49 @@ async def test_hub2_sensor(hass: HomeAssistant) -> None:
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_hubmini_matter_sensor(hass: HomeAssistant) -> None:
"""Test setting up creates the sensor for HubMini Matter."""
await async_setup_component(hass, DOMAIN, {})
inject_bluetooth_service_info(hass, HUBMINI_MATTER_SERVICE_INFO)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "AA:BB:CC:DD:EE:FF",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: "hubmini_matter",
},
unique_id="aabbccddeeff",
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all("sensor")) == 3
temperature_sensor = hass.states.get("sensor.test_name_temperature")
temperature_sensor_attrs = temperature_sensor.attributes
assert temperature_sensor.state == "24.1"
assert temperature_sensor_attrs[ATTR_FRIENDLY_NAME] == "test-name Temperature"
assert temperature_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "°C"
assert temperature_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
humidity_sensor = hass.states.get("sensor.test_name_humidity")
humidity_sensor_attrs = humidity_sensor.attributes
assert humidity_sensor.state == "53"
assert humidity_sensor_attrs[ATTR_FRIENDLY_NAME] == "test-name Humidity"
assert humidity_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "%"
assert humidity_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
rssi_sensor = hass.states.get("sensor.test_name_bluetooth_signal")
rssi_sensor_attrs = rssi_sensor.attributes
assert rssi_sensor.state == "-60"
assert rssi_sensor_attrs[ATTR_FRIENDLY_NAME] == "test-name Bluetooth signal"
assert rssi_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "dBm"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()