Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00
Code Activity

Add slow mode option for SwitchBot curtains (#155272)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Leonardo Merza
2026-02-09 07:30:14 -05:00
committed by GitHub
parent de8a26c5b0
commit a56114d84a
11 changed files with 356 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
homeassistant/components/switchbot/__init__.py
View File

@@ -1,6 +1,9 @@
"""Support for Switchbot devices."""
from __future__ import annotations
import logging
from typing import Any
import switchbot
@@ -20,10 +23,12 @@ from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
from .const import (
CONF_CURTAIN_SPEED,
CONF_ENCRYPTION_KEY,
CONF_KEY_ID,
CONF_RETRY_COUNT,
CONNECTABLE_SUPPORTED_MODEL_TYPES,
DEFAULT_CURTAIN_SPEED,
DEFAULT_RETRY_COUNT,
DOMAIN,
ENCRYPTED_MODELS,
@@ -185,12 +190,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: SwitchbotConfigEntry) ->
data={**entry.data, CONF_ADDRESS: mac},
)
if not entry.options:
hass.config_entries.async_update_entry(
entry,
options={CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT},
)
sensor_type: str = entry.data[CONF_SENSOR_TYPE]
switchbot_model = HASS_SENSOR_TYPE_TO_SWITCHBOT_MODEL[sensor_type]
# connectable means we can make connections to the device
@@ -241,6 +240,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: SwitchbotConfigEntry) ->
entry.data.get(CONF_NAME, entry.title),
connectable,
switchbot_model,
entry,
)
entry.async_on_unload(coordinator.async_start())
if not await coordinator.async_wait_ready():
@@ -258,6 +258,38 @@ async def async_setup_entry(hass: HomeAssistant, entry: SwitchbotConfigEntry) ->
return True
async def async_migrate_entry(hass: HomeAssistant, entry: SwitchbotConfigEntry) -> bool:
"""Migrate old entry."""
version = entry.version
minor_version = entry.minor_version
_LOGGER.debug("Migrating from version %s.%s", version, minor_version)
if version > 1:
return False
if version == 1 and minor_version < 2:
new_options: dict[str, Any] = {**entry.options}
if CONF_RETRY_COUNT not in new_options:
new_options[CONF_RETRY_COUNT] = DEFAULT_RETRY_COUNT
sensor_type = entry.data.get(CONF_SENSOR_TYPE)
if (
sensor_type == SupportedModels.CURTAIN
and CONF_CURTAIN_SPEED not in new_options
):
new_options[CONF_CURTAIN_SPEED] = DEFAULT_CURTAIN_SPEED
hass.config_entries.async_update_entry(
entry,
options=new_options,
minor_version=2,
)
_LOGGER.debug("Migration to version %s.2 successful", version)
return True
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)

36
homeassistant/components/switchbot/config_flow.py
View File

