Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-24 21:06:19 +00:00
Code Activity

Add switchbot relayswitch 2PM (#146140)

Browse Source
This commit is contained in:
Retha Runolfsson
2025-09-16 21:59:13 +08:00
committed by GitHub
parent e70b147c0c
commit aadaf87c16
8 changed files with 349 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -98,6 +98,7 @@ PLATFORMS_BY_TYPE = {
SupportedModels.RGBICWW_FLOOR_LAMP.value: [Platform.LIGHT, Platform.SENSOR],
SupportedModels.RGBICWW_STRIP_LIGHT.value: [Platform.LIGHT, Platform.SENSOR],
SupportedModels.PLUG_MINI_EU.value: [Platform.SWITCH, Platform.SENSOR],
SupportedModels.RELAY_SWITCH_2PM.value: [Platform.SWITCH, Platform.SENSOR],
}
CLASS_BY_DEVICE = {
SupportedModels.CEILING_LIGHT.value: switchbot.SwitchbotCeilingLight,
@@ -129,6 +130,7 @@ CLASS_BY_DEVICE = {
SupportedModels.RGBICWW_FLOOR_LAMP.value: switchbot.SwitchbotRgbicLight,
SupportedModels.RGBICWW_STRIP_LIGHT.value: switchbot.SwitchbotRgbicLight,
SupportedModels.PLUG_MINI_EU.value: switchbot.SwitchbotRelaySwitch,
SupportedModels.RELAY_SWITCH_2PM.value: switchbot.SwitchbotRelaySwitch2PM,
}

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

@@ -54,6 +54,7 @@ class SupportedModels(StrEnum):
RGBICWW_STRIP_LIGHT = "rgbicww_strip_light"
RGBICWW_FLOOR_LAMP = "rgbicww_floor_lamp"
PLUG_MINI_EU = "plug_mini_eu"
RELAY_SWITCH_2PM = "relay_switch_2pm"
CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@@ -87,6 +88,7 @@ CONNECTABLE_SUPPORTED_MODEL_TYPES = {
SwitchbotModel.RGBICWW_STRIP_LIGHT: SupportedModels.RGBICWW_STRIP_LIGHT,
SwitchbotModel.RGBICWW_FLOOR_LAMP: SupportedModels.RGBICWW_FLOOR_LAMP,
SwitchbotModel.PLUG_MINI_EU: SupportedModels.PLUG_MINI_EU,
SwitchbotModel.RELAY_SWITCH_2PM: SupportedModels.RELAY_SWITCH_2PM,
}
NON_CONNECTABLE_SUPPORTED_MODEL_TYPES = {
@@ -121,6 +123,7 @@ ENCRYPTED_MODELS = {
SwitchbotModel.RGBICWW_STRIP_LIGHT,
SwitchbotModel.RGBICWW_FLOOR_LAMP,
SwitchbotModel.PLUG_MINI_EU,
SwitchbotModel.RELAY_SWITCH_2PM,
}
ENCRYPTED_SWITCHBOT_MODEL_TO_CLASS: dict[
@@ -140,6 +143,7 @@ ENCRYPTED_SWITCHBOT_MODEL_TO_CLASS: dict[
SwitchbotModel.RGBICWW_STRIP_LIGHT: switchbot.SwitchbotRgbicLight,
SwitchbotModel.RGBICWW_FLOOR_LAMP: switchbot.SwitchbotRgbicLight,
SwitchbotModel.PLUG_MINI_EU: switchbot.SwitchbotRelaySwitch,
SwitchbotModel.RELAY_SWITCH_2PM: switchbot.SwitchbotRelaySwitch2PM,
}
HASS_SENSOR_TYPE_TO_SWITCHBOT_MODEL = {

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

@@ -6,6 +6,7 @@ from collections.abc import Callable, Coroutine, Mapping
import logging
from typing import Any, Concatenate
import switchbot
from switchbot import Switchbot, SwitchbotDevice
from switchbot.devices.device import SwitchbotOperationError
@@ -46,6 +47,7 @@ class SwitchbotEntity(
model=coordinator.model, # Sometimes the modelName is missing from the advertisement data
name=coordinator.device_name,
)
self._channel: int | None = None
if ":" not in self._address:
# MacOS Bluetooth addresses are not mac addresses
return
@@ -60,6 +62,8 @@ class SwitchbotEntity(
@property
def parsed_data(self) -> dict[str, Any]:
"""Return parsed device data for this entity."""
if isinstance(self.coordinator.device, switchbot.SwitchbotRelaySwitch2PM):
return self.coordinator.device.get_parsed_data(self._channel)
return self.coordinator.device.parsed_data
@property

42
homeassistant/components/switchbot/sensor.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import switchbot
from switchbot import HumidifierWaterLevel
from switchbot.const.air_purifier import AirQualityLevel
@@ -25,8 +26,10 @@ from homeassistant.const import (
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotEntity
@@ -133,13 +136,22 @@ async def async_setup_entry(
) -> None:
"""Set up Switchbot sensor based on a config entry."""
coordinator = entry.runtime_data
entities = [
SwitchBotSensor(coordinator, sensor)
for sensor in coordinator.device.parsed_data
if sensor in SENSOR_TYPES
]
entities.append(SwitchbotRSSISensor(coordinator, "rssi"))
async_add_entities(entities)
sensor_entities: list[SensorEntity] = []
if isinstance(coordinator.device, switchbot.SwitchbotRelaySwitch2PM):
sensor_entities.extend(
SwitchBotSensor(coordinator, sensor, channel)
for channel in range(1, coordinator.device.channel + 1)
for sensor in coordinator.device.get_parsed_data(channel)
if sensor in SENSOR_TYPES
)
else:
sensor_entities.extend(
SwitchBotSensor(coordinator, sensor)
for sensor in coordinator.device.parsed_data
if sensor in SENSOR_TYPES
)
sensor_entities.append(SwitchbotRSSISensor(coordinator, "rssi"))
async_add_entities(sensor_entities)
class SwitchBotSensor(SwitchbotEntity, SensorEntity):
@@ -149,13 +161,27 @@ class SwitchBotSensor(SwitchbotEntity, SensorEntity):
self,
coordinator: SwitchbotDataUpdateCoordinator,
sensor: str,
channel: int | None = None,
) -> None:
"""Initialize the Switchbot sensor."""
super().__init__(coordinator)
self._sensor = sensor
self._attr_unique_id = f"{coordinator.base_unique_id}-{sensor}"
self._channel = channel
self.entity_description = SENSOR_TYPES[sensor]
if channel:
self._attr_unique_id = f"{coordinator.base_unique_id}-{sensor}-{channel}"
self._attr_device_info = DeviceInfo(
identifiers={
(DOMAIN, f"{coordinator.base_unique_id}-channel-{channel}")
},
manufacturer="SwitchBot",
model_id="RelaySwitch2PM",
name=f"{coordinator.device_name} Channel {channel}",
)
else:
self._attr_unique_id = f"{coordinator.base_unique_id}-{sensor}"
@property
def native_value(self) -> str | int | None:
"""Return the state of the sensor."""

63
homeassistant/components/switchbot/switch.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
import switchbot
@@ -9,13 +10,16 @@ import switchbot
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.const import STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from .const import DOMAIN
from .coordinator import SwitchbotConfigEntry, SwitchbotDataUpdateCoordinator
from .entity import SwitchbotSwitchedEntity
from .entity import SwitchbotSwitchedEntity, exception_handler
PARALLEL_UPDATES = 0
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
@@ -24,7 +28,16 @@ async def async_setup_entry(
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Switchbot based on a config entry."""
async_add_entities([SwitchBotSwitch(entry.runtime_data)])
coordinator = entry.runtime_data
if isinstance(coordinator.device, switchbot.SwitchbotRelaySwitch2PM):
entries = [
SwitchbotMultiChannelSwitch(coordinator, channel)
for channel in range(1, coordinator.device.channel + 1)
]
async_add_entities(entries)
else:
async_add_entities([SwitchBotSwitch(coordinator)])
class SwitchBotSwitch(SwitchbotSwitchedEntity, SwitchEntity, RestoreEntity):
@@ -67,3 +80,49 @@ class SwitchBotSwitch(SwitchbotSwitchedEntity, SwitchEntity, RestoreEntity):
**super().extra_state_attributes,
"switch_mode": self._device.switch_mode(),
}
class SwitchbotMultiChannelSwitch(SwitchbotSwitchedEntity, SwitchEntity):
"""Representation of a Switchbot multi-channel switch."""
_attr_device_class = SwitchDeviceClass.SWITCH
_device: switchbot.Switchbot
_attr_name = None
def __init__(
self, coordinator: SwitchbotDataUpdateCoordinator, channel: int
) -> None:
"""Initialize the Switchbot."""
super().__init__(coordinator)
self._channel = channel
self._attr_unique_id = f"{coordinator.base_unique_id}-{channel}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{coordinator.base_unique_id}-channel-{channel}")},
manufacturer="SwitchBot",
model_id="RelaySwitch2PM",
name=f"{coordinator.device_name} Channel {channel}",
)
@property
def is_on(self) -> bool | None:
"""Return true if device is on."""
return self._device.is_on(self._channel)
@exception_handler
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn device on."""
_LOGGER.debug(
"Turn Switchbot device on %s, channel %d", self._address, self._channel
)
await self._device.turn_on(self._channel)
self.async_write_ha_state()
@exception_handler
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn device off."""
_LOGGER.debug(
"Turn Switchbot device off %s, channel %d", self._address, self._channel
)
await self._device.turn_off(self._channel)
self.async_write_ha_state()

25
tests/components/switchbot/__init__.py
View File

@@ -1080,3 +1080,28 @@ PLUG_MINI_EU_SERVICE_INFO = BluetoothServiceInfoBleak(
connectable=True,
tx_power=-127,
)
RELAY_SWITCH_2PM_SERVICE_INFO = BluetoothServiceInfoBleak(
name="Relay Switch 2PM",
manufacturer_data={
2409: b"\xc0N0\xdd\xb9\xf2\x8a\xc1\x00\x00\x00\x00\x00F\x00\x00"
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"=\x00\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="Relay Switch 2PM",
manufacturer_data={
2409: b"\xc0N0\xdd\xb9\xf2\x8a\xc1\x00\x00\x00\x00\x00F\x00\x00"
},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"=\x00\x00\x00"},
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
),
device=generate_ble_device("AA:BB:CC:DD:EE:FF", "Relay Switch 2PM"),
time=0,
connectable=True,
tx_power=-127,
)

111
tests/components/switchbot/test_sensor.py
View File

@@ -29,6 +29,7 @@ from . import (
HUBMINI_MATTER_SERVICE_INFO,
LEAK_SERVICE_INFO,
PLUG_MINI_EU_SERVICE_INFO,
RELAY_SWITCH_2PM_SERVICE_INFO,
REMOTE_SERVICE_INFO,
WOHAND_SERVICE_INFO,
WOHUB2_SERVICE_INFO,
@@ -617,3 +618,113 @@ async def test_plug_mini_eu_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_relay_switch_2pm_sensor(hass: HomeAssistant) -> None:
"""Test setting up creates the relay switch 2PM sensor."""
await async_setup_component(hass, DOMAIN, {})
inject_bluetooth_service_info(hass, RELAY_SWITCH_2PM_SERVICE_INFO)
with patch(
"homeassistant.components.switchbot.switch.switchbot.SwitchbotRelaySwitch2PM.get_basic_info",
new=AsyncMock(
return_value={
1: {
"power": 4.9,
"current": 0.1,
"voltage": 25,
"energy": 0.2,
},
2: {
"power": 7.9,
"current": 0.6,
"voltage": 25,
"energy": 2.5,
},
}
),
):
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_ADDRESS: "aa:bb:cc:dd:ee:ff",
CONF_NAME: "test-name",
CONF_SENSOR_TYPE: "relay_switch_2pm",
CONF_KEY_ID: "ff",
CONF_ENCRYPTION_KEY: "ffffffffffffffffffffffffffffffff",
},
unique_id="aabbccddeeaa",
)
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")) == 9
power_sensor_1 = hass.states.get("sensor.test_name_channel_1_power")
power_sensor_attrs = power_sensor_1.attributes
assert power_sensor_1.state == "4.9"
assert power_sensor_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 1 Power"
assert power_sensor_attrs[ATTR_UNIT_OF_MEASUREMENT] == "W"
assert power_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
voltage_sensor_1 = hass.states.get("sensor.test_name_channel_1_voltage")
voltage_sensor_1_attrs = voltage_sensor_1.attributes
assert voltage_sensor_1.state == "25"
assert voltage_sensor_1_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 1 Voltage"
assert voltage_sensor_1_attrs[ATTR_UNIT_OF_MEASUREMENT] == "V"
assert voltage_sensor_1_attrs[ATTR_STATE_CLASS] == "measurement"
current_sensor_1 = hass.states.get("sensor.test_name_channel_1_current")
current_sensor_1_attrs = current_sensor_1.attributes
assert current_sensor_1.state == "0.1"
assert current_sensor_1_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 1 Current"
assert current_sensor_1_attrs[ATTR_UNIT_OF_MEASUREMENT] == "A"
assert current_sensor_1_attrs[ATTR_STATE_CLASS] == "measurement"
energy_sensor_1 = hass.states.get("sensor.test_name_channel_1_energy")
energy_sensor_1_attrs = energy_sensor_1.attributes
assert energy_sensor_1.state == "0.2"
assert energy_sensor_1_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 1 Energy"
assert energy_sensor_1_attrs[ATTR_UNIT_OF_MEASUREMENT] == "kWh"
assert energy_sensor_1_attrs[ATTR_STATE_CLASS] == "total_increasing"
power_sensor_2 = hass.states.get("sensor.test_name_channel_2_power")
power_sensor_2_attrs = power_sensor_2.attributes
assert power_sensor_2.state == "7.9"
assert power_sensor_2_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 2 Power"
assert power_sensor_2_attrs[ATTR_UNIT_OF_MEASUREMENT] == "W"
assert power_sensor_2_attrs[ATTR_STATE_CLASS] == "measurement"
voltage_sensor_2 = hass.states.get("sensor.test_name_channel_2_voltage")
voltage_sensor_2_attrs = voltage_sensor_2.attributes
assert voltage_sensor_2.state == "25"
assert voltage_sensor_2_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 2 Voltage"
assert voltage_sensor_2_attrs[ATTR_UNIT_OF_MEASUREMENT] == "V"
assert voltage_sensor_2_attrs[ATTR_STATE_CLASS] == "measurement"
current_sensor_2 = hass.states.get("sensor.test_name_channel_2_current")
current_sensor_2_attrs = current_sensor_2.attributes
assert current_sensor_2.state == "0.6"
assert current_sensor_2_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 2 Current"
assert current_sensor_2_attrs[ATTR_UNIT_OF_MEASUREMENT] == "A"
assert current_sensor_2_attrs[ATTR_STATE_CLASS] == "measurement"
energy_sensor_2 = hass.states.get("sensor.test_name_channel_2_energy")
energy_sensor_2_attrs = energy_sensor_2.attributes
assert energy_sensor_2.state == "2.5"
assert energy_sensor_2_attrs[ATTR_FRIENDLY_NAME] == "test-name Channel 2 Energy"
assert energy_sensor_2_attrs[ATTR_UNIT_OF_MEASUREMENT] == "kWh"
assert energy_sensor_2_attrs[ATTR_STATE_CLASS] == "total_increasing"
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 rssi_sensor_attrs[ATTR_STATE_CLASS] == "measurement"
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()

109
tests/components/switchbot/test_switch.py
View File

@@ -17,7 +17,11 @@ from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
from . import PLUG_MINI_EU_SERVICE_INFO, WOHAND_SERVICE_INFO
from . import (
PLUG_MINI_EU_SERVICE_INFO,
RELAY_SWITCH_2PM_SERVICE_INFO,
WOHAND_SERVICE_INFO,
)
from tests.common import MockConfigEntry, mock_restore_cache
from tests.components.bluetooth import inject_bluetooth_service_info
@@ -152,3 +156,106 @@ async def test_relay_switch_control(
)
mocked_instance.assert_awaited_once()
@pytest.mark.parametrize(
("service", "mock_method"),
[(SERVICE_TURN_ON, "turn_on"), (SERVICE_TURN_OFF, "turn_off")],
)
async def test_relay_switch_2pm_control(
hass: HomeAssistant,
mock_entry_encrypted_factory: Callable[[str], MockConfigEntry],
service: str,
mock_method: str,
) -> None:
"""Test Relay Switch 2PM control."""
inject_bluetooth_service_info(hass, RELAY_SWITCH_2PM_SERVICE_INFO)
entry = mock_entry_encrypted_factory(sensor_type="relay_switch_2pm")
entry.add_to_hass(hass)
mocked_instance = AsyncMock(return_value=True)
with patch.multiple(
"homeassistant.components.switchbot.switch.switchbot.SwitchbotRelaySwitch2PM",
update=AsyncMock(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_1 = "switch.test_name_channel_1"
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id_1},
blocking=True,
)
mocked_instance.assert_called_with(1)
entity_id_2 = "switch.test_name_channel_2"
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: entity_id_2},
blocking=True,
)
mocked_instance.assert_called_with(2)
@pytest.mark.parametrize(
("exception", "error_message"),
[
(
SwitchbotOperationError("Operation failed"),
"An error occurred while performing the action: Operation failed",
),
],
)
@pytest.mark.parametrize(
("service", "mock_method"),
[
(SERVICE_TURN_ON, "turn_on"),
(SERVICE_TURN_OFF, "turn_off"),
],
)
@pytest.mark.parametrize(
"entry_id",
[
"switch.test_name_channel_1",
"switch.test_name_channel_2",
],
)
async def test_relay_switch_2pm_exception(
hass: HomeAssistant,
mock_entry_encrypted_factory: Callable[[str], MockConfigEntry],
exception: Exception,
error_message: str,
service: str,
mock_method: str,
entry_id: str,
) -> None:
"""Test Relay Switch 2PM exception handling."""
inject_bluetooth_service_info(hass, RELAY_SWITCH_2PM_SERVICE_INFO)
entry = mock_entry_encrypted_factory(sensor_type="relay_switch_2pm")
entry.add_to_hass(hass)
with patch.multiple(
"homeassistant.components.switchbot.switch.switchbot.SwitchbotRelaySwitch2PM",
update=AsyncMock(return_value=None),
**{mock_method: AsyncMock(side_effect=exception)},
):
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError, match=error_message):
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: entry_id},
blocking=True,
)