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

Improve typing in Shelly integration (#52544)

Browse source
This commit is contained in:
Maciej Bieniek 2021-07-21 19:11:44 +02:00 • committed by GitHub
parent f128bc9ef8
commit 772cbd59d7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 365 additions and 245 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -74,6 +74,7 @@ homeassistant.components.remote.*
homeassistant.components.scene.*
homeassistant.components.select.*
homeassistant.components.sensor.*
homeassistant.components.shelly.*
homeassistant.components.slack.*
homeassistant.components.sonos.media_player
homeassistant.components.ssdp.*

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

@ -1,7 +1,10 @@
"""The Shelly integration."""
from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import Any, Final, cast
import aioshelly
import async_timeout
@ -15,10 +18,11 @@ from homeassistant.const import (
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import aiohttp_client, device_registry, update_coordinator
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import (
AIOSHELLY_DEVICE_TIMEOUT_SEC,
@ -43,19 +47,19 @@ from .const import (
)
from .utils import get_coap_context, get_device_name, get_device_sleep_period
PLATFORMS = ["binary_sensor", "cover", "light", "sensor", "switch"]
SLEEPING_PLATFORMS = ["binary_sensor", "sensor"]
_LOGGER = logging.getLogger(__name__)
PLATFORMS: Final = ["binary_sensor", "cover", "light", "sensor", "switch"]
SLEEPING_PLATFORMS: Final = ["binary_sensor", "sensor"]
_LOGGER: Final = logging.getLogger(__name__)
COAP_SCHEMA = vol.Schema(
COAP_SCHEMA: Final = vol.Schema(
{
vol.Optional(CONF_COAP_PORT, default=DEFAULT_COAP_PORT): cv.port,
}
)
CONFIG_SCHEMA = vol.Schema({DOMAIN: COAP_SCHEMA}, extra=vol.ALLOW_EXTRA)
CONFIG_SCHEMA: Final = vol.Schema({DOMAIN: COAP_SCHEMA}, extra=vol.ALLOW_EXTRA)
async def async_setup(hass: HomeAssistant, config: dict):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Shelly component."""
hass.data[DOMAIN] = {DATA_CONFIG_ENTRY: {}}
@ -113,7 +117,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
sleep_period = entry.data.get("sleep_period")
@callback
def _async_device_online(_):
def _async_device_online(_: Any) -> None:
_LOGGER.debug("Device %s is online, resuming setup", entry.title)
hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][DEVICE] = None
@ -153,7 +157,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_device_setup(
hass: HomeAssistant, entry: ConfigEntry, device: aioshelly.Device
):
) -> None:
"""Set up a device that is online."""
device_wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][
COAP
@ -174,9 +178,11 @@ async def async_device_setup(
class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
"""Wrapper for a Shelly device with Home Assistant specific functions."""
def __init__(self, hass, entry, device: aioshelly.Device):
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, device: aioshelly.Device
) -> None:
"""Initialize the Shelly device wrapper."""
self.device_id = None
self.device_id: str | None = None
sleep_period = entry.data["sleep_period"]
if sleep_period:
@ -205,7 +211,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop)
@callback
def _async_device_updates_handler(self):
def _async_device_updates_handler(self) -> None:
"""Handle device updates."""
if not self.device.initialized:
return
@ -258,7 +264,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
self.name,
)
async def _async_update_data(self):
async def _async_update_data(self) -> None:
"""Fetch data."""
if self.entry.data.get("sleep_period"):
# Sleeping device, no point polling it, just mark it unavailable
@ -267,21 +273,21 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
_LOGGER.debug("Polling Shelly Device - %s", self.name)
try:
async with async_timeout.timeout(POLLING_TIMEOUT_SEC):
return await self.device.update()
await self.device.update()
except OSError as err:
raise update_coordinator.UpdateFailed("Error fetching data") from err
@property
def model(self):
def model(self) -> str:
"""Model of the device."""
return self.entry.data["model"]
return cast(str, self.entry.data["model"])
@property
def mac(self):
def mac(self) -> str:
"""Mac address of the device."""
return self.entry.unique_id
return cast(str, self.entry.unique_id)
async def async_setup(self):
async 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 ""
@ -298,7 +304,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
self.device_id = entry.id
self.device.subscribe_updates(self.async_set_updated_data)
def shutdown(self):
def shutdown(self) -> None:
"""Shutdown the wrapper."""
if self.device:
self.device.shutdown()
@ -306,7 +312,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
self.device = None
@callback
def _handle_ha_stop(self, _):
def _handle_ha_stop(self, _event: Event) -> None:
"""Handle Home Assistant stopping."""
_LOGGER.debug("Stopping ShellyDeviceWrapper for %s", self.name)
self.shutdown()
@ -315,7 +321,7 @@ class ShellyDeviceWrapper(update_coordinator.DataUpdateCoordinator):
class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
"""Rest Wrapper for a Shelly device with Home Assistant specific functions."""
def __init__(self, hass, device: aioshelly.Device):
def __init__(self, hass: HomeAssistant, device: aioshelly.Device) -> None:
"""Initialize the Shelly device wrapper."""
if (
device.settings["device"]["type"]
@ -335,22 +341,22 @@ class ShellyDeviceRestWrapper(update_coordinator.DataUpdateCoordinator):
)
self.device = device
async def _async_update_data(self):
async def _async_update_data(self) -> None:
"""Fetch data."""
try:
async with async_timeout.timeout(AIOSHELLY_DEVICE_TIMEOUT_SEC):
_LOGGER.debug("REST update for %s", self.name)
return await self.device.update_status()
await self.device.update_status()
except OSError as err:
raise update_coordinator.UpdateFailed("Error fetching data") from err
@property
def mac(self):
def mac(self) -> str:
"""Mac address of the device."""
return self.device.settings["device"]["mac"]
return cast(str, self.device.settings["device"]["mac"])
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
device = hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id].get(DEVICE)
if device is not None:
@ -370,3 +376,21 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
hass.data[DOMAIN][DATA_CONFIG_ENTRY].pop(entry.entry_id)
return unload_ok
def get_device_wrapper(
hass: HomeAssistant, device_id: str
) -> ShellyDeviceWrapper | None:
"""Get a Shelly 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)
if wrapper and wrapper.device_id == device_id:
return wrapper
return None

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

@ -1,4 +1,8 @@
"""Binary sensor for Shelly."""
from __future__ import annotations
from typing import Final
from homeassistant.components.binary_sensor import (
DEVICE_CLASS_CONNECTIVITY,
DEVICE_CLASS_GAS,
@ -12,6 +16,9 @@ from homeassistant.components.binary_sensor import (
STATE_ON,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .entity import (
BlockAttributeDescription,
@ -24,7 +31,7 @@ from .entity import (
)
from .utils import is_momentary_input
SENSORS = {
SENSORS: Final = {
("device", "overtemp"): BlockAttributeDescription(
name="Overheating", device_class=DEVICE_CLASS_PROBLEM
),
@ -83,7 +90,7 @@ SENSORS = {
),
}
REST_SENSORS = {
REST_SENSORS: Final = {
"cloud": RestAttributeDescription(
name="Cloud",
value=lambda status, _: status["cloud"]["connected"],
@ -103,7 +110,11 @@ REST_SENSORS = {
}
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensors for device."""
if config_entry.data["sleep_period"]:
await async_setup_entry_attribute_entities(
@ -130,7 +141,7 @@ class ShellyBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity):
"""Shelly binary sensor entity."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if sensor state is on."""
return bool(self.attribute_value)
@ -139,7 +150,7 @@ class ShellyRestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity):
"""Shelly REST binary sensor entity."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if REST sensor state is on."""
return bool(self.attribute_value)
@ -150,7 +161,7 @@ class ShellySleepingBinarySensor(
"""Represent a shelly sleeping binary sensor."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if sensor state is on."""
if self.block is not None:
return bool(self.attribute_value)

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

@ -1,6 +1,9 @@
"""Config flow for Shelly integration."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Final, cast
import aiohttp
import aioshelly
@ -14,19 +17,23 @@ from homeassistant.const import (
CONF_USERNAME,
HTTP_UNAUTHORIZED,
)
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
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
HOST_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
HOST_SCHEMA: Final = vol.Schema({vol.Required(CONF_HOST): str})
HTTP_CONNECT_ERRORS = (asyncio.TimeoutError, aiohttp.ClientError)
HTTP_CONNECT_ERRORS: Final = (asyncio.TimeoutError, aiohttp.ClientError)
async def validate_input(hass: core.HomeAssistant, host, data):
async def validate_input(
hass: core.HomeAssistant, host: str, 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.
@ -60,15 +67,17 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
host = None
info = None
device_info = None
host: str = ""
info: dict[str, Any] = {}
device_info: dict[str, Any] = {}
async def async_step_user(self, user_input=None):
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
host = user_input[CONF_HOST]
host: str = user_input[CONF_HOST]
try:
info = await self._async_get_info(host)
except HTTP_CONNECT_ERRORS:
@ -106,9 +115,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=HOST_SCHEMA, errors=errors
)
async def async_step_credentials(self, user_input=None):
async def async_step_credentials(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the credentials step."""
errors = {}
errors: dict[str, str] = {}
if user_input is not None:
try:
device_info = await validate_input(self.hass, self.host, user_input)
@ -146,7 +157,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="credentials", data_schema=schema, errors=errors
)
async def async_step_zeroconf(self, discovery_info):
async def async_step_zeroconf(
self, discovery_info: DiscoveryInfoType
) -> FlowResult:
"""Handle zeroconf discovery."""
try:
self.info = info = await self._async_get_info(discovery_info["host"])
@ -173,9 +186,11 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm_discovery()
async def async_step_confirm_discovery(self, user_input=None):
async def async_step_confirm_discovery(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle discovery confirm."""
errors = {}
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"],
@ -199,10 +214,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def _async_get_info(self, host):
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 await aioshelly.get_info(
aiohttp_client.async_get_clientsession(self.hass),
host,
return cast(
Dict[str, Any],
await aioshelly.get_info(
aiohttp_client.async_get_clientsession(self.hass),
host,
),
)

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

@ -1,34 +1,37 @@
"""Constants for the Shelly integration."""
from __future__ import annotations
COAP = "coap"
DATA_CONFIG_ENTRY = "config_entry"
DEVICE = "device"
DOMAIN = "shelly"
REST = "rest"
from typing import Final
CONF_COAP_PORT = "coap_port"
DEFAULT_COAP_PORT = 5683
COAP: Final = "coap"
DATA_CONFIG_ENTRY: Final = "config_entry"
DEVICE: Final = "device"
DOMAIN: Final = "shelly"
REST: Final = "rest"
CONF_COAP_PORT: Final = "coap_port"
DEFAULT_COAP_PORT: Final = 5683
# Used in "_async_update_data" as timeout for polling data from devices.
POLLING_TIMEOUT_SEC = 18
POLLING_TIMEOUT_SEC: Final = 18
# Refresh interval for REST sensors
REST_SENSORS_UPDATE_INTERVAL = 60
REST_SENSORS_UPDATE_INTERVAL: Final = 60
# Timeout used for aioshelly calls
AIOSHELLY_DEVICE_TIMEOUT_SEC = 10
AIOSHELLY_DEVICE_TIMEOUT_SEC: Final = 10
# Multiplier used to calculate the "update_interval" for sleeping devices.
SLEEP_PERIOD_MULTIPLIER = 1.2
SLEEP_PERIOD_MULTIPLIER: Final = 1.2
# Multiplier used to calculate the "update_interval" for non-sleeping devices.
UPDATE_PERIOD_MULTIPLIER = 2.2
UPDATE_PERIOD_MULTIPLIER: Final = 2.2
# Shelly Air - Maximum work hours before lamp replacement
SHAIR_MAX_WORK_HOURS = 9000
SHAIR_MAX_WORK_HOURS: Final = 9000
# Map Shelly input events
INPUTS_EVENTS_DICT = {
INPUTS_EVENTS_DICT: Final = {
"S": "single",
"SS": "double",
"SSS": "triple",
@ -38,28 +41,20 @@ INPUTS_EVENTS_DICT = {
}
# List of battery devices that maintain a permanent WiFi connection
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION = ["SHMOS-01"]
BATTERY_DEVICES_WITH_PERMANENT_CONNECTION: Final = ["SHMOS-01"]
EVENT_SHELLY_CLICK = "shelly.click"
EVENT_SHELLY_CLICK: Final = "shelly.click"
ATTR_CLICK_TYPE = "click_type"
ATTR_CHANNEL = "channel"
ATTR_DEVICE = "device"
CONF_SUBTYPE = "subtype"
ATTR_CLICK_TYPE: Final = "click_type"
ATTR_CHANNEL: Final = "channel"
ATTR_DEVICE: Final = "device"
CONF_SUBTYPE: Final = "subtype"
BASIC_INPUTS_EVENTS_TYPES = {
"single",
"long",
}
BASIC_INPUTS_EVENTS_TYPES: Final = {"single", "long"}
SHBTN_INPUTS_EVENTS_TYPES = {
"single",
"double",
"triple",
"long",
}
SHBTN_INPUTS_EVENTS_TYPES: Final = {"single", "double", "triple", "long"}
SUPPORTED_INPUTS_EVENTS_TYPES = SHIX3_1_INPUTS_EVENTS_TYPES = {
SUPPORTED_INPUTS_EVENTS_TYPES: Final = {
"single",
"double",
"triple",
@ -68,23 +63,20 @@ SUPPORTED_INPUTS_EVENTS_TYPES = SHIX3_1_INPUTS_EVENTS_TYPES = {
"long_single",
}
INPUTS_EVENTS_SUBTYPES = {
"button": 1,
"button1": 1,
"button2": 2,
"button3": 3,
}
SHIX3_1_INPUTS_EVENTS_TYPES = SUPPORTED_INPUTS_EVENTS_TYPES
SHBTN_MODELS = ["SHBTN-1", "SHBTN-2"]
INPUTS_EVENTS_SUBTYPES: Final = {"button": 1, "button1": 1, "button2": 2, "button3": 3}
STANDARD_RGB_EFFECTS = {
SHBTN_MODELS: Final = ["SHBTN-1", "SHBTN-2"]
STANDARD_RGB_EFFECTS: Final = {
0: "Off",
1: "Meteor Shower",
2: "Gradual Change",
3: "Flash",
}
SHBLB_1_RGB_EFFECTS = {
SHBLB_1_RGB_EFFECTS: Final = {
0: "Off",
1: "Meteor Shower",
2: "Gradual Change",
@ -95,8 +87,8 @@ SHBLB_1_RGB_EFFECTS = {
}
# Kelvin value for colorTemp
KELVIN_MAX_VALUE = 6500
KELVIN_MIN_VALUE_WHITE = 2700
KELVIN_MIN_VALUE_COLOR = 3000
KELVIN_MAX_VALUE: Final = 6500
KELVIN_MIN_VALUE_WHITE: Final = 2700
KELVIN_MIN_VALUE_COLOR: Final = 3000
UPTIME_DEVIATION = 5
UPTIME_DEVIATION: Final = 5

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

@ -1,4 +1,8 @@
"""Cover for Shelly."""
from __future__ import annotations
from typing import Any, cast
from aioshelly import Block
from homeassistant.components.cover import (
@ -10,14 +14,20 @@ from homeassistant.components.cover import (
SUPPORT_STOP,
CoverEntity,
)
from homeassistant.core import callback
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
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up cover for device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][COAP]
blocks = [block for block in wrapper.device.blocks if block.type == "roller"]
@ -36,72 +46,72 @@ class ShellyCover(ShellyBlockEntity, CoverEntity):
def __init__(self, wrapper: ShellyDeviceWrapper, block: Block) -> None:
"""Initialize light."""
super().__init__(wrapper, block)
self.control_result = None
self._supported_features = SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP
self.control_result: dict[str, Any] | None = None
self._supported_features: int = SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP
if self.wrapper.device.settings["rollers"][0]["positioning"]:
self._supported_features |= SUPPORT_SET_POSITION
@property
def is_closed(self):
def is_closed(self) -> bool:
"""If cover is closed."""
if self.control_result:
return self.control_result["current_pos"] == 0
return cast(bool, self.control_result["current_pos"] == 0)
return self.block.rollerPos == 0
return cast(bool, self.block.rollerPos == 0)
@property
def current_cover_position(self):
def current_cover_position(self) -> int:
"""Position of the cover."""
if self.control_result:
return self.control_result["current_pos"]
return cast(int, self.control_result["current_pos"])
return self.block.rollerPos
return cast(int, self.block.rollerPos)
@property
def is_closing(self):
def is_closing(self) -> bool:
"""Return if the cover is closing."""
if self.control_result:
return self.control_result["state"] == "close"
return cast(bool, self.control_result["state"] == "close")
return self.block.roller == "close"
return cast(bool, self.block.roller == "close")
@property
def is_opening(self):
def is_opening(self) -> bool:
"""Return if the cover is opening."""
if self.control_result:
return self.control_result["state"] == "open"
return cast(bool, self.control_result["state"] == "open")
return self.block.roller == "open"
return cast(bool, self.block.roller == "open")
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag supported features."""
return self._supported_features
async def async_close_cover(self, **kwargs):
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close cover."""
self.control_result = await self.set_state(go="close")
self.async_write_ha_state()
async def async_open_cover(self, **kwargs):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open cover."""
self.control_result = await self.set_state(go="open")
self.async_write_ha_state()
async def async_set_cover_position(self, **kwargs):
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
self.control_result = await self.set_state(
go="to_pos", roller_pos=kwargs[ATTR_POSITION]
)
self.async_write_ha_state()
async def async_stop_cover(self, **_kwargs):
async def async_stop_cover(self, **_kwargs: Any) -> None:
"""Stop the cover."""
self.control_result = await self.set_state(go="stop")
self.async_write_ha_state()
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""When device updates, clear control result that overrides state."""
self.control_result = None
super()._update_callback()

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

@ -1,6 +1,8 @@
"""Provides device triggers for Shelly."""
from __future__ import annotations
from typing import Any, Final
import voluptuous as vol
from homeassistant.components.automation import AutomationActionType
@ -20,6 +22,7 @@ from homeassistant.const import (
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.typing import ConfigType
from . import get_device_wrapper
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
@ -31,9 +34,9 @@ from .const import (
SHBTN_MODELS,
SUPPORTED_INPUTS_EVENTS_TYPES,
)
from .utils import get_device_wrapper, get_input_triggers
from .utils import get_input_triggers
TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
TRIGGER_SCHEMA: Final = DEVICE_TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(SUPPORTED_INPUTS_EVENTS_TYPES),
vol.Required(CONF_SUBTYPE): vol.In(INPUTS_EVENTS_SUBTYPES),
@ -41,7 +44,9 @@ TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
)
async def async_validate_trigger_config(hass, config):
async def async_validate_trigger_config(
hass: HomeAssistant, config: dict[str, Any]
) -> dict[str, Any]:
"""Validate config."""
config = TRIGGER_SCHEMA(config)
@ -62,7 +67,9 @@ async def async_validate_trigger_config(hass, config):
)
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> list[dict]:
async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
"""List device triggers for Shelly devices."""
triggers = []

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

@ -4,31 +4,39 @@ from __future__ import annotations
import asyncio
from dataclasses import dataclass
import logging
from typing import Any, Callable
from typing import Any, Callable, Final, cast
import aioshelly
import async_timeout
from homeassistant.components.sensor import ATTR_STATE_CLASS
from homeassistant.core import callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import (
device_registry,
entity,
entity_registry,
update_coordinator,
)
from homeassistant.helpers.entity import DeviceInfo
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
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
async def async_setup_entry_attribute_entities(
hass, config_entry, async_add_entities, sensors, sensor_class
):
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
sensors: dict[tuple[str, str], BlockAttributeDescription],
sensor_class: Callable,
) -> None:
"""Set up entities for attributes."""
wrapper: ShellyDeviceWrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry.entry_id
@ -45,8 +53,12 @@ async def async_setup_entry_attribute_entities(
async def async_setup_block_attribute_entities(
hass, async_add_entities, wrapper, sensors, sensor_class
):
hass: HomeAssistant,
async_add_entities: AddEntitiesCallback,
wrapper: ShellyDeviceWrapper,
sensors: dict[tuple[str, str], BlockAttributeDescription],
sensor_class: Callable,
) -> None:
"""Set up entities for block attributes."""
blocks = []
@ -82,8 +94,13 @@ async def async_setup_block_attribute_entities(
async def async_restore_block_attribute_entities(
hass, config_entry, async_add_entities, wrapper, sensors, sensor_class
):
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
wrapper: ShellyDeviceWrapper,
sensors: dict[tuple[str, str], BlockAttributeDescription],
sensor_class: Callable,
) -> None:
"""Restore block attributes entities."""
entities = []
@ -117,8 +134,12 @@ async def async_restore_block_attribute_entities(
async def async_setup_entry_rest(
hass, config_entry, async_add_entities, sensors, sensor_class
):
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
sensors: dict[str, RestAttributeDescription],
sensor_class: Callable,
) -> None:
"""Set up entities for REST sensors."""
wrapper: ShellyDeviceRestWrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][
config_entry.entry_id
@ -177,53 +198,53 @@ class RestAttributeDescription:
class ShellyBlockEntity(entity.Entity):
"""Helper class to represent a block."""
def __init__(self, wrapper: ShellyDeviceWrapper, block):
def __init__(self, wrapper: ShellyDeviceWrapper, block: aioshelly.Block) -> None:
"""Initialize Shelly entity."""
self.wrapper = wrapper
self.block = block
self._name: str | None = get_entity_name(wrapper.device, block)
self._name = get_entity_name(wrapper.device, block)
@property
def name(self):
def name(self) -> str:
"""Name of entity."""
return self._name
@property
def should_poll(self):
def should_poll(self) -> bool:
"""If device should be polled."""
return False
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Device info."""
return {
"connections": {(device_registry.CONNECTION_NETWORK_MAC, self.wrapper.mac)}
}
@property
def available(self):
def available(self) -> bool:
"""Available."""
return self.wrapper.last_update_success
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return unique ID of entity."""
return f"{self.wrapper.mac}-{self.block.description}"
async def async_added_to_hass(self):
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):
async def async_update(self) -> None:
"""Update entity with latest info."""
await self.wrapper.async_request_refresh()
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""Handle device update."""
self.async_write_ha_state()
async def set_state(self, **kwargs):
async def set_state(self, **kwargs: Any) -> Any:
"""Set block state (HTTP request)."""
_LOGGER.debug("Setting state for entity %s, state: %s", self.name, kwargs)
try:
@ -261,16 +282,16 @@ class ShellyBlockAttributeEntity(ShellyBlockEntity, entity.Entity):
unit = unit(block.info(attribute))
self._unit: None | str | Callable[[dict], str] = unit
self._unique_id: None | str = f"{super().unique_id}-{self.attribute}"
self._unique_id: str = f"{super().unique_id}-{self.attribute}"
self._name = get_entity_name(wrapper.device, block, self.description.name)
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return unique ID of entity."""
return self._unique_id
@property
def name(self):
def name(self) -> str:
"""Name of sensor."""
return self._name
@ -280,27 +301,27 @@ class ShellyBlockAttributeEntity(ShellyBlockEntity, entity.Entity):
return self.description.default_enabled
@property
def attribute_value(self):
def attribute_value(self) -> StateType:
"""Value of sensor."""
value = getattr(self.block, self.attribute)
if value is None:
return None
return self.description.value(value)
return cast(StateType, self.description.value(value))
@property
def device_class(self):
def device_class(self) -> str | None:
"""Device class of sensor."""
return self.description.device_class
@property
def icon(self):
def icon(self) -> str | None:
"""Icon of sensor."""
return self.description.icon
@property
def available(self):
def available(self) -> bool:
"""Available."""
available = super().available
@ -310,7 +331,7 @@ class ShellyBlockAttributeEntity(ShellyBlockEntity, entity.Entity):
return self.description.available(self.block)
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self.description.extra_state_attributes is None:
return None
@ -336,12 +357,12 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
self._last_value = None
@property
def name(self):
def name(self) -> str:
"""Name of sensor."""
return self._name
@property
def device_info(self):
def device_info(self) -> DeviceInfo:
"""Device info."""
return {
"connections": {(device_registry.CONNECTION_NETWORK_MAC, self.wrapper.mac)}
@ -353,35 +374,36 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity):
return self.description.default_enabled
@property
def available(self):
def available(self) -> bool:
"""Available."""
return self.wrapper.last_update_success
@property
def attribute_value(self):
def attribute_value(self) -> StateType:
"""Value of sensor."""
self._last_value = self.description.value(
self.wrapper.device.status, self._last_value
)
if callable(self.description.value):
self._last_value = self.description.value(
self.wrapper.device.status, self._last_value
)
return self._last_value
@property
def device_class(self):
def device_class(self) -> str | None:
"""Device class of sensor."""
return self.description.device_class
@property
def icon(self):
def icon(self) -> str | None:
"""Icon of sensor."""
return self.description.icon
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return unique ID of entity."""
return f"{self.wrapper.mac}-{self.attribute}"
@property
def extra_state_attributes(self) -> dict | None:
def extra_state_attributes(self) -> dict[str, Any] | None:
"""Return the state attributes."""
if self.description.extra_state_attributes is None:
return None
@ -400,11 +422,11 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
attribute: str,
description: BlockAttributeDescription,
entry: entity_registry.RegistryEntry | None = None,
sensors: set | None = None,
sensors: dict[tuple[str, str], BlockAttributeDescription] | None = None,
) -> None:
"""Initialize the sleeping sensor."""
self.sensors = sensors
self.last_state = None
self.last_state: StateType = None
self.wrapper = wrapper
self.attribute = attribute
self.block = block
@ -421,9 +443,9 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
)
elif entry is not None:
self._unique_id = entry.unique_id
self._name = entry.original_name
self._name = cast(str, entry.original_name)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
@ -434,7 +456,7 @@ class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEnti
self.description.state_class = last_state.attributes.get(ATTR_STATE_CLASS)
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""Handle device update."""
if (
self.block is not None

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

@ -3,7 +3,7 @@ from __future__ import annotations
import asyncio
import logging
from typing import Any
from typing import Any, Final, cast
from aioshelly import Block
import async_timeout
@ -23,7 +23,9 @@ from homeassistant.components.light import (
LightEntity,
brightness_supported,
)
from homeassistant.core import callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.color import (
color_temperature_kelvin_to_mired,
color_temperature_mired_to_kelvin,
@ -44,10 +46,14 @@ from .const import (
from .entity import ShellyBlockEntity
from .utils import async_remove_shelly_entity
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up lights for device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][COAP]
@ -78,12 +84,12 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
def __init__(self, wrapper: ShellyDeviceWrapper, block: Block) -> None:
"""Initialize light."""
super().__init__(wrapper, block)
self.control_result = None
self.mode_result = None
self._supported_color_modes = set()
self._supported_features = 0
self._min_kelvin = KELVIN_MIN_VALUE_WHITE
self._max_kelvin = KELVIN_MAX_VALUE
self.control_result: dict[str, Any] | None = None
self.mode_result: dict[str, Any] | None = None
self._supported_color_modes: set[str] = set()
self._supported_features: int = 0
self._min_kelvin: int = KELVIN_MIN_VALUE_WHITE
self._max_kelvin: int = KELVIN_MAX_VALUE
if hasattr(block, "red") and hasattr(block, "green") and hasattr(block, "blue"):
self._min_kelvin = KELVIN_MIN_VALUE_COLOR
@ -113,18 +119,18 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
def is_on(self) -> bool:
"""If light is on."""
if self.control_result:
return self.control_result["ison"]
return cast(bool, self.control_result["ison"])
return self.block.output
return bool(self.block.output)
@property
def mode(self) -> str | None:
def mode(self) -> str:
"""Return the color mode of the light."""
if self.mode_result:
return self.mode_result["mode"]
return cast(str, self.mode_result["mode"])
if hasattr(self.block, "mode"):
return self.block.mode
return cast(str, self.block.mode)
if (
hasattr(self.block, "red")
@ -136,7 +142,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return "white"
@property
def brightness(self) -> int | None:
def brightness(self) -> int:
"""Return the brightness of this light between 0..255."""
if self.mode == "color":
if self.control_result:
@ -152,7 +158,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return round(255 * brightness_pct / 100)
@property
def color_mode(self) -> str | None:
def color_mode(self) -> str:
"""Return the color mode of the light."""
if self.mode == "color":
if hasattr(self.block, "white"):
@ -191,7 +197,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return (*self.rgb_color, white)
@property
def color_temp(self) -> int | None:
def color_temp(self) -> int:
"""Return the CT color value in mireds."""
if self.control_result:
color_temp = self.control_result["temp"]
@ -244,7 +250,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return STANDARD_RGB_EFFECTS[effect_index]
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on light."""
if self.block.type == "relay":
self.control_result = await self.set_state(turn="on")
@ -304,12 +310,12 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
self.async_write_ha_state()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off light."""
self.control_result = await self.set_state(turn="off")
self.async_write_ha_state()
async def set_light_mode(self, set_mode):
async def set_light_mode(self, set_mode: str | None) -> bool:
"""Change device mode color/white if mode has changed."""
if set_mode is None or self.mode == set_mode:
return True
@ -331,7 +337,7 @@ class ShellyLight(ShellyBlockEntity, LightEntity):
return True
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""When device updates, clear control & mode result that overrides state."""
self.control_result = None
self.mode_result = None

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

@ -1,8 +1,13 @@
"""Describe Shelly logbook events."""
from __future__ import annotations
from typing import Callable
from homeassistant.const import ATTR_DEVICE_ID
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.typing import EventType
from . import get_device_wrapper
from .const import (
ATTR_CHANNEL,
ATTR_CLICK_TYPE,
@ -10,15 +15,18 @@ from .const import (
DOMAIN,
EVENT_SHELLY_CLICK,
)
from .utils import get_device_name, get_device_wrapper
from .utils import get_device_name
@callback
def async_describe_events(hass, async_describe_event):
def async_describe_events(
hass: HomeAssistant,
async_describe_event: Callable[[str, str, Callable[[EventType], dict]], None],
) -> None:
"""Describe logbook events."""
@callback
def async_describe_shelly_click_event(event):
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:

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

@ -1,6 +1,11 @@
"""Sensor for Shelly."""
from __future__ import annotations
from typing import Final, cast
from homeassistant.components import sensor
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONCENTRATION_PARTS_PER_MILLION,
DEGREE,
@ -12,6 +17,9 @@ from homeassistant.const import (
POWER_WATT,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import SHAIR_MAX_WORK_HOURS
from .entity import (
@ -25,7 +33,7 @@ from .entity import (
)
from .utils import get_device_uptime, temperature_unit
SENSORS = {
SENSORS: Final = {
("device", "battery"): BlockAttributeDescription(
name="Battery",
unit=PERCENTAGE,
@ -153,7 +161,7 @@ SENSORS = {
value=lambda value: round(value, 1),
device_class=sensor.DEVICE_CLASS_TEMPERATURE,
state_class=sensor.STATE_CLASS_MEASUREMENT,
available=lambda block: block.extTemp != 999,
available=lambda block: cast(bool, block.extTemp != 999),
),
("sensor", "humidity"): BlockAttributeDescription(
name="Humidity",
@ -161,7 +169,7 @@ SENSORS = {
value=lambda value: round(value, 1),
device_class=sensor.DEVICE_CLASS_HUMIDITY,
state_class=sensor.STATE_CLASS_MEASUREMENT,
available=lambda block: block.extTemp != 999,
available=lambda block: cast(bool, block.extTemp != 999),
),
("sensor", "luminosity"): BlockAttributeDescription(
name="Luminosity",
@ -199,7 +207,7 @@ SENSORS = {
),
}
REST_SENSORS = {
REST_SENSORS: Final = {
"rssi": RestAttributeDescription(
name="RSSI",
unit=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
@ -217,7 +225,11 @@ REST_SENSORS = {
}
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up sensors for device."""
if config_entry.data["sleep_period"]:
await async_setup_entry_attribute_entities(
@ -236,36 +248,36 @@ class ShellySensor(ShellyBlockAttributeEntity, SensorEntity):
"""Represent a shelly sensor."""
@property
def state(self):
def state(self) -> StateType:
"""Return value of sensor."""
return self.attribute_value
@property
def state_class(self):
def state_class(self) -> str | None:
"""State class of sensor."""
return self.description.state_class
@property
def unit_of_measurement(self):
def unit_of_measurement(self) -> str | None:
"""Return unit of sensor."""
return self._unit
return cast(str, self._unit)
class ShellyRestSensor(ShellyRestAttributeEntity, SensorEntity):
"""Represent a shelly REST sensor."""
@property
def state(self):
def state(self) -> StateType:
"""Return value of sensor."""
return self.attribute_value
@property
def state_class(self):
def state_class(self) -> str | None:
"""State class of sensor."""
return self.description.state_class
@property
def unit_of_measurement(self):
def unit_of_measurement(self) -> str | None:
"""Return unit of sensor."""
return self.description.unit
@ -274,7 +286,7 @@ class ShellySleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity):
"""Represent a shelly sleeping sensor."""
@property
def state(self):
def state(self) -> StateType:
"""Return value of sensor."""
if self.block is not None:
return self.attribute_value
@ -282,11 +294,11 @@ class ShellySleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity):
return self.last_state
@property
def state_class(self):
def state_class(self) -> str | None:
"""State class of sensor."""
return self.description.state_class
@property
def unit_of_measurement(self):
def unit_of_measurement(self) -> str | None:
"""Return unit of sensor."""
return self._unit
return cast(str, self._unit)

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

@ -1,8 +1,14 @@
"""Switch for Shelly."""
from __future__ import annotations
from typing import Any, cast
from aioshelly import Block
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import callback
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
@ -10,7 +16,11 @@ from .entity import ShellyBlockEntity
from .utils import async_remove_shelly_entity
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up switches for device."""
wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][COAP]
@ -50,28 +60,28 @@ class RelaySwitch(ShellyBlockEntity, SwitchEntity):
def __init__(self, wrapper: ShellyDeviceWrapper, block: Block) -> None:
"""Initialize relay switch."""
super().__init__(wrapper, block)
self.control_result = None
self.control_result: dict[str, Any] | None = None
@property
def is_on(self) -> bool:
"""If switch is on."""
if self.control_result:
return self.control_result["ison"]
return cast(bool, self.control_result["ison"])
return self.block.output
return bool(self.block.output)
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on relay."""
self.control_result = await self.set_state(turn="on")
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off relay."""
self.control_result = await self.set_state(turn="off")
self.async_write_ha_state()
@callback
def _update_callback(self):
def _update_callback(self) -> None:
"""When device updates, clear control result that overrides state."""
self.control_result = None
super()._update_callback()

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

@ -3,19 +3,19 @@ from __future__ import annotations
from datetime import datetime, timedelta
import logging
from typing import Any, Final, cast
import aioshelly
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import singleton
from homeassistant.helpers.typing import EventType
from homeassistant.util.dt import utcnow
from .const import (
BASIC_INPUTS_EVENTS_TYPES,
COAP,
CONF_COAP_PORT,
DATA_CONFIG_ENTRY,
DEFAULT_COAP_PORT,
DOMAIN,
SHBTN_INPUTS_EVENTS_TYPES,
@ -24,10 +24,12 @@ from .const import (
UPTIME_DEVIATION,
)
_LOGGER = logging.getLogger(__name__)
_LOGGER: Final = logging.getLogger(__name__)
async def async_remove_shelly_entity(hass, domain, unique_id):
async def async_remove_shelly_entity(
hass: HomeAssistant, domain: str, unique_id: str
) -> None:
"""Remove a Shelly entity."""
entity_reg = await hass.helpers.entity_registry.async_get_registry()
entity_id = entity_reg.async_get_entity_id(domain, DOMAIN, unique_id)
@ -36,7 +38,7 @@ async def async_remove_shelly_entity(hass, domain, unique_id):
entity_reg.async_remove(entity_id)
def temperature_unit(block_info: dict) -> str:
def temperature_unit(block_info: dict[str, Any]) -> str:
"""Detect temperature unit."""
if block_info[aioshelly.BLOCK_VALUE_UNIT] == "F":
return TEMP_FAHRENHEIT
@ -45,7 +47,7 @@ def temperature_unit(block_info: dict) -> str:
def get_device_name(device: aioshelly.Device) -> str:
"""Naming for device."""
return device.settings["name"] or device.settings["device"]["hostname"]
return cast(str, device.settings["name"] or device.settings["device"]["hostname"])
def get_number_of_channels(device: aioshelly.Device, block: aioshelly.Block) -> int:
@ -96,7 +98,7 @@ def get_device_channel_name(
):
return entity_name
channel_name = None
channel_name: str | None = None
mode = block.type + "s"
if mode in device.settings:
channel_name = device.settings[mode][int(block.channel)].get("name")
@ -112,7 +114,7 @@ def get_device_channel_name(
return f"{entity_name} channel {chr(int(block.channel)+base)}"
def is_momentary_input(settings: dict, block: aioshelly.Block) -> bool:
def is_momentary_input(settings: dict[str, Any], block: aioshelly.Block) -> bool:
"""Return true if 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:
@ -134,7 +136,7 @@ def is_momentary_input(settings: dict, block: aioshelly.Block) -> bool:
return button_type in ["momentary", "momentary_on_release"]
def get_device_uptime(status: dict, last_uptime: str) -> str:
def get_device_uptime(status: dict[str, Any], last_uptime: str) -> str:
"""Return device uptime string, tolerate up to 5 seconds deviation."""
delta_uptime = utcnow() - timedelta(seconds=status["uptime"])
@ -178,22 +180,8 @@ def get_input_triggers(
return triggers
def get_device_wrapper(hass: HomeAssistant, device_id: str):
"""Get a Shelly 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 = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(COAP)
if wrapper and wrapper.device_id == device_id:
return wrapper
return None
@singleton.singleton("shelly_coap")
async def get_coap_context(hass):
async def get_coap_context(hass: HomeAssistant) -> aioshelly.COAP:
"""Get CoAP context to be used in all Shelly devices."""
context = aioshelly.COAP()
if DOMAIN in hass.data:
@ -204,7 +192,7 @@ async def get_coap_context(hass):
await context.initialize(port)
@callback
def shutdown_listener(ev):
def shutdown_listener(ev: EventType) -> None:
context.close()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, shutdown_listener)
@ -212,7 +200,7 @@ async def get_coap_context(hass):
return context
def get_device_sleep_period(settings: dict) -> int:
def get_device_sleep_period(settings: dict[str, Any]) -> int:
"""Return the device sleep period in seconds or 0 for non sleeping devices."""
sleep_period = 0

11
mypy.ini
View file

@ -825,6 +825,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.shelly.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.slack.*]
check_untyped_defs = true
disallow_incomplete_defs = true