@@ -34,14 +34,19 @@ from homeassistant.const import (
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow
from homeassistant.helpers import selector
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import (
CONF_CURTAIN_SPEED,
CONF_ENCRYPTION_KEY,
CONF_KEY_ID,
CONF_LOCK_NIGHTLATCH,
CONF_RETRY_COUNT,
CONNECTABLE_SUPPORTED_MODEL_TYPES,
CURTAIN_SPEED_MAX,
CURTAIN_SPEED_MIN,
DEFAULT_CURTAIN_SPEED,
DEFAULT_LOCK_NIGHTLATCH,
DEFAULT_RETRY_COUNT,
DOMAIN,
@@ -75,6 +80,7 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Switchbot."""
VERSION = 1
MINOR_VERSION = 2
@staticmethod
@callback
@@ -130,13 +136,20 @@ class SwitchbotConfigFlow(ConfigFlow, domain=DOMAIN):
discovery = self._discovered_adv
name = name_from_discovery(discovery)
model_name = discovery.data["modelName"]
sensor_type = SUPPORTED_MODEL_TYPES[model_name]
options: dict[str, Any] = {CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT}
if sensor_type == SupportedModels.CURTAIN:
options[CONF_CURTAIN_SPEED] = DEFAULT_CURTAIN_SPEED
return self.async_create_entry(
title=name,
data={
**user_input,
CONF_ADDRESS: discovery.address,
CONF_SENSOR_TYPE: str(SUPPORTED_MODEL_TYPES[model_name]),
CONF_SENSOR_TYPE: str(sensor_type),
},
options=options,
)
async def async_step_confirm(
@@ -455,5 +468,26 @@ class SwitchbotOptionsFlowHandler(OptionsFlow):
): bool
}
)
if (
CONF_SENSOR_TYPE in self.config_entry.data
and self.config_entry.data[CONF_SENSOR_TYPE] == SupportedModels.CURTAIN
):
options.update(
{
vol.Optional(
CONF_CURTAIN_SPEED,
default=self.config_entry.options.get(
CONF_CURTAIN_SPEED, DEFAULT_CURTAIN_SPEED
),
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=CURTAIN_SPEED_MIN,
max=CURTAIN_SPEED_MAX,
step=1,
mode=selector.NumberSelectorMode.SLIDER,
)
)
}
)
return self.async_show_form(step_id="init", data_schema=vol.Schema(options))

4
homeassistant/components/switchbot/const.py
View File

@@ -182,9 +182,13 @@ HASS_SENSOR_TYPE_TO_SWITCHBOT_MODEL = {
# Config Defaults
DEFAULT_RETRY_COUNT = 3
DEFAULT_LOCK_NIGHTLATCH = False
DEFAULT_CURTAIN_SPEED = 255
CURTAIN_SPEED_MIN = 0
CURTAIN_SPEED_MAX = 255
# Config Options
CONF_RETRY_COUNT = "retry_count"
CONF_KEY_ID = "key_id"
CONF_ENCRYPTION_KEY = "encryption_key"
CONF_LOCK_NIGHTLATCH = "lock_force_nightlatch"
CONF_CURTAIN_SPEED = "curtain_speed"

2
homeassistant/components/switchbot/coordinator.py
View File

@@ -41,6 +41,7 @@ class SwitchbotDataUpdateCoordinator(ActiveBluetoothDataUpdateCoordinator[None])
device_name: str,
connectable: bool,
model: SwitchbotModel,
config_entry: ConfigEntry,
) -> None:
"""Initialize global switchbot data updater."""
super().__init__(
@@ -57,6 +58,7 @@ class SwitchbotDataUpdateCoordinator(ActiveBluetoothDataUpdateCoordinator[None])
self.device_name = device_name
self.base_unique_id = base_unique_id
self.model = model
self.config_entry = config_entry
self._ready_event = asyncio.Event()
self._was_unavailable = True

16
homeassistant/components/switchbot/cover.py
View File

@@ -20,6 +20,7 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from .const import CONF_CURTAIN_SPEED, DEFAULT_CURTAIN_SPEED
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity, exception_handler
@@ -64,6 +65,15 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
super().__init__(coordinator)
self._attr_is_closed = None
@callback
def _get_curtain_speed(self) -> int:
"""Return the configured curtain speed."""
return int(
self.coordinator.config_entry.options.get(
CONF_CURTAIN_SPEED, DEFAULT_CURTAIN_SPEED
)
)
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added."""
await super().async_added_to_hass()
@@ -83,7 +93,8 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
"""Open the curtain."""
_LOGGER.debug("Switchbot to open curtain %s", self._address)
self._last_run_success = bool(await self._device.open())
speed = self._get_curtain_speed()
self._last_run_success = bool(await self._device.open(speed))
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()
@@ -93,7 +104,8 @@ class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity):
"""Close the curtain."""
_LOGGER.debug("Switchbot to close the curtain %s", self._address)
self._last_run_success = bool(await self._device.close())
speed = self._get_curtain_speed()
self._last_run_success = bool(await self._device.close(speed))
self._attr_is_opening = self._device.is_opening()
self._attr_is_closing = self._device.is_closing()
self.async_write_ha_state()

2
homeassistant/components/switchbot/strings.json
View File

@@ -358,10 +358,12 @@
"step": {
"init": {
"data": {
"curtain_speed": "Curtain movement speed",
"lock_force_nightlatch": "Force Nightlatch operation mode",
"retry_count": "Retry count"
},
"data_description": {
"curtain_speed": "Speed for curtain open and close operations (1-255, where 1 is slowest and 255 is fastest)",
"lock_force_nightlatch": "Force Nightlatch operation mode even if Nightlatch is not detected",
"retry_count": "How many times to retry sending commands to your SwitchBot devices"
}

65
tests/components/switchbot/conftest.py
View File

@@ -3,9 +3,14 @@
import pytest
from homeassistant.components.switchbot.const import (
CONF_CURTAIN_SPEED,
CONF_ENCRYPTION_KEY,
CONF_KEY_ID,
CONF_RETRY_COUNT,
DEFAULT_CURTAIN_SPEED,
DEFAULT_RETRY_COUNT,
DOMAIN,
SupportedModels,
)
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_SENSOR_TYPE
@@ -20,28 +25,48 @@ def mock_bluetooth(enable_bluetooth: None) -> None:
@pytest.fixture
def mock_entry_factory():
"""Fixture to create a MockConfigEntry with a customizable sensor type."""
return lambda sensor_type="curtain": MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: sensor_type,
},
unique_id="aabbccddeeff",
)
def _create_entry(sensor_type: str = "curtain") -> MockConfigEntry:
options: dict[str, int] = {CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT}
if sensor_type == SupportedModels.CURTAIN:
options[CONF_CURTAIN_SPEED] = DEFAULT_CURTAIN_SPEED
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: sensor_type,
},
unique_id="aabbccddeeff",
version=1,
minor_version=2,
options=options,
)
return _create_entry
@pytest.fixture
def mock_entry_encrypted_factory():
"""Fixture to create a MockConfigEntry with an encryption key and a customizable sensor type."""
return lambda sensor_type="lock": MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: sensor_type,
CONF_KEY_ID: "ff",
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
},
unique_id="aabbccddeeff",
)
def _create_entry(sensor_type: str = "lock") -> MockConfigEntry:
options: dict[str, int] = {CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT}
if sensor_type == SupportedModels.CURTAIN:
options[CONF_CURTAIN_SPEED] = DEFAULT_CURTAIN_SPEED
return MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: sensor_type,
CONF_KEY_ID: "ff",
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
},
unique_id="aabbccddeeff",
version=1,
minor_version=2,
options=options,
)
return _create_entry

