evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #56472 from home-assistant/shelly-gen2

Browse source
This commit is contained in:
Paulus Schoutsen 2021-09-20 20:19:19 -07:00 • committed by GitHub
commit 9059ee6604
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 1527 additions and 351 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

293
homeassistant/components/shelly/__init__.py
View file

@ -7,6 +7,8 @@ import logging
from typing import Any, Final, cast
import aioshelly
from aioshelly.block_device import BlockDevice
from aioshelly.rpc_device import RpcDevice
import async_timeout
import voluptuous as vol
@ -29,8 +31,9 @@ from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
ATTR_GENERATION,
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION,
COAP,
BLOCK,
CONF_COAP_PORT,
DATA_CONFIG_ENTRY,
DEFAULT_COAP_PORT,
@ -41,14 +44,24 @@ from .const import (
POLLING_TIMEOUT_SEC,
REST,
REST_SENSORS_UPDATE_INTERVAL,
RPC,
RPC_INPUTS_EVENTS_TYPES,
RPC_RECONNECT_INTERVAL,
SHBTN_MODELS,
SLEEP_PERIOD_MULTIPLIER,
UPDATE_PERIOD_MULTIPLIER,
)
from .utils import get_coap_context, get_device_name, get_device_sleep_period
from .utils import (
get_block_device_name,
get_block_device_sleep_period,
get_coap_context,
get_device_entry_gen,
get_rpc_device_name,
)
PLATFORMS: Final = ["binary_sensor", "cover", "light", "sensor", "switch"]
SLEEPING_PLATFORMS: Final = ["binary_sensor", "sensor"]
BLOCK_PLATFORMS: Final = ["binary_sensor", "cover", "light", "sensor", "switch"]
BLOCK_SLEEPING_PLATFORMS: Final = ["binary_sensor", "sensor"]
RPC_PLATFORMS: Final = ["binary_sensor", "light", "sensor", "switch"]
_LOGGER: Final = logging.getLogger(__name__)
COAP_SCHEMA: Final = vol.Schema(
@ -87,9 +100,17 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id] = {}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][DEVICE] = None
if get_device_entry_gen(entry) == 2:
return await async_setup_rpc_entry(hass, entry)
return await async_setup_block_entry(hass, entry)
async def async_setup_block_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Shelly block based device from a config entry."""
temperature_unit = "C" if hass.config.units.is_metric else "F"
options = aioshelly.ConnectionOptions(
options = aioshelly.common.ConnectionOptions(
entry.data[CONF_HOST],
entry.data.get(CONF_USERNAME),
entry.data.get(CONF_PASSWORD),
@ -98,14 +119,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
coap_context = await get_coap_context(hass)
device = await aioshelly.Device.create(
device = await BlockDevice.create(
aiohttp_client.async_get_clientsession(hass),
coap_context,
options,
False,
)
dev_reg = await device_registry.async_get_registry(hass)
dev_reg = device_registry.async_get(hass)
device_entry = None
if entry.unique_id is not None:
device_entry = dev_reg.async_get_device(
@ -123,22 +144,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if sleep_period is None:
data = {**entry.data}
data["sleep_period"] = get_device_sleep_period(device.settings)
data["sleep_period"] = get_block_device_sleep_period(device.settings)
data["model"] = device.settings["device"]["type"]
hass.config_entries.async_update_entry(entry, data=data)
hass.async_create_task(async_device_setup(hass, entry, device))
hass.async_create_task(async_block_device_setup(hass, entry, device))
if sleep_period == 0:
# Not a sleeping device, finish setup
_LOGGER.debug("Setting up online device %s", entry.title)
_LOGGER.debug("Setting up online block device %s", entry.title)
try:
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
await device.initialize(True)
await device.initialize()
except (asyncio.TimeoutError, OSError) as err:
raise ConfigEntryNotReady from err
await async_device_setup(hass, entry, device)
await async_block_device_setup(hass, entry, device)
elif sleep_period is None or device_entry is None:
# Need to get sleep info or first time sleeping device setup, wait for device
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][DEVICE] = device
@ -146,40 +167,66 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"Setup for device %s will resume when device is online", entry.title
)
device.subscribe_updates(_async_device_online)
await device.coap_request("s")
else:
# Restore sensors for sleeping device
_LOGGER.debug("Setting up offline device %s", entry.title)
await async_device_setup(hass, entry, device)
_LOGGER.debug("Setting up offline block device %s", entry.title)
await async_block_device_setup(hass, entry, device)
return True
async def async_device_setup(
hass: HomeAssistant, entry: ConfigEntry, device: aioshelly.Device
async def async_block_device_setup(
hass: HomeAssistant, entry: ConfigEntry, device: BlockDevice
) -> None:
"""Set up a device that is online."""
"""Set up a block based device that is online."""
device_wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][
COAP
] = ShellyDeviceWrapper(hass, entry, device)
await device_wrapper.async_setup()
BLOCK
] = BlockDeviceWrapper(hass, entry, device)
device_wrapper.async_setup()
platforms = SLEEPING_PLATFORMS
platforms = BLOCK_SLEEPING_PLATFORMS
if not entry.data.get("sleep_period"):
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][
REST
] = ShellyDeviceRestWrapper(hass, device)
platforms = PLATFORMS
platforms = BLOCK_PLATFORMS
hass.config_entries.async_setup_platforms(entry, platforms)
class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
"""Wrapper for a Shelly device with Home Assistant specific functions."""
async def async_setup_rpc_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Shelly RPC based device from a config entry."""
options = aioshelly.common.ConnectionOptions(
entry.data[CONF_HOST],
entry.data.get(CONF_USERNAME),
entry.data.get(CONF_PASSWORD),
)
_LOGGER.debug("Setting up online RPC device %s", entry.title)
try:
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
device = await RpcDevice.create(
aiohttp_client.async_get_clientsession(hass), options
)
except (asyncio.TimeoutError, OSError) as err:
raise ConfigEntryNotReady from err
device_wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][
RPC
] = RpcDeviceWrapper(hass, entry, device)
device_wrapper.async_setup()
hass.config_entries.async_setup_platforms(entry, RPC_PLATFORMS)
return True
class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator):
"""Wrapper for a Shelly block based device with Home Assistant specific functions."""
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, device: aioshelly.Device
self, hass: HomeAssistant, entry: ConfigEntry, device: BlockDevice
) -> None:
"""Initialize the Shelly device wrapper."""
self.device_id: str | None = None
@ -192,7 +239,9 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
UPDATE_PERIOD_MULTIPLIER * device.settings["coiot"]["update_period"]
)
device_name = get_device_name(device) if device.initialized else entry.title
device_name = (
get_block_device_name(device) if device.initialized else entry.title
)
super().__init__(
hass,
_LOGGER,
@ -203,12 +252,14 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
self.entry = entry
self.device = device
self._async_remove_device_updates_handler = self.async_add_listener(
self._async_device_updates_handler
entry.async_on_unload(
self.async_add_listener(self._async_device_updates_handler)
)
self._last_input_events_count: dict = {}
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
)
@callback
def _async_device_updates_handler(self) -> None:
@ -216,6 +267,8 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
if not self.device.initialized:
return
assert self.device.blocks
# For buttons which are battery powered - set initial value for last_event_count
if self.model in SHBTN_MODELS and self._last_input_events_count.get(1) is None:
for block in self.device.blocks:
@ -255,6 +308,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
ATTR_DEVICE: self.device.settings["device"]["hostname"],
ATTR_CHANNEL: channel,
ATTR_CLICK_TYPE: INPUTS_EVENTS_DICT[event_type],
ATTR_GENERATION: 1,
},
)
else:
@ -270,7 +324,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
# Sleeping device, no point polling it, just mark it unavailable
raise update_coordinator.UpdateFailed("Sleeping device did not update")
_LOGGER.debug("Polling Shelly Device - %s", self.name)
_LOGGER.debug("Polling Shelly Block Device - %s", self.name)
try:
async with async_timeout.timeout(POLLING_TIMEOUT_SEC):
await self.device.update()
@ -287,18 +341,16 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
"""Mac address of the device."""
return cast(str, self.entry.unique_id)
async def async_setup(self) -> None:
def async_setup(self) -> None:
"""Set up the wrapper."""
dev_reg = await device_registry.async_get_registry(self.hass)
sw_version = self.device.settings["fw"] if self.device.initialized else ""
dev_reg = device_registry.async_get(self.hass)
sw_version = self.device.firmware_version if self.device.initialized else ""
entry = dev_reg.async_get_or_create(
config_entry_id=self.entry.entry_id,
name=self.name,
connections={(device_registry.CONNECTION_NETWORK_MAC, self.mac)},
# This is duplicate but otherwise via_device can't work
identifiers={(DOMAIN, self.mac)},
manufacturer="Shelly",
model=aioshelly.MODEL_NAMES.get(self.model, self.model),
model=aioshelly.const.MODEL_NAMES.get(self.model, self.model),
sw_version=sw_version,
)
self.device_id = entry.id
@ -306,22 +358,19 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
def shutdown(self) -> None:
"""Shutdown the wrapper."""
if self.device:
self.device.shutdown()
self._async_remove_device_updates_handler()
self.device = None
self.device.shutdown()
@callback
def _handle_ha_stop(self, _event: Event) -> None:
"""Handle Home Assistant stopping."""
_LOGGER.debug("Stopping ShellyDeviceWrapper for %s", self.name)
_LOGGER.debug("Stopping BlockDeviceWrapper for %s", self.name)
self.shutdown()
class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
"""Rest Wrapper for a Shelly device with Home Assistant specific functions."""
def __init__(self, hass: HomeAssistant, device: aioshelly.Device) -> None:
def __init__(self, hass: HomeAssistant, device: BlockDevice) -> None:
"""Initialize the Shelly device wrapper."""
if (
device.settings["device"]["type"]
@ -336,7 +385,7 @@ class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
super().__init__(
hass,
_LOGGER,
name=get_device_name(device),
name=get_block_device_name(device),
update_interval=timedelta(seconds=update_interval),
)
self.device = device
@ -358,39 +407,171 @@ class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
if get_device_entry_gen(entry) == 2:
unload_ok = await hass.config_entries.async_unload_platforms(
entry, RPC_PLATFORMS
)
if unload_ok:
await hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][RPC].shutdown()
hass.data[DOMAIN][DATA_CONFIG_ENTRY].pop(entry.entry_id)
return unload_ok
device = hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id].get(DEVICE)
if device is not None:
# If device is present, device wrapper is not setup yet
device.shutdown()
return True
platforms = SLEEPING_PLATFORMS
platforms = BLOCK_SLEEPING_PLATFORMS
if not entry.data.get("sleep_period"):
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][REST] = None
platforms = PLATFORMS
platforms = BLOCK_PLATFORMS
unload_ok = await hass.config_entries.async_unload_platforms(entry, platforms)
if unload_ok:
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][COAP].shutdown()
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][BLOCK].shutdown()
hass.data[DOMAIN][DATA_CONFIG_ENTRY].pop(entry.entry_id)
return unload_ok
def get_device_wrapper(
def get_block_device_wrapper(
hass: HomeAssistant, device_id: str
) -> ShellyDeviceWrapper | None:
"""Get a Shelly device wrapper for the given device id."""
) -> BlockDeviceWrapper | None:
"""Get a Shelly block device wrapper for the given device id."""
if not hass.data.get(DOMAIN):
return None
for config_entry in hass.data[DOMAIN][DATA_CONFIG_ENTRY]:
wrapper: ShellyDeviceWrapper | None = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry
].get(COAP)
dev_reg = device_registry.async_get(hass)
if device := dev_reg.async_get(device_id):
for config_entry in device.config_entries:
if not hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry):
continue
if wrapper and wrapper.device_id == device_id:
return wrapper
if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(BLOCK):
return cast(BlockDeviceWrapper, wrapper)
return None
def get_rpc_device_wrapper(
hass: HomeAssistant, device_id: str
) -> RpcDeviceWrapper | None:
"""Get a Shelly RPC device wrapper for the given device id."""
if not hass.data.get(DOMAIN):
return None
dev_reg = device_registry.async_get(hass)
if device := dev_reg.async_get(device_id):
for config_entry in device.config_entries:
if not hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry):
continue
if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(RPC):
return cast(RpcDeviceWrapper, wrapper)
return None
class RpcDeviceWrapper(update_coordinator.DataUpdateCoordinator):
"""Wrapper for a Shelly RPC based device with Home Assistant specific functions."""
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, device: RpcDevice
) -> None:
"""Initialize the Shelly device wrapper."""
self.device_id: str | None = None
device_name = get_rpc_device_name(device) if device.initialized else entry.title
super().__init__(
hass,
_LOGGER,
name=device_name,
update_interval=timedelta(seconds=RPC_RECONNECT_INTERVAL),
)
self.entry = entry
self.device = device
entry.async_on_unload(
self.async_add_listener(self._async_device_updates_handler)
)
self._last_event: dict[str, Any] | None = None
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
)
@callback
def _async_device_updates_handler(self) -> None:
"""Handle device updates."""
if (
not self.device.initialized
or not self.device.event
or self.device.event == self._last_event
):
return
self._last_event = self.device.event
for event in self.device.event["events"]:
if event.get("event") not in RPC_INPUTS_EVENTS_TYPES:
continue
self.hass.bus.async_fire(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: self.device_id,
ATTR_DEVICE: self.device.hostname,
ATTR_CHANNEL: event["id"] + 1,
ATTR_CLICK_TYPE: event["event"],
ATTR_GENERATION: 2,
},
)
async def _async_update_data(self) -> None:
"""Fetch data."""
if self.device.connected:
return
try:
_LOGGER.debug("Reconnecting to Shelly RPC Device - %s", self.name)
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
await self.device.initialize()
except OSError as err:
raise update_coordinator.UpdateFailed("Device disconnected") from err
@property
def model(self) -> str:
"""Model of the device."""
return cast(str, self.entry.data["model"])
@property
def mac(self) -> str:
"""Mac address of the device."""
return cast(str, self.entry.unique_id)
def async_setup(self) -> None:
"""Set up the wrapper."""
dev_reg = device_registry.async_get(self.hass)
sw_version = self.device.firmware_version if self.device.initialized else ""
entry = dev_reg.async_get_or_create(
config_entry_id=self.entry.entry_id,
name=self.name,
connections={(device_registry.CONNECTION_NETWORK_MAC, self.mac)},
manufacturer="Shelly",
model=aioshelly.const.MODEL_NAMES.get(self.model, self.model),
sw_version=sw_version,
)
self.device_id = entry.id
self.device.subscribe_updates(self.async_set_updated_data)
async def shutdown(self) -> None:
"""Shutdown the wrapper."""
await self.device.shutdown()
async def _handle_ha_stop(self, _event: Event) -> None:
"""Handle Home Assistant stopping."""
_LOGGER.debug("Stopping RpcDeviceWrapper for %s", self.name)
await self.shutdown()

