From 772cbd59d7b9593d35ed9425fc62b4cc61958884 Mon Sep 17 00:00:00 2001 From: Maciej Bieniek Date: Wed, 21 Jul 2021 19:11:44 +0200 Subject: [PATCH] Improve typing in Shelly integration (#52544) --- .strict-typing | 1 + homeassistant/components/shelly/__init__.py | 78 ++++++++---- .../components/shelly/binary_sensor.py | 23 +++- .../components/shelly/config_flow.py | 56 ++++++--- homeassistant/components/shelly/const.py | 80 ++++++------ homeassistant/components/shelly/cover.py | 54 ++++---- .../components/shelly/device_trigger.py | 15 ++- homeassistant/components/shelly/entity.py | 116 +++++++++++------- homeassistant/components/shelly/light.py | 50 ++++---- homeassistant/components/shelly/logbook.py | 16 ++- homeassistant/components/shelly/sensor.py | 44 ++++--- homeassistant/components/shelly/switch.py | 26 ++-- homeassistant/components/shelly/utils.py | 40 +++--- mypy.ini | 11 ++ 14 files changed, 365 insertions(+), 245 deletions(-) diff --git a/.strict-typing b/.strict-typing index 40fb375ca16..735d9ca4a64 100644 --- a/.strict-typing +++ b/.strict-typing @@ -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.* diff --git a/homeassistant/components/shelly/__init__.py b/homeassistant/components/shelly/__init__.py index 425ff11399b..48e27203288 100644 --- a/homeassistant/components/shelly/__init__.py +++ b/homeassistant/components/shelly/__init__.py @@ -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 diff --git a/homeassistant/components/shelly/binary_sensor.py b/homeassistant/components/shelly/binary_sensor.py index 385b3b30c36..dd1b3a9d66d 100644 --- a/homeassistant/components/shelly/binary_sensor.py +++ b/homeassistant/components/shelly/binary_sensor.py @@ -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) diff --git a/homeassistant/components/shelly/config_flow.py b/homeassistant/components/shelly/config_flow.py index 5bf8277066c..c4ddbc0b0aa 100644 --- a/homeassistant/components/shelly/config_flow.py +++ b/homeassistant/components/shelly/config_flow.py @@ -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, + ), ) diff --git a/homeassistant/components/shelly/const.py b/homeassistant/components/shelly/const.py index 119ae478bb7..2c401829c30 100644 --- a/homeassistant/components/shelly/const.py +++ b/homeassistant/components/shelly/const.py @@ -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 diff --git a/homeassistant/components/shelly/cover.py b/homeassistant/components/shelly/cover.py index dc2dba654f3..73b8b1baae3 100644 --- a/homeassistant/components/shelly/cover.py +++ b/homeassistant/components/shelly/cover.py @@ -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() diff --git a/homeassistant/components/shelly/device_trigger.py b/homeassistant/components/shelly/device_trigger.py index e767f49bcbb..bcb909555a9 100644 --- a/homeassistant/components/shelly/device_trigger.py +++ b/homeassistant/components/shelly/device_trigger.py @@ -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 = [] diff --git a/homeassistant/components/shelly/entity.py b/homeassistant/components/shelly/entity.py index 744272ccf91..743dd07414e 100644 --- a/homeassistant/components/shelly/entity.py +++ b/homeassistant/components/shelly/entity.py @@ -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 diff --git a/homeassistant/components/shelly/light.py b/homeassistant/components/shelly/light.py index 8314650d548..047a105a30f 100644 --- a/homeassistant/components/shelly/light.py +++ b/homeassistant/components/shelly/light.py @@ -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 diff --git a/homeassistant/components/shelly/logbook.py b/homeassistant/components/shelly/logbook.py index 78a5c279a93..5b0ada6f166 100644 --- a/homeassistant/components/shelly/logbook.py +++ b/homeassistant/components/shelly/logbook.py @@ -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: diff --git a/homeassistant/components/shelly/sensor.py b/homeassistant/components/shelly/sensor.py index 8a435c3e50f..96ff6e55f8d 100644 --- a/homeassistant/components/shelly/sensor.py +++ b/homeassistant/components/shelly/sensor.py @@ -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) diff --git a/homeassistant/components/shelly/switch.py b/homeassistant/components/shelly/switch.py index 6f3dd0b0136..3e35ba878e4 100644 --- a/homeassistant/components/shelly/switch.py +++ b/homeassistant/components/shelly/switch.py @@ -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() diff --git a/homeassistant/components/shelly/utils.py b/homeassistant/components/shelly/utils.py index 37b34dfe9e8..d8ce5ae9e45 100644 --- a/homeassistant/components/shelly/utils.py +++ b/homeassistant/components/shelly/utils.py @@ -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 diff --git a/mypy.ini b/mypy.ini index cf85d2e60a9..22cd7f4a1e0 100644 --- a/mypy.ini +++ b/mypy.ini @@ -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