Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-12-27 14:31:13 +00:00
Code Activity

Add support model [relay switch 2pm] for switchbot cloud (#148381)

Browse Source
This commit is contained in:
Samuel Xiao
2025-09-30 23:49:32 +08:00
committed by GitHub
parent aeadc0c4b0
commit dcb8d4f702
4 changed files with 263 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
homeassistant/components/switchbot_cloud/__init__.py
View File

@@ -187,6 +187,15 @@ async def make_device_data(
devices_data.buttons.append((device, coordinator))
else:
devices_data.switches.append((device, coordinator))
if isinstance(device, Device) and device.device_type in [
"Relay Switch 2PM",
]:
coordinator = await coordinator_for_device(
hass, entry, api, device, coordinators_by_id
)
devices_data.sensors.append((device, coordinator))
devices_data.switches.append((device, coordinator))
if isinstance(device, Device) and device.device_type.startswith("Air Purifier"):
coordinator = await coordinator_for_device(
hass, entry, api, device, coordinators_by_id

127
homeassistant/components/switchbot_cloud/sensor.py
View File

@@ -4,7 +4,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from switchbot_api import Device, SwitchBotAPI
from switchbot_api import Device, Remote, SwitchBotAPI
from homeassistant.components.sensor import (
SensorDeviceClass,
@@ -22,7 +22,8 @@ from homeassistant.const import (
UnitOfPower,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import SwitchbotCloudData
@@ -41,6 +42,12 @@ SENSOR_TYPE_USED_ELECTRICITY = "usedElectricity"
SENSOR_TYPE_LIGHTLEVEL = "lightLevel"
RELAY_SWITCH_2PM_SENSOR_TYPE_POWER = "Power"
RELAY_SWITCH_2PM_SENSOR_TYPE_VOLTAGE = "Voltage"
RELAY_SWITCH_2PM_SENSOR_TYPE_CURRENT = "ElectricCurrent"
RELAY_SWITCH_2PM_SENSOR_TYPE_ELECTRICITY = "UsedElectricity"
@dataclass(frozen=True, kw_only=True)
class SwitchbotCloudSensorEntityDescription(SensorEntityDescription):
"""Plug Mini Eu UsedElectricity Sensor EntityDescription."""
@@ -113,6 +120,34 @@ CO2_DESCRIPTION = SensorEntityDescription(
native_unit_of_measurement=CONCENTRATION_PARTS_PER_MILLION,
)
RELAY_SWITCH_2PM_POWER_DESCRIPTION = SensorEntityDescription(
key=RELAY_SWITCH_2PM_SENSOR_TYPE_POWER,
device_class=SensorDeviceClass.POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.WATT,
)
RELAY_SWITCH_2PM_VOLTAGE_DESCRIPTION = SensorEntityDescription(
key=RELAY_SWITCH_2PM_SENSOR_TYPE_VOLTAGE,
device_class=SensorDeviceClass.VOLTAGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
)
RELAY_SWITCH_2PM_CURRENT_DESCRIPTION = SensorEntityDescription(
key=RELAY_SWITCH_2PM_SENSOR_TYPE_CURRENT,
device_class=SensorDeviceClass.CURRENT,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.MILLIAMPERE,
)
RELAY_SWITCH_2PM_ElECTRICITY_DESCRIPTION = SensorEntityDescription(
key=RELAY_SWITCH_2PM_SENSOR_TYPE_ELECTRICITY,
device_class=SensorDeviceClass.ENERGY,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
)
LIGHTLEVEL_DESCRIPTION = SensorEntityDescription(
key="lightLevel",
translation_key="light_level",
@@ -175,6 +210,12 @@ SENSOR_DESCRIPTIONS_BY_DEVICE_TYPES = {
"Smart Lock Lite": (BATTERY_DESCRIPTION,),
"Smart Lock Pro": (BATTERY_DESCRIPTION,),
"Smart Lock Ultra": (BATTERY_DESCRIPTION,),
"Relay Switch 2PM": (
RELAY_SWITCH_2PM_POWER_DESCRIPTION,
RELAY_SWITCH_2PM_VOLTAGE_DESCRIPTION,
RELAY_SWITCH_2PM_CURRENT_DESCRIPTION,
RELAY_SWITCH_2PM_ElECTRICITY_DESCRIPTION,
),
"Curtain": (BATTERY_DESCRIPTION,),
"Curtain3": (BATTERY_DESCRIPTION,),
"Roller Shade": (BATTERY_DESCRIPTION,),
@@ -203,12 +244,25 @@ async def async_setup_entry(
) -> None:
"""Set up SwitchBot Cloud entry."""
data: SwitchbotCloudData = hass.data[DOMAIN][config.entry_id]
async_add_entities(
SwitchBotCloudSensor(data.api, device, coordinator, description)
for device, coordinator in data.devices.sensors
for description in SENSOR_DESCRIPTIONS_BY_DEVICE_TYPES[device.device_type]
)
entities: list[SwitchBotCloudSensor] = []
for device, coordinator in data.devices.sensors:
for description in SENSOR_DESCRIPTIONS_BY_DEVICE_TYPES[device.device_type]:
if device.device_type == "Relay Switch 2PM":
entities.append(
SwitchBotCloudRelaySwitch2PMSensor(
data.api, device, coordinator, description, "1"
)
)
entities.append(
SwitchBotCloudRelaySwitch2PMSensor(
data.api, device, coordinator, description, "2"
)
)
else:
entities.append(
_async_make_entity(data.api, device, coordinator, description)
)
async_add_entities(entities)
class SwitchBotCloudSensor(SwitchBotCloudEntity, SensorEntity):
@@ -230,14 +284,49 @@ class SwitchBotCloudSensor(SwitchBotCloudEntity, SensorEntity):
"""Set attributes from coordinator data."""
if not self.coordinator.data:
return
if isinstance(
self.entity_description,
SwitchbotCloudSensorEntityDescription,
):
self._attr_native_value = self.entity_description.value_fn(
self.coordinator.data
)
else:
self._attr_native_value = self.coordinator.data.get(
self.entity_description.key
)
self._attr_native_value = self.coordinator.data.get(self.entity_description.key)
class SwitchBotCloudRelaySwitch2PMSensor(SwitchBotCloudSensor):
"""Representation of a SwitchBot Cloud Relay Switch 2PM sensor entity."""
def __init__(
self,
api: SwitchBotAPI,
device: Device,
coordinator: SwitchBotCoordinator,
description: SensorEntityDescription,
channel: str,
) -> None:
"""Initialize SwitchBot Cloud sensor entity."""
super().__init__(api, device, coordinator, description)
self.entity_description = description
self._channel = channel
self._attr_unique_id = f"{device.device_id}-{description.key}-{channel}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{device.device_name}-channel-{channel}")},
manufacturer="SwitchBot",
model=device.device_type,
model_id="RelaySwitch2PM",
name=f"{device.device_name} Channel {channel}",
)
def _set_attributes(self) -> None:
"""Set attributes from coordinator data."""
if not self.coordinator.data:
return
self._attr_native_value = self.coordinator.data.get(
f"switch{self._channel}{self.entity_description.key.strip()}"
)
@callback
def _async_make_entity(
api: SwitchBotAPI,
device: Device | Remote,
coordinator: SwitchBotCoordinator,
description: SensorEntityDescription,
) -> SwitchBotCloudSensor:
"""Make a SwitchBotCloudSensor or SwitchBotCloudRelaySwitch2PMSensor."""
return SwitchBotCloudSensor(api, device, coordinator, description)

70
homeassistant/components/switchbot_cloud/switch.py
View File

@@ -1,5 +1,6 @@
"""Support for SwitchBot switch."""
import asyncio
from typing import Any
from switchbot_api import CommonCommands, Device, PowerState, Remote, SwitchBotAPI
@@ -7,10 +8,11 @@ from switchbot_api import CommonCommands, Device, PowerState, Remote, SwitchBotA
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import SwitchbotCloudData
from .const import DOMAIN
from .const import AFTER_COMMAND_REFRESH, DOMAIN
from .coordinator import SwitchBotCoordinator
from .entity import SwitchBotCloudEntity
@@ -22,10 +24,19 @@ async def async_setup_entry(
) -> None:
"""Set up SwitchBot Cloud entry."""
data: SwitchbotCloudData = hass.data[DOMAIN][config.entry_id]
async_add_entities(
_async_make_entity(data.api, device, coordinator)
for device, coordinator in data.devices.switches
)
entities: list[SwitchBotCloudSwitch] = []
for device, coordinator in data.devices.switches:
if device.device_type == "Relay Switch 2PM":
entities.append(
SwitchBotCloudRelaySwitch2PMSwitch(data.api, device, coordinator, "1")
)
entities.append(
SwitchBotCloudRelaySwitch2PMSwitch(data.api, device, coordinator, "2")
)
else:
entities.append(_async_make_entity(data.api, device, coordinator))
async_add_entities(entities)
class SwitchBotCloudSwitch(SwitchBotCloudEntity, SwitchEntity):
@@ -76,6 +87,54 @@ class SwitchBotCloudRelaySwitchSwitch(SwitchBotCloudSwitch):
self._attr_is_on = self.coordinator.data.get("switchStatus") == 1
class SwitchBotCloudRelaySwitch2PMSwitch(SwitchBotCloudSwitch):
"""Representation of a SwitchBot relay switch."""
def __init__(
self,
api: SwitchBotAPI,
device: Device | Remote,
coordinator: SwitchBotCoordinator,
channel: str,
) -> None:
"""Init SwitchBotCloudRelaySwitch2PMSwitch."""
super().__init__(api, device, coordinator)
self._channel = channel
self._device_id = device.device_id
self._attr_unique_id = f"{device.device_id}-{channel}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"{device.device_name}-channel-{channel}")},
manufacturer="SwitchBot",
model=device.device_type,
model_id="RelaySwitch2PM",
name=f"{device.device_name} Channel {channel}",
)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
await self._api.send_command(
self._device_id, command=CommonCommands.ON, parameters=self._channel
)
await asyncio.sleep(AFTER_COMMAND_REFRESH)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
await self._api.send_command(
self._device_id, command=CommonCommands.OFF, parameters=self._channel
)
await asyncio.sleep(AFTER_COMMAND_REFRESH)
await self.coordinator.async_request_refresh()
def _set_attributes(self) -> None:
"""Set attributes from coordinator data."""
if self.coordinator.data is None:
return
self._attr_is_on = (
self.coordinator.data.get(f"switch{self._channel}Status") == 1
)
@callback
def _async_make_entity(
api: SwitchBotAPI, device: Device | Remote, coordinator: SwitchBotCoordinator
@@ -89,4 +148,5 @@ def _async_make_entity(
return SwitchBotCloudPlugSwitch(api, device, coordinator)
if "Bot" in device.device_type:
return SwitchBotCloudSwitch(api, device, coordinator)
raise NotImplementedError(f"Unsupported device type: {device.device_type}")

81
tests/components/switchbot_cloud/test_switch.py
View File

@@ -13,6 +13,7 @@ from homeassistant.const import (
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
@@ -108,3 +109,83 @@ async def test_pressmode_bot_no_switch_entity(
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
assert not hass.states.async_entity_ids(SWITCH_DOMAIN)
async def test_switch_relay_2pm_turn_on(
hass: HomeAssistant, mock_list_devices, mock_get_status
) -> None:
"""Test switch relay 2pm turn on."""
mock_list_devices.return_value = [
Device(
version="V1.0",
deviceId="relay-switch-id-1",
deviceName="relay-switch-1",
deviceType="Relay Switch 2PM",
hubDeviceId="test-hub-id",
),
]
mock_get_status.return_value = {"switchStatus": 0}
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
entity_id = "switch.relay_switch_1_channel_1"
assert hass.states.get(entity_id).state == STATE_OFF
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: entity_id}, blocking=True
)
mock_send_command.assert_called_once()
async def test_switch_relay_2pm_turn_off(
hass: HomeAssistant, mock_list_devices, mock_get_status
) -> None:
"""Test switch relay 2pm turn off."""
mock_list_devices.return_value = [
Device(
version="V1.0",
deviceId="relay-switch-id-1",
deviceName="relay-switch-1",
deviceType="Relay Switch 2PM",
hubDeviceId="test-hub-id",
),
]
mock_get_status.return_value = {"switchStatus": 0}
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
entity_id = "switch.relay_switch_1_channel_1"
assert hass.states.get(entity_id).state == STATE_OFF
with patch.object(SwitchBotAPI, "send_command") as mock_send_command:
await hass.services.async_call(
SWITCH_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: entity_id}, blocking=True
)
mock_send_command.assert_called_once()
async def test_switch_relay_2pm_coordination_is_none(
hass: HomeAssistant, mock_list_devices, mock_get_status
) -> None:
"""Test switch relay 2pm coordination is none."""
mock_list_devices.return_value = [
Device(
version="V1.0",
deviceId="relay-switch-id-1",
deviceName="relay-switch-1",
deviceType="Relay Switch 2PM",
hubDeviceId="test-hub-id",
),
]
mock_get_status.return_value = None
entry = await configure_integration(hass)
assert entry.state is ConfigEntryState.LOADED
entity_id = "switch.relay_switch_1_channel_1"
assert hass.states.get(entity_id).state == STATE_UNKNOWN