86
homeassistant/components/shelly/binary_sensor.py
View file

@ -24,13 +24,20 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import (
BlockAttributeDescription,
RestAttributeDescription,
RpcAttributeDescription,
ShellyBlockAttributeEntity,
ShellyRestAttributeEntity,
ShellyRpcAttributeEntity,
ShellySleepingBlockAttributeEntity,
async_setup_entry_attribute_entities,
async_setup_entry_rest,
async_setup_entry_rpc,
)
from .utils import (
get_device_entry_gen,
is_block_momentary_input,
is_rpc_momentary_input,
)
from .utils import is_momentary_input
SENSORS: Final = {
("device", "overtemp"): BlockAttributeDescription(
@ -48,7 +55,7 @@ SENSORS: Final = {
("sensor", "dwIsOpened"): BlockAttributeDescription(
name="Door",
device_class=DEVICE_CLASS_OPENING,
available=lambda block: cast(bool, block.dwIsOpened != -1),
available=lambda block: cast(int, block.dwIsOpened) != -1,
),
("sensor", "flood"): BlockAttributeDescription(
name="Flood", device_class=DEVICE_CLASS_MOISTURE
@ -69,19 +76,19 @@ SENSORS: Final = {
name="Input",
device_class=DEVICE_CLASS_POWER,
default_enabled=False,
removal_condition=is_momentary_input,
removal_condition=is_block_momentary_input,
),
("relay", "input"): BlockAttributeDescription(
name="Input",
device_class=DEVICE_CLASS_POWER,
default_enabled=False,
removal_condition=is_momentary_input,
removal_condition=is_block_momentary_input,
),
("device", "input"): BlockAttributeDescription(
name="Input",
device_class=DEVICE_CLASS_POWER,
default_enabled=False,
removal_condition=is_momentary_input,
removal_condition=is_block_momentary_input,
),
("sensor", "extInput"): BlockAttributeDescription(
name="External Input",
@ -112,6 +119,41 @@ REST_SENSORS: Final = {
),
}
RPC_SENSORS: Final = {
"input": RpcAttributeDescription(
key="input",
name="Input",
value=lambda status, _: status["state"],
device_class=DEVICE_CLASS_POWER,
default_enabled=False,
removal_condition=is_rpc_momentary_input,
),
"cloud": RpcAttributeDescription(
key="cloud",
name="Cloud",
value=lambda status, _: status["connected"],
device_class=DEVICE_CLASS_CONNECTIVITY,
default_enabled=False,
),
"fwupdate": RpcAttributeDescription(
key="sys",
name="Firmware Update",
device_class=DEVICE_CLASS_UPDATE,
value=lambda status, _: status["available_updates"],
default_enabled=False,
extra_state_attributes=lambda status: {
"latest_stable_version": status["available_updates"].get(
"stable",
{"version": ""},
)["version"],
"beta_version": status["available_updates"].get(
"beta",
{"version": ""},
)["version"],
},
),
}
async def async_setup_entry(
hass: HomeAssistant,
@ -119,29 +161,34 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensors for device."""
if get_device_entry_gen(config_entry) == 2:
return await async_setup_entry_rpc(
hass, config_entry, async_add_entities, RPC_SENSORS, RpcBinarySensor
)
if config_entry.data["sleep_period"]:
await async_setup_entry_attribute_entities(
hass,
config_entry,
async_add_entities,
SENSORS,
ShellySleepingBinarySensor,
BlockSleepingBinarySensor,
)
else:
await async_setup_entry_attribute_entities(
hass, config_entry, async_add_entities, SENSORS, ShellyBinarySensor
hass, config_entry, async_add_entities, SENSORS, BlockBinarySensor
)
await async_setup_entry_rest(
hass,
config_entry,
async_add_entities,
REST_SENSORS,
ShellyRestBinarySensor,
RestBinarySensor,
)
class ShellyBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity):
"""Shelly binary sensor entity."""
class BlockBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity):
"""Represent a block binary sensor entity."""
@property
def is_on(self) -> bool:
@ -149,8 +196,8 @@ class ShellyBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity):
return bool(self.attribute_value)
class ShellyRestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity):
"""Shelly REST binary sensor entity."""
class RestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity):
"""Represent a REST binary sensor entity."""
@property
def is_on(self) -> bool:
@ -158,10 +205,17 @@ class ShellyRestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity):
return bool(self.attribute_value)
class ShellySleepingBinarySensor(
ShellySleepingBlockAttributeEntity, BinarySensorEntity
):
"""Represent a shelly sleeping binary sensor."""
class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity):
"""Represent a RPC binary sensor entity."""
@property
def is_on(self) -> bool:
"""Return true if RPC sensor state is on."""
return bool(self.attribute_value)
class BlockSleepingBinarySensor(ShellySleepingBlockAttributeEntity, BinarySensorEntity):
"""Represent a block sleeping binary sensor."""
@property
def is_on(self) -> bool:

113
homeassistant/components/shelly/config_flow.py
View file

@ -3,26 +3,37 @@ from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Final, cast
from typing import Any, Final
import aiohttp
import aioshelly
from aioshelly.block_device import BlockDevice
from aioshelly.rpc_device import RpcDevice
import async_timeout
import voluptuous as vol
from homeassistant import config_entries, core
from homeassistant import config_entries
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_USERNAME,
HTTP_UNAUTHORIZED,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import AIOSHELLY_DEVICE_TIMEOUT_SEC, DOMAIN
from .utils import get_coap_context, get_device_sleep_period
from .utils import (
get_block_device_name,
get_block_device_sleep_period,
get_coap_context,
get_info_auth,
get_info_gen,
get_model_name,
get_rpc_device_name,
)
_LOGGER: Final = logging.getLogger(__name__)
@ -32,34 +43,49 @@ HTTP_CONNECT_ERRORS: Final = (asyncio.TimeoutError, aiohttp.ClientError)
async def validate_input(
hass: core.HomeAssistant, host: str, data: dict[str, Any]
hass: HomeAssistant,
host: str,
info: dict[str, Any],
data: dict[str, Any],
) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
Data has the keys from HOST_SCHEMA with values provided by the user.
"""
options = aioshelly.ConnectionOptions(
host, data.get(CONF_USERNAME), data.get(CONF_PASSWORD)
options = aioshelly.common.ConnectionOptions(
host,
data.get(CONF_USERNAME),
data.get(CONF_PASSWORD),
)
coap_context = await get_coap_context(hass)
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
device = await aioshelly.Device.create(
if get_info_gen(info) == 2:
rpc_device = await RpcDevice.create(
aiohttp_client.async_get_clientsession(hass),
options,
)
await rpc_device.shutdown()
return {
"title": get_rpc_device_name(rpc_device),
"sleep_period": 0,
"model": rpc_device.model,
"gen": 2,
}
# Gen1
coap_context = await get_coap_context(hass)
block_device = await BlockDevice.create(
aiohttp_client.async_get_clientsession(hass),
coap_context,
options,
)
device.shutdown()
# Return info that you want to store in the config entry.
return {
"title": device.settings["name"],
"hostname": device.settings["device"]["hostname"],
"sleep_period": get_device_sleep_period(device.settings),
"model": device.settings["device"]["type"],
}
block_device.shutdown()
return {
"title": get_block_device_name(block_device),
"sleep_period": get_block_device_sleep_period(block_device.settings),
"model": block_device.model,
"gen": 1,
}
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@ -79,23 +105,25 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if user_input is not None:
host: str = user_input[CONF_HOST]
try:
info = await self._async_get_info(host)
self.info = await self._async_get_info(host)
except HTTP_CONNECT_ERRORS:
errors["base"] = "cannot_connect"
except aioshelly.FirmwareUnsupported:
except aioshelly.exceptions.FirmwareUnsupported:
return self.async_abort(reason="unsupported_firmware")
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(info["mac"])
await self.async_set_unique_id(self.info["mac"])
self._abort_if_unique_id_configured({CONF_HOST: host})
self.host = host
if info["auth"]:
if get_info_auth(self.info):
return await self.async_step_credentials()
try:
device_info = await validate_input(self.hass, self.host, {})
device_info = await validate_input(
self.hass, self.host, self.info, {}
)
except HTTP_CONNECT_ERRORS:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
@ -103,11 +131,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=device_info["title"] or device_info["hostname"],
title=device_info["title"],
data={
**user_input,
"sleep_period": device_info["sleep_period"],
"model": device_info["model"],
"gen": device_info["gen"],
},
)
@ -122,7 +151,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
if user_input is not None:
try:
device_info = await validate_input(self.hass, self.host, user_input)
device_info = await validate_input(
self.hass, self.host, self.info, user_input
)
except aiohttp.ClientResponseError as error:
if error.status == HTTP_UNAUTHORIZED:
errors["base"] = "invalid_auth"
@ -135,12 +166,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "unknown"
else:
return self.async_create_entry(
title=device_info["title"] or device_info["hostname"],
title=device_info["title"],
data={
**user_input,
CONF_HOST: self.host,
"sleep_period": device_info["sleep_period"],
"model": device_info["model"],
"gen": device_info["gen"],
},
)
else:
@ -162,13 +194,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
) -> FlowResult:
"""Handle zeroconf discovery."""
try:
self.info = info = await self._async_get_info(discovery_info["host"])
self.info = await self._async_get_info(discovery_info["host"])
except HTTP_CONNECT_ERRORS:
return self.async_abort(reason="cannot_connect")
except aioshelly.FirmwareUnsupported:
except aioshelly.exceptions.FirmwareUnsupported:
return self.async_abort(reason="unsupported_firmware")
await self.async_set_unique_id(info["mac"])
await self.async_set_unique_id(self.info["mac"])
self._abort_if_unique_id_configured({CONF_HOST: discovery_info["host"]})
self.host = discovery_info["host"]
@ -176,11 +208,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"name": discovery_info.get("name", "").split(".")[0]
}
if info["auth"]:
if get_info_auth(self.info):
return await self.async_step_credentials()
try:
self.device_info = await validate_input(self.hass, self.host, {})
self.device_info = await validate_input(self.hass, self.host, self.info, {})
except HTTP_CONNECT_ERRORS:
return self.async_abort(reason="cannot_connect")
@ -193,11 +225,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
if user_input is not None:
return self.async_create_entry(
title=self.device_info["title"] or self.device_info["hostname"],
title=self.device_info["title"],
data={
"host": self.host,
"sleep_period": self.device_info["sleep_period"],
"model": self.device_info["model"],
"gen": self.device_info["gen"],
},
)
@ -206,9 +239,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="confirm_discovery",
description_placeholders={
"model": aioshelly.MODEL_NAMES.get(
self.info["type"], self.info["type"]
),
"model": get_model_name(self.info),
"host": self.host,
},
errors=errors,
@ -217,10 +248,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_get_info(self, host: str) -> dict[str, Any]:
"""Get info from shelly device."""
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
return cast(
Dict[str, Any],
await aioshelly.get_info(
aiohttp_client.async_get_clientsession(self.hass),
host,
),
return await aioshelly.common.get_info(
aiohttp_client.async_get_clientsession(self.hass), host
)

31
homeassistant/components/shelly/const.py
View file

@ -4,11 +4,12 @@ from __future__ import annotations
import re
from typing import Final
COAP: Final = "coap"
BLOCK: Final = "block"
DATA_CONFIG_ENTRY: Final = "config_entry"
DEVICE: Final = "device"
DOMAIN: Final = "shelly"
REST: Final = "rest"
RPC: Final = "rpc"
CONF_COAP_PORT: Final = "coap_port"
DEFAULT_COAP_PORT: Final = 5683
@ -44,6 +45,9 @@ SLEEP_PERIOD_MULTIPLIER: Final = 1.2
# Multiplier used to calculate the "update_interval" for non-sleeping devices.
UPDATE_PERIOD_MULTIPLIER: Final = 2.2
# Reconnect interval for GEN2 devices
RPC_RECONNECT_INTERVAL = 60
# Shelly Air - Maximum work hours before lamp replacement
SHAIR_MAX_WORK_HOURS: Final = 9000
@ -60,18 +64,28 @@ INPUTS_EVENTS_DICT: Final = {
# List of battery devices that maintain a permanent WiFi connection
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION: Final = ["SHMOS-01"]
# Button/Click events for Block & RPC devices
EVENT_SHELLY_CLICK: Final = "shelly.click"
ATTR_CLICK_TYPE: Final = "click_type"
ATTR_CHANNEL: Final = "channel"
ATTR_DEVICE: Final = "device"
ATTR_GENERATION: Final = "generation"
CONF_SUBTYPE: Final = "subtype"
BASIC_INPUTS_EVENTS_TYPES: Final = {"single", "long"}
SHBTN_INPUTS_EVENTS_TYPES: Final = {"single", "double", "triple", "long"}
SUPPORTED_INPUTS_EVENTS_TYPES: Final = {
RPC_INPUTS_EVENTS_TYPES: Final = {
"btn_down",
"btn_up",
"single_push",
"double_push",
"long_push",
}
BLOCK_INPUTS_EVENTS_TYPES: Final = {
"single",
"double",
"triple",
@ -80,9 +94,15 @@ SUPPORTED_INPUTS_EVENTS_TYPES: Final = {
"long_single",
}
SHIX3_1_INPUTS_EVENTS_TYPES = SUPPORTED_INPUTS_EVENTS_TYPES
SHIX3_1_INPUTS_EVENTS_TYPES = BLOCK_INPUTS_EVENTS_TYPES
INPUTS_EVENTS_SUBTYPES: Final = {"button": 1, "button1": 1, "button2": 2, "button3": 3}
INPUTS_EVENTS_SUBTYPES: Final = {
"button": 1,
"button1": 1,
"button2": 2,
"button3": 3,
"button4": 4,
}
SHBTN_MODELS: Final = ["SHBTN-1", "SHBTN-2"]
@ -109,3 +129,6 @@ KELVIN_MIN_VALUE_WHITE: Final = 2700
KELVIN_MIN_VALUE_COLOR: Final = 3000
UPTIME_DEVIATION: Final = 5
# Max RPC switch/input key instances
MAX_RPC_KEY_INSTANCES = 4

16
homeassistant/components/shelly/cover.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
from typing import Any, cast
from aioshelly import Block
from aioshelly.block_device import Block
from homeassistant.components.cover import (
ATTR_POSITION,
@ -18,8 +18,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ShellyDeviceWrapper
from .const import COAP, DATA_CONFIG_ENTRY, DOMAIN
from . import BlockDeviceWrapper
from .const import BLOCK, DATA_CONFIG_ENTRY, DOMAIN
from .entity import ShellyBlockEntity
@ -29,7 +29,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up cover for device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][COAP]
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][BLOCK]
blocks = [block for block in wrapper.device.blocks if block.type == "roller"]
if not blocks:
@ -43,7 +43,7 @@ class ShellyCover(ShellyBlockEntity, CoverEntity):
_attr_device_class = DEVICE_CLASS_SHUTTER
def __init__(self, wrapper: ShellyDeviceWrapper, block: Block) -> None:
def __init__(self, wrapper: BlockDeviceWrapper, block: Block) -> None:
"""Initialize light."""
super().__init__(wrapper, block)
self.control_result: dict[str, Any] | None = None
@ -57,7 +57,7 @@ class ShellyCover(ShellyBlockEntity, CoverEntity):
if self.control_result:
return cast(bool, self.control_result["current_pos"] == 0)
return cast(bool, self.block.rollerPos == 0)
return cast(int, self.block.rollerPos) == 0
@property
def current_cover_position(self) -> int:
@ -73,7 +73,7 @@ class ShellyCover(ShellyBlockEntity, CoverEntity):
if self.control_result:
return cast(bool, self.control_result["state"] == "close")
return cast(bool, self.block.roller == "close")
return self.block.roller == "close"
@property
def is_opening(self) -> bool:
@ -81,7 +81,7 @@ class ShellyCover(ShellyBlockEntity, CoverEntity):
if self.control_result:
return cast(bool, self.control_result["state"] == "open")
return cast(bool, self.block.roller == "open")
return self.block.roller == "open"
@property
def supported_features(self) -> int:

105
homeassistant/components/shelly/device_trigger.py
View file

@ -25,28 +25,52 @@ from homeassistant.const import (
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.typing import ConfigType
from . import get_device_wrapper
from . import get_block_device_wrapper, get_rpc_device_wrapper
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
BLOCK_INPUTS_EVENTS_TYPES,
CONF_SUBTYPE,
DOMAIN,
EVENT_SHELLY_CLICK,
INPUTS_EVENTS_SUBTYPES,
SHBTN_INPUTS_EVENTS_TYPES,
RPC_INPUTS_EVENTS_TYPES,
SHBTN_MODELS,
SUPPORTED_INPUTS_EVENTS_TYPES,
)
from .utils import get_input_triggers
from .utils import (
get_block_input_triggers,
get_rpc_input_triggers,
get_shbtn_input_triggers,
)
TRIGGER_SCHEMA: Final = DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(SUPPORTED_INPUTS_EVENTS_TYPES),
vol.Required(CONF_TYPE): vol.In(
RPC_INPUTS_EVENTS_TYPES | BLOCK_INPUTS_EVENTS_TYPES
),
vol.Required(CONF_SUBTYPE): vol.In(INPUTS_EVENTS_SUBTYPES),
}
)
def append_input_triggers(
triggers: list[dict[str, Any]],
input_triggers: list[tuple[str, str]],
device_id: str,
) -> None:
"""Add trigger to triggers list."""
for trigger, subtype in input_triggers:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
async def async_validate_trigger_config(
hass: HomeAssistant, config: dict[str, Any]
) -> dict[str, Any]:
@ -54,17 +78,29 @@ async def async_validate_trigger_config(
config = TRIGGER_SCHEMA(config)
# if device is available verify parameters against device capabilities
wrapper = get_device_wrapper(hass, config[CONF_DEVICE_ID])
if not wrapper or not wrapper.device.initialized:
return config
trigger = (config[CONF_TYPE], config[CONF_SUBTYPE])
for block in wrapper.device.blocks:
input_triggers = get_input_triggers(wrapper.device, block)
if config[CONF_TYPE] in RPC_INPUTS_EVENTS_TYPES:
rpc_wrapper = get_rpc_device_wrapper(hass, config[CONF_DEVICE_ID])
if not rpc_wrapper or not rpc_wrapper.device.initialized:
return config
input_triggers = get_rpc_input_triggers(rpc_wrapper.device)
if trigger in input_triggers:
return config
elif config[CONF_TYPE] in BLOCK_INPUTS_EVENTS_TYPES:
block_wrapper = get_block_device_wrapper(hass, config[CONF_DEVICE_ID])
if not block_wrapper or not block_wrapper.device.initialized:
return config
assert block_wrapper.device.blocks
for block in block_wrapper.device.blocks:
input_triggers = get_block_input_triggers(block_wrapper.device, block)
if trigger in input_triggers:
return config
raise InvalidDeviceAutomationConfig(
f"Invalid ({CONF_TYPE},{CONF_SUBTYPE}): {trigger}"
)
@ -74,40 +110,28 @@ async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
"""List device triggers for Shelly devices."""
triggers = []
triggers: list[dict[str, Any]] = []
wrapper = get_device_wrapper(hass, device_id)
if not wrapper:
raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}")
if wrapper.model in SHBTN_MODELS:
for trigger in SHBTN_INPUTS_EVENTS_TYPES:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: "button",
}
)
if rpc_wrapper := get_rpc_device_wrapper(hass, device_id):
input_triggers = get_rpc_input_triggers(rpc_wrapper.device)
append_input_triggers(triggers, input_triggers, device_id)
return triggers
for block in wrapper.device.blocks:
input_triggers = get_input_triggers(wrapper.device, block)
if block_wrapper := get_block_device_wrapper(hass, device_id):
if block_wrapper.model in SHBTN_MODELS:
input_triggers = get_shbtn_input_triggers()
append_input_triggers(triggers, input_triggers, device_id)
return triggers
for trigger, subtype in input_triggers:
triggers.append(
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: trigger,
CONF_SUBTYPE: subtype,
}
)
assert block_wrapper.device.blocks
return triggers
for block in block_wrapper.device.blocks:
input_triggers = get_block_input_triggers(block_wrapper.device, block)
append_input_triggers(triggers, input_triggers, device_id)
return triggers
raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}")
async def async_attach_trigger(
@ -126,6 +150,7 @@ async def async_attach_trigger(
ATTR_CLICK_TYPE: config[CONF_TYPE],
},
}
event_config = event_trigger.TRIGGER_SCHEMA(event_config)
return await event_trigger.async_attach_trigger(
hass, event_config, action, automation_info, platform_type="device"

225
homeassistant/components/shelly/entity.py
View file

@ -6,7 +6,7 @@ from dataclasses import dataclass
import logging
from typing import Any, Callable, Final, cast
import aioshelly
from aioshelly.block_device import Block
import async_timeout
from homeassistant.components.sensor import ATTR_STATE_CLASS
@ -23,9 +23,21 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import StateType
from . import ShellyDeviceRestWrapper, ShellyDeviceWrapper
from .const import AIOSHELLY_DEVICE_TIMEOUT_SEC, COAP, DATA_CONFIG_ENTRY, DOMAIN, REST
from .utils import async_remove_shelly_entity, get_entity_name
from . import BlockDeviceWrapper, RpcDeviceWrapper, ShellyDeviceRestWrapper
from .const import (
AIOSHELLY_DEVICE_TIMEOUT_SEC,
BLOCK,
DATA_CONFIG_ENTRY,
DOMAIN,
REST,
RPC,
)
from .utils import (
async_remove_shelly_entity,
get_block_entity_name,
get_rpc_entity_name,
get_rpc_key_instances,
)
_LOGGER: Final = logging.getLogger(__name__)
@ -38,9 +50,9 @@ async def async_setup_entry_attribute_entities(
sensor_class: Callable,
) -> None:
"""Set up entities for attributes."""
wrapper: ShellyDeviceWrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
wrapper: BlockDeviceWrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry.entry_id
][COAP]
][BLOCK]
if wrapper.device.initialized:
await async_setup_block_attribute_entities(
@ -55,13 +67,15 @@ async def async_setup_entry_attribute_entities(
async def async_setup_block_attribute_entities(
hass: HomeAssistant,
async_add_entities: AddEntitiesCallback,
wrapper: ShellyDeviceWrapper,
wrapper: BlockDeviceWrapper,
sensors: dict[tuple[str, str], BlockAttributeDescription],
sensor_class: Callable,
) -> None:
"""Set up entities for block attributes."""
blocks = []
assert wrapper.device.blocks
for block in wrapper.device.blocks:
for sensor_id in block.sensor_ids:
description = sensors.get((block.type, sensor_id))
@ -97,7 +111,7 @@ async def async_restore_block_attribute_entities(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
wrapper: ShellyDeviceWrapper,
wrapper: BlockDeviceWrapper,
sensors: dict[tuple[str, str], BlockAttributeDescription],
sensor_class: Callable,
) -> None:
@ -133,6 +147,45 @@ async def async_restore_block_attribute_entities(
async_add_entities(entities)
async def async_setup_entry_rpc(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
sensors: dict[str, RpcAttributeDescription],
sensor_class: Callable,
) -> None:
"""Set up entities for REST sensors."""
wrapper: RpcDeviceWrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry.entry_id
][RPC]
entities = []
for sensor_id in sensors:
description = sensors[sensor_id]
key_instances = get_rpc_key_instances(wrapper.device.status, description.key)
for key in key_instances:
# Filter and remove entities that according to settings should not create an entity
if description.removal_condition and description.removal_condition(
wrapper.device.config, key
):
domain = sensor_class.__module__.split(".")[-1]
unique_id = f"{wrapper.mac}-{key}-{sensor_id}"
await async_remove_shelly_entity(hass, domain, unique_id)
else:
entities.append((key, sensor_id, description))
if not entities:
return
async_add_entities(
[
sensor_class(wrapper, key, sensor_id, description)
for key, sensor_id, description in entities
]
)
async def async_setup_entry_rest(
hass: HomeAssistant,
config_entry: ConfigEntry,
@ -175,10 +228,27 @@ class BlockAttributeDescription:
device_class: str | None = None
state_class: str | None = None
default_enabled: bool = True
available: Callable[[aioshelly.Block], bool] | None = None
available: Callable[[Block], bool] | None = None
# Callable (settings, block), return true if entity should be removed
removal_condition: Callable[[dict, aioshelly.Block], bool] | None = None
extra_state_attributes: Callable[[aioshelly.Block], dict | None] | None = None
removal_condition: Callable[[dict, Block], bool] | None = None
extra_state_attributes: Callable[[Block], dict | None] | None = None
@dataclass
class RpcAttributeDescription:
"""Class to describe a RPC sensor."""
key: str
name: str
icon: str | None = None
unit: str | None = None
value: Callable[[dict, Any], Any] | None = None
device_class: str | None = None
state_class: str | None = None
default_enabled: bool = True
available: Callable[[dict], bool] | None = None
removal_condition: Callable[[dict, str], bool] | None = None
extra_state_attributes: Callable[[dict], dict | None] | None = None
@dataclass
@ -196,13 +266,13 @@ class RestAttributeDescription:
class ShellyBlockEntity(entity.Entity):
"""Helper class to represent a block."""
"""Helper class to represent a block entity."""
def __init__(self, wrapper: ShellyDeviceWrapper, block: aioshelly.Block) -> None:
def __init__(self, wrapper: BlockDeviceWrapper, block: Block) -> None:
"""Initialize Shelly entity."""
self.wrapper = wrapper
self.block = block
self._name = get_entity_name(wrapper.device, block)
self._name = get_block_entity_name(wrapper.device, block)
@property
def name(self) -> str:
@ -261,13 +331,68 @@ class ShellyBlockEntity(entity.Entity):
return None
class ShellyRpcEntity(entity.Entity):
"""Helper class to represent a rpc entity."""
def __init__(self, wrapper: RpcDeviceWrapper, key: str) -> None:
"""Initialize Shelly entity."""
self.wrapper = wrapper
self.key = key
self._attr_should_poll = False
self._attr_device_info = {
"connections": {(device_registry.CONNECTION_NETWORK_MAC, wrapper.mac)}
}
self._attr_unique_id = f"{wrapper.mac}-{key}"
self._attr_name = get_rpc_entity_name(wrapper.device, key)
@property
def available(self) -> bool:
"""Available."""
return self.wrapper.device.connected
async def async_added_to_hass(self) -> None:
"""When entity is added to HASS."""
self.async_on_remove(self.wrapper.async_add_listener(self._update_callback))
async def async_update(self) -> None:
"""Update entity with latest info."""
await self.wrapper.async_request_refresh()
@callback
def _update_callback(self) -> None:
"""Handle device update."""
self.async_write_ha_state()
async def call_rpc(self, method: str, params: Any) -> Any:
"""Call RPC method."""
_LOGGER.debug(
"Call RPC for entity %s, method: %s, params: %s",
self.name,
method,
params,
)
try:
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
return await self.wrapper.device.call_rpc(method, params)
except asyncio.TimeoutError as err:
_LOGGER.error(
"Call RPC for entity %s failed, method: %s, params: %s, error: %s",
self.name,
method,
params,
repr(err),
)
self.wrapper.last_update_success = False
return None
class ShellyBlockAttributeEntity(ShellyBlockEntity, entity.Entity):
"""Helper class to represent a block attribute."""
def __init__(
self,
wrapper: ShellyDeviceWrapper,
block: aioshelly.Block,
wrapper: BlockDeviceWrapper,
block: Block,
attribute: str,
description: BlockAttributeDescription,
) -> None:
@ -283,7 +408,7 @@ class ShellyBlockAttributeEntity(ShellyBlockEntity, entity.Entity):
self._unit: None | str | Callable[[dict], str] = unit
self._unique_id: str = f"{super().unique_id}-{self.attribute}"
self._name = get_entity_name(wrapper.device, block, self.description.name)
self._name = get_block_entity_name(wrapper.device, block, self.description.name)
@property
def unique_id(self) -> str:
@ -344,7 +469,7 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
def __init__(
self,
wrapper: ShellyDeviceWrapper,
wrapper: BlockDeviceWrapper,
attribute: str,
description: RestAttributeDescription,
) -> None:
@ -353,7 +478,7 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
self.wrapper = wrapper
self.attribute = attribute
self.description = description
self._name = get_entity_name(wrapper.device, None, self.description.name)
self._name = get_block_entity_name(wrapper.device, None, self.description.name)
self._last_value = None
@property
@ -411,14 +536,66 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
return self.description.extra_state_attributes(self.wrapper.device.status)
class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity):
"""Helper class to represent a rpc attribute."""
def __init__(
self,
wrapper: RpcDeviceWrapper,
key: str,
attribute: str,
description: RpcAttributeDescription,
) -> None:
"""Initialize sensor."""
super().__init__(wrapper, key)
self.attribute = attribute
self.description = description
self._attr_unique_id = f"{super().unique_id}-{attribute}"
self._attr_name = get_rpc_entity_name(wrapper.device, key, description.name)
self._attr_entity_registry_enabled_default = description.default_enabled
self._attr_device_class = description.device_class
self._attr_icon = description.icon
self._last_value = None
@property
def attribute_value(self) -> StateType:
"""Value of sensor."""
if callable(self.description.value):
self._last_value = self.description.value(
self.wrapper.device.status[self.key], self._last_value
)
return self._last_value
@property
def available(self) -> bool:
"""Available."""
available = super().available
if not available or not self.description.available:
return available
return self.description.available(self.wrapper.device.status[self.key])
@property
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self.description.extra_state_attributes is None:
return None
return self.description.extra_state_attributes(
self.wrapper.device.status[self.key]
)
class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEntity):
"""Represent a shelly sleeping block attribute entity."""
# pylint: disable=super-init-not-called
def __init__(
self,
wrapper: ShellyDeviceWrapper,
block: aioshelly.Block,
wrapper: BlockDeviceWrapper,
block: Block | None,
attribute: str,
description: BlockAttributeDescription,
entry: entity_registry.RegistryEntry | None = None,
@ -429,7 +606,7 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
self.last_state: StateType = None
self.wrapper = wrapper
self.attribute = attribute
self.block = block
self.block: Block | None = block # type: ignore[assignment]
self.description = description
self._unit = self.description.unit
@ -438,7 +615,7 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
self._unit = self._unit(block.info(attribute))
self._unique_id = f"{self.wrapper.mac}-{block.description}-{attribute}"
self._name = get_entity_name(
self._name = get_block_entity_name(
self.wrapper.device, block, self.description.name
)
elif entry is not None:
@ -468,6 +645,8 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
_, entity_block, entity_sensor = self.unique_id.split("-")
assert self.wrapper.device.blocks
for block in self.wrapper.device.blocks:
if block.description != entity_block:
continue

106
homeassistant/components/shelly/light.py
View file

@ -5,7 +5,7 @@ import asyncio
import logging
from typing import Any, Final, cast
from aioshelly import Block
from aioshelly.block_device import Block
import async_timeout
from homeassistant.components.light import (
@ -33,10 +33,10 @@ from homeassistant.util.color import (
color_temperature_mired_to_kelvin,
)
from . import ShellyDeviceWrapper
from . import BlockDeviceWrapper, RpcDeviceWrapper
from .const import (
AIOSHELLY_DEVICE_TIMEOUT_SEC,
COAP,
BLOCK,
DATA_CONFIG_ENTRY,
DOMAIN,
FIRMWARE_PATTERN,
@ -46,11 +46,18 @@ from .const import (
LIGHT_TRANSITION_MIN_FIRMWARE_DATE,
MAX_TRANSITION_TIME,
MODELS_SUPPORTING_LIGHT_TRANSITION,
RPC,
SHBLB_1_RGB_EFFECTS,
STANDARD_RGB_EFFECTS,
)
from .entity import ShellyBlockEntity
from .utils import async_remove_shelly_entity
from .entity import ShellyBlockEntity, ShellyRpcEntity
from .utils import (
async_remove_shelly_entity,
get_device_entry_gen,
get_rpc_key_ids,
is_block_channel_type_light,
is_rpc_channel_type_light,
)
_LOGGER: Final = logging.getLogger(__name__)
@ -61,33 +68,70 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up lights for device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][COAP]
if get_device_entry_gen(config_entry) == 2:
return await async_setup_rpc_entry(hass, config_entry, async_add_entities)
return await async_setup_block_entry(hass, config_entry, async_add_entities)
async def async_setup_block_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up entities for block device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][BLOCK]
blocks = []
assert wrapper.device.blocks
for block in wrapper.device.blocks:
if block.type == "light":
blocks.append(block)
elif block.type == "relay":
appliance_type = wrapper.device.settings["relays"][int(block.channel)].get(
"appliance_type"
)
if appliance_type and appliance_type.lower() == "light":
blocks.append(block)
unique_id = (
f'{wrapper.device.shelly["mac"]}-{block.type}_{block.channel}'
)
await async_remove_shelly_entity(hass, "switch", unique_id)
if not is_block_channel_type_light(
wrapper.device.settings, int(block.channel)
):
continue
blocks.append(block)
assert wrapper.device.shelly
unique_id = f"{wrapper.mac}-{block.type}_{block.channel}"
await async_remove_shelly_entity(hass, "switch", unique_id)
if not blocks:
return
async_add_entities(ShellyLight(wrapper, block) for block in blocks)
async_add_entities(BlockShellyLight(wrapper, block) for block in blocks)
class ShellyLight(ShellyBlockEntity, LightEntity):
"""Switch that controls a relay block on Shelly devices."""
async def async_setup_rpc_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up entities for RPC device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][RPC]
switch_key_ids = get_rpc_key_ids(wrapper.device.status, "switch")
def __init__(self, wrapper: ShellyDeviceWrapper, block: Block) -> None:
switch_ids = []
for id_ in switch_key_ids:
if not is_rpc_channel_type_light(wrapper.device.config, id_):
continue
switch_ids.append(id_)
unique_id = f"{wrapper.mac}-switch:{id_}"
await async_remove_shelly_entity(hass, "switch", unique_id)
if not switch_ids:
return
async_add_entities(RpcShellyLight(wrapper, id_) for id_ in switch_ids)
class BlockShellyLight(ShellyBlockEntity, LightEntity):
"""Entity that controls a light on block based Shelly devices."""
def __init__(self, wrapper: BlockDeviceWrapper, block: Block) -> None:
"""Initialize light."""
super().__init__(wrapper, block)
self.control_result: dict[str, Any] | None = None
@ -117,7 +161,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
self._supported_features |= SUPPORT_EFFECT
if wrapper.model in MODELS_SUPPORTING_LIGHT_TRANSITION:
match = FIRMWARE_PATTERN.search(wrapper.device.settings.get("fw"))
match = FIRMWARE_PATTERN.search(wrapper.device.settings.get("fw", ""))
if (
match is not None
and int(match[0]) >= LIGHT_TRANSITION_MIN_FIRMWARE_DATE
@ -369,3 +413,25 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
self.control_result = None
self.mode_result = None
super()._update_callback()
class RpcShellyLight(ShellyRpcEntity, LightEntity):
"""Entity that controls a light on RPC based Shelly devices."""
def __init__(self, wrapper: RpcDeviceWrapper, id_: int) -> None:
"""Initialize light."""
super().__init__(wrapper, f"switch:{id_}")
self._id = id_
@property
def is_on(self) -> bool:
"""If light is on."""
return bool(self.wrapper.device.status[self.key]["output"])
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on light."""
await self.call_rpc("Switch.Set", {"id": self._id, "on": True})
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off light."""
await self.call_rpc("Switch.Set", {"id": self._id, "on": False})

32
homeassistant/components/shelly/logbook.py
View file

@ -7,15 +7,17 @@ from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.typing import EventType
from . import get_device_wrapper
from . import get_block_device_wrapper, get_rpc_device_wrapper
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
ATTR_DEVICE,
BLOCK_INPUTS_EVENTS_TYPES,
DOMAIN,
EVENT_SHELLY_CLICK,
RPC_INPUTS_EVENTS_TYPES,
)
from .utils import get_device_name
from .utils import get_block_device_name, get_rpc_entity_name
@callback
@ -27,19 +29,27 @@ def async_describe_events(
@callback
def async_describe_shelly_click_event(event: EventType) -> dict[str, str]:
"""Describe shelly.click logbook event."""
wrapper = get_device_wrapper(hass, event.data[ATTR_DEVICE_ID])
if wrapper and wrapper.device.initialized:
device_name = get_device_name(wrapper.device)
else:
device_name = event.data[ATTR_DEVICE]
channel = event.data[ATTR_CHANNEL]
"""Describe shelly.click logbook event (block device)."""
device_id = event.data[ATTR_DEVICE_ID]
click_type = event.data[ATTR_CLICK_TYPE]
channel = event.data[ATTR_CHANNEL]
input_name = f"{event.data[ATTR_DEVICE]} channel {channel}"
if click_type in RPC_INPUTS_EVENTS_TYPES:
rpc_wrapper = get_rpc_device_wrapper(hass, device_id)
if rpc_wrapper and rpc_wrapper.device.initialized:
key = f"input:{channel-1}"
input_name = get_rpc_entity_name(rpc_wrapper.device, key)
elif click_type in BLOCK_INPUTS_EVENTS_TYPES:
block_wrapper = get_block_device_wrapper(hass, device_id)
if block_wrapper and block_wrapper.device.initialized:
device_name = get_block_device_name(block_wrapper.device)
input_name = f"{device_name} channel {channel}"
return {
"name": "Shelly",
"message": f"'{click_type}' click event for {device_name} channel {channel} was fired.",
"message": f"'{click_type}' click event for {input_name} Input was fired.",
}
async_describe_event(DOMAIN, EVENT_SHELLY_CLICK, async_describe_shelly_click_event)

2
homeassistant/components/shelly/manifest.json
View file

@ -3,7 +3,7 @@
"name": "Shelly",
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/shelly",
"requirements": ["aioshelly==0.6.4"],
"requirements": ["aioshelly==1.0.1"],
"zeroconf": [
{
"type": "_http._tcp.local.",

113
homeassistant/components/shelly/sensor.py
View file

@ -16,6 +16,7 @@ from homeassistant.const import (
PERCENTAGE,
POWER_WATT,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
TEMP_CELSIUS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -25,13 +26,16 @@ from .const import SHAIR_MAX_WORK_HOURS
from .entity import (
BlockAttributeDescription,
RestAttributeDescription,
RpcAttributeDescription,
ShellyBlockAttributeEntity,
ShellyRestAttributeEntity,
ShellyRpcAttributeEntity,
ShellySleepingBlockAttributeEntity,
async_setup_entry_attribute_entities,
async_setup_entry_rest,
async_setup_entry_rpc,
)
from .utils import get_device_uptime, temperature_unit
from .utils import get_device_entry_gen, get_device_uptime, temperature_unit
SENSORS: Final = {
("device", "battery"): BlockAttributeDescription(
@ -40,7 +44,7 @@ SENSORS: Final = {
device_class=sensor.DEVICE_CLASS_BATTERY,
state_class=sensor.STATE_CLASS_MEASUREMENT,
removal_condition=lambda settings, _: settings.get("external_power") == 1,
available=lambda block: cast(bool, block.battery != -1),
available=lambda block: cast(int, block.battery) != -1,
),
("device", "deviceTemp"): BlockAttributeDescription(
name="Device Temperature",
@ -162,7 +166,7 @@ SENSORS: Final = {
value=lambda value: round(value, 1),
device_class=sensor.DEVICE_CLASS_TEMPERATURE,
state_class=sensor.STATE_CLASS_MEASUREMENT,
available=lambda block: cast(bool, block.extTemp != 999),
available=lambda block: cast(int, block.extTemp) != 999,
),
("sensor", "humidity"): BlockAttributeDescription(
name="Humidity",
@ -170,14 +174,14 @@ SENSORS: Final = {
value=lambda value: round(value, 1),
device_class=sensor.DEVICE_CLASS_HUMIDITY,
state_class=sensor.STATE_CLASS_MEASUREMENT,
available=lambda block: cast(bool, block.extTemp != 999),
available=lambda block: cast(int, block.extTemp) != 999,
),
("sensor", "luminosity"): BlockAttributeDescription(
name="Luminosity",
unit=LIGHT_LUX,
device_class=sensor.DEVICE_CLASS_ILLUMINANCE,
state_class=sensor.STATE_CLASS_MEASUREMENT,
available=lambda block: cast(bool, block.luminosity != -1),
available=lambda block: cast(int, block.luminosity) != -1,
),
("sensor", "tilt"): BlockAttributeDescription(
name="Tilt",
@ -191,7 +195,7 @@ SENSORS: Final = {
icon="mdi:progress-wrench",
value=lambda value: round(100 - (value / 3600 / SHAIR_MAX_WORK_HOURS), 1),
extra_state_attributes=lambda block: {
"Operational hours": round(block.totalWorkTime / 3600, 1)
"Operational hours": round(cast(int, block.totalWorkTime) / 3600, 1)
},
),
("adc", "adc"): BlockAttributeDescription(
@ -220,7 +224,60 @@ REST_SENSORS: Final = {
),
"uptime": RestAttributeDescription(
name="Uptime",
value=get_device_uptime,
value=lambda status, last: get_device_uptime(status["uptime"], last),
device_class=sensor.DEVICE_CLASS_TIMESTAMP,
default_enabled=False,
),
}
RPC_SENSORS: Final = {
"power": RpcAttributeDescription(
key="switch",
name="Power",
unit=POWER_WATT,
value=lambda status, _: round(float(status["apower"]), 1),
device_class=sensor.DEVICE_CLASS_POWER,
state_class=sensor.STATE_CLASS_MEASUREMENT,
),
"voltage": RpcAttributeDescription(
key="switch",
name="Voltage",
unit=ELECTRIC_POTENTIAL_VOLT,
value=lambda status, _: round(float(status["voltage"]), 1),
device_class=sensor.DEVICE_CLASS_VOLTAGE,
state_class=sensor.STATE_CLASS_MEASUREMENT,
),
"energy": RpcAttributeDescription(
key="switch",
name="Energy",
unit=ENERGY_KILO_WATT_HOUR,
value=lambda status, _: round(status["aenergy"]["total"] / 1000, 2),
device_class=sensor.DEVICE_CLASS_ENERGY,
state_class=sensor.STATE_CLASS_TOTAL_INCREASING,
),
"temperature": RpcAttributeDescription(
key="switch",
name="Temperature",
unit=TEMP_CELSIUS,
value=lambda status, _: round(status["temperature"]["tC"], 1),
device_class=sensor.DEVICE_CLASS_TEMPERATURE,
state_class=sensor.STATE_CLASS_MEASUREMENT,
default_enabled=False,
),
"rssi": RpcAttributeDescription(
key="wifi",
name="RSSI",
unit=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
value=lambda status, _: status["rssi"],
device_class=sensor.DEVICE_CLASS_SIGNAL_STRENGTH,
state_class=sensor.STATE_CLASS_MEASUREMENT,
default_enabled=False,
),
"uptime": RpcAttributeDescription(
key="sys",
name="Uptime",
value=lambda status, last: get_device_uptime(status["uptime"], last),
device_class=sensor.DEVICE_CLASS_TIMESTAMP,
default_enabled=False,
),
@ -233,21 +290,26 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensors for device."""
if get_device_entry_gen(config_entry) == 2:
return await async_setup_entry_rpc(
hass, config_entry, async_add_entities, RPC_SENSORS, RpcSensor
)
if config_entry.data["sleep_period"]:
await async_setup_entry_attribute_entities(
hass, config_entry, async_add_entities, SENSORS, ShellySleepingSensor
hass, config_entry, async_add_entities, SENSORS, BlockSleepingSensor
)
else:
await async_setup_entry_attribute_entities(
hass, config_entry, async_add_entities, SENSORS, ShellySensor
hass, config_entry, async_add_entities, SENSORS, BlockSensor
)
await async_setup_entry_rest(
hass, config_entry, async_add_entities, REST_SENSORS, ShellyRestSensor
hass, config_entry, async_add_entities, REST_SENSORS, RestSensor
)
class ShellySensor(ShellyBlockAttributeEntity, SensorEntity):
"""Represent a shelly sensor."""
class BlockSensor(ShellyBlockAttributeEntity, SensorEntity):
"""Represent a block sensor."""
@property
def native_value(self) -> StateType:
@ -265,8 +327,8 @@ class ShellySensor(ShellyBlockAttributeEntity, SensorEntity):
return cast(str, self._unit)
class ShellyRestSensor(ShellyRestAttributeEntity, SensorEntity):
"""Represent a shelly REST sensor."""
class RestSensor(ShellyRestAttributeEntity, SensorEntity):
"""Represent a REST sensor."""
@property
def native_value(self) -> StateType:
@ -284,8 +346,27 @@ class ShellyRestSensor(ShellyRestAttributeEntity, SensorEntity):
return self.description.unit
class ShellySleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity):
"""Represent a shelly sleeping sensor."""
class RpcSensor(ShellyRpcAttributeEntity, SensorEntity):
"""Represent a RPC sensor."""
@property
def native_value(self) -> StateType:
"""Return value of sensor."""
return self.attribute_value
@property
def state_class(self) -> str | None:
"""State class of sensor."""
return self.description.state_class
@property
def native_unit_of_measurement(self) -> str | None:
"""Return unit of sensor."""
return self.description.unit
class BlockSleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity):
"""Represent a block sleeping sensor."""
@property
def native_value(self) -> StateType:

10
homeassistant/components/shelly/strings.json
View file

@ -33,7 +33,8 @@
"button": "Button",
"button1": "First button",
"button2": "Second button",
"button3": "Third button"
"button3": "Third button",
"button4": "Fourth button"
},
"trigger_type": {
"single": "{subtype} single clicked",
@ -41,7 +42,12 @@
"triple": "{subtype} triple clicked",
"long": " {subtype} long clicked",
"single_long": "{subtype} single clicked and then long clicked",
"long_single": "{subtype} long clicked and then single clicked"
"long_single": "{subtype} long clicked and then single clicked",
"btn_down": "{subtype} button down",
"btn_up": "{subtype} button up",
"single_push": "{subtype} single push",
"double_push": "{subtype} double push",
"long_push": " {subtype} long push"
}
}
}

107
homeassistant/components/shelly/switch.py
View file

@ -3,17 +3,23 @@ from __future__ import annotations
from typing import Any, cast
from aioshelly import Block
from aioshelly.block_device import Block
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import ShellyDeviceWrapper
from .const import COAP, DATA_CONFIG_ENTRY, DOMAIN
from .entity import ShellyBlockEntity
from .utils import async_remove_shelly_entity
from . import BlockDeviceWrapper, RpcDeviceWrapper
from .const import BLOCK, DATA_CONFIG_ENTRY, DOMAIN, RPC
from .entity import ShellyBlockEntity, ShellyRpcEntity
from .utils import (
async_remove_shelly_entity,
get_device_entry_gen,
get_rpc_key_ids,
is_block_channel_type_light,
is_rpc_channel_type_light,
)
async def async_setup_entry(
@ -22,7 +28,19 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up switches for device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][COAP]
if get_device_entry_gen(config_entry) == 2:
return await async_setup_rpc_entry(hass, config_entry, async_add_entities)
return await async_setup_block_entry(hass, config_entry, async_add_entities)
async def async_setup_block_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up entities for block device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][BLOCK]
# In roller mode the relay blocks exist but do not contain required info
if (
@ -32,32 +50,51 @@ async def async_setup_entry(
return
relay_blocks = []
assert wrapper.device.blocks
for block in wrapper.device.blocks:
if block.type == "relay":
appliance_type = wrapper.device.settings["relays"][int(block.channel)].get(
"appliance_type"
)
if not appliance_type or appliance_type.lower() != "light":
relay_blocks.append(block)
unique_id = (
f'{wrapper.device.shelly["mac"]}-{block.type}_{block.channel}'
)
await async_remove_shelly_entity(
hass,
"light",
unique_id,
)
if block.type != "relay" or is_block_channel_type_light(
wrapper.device.settings, int(block.channel)
):
continue
relay_blocks.append(block)
unique_id = f"{wrapper.mac}-{block.type}_{block.channel}"
await async_remove_shelly_entity(hass, "light", unique_id)
if not relay_blocks:
return
async_add_entities(RelaySwitch(wrapper, block) for block in relay_blocks)
async_add_entities(BlockRelaySwitch(wrapper, block) for block in relay_blocks)
class RelaySwitch(ShellyBlockEntity, SwitchEntity):
"""Switch that controls a relay block on Shelly devices."""
async def async_setup_rpc_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up entities for RPC device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][RPC]
switch_key_ids = get_rpc_key_ids(wrapper.device.status, "switch")
def __init__(self, wrapper: ShellyDeviceWrapper, block: Block) -> None:
switch_ids = []
for id_ in switch_key_ids:
if is_rpc_channel_type_light(wrapper.device.config, id_):
continue
switch_ids.append(id_)
unique_id = f"{wrapper.mac}-switch:{id_}"
await async_remove_shelly_entity(hass, "light", unique_id)
if not switch_ids:
return
async_add_entities(RpcRelaySwitch(wrapper, id_) for id_ in switch_ids)
class BlockRelaySwitch(ShellyBlockEntity, SwitchEntity):
"""Entity that controls a relay on Block based Shelly devices."""
def __init__(self, wrapper: BlockDeviceWrapper, block: Block) -> None:
"""Initialize relay switch."""
super().__init__(wrapper, block)
self.control_result: dict[str, Any] | None = None
@ -85,3 +122,25 @@ class RelaySwitch(ShellyBlockEntity, SwitchEntity):
"""When device updates, clear control result that overrides state."""
self.control_result = None
super()._update_callback()
class RpcRelaySwitch(ShellyRpcEntity, SwitchEntity):
"""Entity that controls a relay on RPC based Shelly devices."""
def __init__(self, wrapper: RpcDeviceWrapper, id_: int) -> None:
"""Initialize relay switch."""
super().__init__(wrapper, f"switch:{id_}")
self._id = id_
@property
def is_on(self) -> bool:
"""If switch is on."""
return bool(self.wrapper.device.status[self.key]["output"])
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on relay."""
await self.call_rpc("Switch.Set", {"id": self._id, "on": True})
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off relay."""
await self.call_rpc("Switch.Set", {"id": self._id, "on": False})

10
homeassistant/components/shelly/translations/en.json
View file

@ -33,7 +33,8 @@
"button": "Button",
"button1": "First button",
"button2": "Second button",
"button3": "Third button"
"button3": "Third button",
"button4": "Fourth button"
},
"trigger_type": {
"double": "{subtype} double clicked",
@ -41,7 +42,12 @@
"long_single": "{subtype} long clicked and then single clicked",
"single": "{subtype} single clicked",
"single_long": "{subtype} single clicked and then long clicked",
"triple": "{subtype} triple clicked"
"triple": "{subtype} triple clicked",
"btn_down": "{subtype} button down",
"btn_up": "{subtype} button up",
"single_push": "{subtype} single push",
"double_push": "{subtype} double push",
"long_push": " {subtype} long push"
}
}
}

185
homeassistant/components/shelly/utils.py
View file

@ -5,8 +5,11 @@ from datetime import datetime, timedelta
import logging
from typing import Any, Final, cast
import aioshelly
from aioshelly.block_device import BLOCK_VALUE_UNIT, COAP, Block, BlockDevice
from aioshelly.const import MODEL_NAMES
from aioshelly.rpc_device import RpcDevice
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import singleton
@ -18,6 +21,8 @@ from .const import (
CONF_COAP_PORT,
DEFAULT_COAP_PORT,
DOMAIN,
MAX_RPC_KEY_INSTANCES,
RPC_INPUTS_EVENTS_TYPES,
SHBTN_INPUTS_EVENTS_TYPES,
SHBTN_MODELS,
SHIX3_1_INPUTS_EVENTS_TYPES,
@ -40,18 +45,27 @@ async def async_remove_shelly_entity(
def temperature_unit(block_info: dict[str, Any]) -> str:
"""Detect temperature unit."""
if block_info[aioshelly.BLOCK_VALUE_UNIT] == "F":
if block_info[BLOCK_VALUE_UNIT] == "F":
return TEMP_FAHRENHEIT
return TEMP_CELSIUS
def get_device_name(device: aioshelly.Device) -> str:
def get_block_device_name(device: BlockDevice) -> str:
"""Naming for device."""
return cast(str, device.settings["name"] or device.settings["device"]["hostname"])
def get_number_of_channels(device: aioshelly.Device, block: aioshelly.Block) -> int:
def get_rpc_device_name(device: RpcDevice) -> str:
"""Naming for device."""
# Gen2 does not support setting device name
# AP SSID name is used as a nicely formatted device name
return cast(str, device.config["wifi"]["ap"]["ssid"] or device.hostname)
def get_number_of_channels(device: BlockDevice, block: Block) -> int:
"""Get number of channels for block type."""
assert isinstance(device.shelly, dict)
channels = None
if block.type == "input":
@ -70,13 +84,13 @@ def get_number_of_channels(device: aioshelly.Device, block: aioshelly.Block) ->
return channels or 1
def get_entity_name(
device: aioshelly.Device,
block: aioshelly.Block,
def get_block_entity_name(
device: BlockDevice,
block: Block | None,
description: str | None = None,
) -> str:
"""Naming for switch and sensors."""
channel_name = get_device_channel_name(device, block)
"""Naming for block based switch and sensors."""
channel_name = get_block_channel_name(device, block)
if description:
return f"{channel_name} {description}"
@ -84,12 +98,9 @@ def get_entity_name(
return channel_name
def get_device_channel_name(
device: aioshelly.Device,
block: aioshelly.Block,
) -> str:
def get_block_channel_name(device: BlockDevice, block: Block | None) -> str:
"""Get name based on device and channel name."""
entity_name = get_device_name(device)
entity_name = get_block_device_name(device)
if (
not block
@ -98,8 +109,10 @@ def get_device_channel_name(
):
return entity_name
assert block.channel
channel_name: str | None = None
mode = block.type + "s"
mode = cast(str, block.type) + "s"
if mode in device.settings:
channel_name = device.settings[mode][int(block.channel)].get("name")
@ -114,8 +127,8 @@ def get_device_channel_name(
return f"{entity_name} channel {chr(int(block.channel)+base)}"
def is_momentary_input(settings: dict[str, Any], block: aioshelly.Block) -> bool:
"""Return true if input button settings is set to a momentary type."""
def is_block_momentary_input(settings: dict[str, Any], block: Block) -> bool:
"""Return true if block input button settings is set to a momentary type."""
# Shelly Button type is fixed to momentary and no btn_type
if settings["device"]["type"] in SHBTN_MODELS:
return True
@ -136,9 +149,9 @@ def is_momentary_input(settings: dict[str, Any], block: aioshelly.Block) -> bool
return button_type in ["momentary", "momentary_on_release"]
def get_device_uptime(status: dict[str, Any], last_uptime: str | None) -> str:
def get_device_uptime(uptime: float, last_uptime: str | None) -> str:
"""Return device uptime string, tolerate up to 5 seconds deviation."""
delta_uptime = utcnow() - timedelta(seconds=status["uptime"])
delta_uptime = utcnow() - timedelta(seconds=uptime)
if (
not last_uptime
@ -150,14 +163,14 @@ def get_device_uptime(status: dict[str, Any], last_uptime: str | None) -> str:
return last_uptime
def get_input_triggers(
device: aioshelly.Device, block: aioshelly.Block
def get_block_input_triggers(
device: BlockDevice, block: Block
) -> list[tuple[str, str]]:
"""Return list of input triggers for block."""
if "inputEvent" not in block.sensor_ids or "inputEventCnt" not in block.sensor_ids:
return []
if not is_momentary_input(device.settings, block):
if not is_block_momentary_input(device.settings, block):
return []
triggers = []
@ -165,6 +178,7 @@ def get_input_triggers(
if block.type == "device" or get_number_of_channels(device, block) == 1:
subtype = "button"
else:
assert block.channel
subtype = f"button{int(block.channel)+1}"
if device.settings["device"]["type"] in SHBTN_MODELS:
@ -180,10 +194,20 @@ def get_input_triggers(
return triggers
def get_shbtn_input_triggers() -> list[tuple[str, str]]:
"""Return list of input triggers for SHBTN models."""
triggers = []
for trigger_type in SHBTN_INPUTS_EVENTS_TYPES:
triggers.append((trigger_type, "button"))
return triggers
@singleton.singleton("shelly_coap")
async def get_coap_context(hass: HomeAssistant) -> aioshelly.COAP:
async def get_coap_context(hass: HomeAssistant) -> COAP:
"""Get CoAP context to be used in all Shelly devices."""
context = aioshelly.COAP()
context = COAP()
if DOMAIN in hass.data:
port = hass.data[DOMAIN].get(CONF_COAP_PORT, DEFAULT_COAP_PORT)
else:
@ -200,7 +224,7 @@ async def get_coap_context(hass: HomeAssistant) -> aioshelly.COAP:
return context
def get_device_sleep_period(settings: dict[str, Any]) -> int:
def get_block_device_sleep_period(settings: dict[str, Any]) -> int:
"""Return the device sleep period in seconds or 0 for non sleeping devices."""
sleep_period = 0
@ -210,3 +234,114 @@ def get_device_sleep_period(settings: dict[str, Any]) -> int:
sleep_period *= 60 # hours to minutes
return sleep_period * 60 # minutes to seconds
def get_info_auth(info: dict[str, Any]) -> bool:
"""Return true if device has authorization enabled."""
return cast(bool, info.get("auth") or info.get("auth_en"))
def get_info_gen(info: dict[str, Any]) -> int:
"""Return the device generation from shelly info."""
return int(info.get("gen", 1))
def get_model_name(info: dict[str, Any]) -> str:
"""Return the device model name."""
if get_info_gen(info) == 2:
return cast(str, MODEL_NAMES.get(info["model"], info["model"]))
return cast(str, MODEL_NAMES.get(info["type"], info["type"]))
def get_rpc_channel_name(device: RpcDevice, key: str) -> str:
"""Get name based on device and channel name."""
key = key.replace("input", "switch")
device_name = get_rpc_device_name(device)
entity_name: str | None = device.config[key].get("name", device_name)
if entity_name is None:
return f"{device_name} {key.replace(':', '_')}"
return entity_name
def get_rpc_entity_name(
device: RpcDevice, key: str, description: str | None = None
) -> str:
"""Naming for RPC based switch and sensors."""
channel_name = get_rpc_channel_name(device, key)
if description:
return f"{channel_name} {description}"
return channel_name
def get_device_entry_gen(entry: ConfigEntry) -> int:
"""Return the device generation from config entry."""
return entry.data.get("gen", 1)
def get_rpc_key_instances(keys_dict: dict[str, Any], key: str) -> list[str]:
"""Return list of key instances for RPC device from a dict."""
if key in keys_dict:
return [key]
keys_list: list[str] = []
for i in range(MAX_RPC_KEY_INSTANCES):
key_inst = f"{key}:{i}"
if key_inst not in keys_dict:
return keys_list
keys_list.append(key_inst)
return keys_list
def get_rpc_key_ids(keys_dict: dict[str, Any], key: str) -> list[int]:
"""Return list of key ids for RPC device from a dict."""
key_ids: list[int] = []
for i in range(MAX_RPC_KEY_INSTANCES):
key_inst = f"{key}:{i}"
if key_inst not in keys_dict:
return key_ids
key_ids.append(i)
return key_ids
def is_rpc_momentary_input(config: dict[str, Any], key: str) -> bool:
"""Return true if rpc input button settings is set to a momentary type."""
return cast(bool, config[key]["type"] == "button")
def is_block_channel_type_light(settings: dict[str, Any], channel: int) -> bool:
"""Return true if block channel appliance type is set to light."""
app_type = settings["relays"][channel].get("appliance_type")
return app_type is not None and app_type.lower().startswith("light")
def is_rpc_channel_type_light(config: dict[str, Any], channel: int) -> bool:
"""Return true if rpc channel consumption type is set to light."""
con_types = config["sys"]["ui_data"].get("consumption_types")
return con_types is not None and con_types[channel].lower().startswith("light")
def get_rpc_input_triggers(device: RpcDevice) -> list[tuple[str, str]]:
"""Return list of input triggers for RPC device."""
triggers = []
key_ids = get_rpc_key_ids(device.config, "input")
for id_ in key_ids:
key = f"input:{id_}"
if not is_rpc_momentary_input(device.config, key):
continue
for trigger_type in RPC_INPUTS_EVENTS_TYPES:
subtype = f"button{id_+1}"
triggers.append((trigger_type, subtype))
return triggers

2
requirements_all.txt
View file

@ -240,7 +240,7 @@ aiopylgtv==0.4.0
aiorecollect==1.0.8
# homeassistant.components.shelly
aioshelly==0.6.4
aioshelly==1.0.1
# homeassistant.components.switcher_kis
aioswitcher==2.0.6

2
requirements_test_all.txt
View file

@ -164,7 +164,7 @@ aiopylgtv==0.4.0
aiorecollect==1.0.8
# homeassistant.components.shelly
aioshelly==0.6.4
aioshelly==1.0.1
# homeassistant.components.switcher_kis
aioswitcher==2.0.6

59
tests/components/shelly/conftest.py
View file

@ -3,12 +3,13 @@ from unittest.mock import AsyncMock, Mock, patch
import pytest
from homeassistant.components.shelly import ShellyDeviceWrapper
from homeassistant.components.shelly import BlockDeviceWrapper, RpcDeviceWrapper
from homeassistant.components.shelly.const import (
COAP,
BLOCK,
DATA_CONFIG_ENTRY,
DOMAIN,
EVENT_SHELLY_CLICK,
RPC,
)
from homeassistant.setup import async_setup_component
@ -54,6 +55,14 @@ MOCK_BLOCKS = [
),
]
MOCK_CONFIG = {
"input:0": {"id": 0, "type": "button"},
"switch:0": {"name": "test switch_0"},
"sys": {"ui_data": {}},
"wifi": {
"ap": {"ssid": "Test name"},
},
}
MOCK_SHELLY = {
"mac": "test-mac",
@ -62,6 +71,10 @@ MOCK_SHELLY = {
"num_outputs": 2,
}
MOCK_STATUS = {
"switch:0": {"output": True},
}
@pytest.fixture(autouse=True)
def mock_coap():
@ -104,6 +117,7 @@ async def coap_wrapper(hass):
blocks=MOCK_BLOCKS,
settings=MOCK_SETTINGS,
shelly=MOCK_SHELLY,
firmware_version="some fw string",
update=AsyncMock(),
initialized=True,
)
@ -111,9 +125,44 @@ async def coap_wrapper(hass):
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
COAP
] = ShellyDeviceWrapper(hass, config_entry, device)
BLOCK
] = BlockDeviceWrapper(hass, config_entry, device)
await wrapper.async_setup()
wrapper.async_setup()
return wrapper
@pytest.fixture
async def rpc_wrapper(hass):
"""Setups a coap wrapper with mocked device."""
await async_setup_component(hass, "shelly", {})
config_entry = MockConfigEntry(
domain=DOMAIN,
data={"sleep_period": 0, "model": "SNSW-001P16EU", "gen": 2},
unique_id="12345678",
)
config_entry.add_to_hass(hass)
device = Mock(
call_rpc=AsyncMock(),
config=MOCK_CONFIG,
event={},
shelly=MOCK_SHELLY,
status=MOCK_STATUS,
firmware_version="some fw string",
update=AsyncMock(),
initialized=True,
shutdown=AsyncMock(),
)
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
RPC
] = RpcDeviceWrapper(hass, config_entry, device)
wrapper.async_setup()
return wrapper

93
tests/components/shelly/test_config_flow.py
View file

@ -20,9 +20,13 @@ DISCOVERY_INFO = {
"name": "shelly1pm-12345",
"properties": {"id": "shelly1pm-12345"},
}
MOCK_CONFIG = {
"wifi": {"ap": {"ssid": "Test name"}},
}
async def test_form(hass):
@pytest.mark.parametrize("gen", [1, 2])
async def test_form(hass, gen):
"""Test we get the form."""
await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
@ -32,15 +36,25 @@ async def test_form(hass):
assert result["errors"] == {}
with patch(
"aioshelly.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False, "gen": gen},
), patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
), patch(
"aioshelly.rpc_device.RpcDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
config=MOCK_CONFIG,
shutdown=AsyncMock(),
)
),
), patch(
"homeassistant.components.shelly.async_setup", return_value=True
) as mock_setup, patch(
@ -59,6 +73,7 @@ async def test_form(hass):
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 0,
"gen": gen,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -78,12 +93,13 @@ async def test_title_without_name(hass):
settings["device"] = settings["device"].copy()
settings["device"]["hostname"] = "shelly1pm-12345"
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
), patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=settings,
)
),
@ -105,6 +121,7 @@ async def test_title_without_name(hass):
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 0,
"gen": 1,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -119,7 +136,7 @@ async def test_form_auth(hass):
assert result["errors"] == {}
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": True},
):
result2 = await hass.config_entries.flow.async_configure(
@ -131,9 +148,10 @@ async def test_form_auth(hass):
assert result["errors"] == {}
with patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
@ -155,6 +173,7 @@ async def test_form_auth(hass):
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 0,
"gen": 1,
"username": "test username",
"password": "test password",
}
@ -172,7 +191,7 @@ async def test_form_errors_get_info(hass, error):
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("aioshelly.get_info", side_effect=exc):
with patch("aioshelly.common.get_info", side_effect=exc):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"host": "1.1.1.1"},
@ -193,8 +212,10 @@ async def test_form_errors_test_connection(hass, error):
)
with patch(
"aioshelly.get_info", return_value={"mac": "test-mac", "auth": False}
), patch("aioshelly.Device.create", new=AsyncMock(side_effect=exc)):
"aioshelly.common.get_info", return_value={"mac": "test-mac", "auth": False}
), patch(
"aioshelly.block_device.BlockDevice.create", new=AsyncMock(side_effect=exc)
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"host": "1.1.1.1"},
@ -217,7 +238,7 @@ async def test_form_already_configured(hass):
)
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
):
result2 = await hass.config_entries.flow.async_configure(
@ -252,12 +273,13 @@ async def test_user_setup_ignored_device(hass):
settings["fw"] = "20201124-092534/v1.9.0@57ac4ad8"
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
), patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=settings,
)
),
@ -287,7 +309,10 @@ async def test_form_firmware_unsupported(hass):
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("aioshelly.get_info", side_effect=aioshelly.FirmwareUnsupported):
with patch(
"aioshelly.common.get_info",
side_effect=aioshelly.exceptions.FirmwareUnsupported,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"host": "1.1.1.1"},
@ -313,14 +338,17 @@ async def test_form_auth_errors_test_connection(hass, error):
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch("aioshelly.get_info", return_value={"mac": "test-mac", "auth": True}):
with patch(
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "auth": True},
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{"host": "1.1.1.1"},
)
with patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(side_effect=exc),
):
result3 = await hass.config_entries.flow.async_configure(
@ -336,12 +364,13 @@ async def test_zeroconf(hass):
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
), patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
@ -378,6 +407,7 @@ async def test_zeroconf(hass):
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 0,
"gen": 1,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -388,7 +418,7 @@ async def test_zeroconf_sleeping_device(hass):
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={
"mac": "test-mac",
"type": "SHSW-1",
@ -396,9 +426,10 @@ async def test_zeroconf_sleeping_device(hass):
"sleep_mode": True,
},
), patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings={
"name": "Test name",
"device": {
@ -442,6 +473,7 @@ async def test_zeroconf_sleeping_device(hass):
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 600,
"gen": 1,
}
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
@ -460,7 +492,7 @@ async def test_zeroconf_sleeping_device_error(hass, error):
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={
"mac": "test-mac",
"type": "SHSW-1",
@ -468,7 +500,7 @@ async def test_zeroconf_sleeping_device_error(hass, error):
"sleep_mode": True,
},
), patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(side_effect=exc),
):
result = await hass.config_entries.flow.async_init(
@ -489,7 +521,7 @@ async def test_zeroconf_already_configured(hass):
entry.add_to_hass(hass)
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": False},
):
result = await hass.config_entries.flow.async_init(
@ -506,7 +538,10 @@ async def test_zeroconf_already_configured(hass):
async def test_zeroconf_firmware_unsupported(hass):
"""Test we abort if device firmware is unsupported."""
with patch("aioshelly.get_info", side_effect=aioshelly.FirmwareUnsupported):
with patch(
"aioshelly.common.get_info",
side_effect=aioshelly.exceptions.FirmwareUnsupported,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DISCOVERY_INFO,
@ -519,7 +554,7 @@ async def test_zeroconf_firmware_unsupported(hass):
async def test_zeroconf_cannot_connect(hass):
"""Test we get the form."""
with patch("aioshelly.get_info", side_effect=asyncio.TimeoutError):
with patch("aioshelly.common.get_info", side_effect=asyncio.TimeoutError):
result = await hass.config_entries.flow.async_init(
DOMAIN,
data=DISCOVERY_INFO,
@ -534,7 +569,7 @@ async def test_zeroconf_require_auth(hass):
await setup.async_setup_component(hass, "persistent_notification", {})
with patch(
"aioshelly.get_info",
"aioshelly.common.get_info",
return_value={"mac": "test-mac", "type": "SHSW-1", "auth": True},
):
result = await hass.config_entries.flow.async_init(
@ -546,9 +581,10 @@ async def test_zeroconf_require_auth(hass):
assert result["errors"] == {}
with patch(
"aioshelly.Device.create",
"aioshelly.block_device.BlockDevice.create",
new=AsyncMock(
return_value=Mock(
model="SHSW-1",
settings=MOCK_SETTINGS,
)
),
@ -570,6 +606,7 @@ async def test_zeroconf_require_auth(hass):
"host": "1.1.1.1",
"model": "SHSW-1",
"sleep_period": 0,
"gen": 1,
"username": "test username",
"password": "test password",
}

155
tests/components/shelly/test_device_trigger.py
View file

@ -8,11 +8,11 @@ from homeassistant.components import automation
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.shelly import ShellyDeviceWrapper
from homeassistant.components.shelly import BlockDeviceWrapper
from homeassistant.components.shelly.const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
COAP,
BLOCK,
CONF_SUBTYPE,
DATA_CONFIG_ENTRY,
DOMAIN,
@ -30,8 +30,8 @@ from tests.common import (
)
async def test_get_triggers(hass, coap_wrapper):
"""Test we get the expected triggers from a shelly."""
async def test_get_triggers_block_device(hass, coap_wrapper):
"""Test we get the expected triggers from a shelly block device."""
assert coap_wrapper
expected_triggers = [
{
@ -57,6 +57,54 @@ async def test_get_triggers(hass, coap_wrapper):
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_rpc_device(hass, rpc_wrapper):
"""Test we get the expected triggers from a shelly RPC device."""
assert rpc_wrapper
expected_triggers = [
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "btn_down",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "btn_up",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "double_push",
CONF_SUBTYPE: "button1",
},
{
CONF_PLATFORM: "device",
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_DOMAIN: DOMAIN,
CONF_TYPE: "long_push",
CONF_SUBTYPE: "button1",
},
]
triggers = await async_get_device_automations(
hass, "trigger", rpc_wrapper.device_id
)
assert_lists_same(triggers, expected_triggers)
async def test_get_triggers_button(hass):
"""Test we get the expected triggers from a shelly button."""
await async_setup_component(hass, "shelly", {})
@ -79,10 +127,10 @@ async def test_get_triggers_button(hass):
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id] = {}
coap_wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][
COAP
] = ShellyDeviceWrapper(hass, config_entry, device)
BLOCK
] = BlockDeviceWrapper(hass, config_entry, device)
await coap_wrapper.async_setup()
coap_wrapper.async_setup()
expected_triggers = [
{
@ -136,8 +184,8 @@ async def test_get_triggers_for_invalid_device_id(hass, device_reg, coap_wrapper
await async_get_device_automations(hass, "trigger", invalid_device.id)
async def test_if_fires_on_click_event(hass, calls, coap_wrapper):
"""Test for click_event trigger firing."""
async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper):
"""Test for click_event trigger firing for block device."""
assert coap_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
@ -175,8 +223,47 @@ async def test_if_fires_on_click_event(hass, calls, coap_wrapper):
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
"""Test for click_event with no device."""
async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper):
"""Test for click_event trigger firing for rpc device."""
assert rpc_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: rpc_wrapper.device_id,
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_push"},
},
},
]
},
)
message = {
CONF_DEVICE_ID: rpc_wrapper.device_id,
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_single_push"
async def test_validate_trigger_block_device_not_ready(hass, calls, coap_wrapper):
"""Test validate trigger config when block device is not ready."""
assert coap_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
@ -189,7 +276,7 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: "no_device",
CONF_DEVICE_ID: "device_not_ready",
CONF_TYPE: "single",
CONF_SUBTYPE: "button1",
},
@ -201,7 +288,11 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
]
},
)
message = {CONF_DEVICE_ID: "no_device", ATTR_CLICK_TYPE: "single", ATTR_CHANNEL: 1}
message = {
CONF_DEVICE_ID: "device_not_ready",
ATTR_CLICK_TYPE: "single",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
@ -209,6 +300,44 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper):
assert calls[0].data["some"] == "test_trigger_single_click"
async def test_validate_trigger_rpc_device_not_ready(hass, calls, rpc_wrapper):
"""Test validate trigger config when RPC device is not ready."""
assert rpc_wrapper
await setup.async_setup_component(hass, "persistent_notification", {})
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
CONF_PLATFORM: "device",
CONF_DOMAIN: DOMAIN,
CONF_DEVICE_ID: "device_not_ready",
CONF_TYPE: "single_push",
CONF_SUBTYPE: "button1",
},
"action": {
"service": "test.automation",
"data_template": {"some": "test_trigger_single_push"},
},
},
]
},
)
message = {
CONF_DEVICE_ID: "device_not_ready",
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
}
hass.bus.async_fire(EVENT_SHELLY_CLICK, message)
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "test_trigger_single_push"
async def test_validate_trigger_invalid_triggers(hass, coap_wrapper):
"""Test for click_event with invalid triggers."""
assert coap_wrapper

59
tests/components/shelly/test_logbook.py
View file

@ -13,8 +13,8 @@ from homeassistant.setup import async_setup_component
from tests.components.logbook.test_init import MockLazyEventPartialState
async def test_humanify_shelly_click_event(hass, coap_wrapper):
"""Test humanifying Shelly click event."""
async def test_humanify_shelly_click_event_block_device(hass, coap_wrapper):
"""Test humanifying Shelly click event for block device."""
assert coap_wrapper
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
@ -51,12 +51,63 @@ async def test_humanify_shelly_click_event(hass, coap_wrapper):
assert event1["name"] == "Shelly"
assert event1["domain"] == DOMAIN
assert (
event1["message"] == "'single' click event for Test name channel 1 was fired."
event1["message"]
== "'single' click event for Test name channel 1 Input was fired."
)
assert event2["name"] == "Shelly"
assert event2["domain"] == DOMAIN
assert (
event2["message"]
== "'long' click event for shellyswitch25-12345678 channel 2 was fired."
== "'long' click event for shellyswitch25-12345678 channel 2 Input was fired."
)
async def test_humanify_shelly_click_event_rpc_device(hass, rpc_wrapper):
"""Test humanifying Shelly click event for rpc device."""
assert rpc_wrapper
hass.config.components.add("recorder")
assert await async_setup_component(hass, "logbook", {})
entity_attr_cache = logbook.EntityAttributeCache(hass)
event1, event2 = list(
logbook.humanify(
hass,
[
MockLazyEventPartialState(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: rpc_wrapper.device_id,
ATTR_DEVICE: "shellyplus1pm-12345678",
ATTR_CLICK_TYPE: "single_push",
ATTR_CHANNEL: 1,
},
),
MockLazyEventPartialState(
EVENT_SHELLY_CLICK,
{
ATTR_DEVICE_ID: "no_device_id",
ATTR_DEVICE: "shellypro4pm-12345678",
ATTR_CLICK_TYPE: "btn_down",
ATTR_CHANNEL: 2,
},
),
],
entity_attr_cache,
{},
)
)
assert event1["name"] == "Shelly"
assert event1["domain"] == DOMAIN
assert (
event1["message"]
== "'single_push' click event for test switch_0 Input was fired."
)
assert event2["name"] == "Shelly"
assert event2["domain"] == DOMAIN
assert (
event2["message"]
== "'btn_down' click event for shellypro4pm-12345678 channel 2 Input was fired."
)

74
tests/components/shelly/test_switch.py
View file

@ -11,8 +11,8 @@ from homeassistant.const import (
RELAY_BLOCK_ID = 0
async def test_services(hass, coap_wrapper):
"""Test device turn on/off services."""
async def test_block_device_services(hass, coap_wrapper):
"""Test block device turn on/off services."""
assert coap_wrapper
hass.async_create_task(
@ -37,8 +37,8 @@ async def test_services(hass, coap_wrapper):
assert hass.states.get("switch.test_name_channel_1").state == STATE_OFF
async def test_update(hass, coap_wrapper, monkeypatch):
"""Test device update."""
async def test_block_device_update(hass, coap_wrapper, monkeypatch):
"""Test block device update."""
assert coap_wrapper
hass.async_create_task(
@ -61,8 +61,8 @@ async def test_update(hass, coap_wrapper, monkeypatch):
assert hass.states.get("switch.test_name_channel_1").state == STATE_ON
async def test_no_relay_blocks(hass, coap_wrapper, monkeypatch):
"""Test device without relay blocks."""
async def test_block_device_no_relay_blocks(hass, coap_wrapper, monkeypatch):
"""Test block device without relay blocks."""
assert coap_wrapper
monkeypatch.setattr(coap_wrapper.device.blocks[RELAY_BLOCK_ID], "type", "roller")
@ -73,8 +73,8 @@ async def test_no_relay_blocks(hass, coap_wrapper, monkeypatch):
assert hass.states.get("switch.test_name_channel_1") is None
async def test_device_mode_roller(hass, coap_wrapper, monkeypatch):
"""Test switch device in roller mode."""
async def test_block_device_mode_roller(hass, coap_wrapper, monkeypatch):
"""Test block device in roller mode."""
assert coap_wrapper
monkeypatch.setitem(coap_wrapper.device.settings, "mode", "roller")
@ -83,3 +83,61 @@ async def test_device_mode_roller(hass, coap_wrapper, monkeypatch):
)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_channel_1") is None
async def test_block_device_app_type_light(hass, coap_wrapper, monkeypatch):
"""Test block device in app type set to light mode."""
assert coap_wrapper
monkeypatch.setitem(
coap_wrapper.device.settings["relays"][0], "appliance_type", "light"
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(coap_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
assert hass.states.get("switch.test_name_channel_1") is None
async def test_rpc_device_services(hass, rpc_wrapper, monkeypatch):
"""Test RPC device turn on/off services."""
assert rpc_wrapper
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.test_switch_0"},
blocking=True,
)
assert hass.states.get("switch.test_switch_0").state == STATE_ON
monkeypatch.setitem(rpc_wrapper.device.status["switch:0"], "output", False)
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_OFF,
{ATTR_ENTITY_ID: "switch.test_switch_0"},
blocking=True,
)
rpc_wrapper.async_set_updated_data("")
assert hass.states.get("switch.test_switch_0").state == STATE_OFF
async def test_rpc_device_switch_type_lights_mode(hass, rpc_wrapper, monkeypatch):
"""Test RPC device with switch in consumption type lights mode."""
assert rpc_wrapper
monkeypatch.setitem(
rpc_wrapper.device.config["sys"]["ui_data"],
"consumption_types",
["lights"],
)
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(rpc_wrapper.entry, SWITCH_DOMAIN)
)
await hass.async_block_till_done()
assert hass.states.get("switch.test_switch_0") is None