2
tests/components/switchbot/snapshots/test_diagnostics.ambr
View File

@@ -13,7 +13,7 @@
'discovery_keys': dict({
}),
'domain': 'switchbot',
'minor_version': 1,
'minor_version': 2,
'options': dict({
'retry_count': 3,
}),

40
tests/components/switchbot/test_config_flow.py
View File

@@ -8,6 +8,7 @@ from switchbot import SwitchbotAccountConnectionError, SwitchbotAuthenticationEr
from homeassistant.components.bluetooth import BluetoothScanningMode
from homeassistant.components.switchbot.const import (
CONF_CURTAIN_SPEED,
CONF_ENCRYPTION_KEY,
CONF_KEY_ID,
CONF_LOCK_NIGHTLATCH,
@@ -1265,6 +1266,45 @@ async def test_options_flow_lock_pro(hass: HomeAssistant) -> None:
assert entry.options[CONF_LOCK_NIGHTLATCH] is True
async def test_options_flow_curtain_speed(hass: HomeAssistant) -> None:
"""Test updating curtain speed option."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: "curtain",
},
options={CONF_RETRY_COUNT: 2, CONF_CURTAIN_SPEED: 255},
unique_id="aabbccddeeff",
)
entry.add_to_hass(hass)
with patch_async_setup_entry() as mock_setup_entry:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["errors"] is None
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_RETRY_COUNT: 4,
CONF_CURTAIN_SPEED: 100,
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_RETRY_COUNT] == 4
assert result["data"][CONF_CURTAIN_SPEED] == 100
assert entry.options[CONF_CURTAIN_SPEED] == 100
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.usefixtures("mock_scanners_all_passive")
async def test_user_setup_worelay_switch_1pm_key(hass: HomeAssistant) -> None:
"""Test the user initiated form for a relay switch 1pm."""

58
tests/components/switchbot/test_cover.py
View File

@@ -15,6 +15,11 @@ from homeassistant.components.cover import (
DOMAIN as COVER_DOMAIN,
CoverState,
)
from homeassistant.components.switchbot.const import (
CONF_CURTAIN_SPEED,
CONF_RETRY_COUNT,
DEFAULT_RETRY_COUNT,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_CLOSE_COVER,
@@ -114,7 +119,7 @@ async def test_curtain3_controlling(
)
await hass.async_block_till_done()
mock_open.assert_awaited_once()
mock_open.assert_awaited_once_with(255) # Default speed
state = hass.states.get(entity_id)
assert state.state == CoverState.OPEN
assert state.attributes[ATTR_CURRENT_POSITION] == 95
@@ -132,7 +137,7 @@ async def test_curtain3_controlling(
)
await hass.async_block_till_done()
mock_close.assert_awaited_once()
mock_close.assert_awaited_once_with(255) # Default speed
state = hass.states.get(entity_id)
assert state.state == CoverState.CLOSED
assert state.attributes[ATTR_CURRENT_POSITION] == 12
@@ -171,6 +176,55 @@ async def test_curtain3_controlling(
assert state.attributes[ATTR_CURRENT_POSITION] == 60
async def test_curtain3_custom_speed_controlling(
hass: HomeAssistant, mock_entry_factory: Callable[[str], MockConfigEntry]
) -> None:
"""Test Curtain3 controlling with custom speed."""
inject_bluetooth_service_info(hass, WOCURTAIN3_SERVICE_INFO)
entry = mock_entry_factory(sensor_type="curtain")
entry.add_to_hass(hass)
# Update entry options using async_update_entry
hass.config_entries.async_update_entry(
entry,
options={
CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT,
CONF_CURTAIN_SPEED: 50,
},
)
with (
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotCurtain.open",
new=AsyncMock(return_value=True),
) as mock_open,
patch(
"homeassistant.components.switchbot.cover.switchbot.SwitchbotCurtain.close",
new=AsyncMock(return_value=True),
) as mock_close,
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entity_id = "cover.test_name"
await hass.services.async_call(
COVER_DOMAIN, SERVICE_OPEN_COVER, {ATTR_ENTITY_ID: entity_id}, blocking=True
)
await hass.async_block_till_done()
mock_open.assert_awaited_once_with(50)
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
mock_close.assert_awaited_once_with(50)
async def test_blindtilt_setup(
hass: HomeAssistant, mock_entry_factory: Callable[[str], MockConfigEntry]
) -> None:

119
tests/components/switchbot/test_init.py
View File

@@ -5,11 +5,21 @@ from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.switchbot.const import (
CONF_CURTAIN_SPEED,
CONF_RETRY_COUNT,
DEFAULT_CURTAIN_SPEED,
DEFAULT_RETRY_COUNT,
DOMAIN,
)
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_SENSOR_TYPE
from homeassistant.core import HomeAssistant
from . import (
HUBMINI_MATTER_SERVICE_INFO,
LOCK_SERVICE_INFO,
WOCURTAIN_SERVICE_INFO,
WOSENSORTH_SERVICE_INFO,
patch_async_ble_device_from_address,
)
@@ -92,3 +102,112 @@ async def test_coordinator_wait_ready_timeout(
await hass.async_block_till_done()
assert "aa:bb:cc:dd:ee:ff is not advertising state" in caplog.text
@pytest.mark.parametrize(
("sensor_type", "service_info", "expected_options"),
[
(
"curtain",
WOCURTAIN_SERVICE_INFO,
{
CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT,
CONF_CURTAIN_SPEED: DEFAULT_CURTAIN_SPEED,
},
),
(
"hygrometer",
WOSENSORTH_SERVICE_INFO,
{
CONF_RETRY_COUNT: DEFAULT_RETRY_COUNT,
},
),
],
)
async def test_migrate_entry_from_v1_1_to_v1_2(
hass: HomeAssistant,
sensor_type: str,
service_info,
expected_options: dict,
) -> None:
"""Test migration from version 1.1 to 1.2 adds default options."""
inject_bluetooth_service_info(hass, service_info)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: sensor_type,
},
unique_id="aabbccddeeff",
version=1,
minor_version=1,
options={},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.version == 1
assert entry.minor_version == 2
assert entry.options == expected_options
async def test_migrate_entry_preserves_existing_options(
hass: HomeAssistant,
) -> None:
"""Test migration preserves existing options."""
inject_bluetooth_service_info(hass, WOCURTAIN_SERVICE_INFO)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: "curtain",
},
unique_id="aabbccddeeff",
version=1,
minor_version=1,
options={CONF_RETRY_COUNT: 5},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.version == 1
assert entry.minor_version == 2
# Existing retry_count should be preserved, curtain_speed added
assert entry.options[CONF_RETRY_COUNT] == 5
assert entry.options[CONF_CURTAIN_SPEED] == DEFAULT_CURTAIN_SPEED
async def test_migrate_entry_fails_for_future_version(
hass: HomeAssistant,
) -> None:
"""Test migration fails for future versions."""
inject_bluetooth_service_info(hass, WOCURTAIN_SERVICE_INFO)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: "curtain",
},
unique_id="aabbccddeeff",
version=2,
minor_version=1,
options={},
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Entry should not be loaded due to failed migration
assert entry.version == 2
assert entry.minor_version == 1