From 7560a77e0e5c858d46c3cf33422a271cf9fa3234 Mon Sep 17 00:00:00 2001 From: Martin Hjelmare Date: Mon, 7 Jun 2021 16:04:04 +0200 Subject: [PATCH] Type mysensors strictly (#51535) --- .strict-typing | 1 + .../components/mysensors/__init__.py | 11 +-- .../components/mysensors/binary_sensor.py | 11 ++- homeassistant/components/mysensors/climate.py | 55 +++++++------ .../components/mysensors/config_flow.py | 8 +- homeassistant/components/mysensors/const.py | 55 +++++++------ homeassistant/components/mysensors/cover.py | 29 +++---- homeassistant/components/mysensors/device.py | 71 +++++++++++------ .../components/mysensors/device_tracker.py | 23 ++++-- homeassistant/components/mysensors/gateway.py | 36 ++++++--- homeassistant/components/mysensors/handler.py | 6 +- homeassistant/components/mysensors/helpers.py | 7 +- homeassistant/components/mysensors/light.py | 77 ++++++++++--------- homeassistant/components/mysensors/notify.py | 27 +++++-- homeassistant/components/mysensors/sensor.py | 41 ++++++---- homeassistant/components/mysensors/switch.py | 52 +++++++++---- mypy.ini | 11 +++ 17 files changed, 329 insertions(+), 192 deletions(-) diff --git a/.strict-typing b/.strict-typing index e11374312da..23b4ed76513 100644 --- a/.strict-typing +++ b/.strict-typing @@ -46,6 +46,7 @@ homeassistant.components.local_ip.* homeassistant.components.lock.* homeassistant.components.mailbox.* homeassistant.components.media_player.* +homeassistant.components.mysensors.* homeassistant.components.nam.* homeassistant.components.network.* homeassistant.components.notify.* diff --git a/homeassistant/components/mysensors/__init__.py b/homeassistant/components/mysensors/__init__.py index 068fd9361fa..2a958cee060 100644 --- a/homeassistant/components/mysensors/__init__.py +++ b/homeassistant/components/mysensors/__init__.py @@ -40,6 +40,7 @@ from .const import ( MYSENSORS_ON_UNLOAD, PLATFORMS_WITH_ENTRY_SUPPORT, DevId, + DiscoveryInfo, SensorType, ) from .device import MySensorsDevice, get_mysensors_devices @@ -70,7 +71,7 @@ def set_default_persistence_file(value: dict) -> dict: return value -def has_all_unique_files(value): +def has_all_unique_files(value: list[dict]) -> list[dict]: """Validate that all persistence files are unique and set if any is set.""" persistence_files = [gateway[CONF_PERSISTENCE_FILE] for gateway in value] schema = vol.Schema(vol.Unique()) @@ -78,17 +79,17 @@ def has_all_unique_files(value): return value -def is_persistence_file(value): +def is_persistence_file(value: str) -> str: """Validate that persistence file path ends in either .pickle or .json.""" if value.endswith((".json", ".pickle")): return value raise vol.Invalid(f"{value} does not end in either `.json` or `.pickle`") -def deprecated(key): +def deprecated(key: str) -> Callable[[dict], dict]: """Mark key as deprecated in configuration.""" - def validator(config): + def validator(config: dict) -> dict: """Check if key is in config, log warning and remove key.""" if key not in config: return config @@ -270,7 +271,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: def setup_mysensors_platform( hass: HomeAssistant, domain: str, # hass platform name - discovery_info: dict[str, list[DevId]], + discovery_info: DiscoveryInfo, device_class: type[MySensorsDevice] | dict[SensorType, type[MySensorsDevice]], device_args: ( None | tuple diff --git a/homeassistant/components/mysensors/binary_sensor.py b/homeassistant/components/mysensors/binary_sensor.py index 8358c3b2ecf..f94b5f71728 100644 --- a/homeassistant/components/mysensors/binary_sensor.py +++ b/homeassistant/components/mysensors/binary_sensor.py @@ -1,4 +1,6 @@ """Support for MySensors binary sensors.""" +from __future__ import annotations + from homeassistant.components import mysensors from homeassistant.components.binary_sensor import ( DEVICE_CLASS_MOISTURE, @@ -17,6 +19,7 @@ from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback +from .const import DiscoveryInfo from .helpers import on_unload SENSORS = { @@ -35,11 +38,11 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, -): +) -> None: """Set up this platform for a specific ConfigEntry(==Gateway).""" @callback - def async_discover(discovery_info): + def async_discover(discovery_info: DiscoveryInfo) -> None: """Discover and add a MySensors binary_sensor.""" mysensors.setup_mysensors_platform( hass, @@ -64,12 +67,12 @@ class MySensorsBinarySensor(mysensors.device.MySensorsEntity, BinarySensorEntity """Representation of a MySensors Binary Sensor child node.""" @property - def is_on(self): + def is_on(self) -> bool: """Return True if the binary sensor is on.""" return self._values.get(self.value_type) == STATE_ON @property - def device_class(self): + def device_class(self) -> str | None: """Return the class of this sensor, from DEVICE_CLASSES.""" pres = self.gateway.const.Presentation device_class = SENSORS.get(pres(self.child_type).name) diff --git a/homeassistant/components/mysensors/climate.py b/homeassistant/components/mysensors/climate.py index 797fcfafcc7..5dd52673581 100644 --- a/homeassistant/components/mysensors/climate.py +++ b/homeassistant/components/mysensors/climate.py @@ -1,4 +1,8 @@ """MySensors platform that offers a Climate (MySensors-HVAC) component.""" +from __future__ import annotations + +from typing import Any + from homeassistant.components import mysensors from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate.const import ( @@ -13,7 +17,7 @@ from homeassistant.components.climate.const import ( SUPPORT_TARGET_TEMPERATURE, SUPPORT_TARGET_TEMPERATURE_RANGE, ) -from homeassistant.components.mysensors.const import MYSENSORS_DISCOVERY +from homeassistant.components.mysensors.const import MYSENSORS_DISCOVERY, DiscoveryInfo from homeassistant.config_entries import ConfigEntry from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS, TEMP_FAHRENHEIT from homeassistant.core import HomeAssistant @@ -43,10 +47,10 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, -): +) -> None: """Set up this platform for a specific ConfigEntry(==Gateway).""" - async def async_discover(discovery_info): + async def async_discover(discovery_info: DiscoveryInfo) -> None: """Discover and add a MySensors climate.""" mysensors.setup_mysensors_platform( hass, @@ -71,7 +75,7 @@ class MySensorsHVAC(mysensors.device.MySensorsEntity, ClimateEntity): """Representation of a MySensors HVAC.""" @property - def supported_features(self): + def supported_features(self) -> int: """Return the list of supported features.""" features = 0 set_req = self.gateway.const.SetReq @@ -87,22 +91,23 @@ class MySensorsHVAC(mysensors.device.MySensorsEntity, ClimateEntity): return features @property - def temperature_unit(self): + def temperature_unit(self) -> str: """Return the unit of measurement.""" return TEMP_CELSIUS if self.hass.config.units.is_metric else TEMP_FAHRENHEIT @property - def current_temperature(self): + def current_temperature(self) -> float | None: """Return the current temperature.""" - value = self._values.get(self.gateway.const.SetReq.V_TEMP) + value: str | None = self._values.get(self.gateway.const.SetReq.V_TEMP) + float_value: float | None = None if value is not None: - value = float(value) + float_value = float(value) - return value + return float_value @property - def target_temperature(self): + def target_temperature(self) -> float | None: """Return the temperature we try to reach.""" set_req = self.gateway.const.SetReq if ( @@ -116,42 +121,46 @@ class MySensorsHVAC(mysensors.device.MySensorsEntity, ClimateEntity): return float(temp) if temp is not None else None @property - def target_temperature_high(self): + def target_temperature_high(self) -> float | None: """Return the highbound target temperature we try to reach.""" set_req = self.gateway.const.SetReq if set_req.V_HVAC_SETPOINT_HEAT in self._values: temp = self._values.get(set_req.V_HVAC_SETPOINT_COOL) return float(temp) if temp is not None else None + return None + @property - def target_temperature_low(self): + def target_temperature_low(self) -> float | None: """Return the lowbound target temperature we try to reach.""" set_req = self.gateway.const.SetReq if set_req.V_HVAC_SETPOINT_COOL in self._values: temp = self._values.get(set_req.V_HVAC_SETPOINT_HEAT) return float(temp) if temp is not None else None - @property - def hvac_mode(self): - """Return current operation ie. heat, cool, idle.""" - return self._values.get(self.value_type) + return None @property - def hvac_modes(self): + def hvac_mode(self) -> str: + """Return current operation ie. heat, cool, idle.""" + return self._values.get(self.value_type, HVAC_MODE_HEAT) + + @property + def hvac_modes(self) -> list[str]: """List of available operation modes.""" return OPERATION_LIST @property - def fan_mode(self): + def fan_mode(self) -> str | None: """Return the fan setting.""" return self._values.get(self.gateway.const.SetReq.V_HVAC_SPEED) @property - def fan_modes(self): + def fan_modes(self) -> list[str]: """List of available fan modes.""" return FAN_LIST - async def async_set_temperature(self, **kwargs): + async def async_set_temperature(self, **kwargs: Any) -> None: """Set new target temperature.""" set_req = self.gateway.const.SetReq temp = kwargs.get(ATTR_TEMPERATURE) @@ -183,7 +192,7 @@ class MySensorsHVAC(mysensors.device.MySensorsEntity, ClimateEntity): self._values[value_type] = value self.async_write_ha_state() - async def async_set_fan_mode(self, fan_mode): + async def async_set_fan_mode(self, fan_mode: str) -> None: """Set new target temperature.""" set_req = self.gateway.const.SetReq self.gateway.set_child_value( @@ -194,7 +203,7 @@ class MySensorsHVAC(mysensors.device.MySensorsEntity, ClimateEntity): self._values[set_req.V_HVAC_SPEED] = fan_mode self.async_write_ha_state() - async def async_set_hvac_mode(self, hvac_mode): + async def async_set_hvac_mode(self, hvac_mode: str) -> None: """Set new target temperature.""" self.gateway.set_child_value( self.node_id, @@ -208,7 +217,7 @@ class MySensorsHVAC(mysensors.device.MySensorsEntity, ClimateEntity): self._values[self.value_type] = hvac_mode self.async_write_ha_state() - async def async_update(self): + async def async_update(self) -> None: """Update the controller with the latest value from a sensor.""" await super().async_update() self._values[self.value_type] = DICT_MYS_TO_HA[self._values[self.value_type]] diff --git a/homeassistant/components/mysensors/config_flow.py b/homeassistant/components/mysensors/config_flow.py index 56af6671958..1abf45dd60f 100644 --- a/homeassistant/components/mysensors/config_flow.py +++ b/homeassistant/components/mysensors/config_flow.py @@ -82,8 +82,8 @@ def _validate_version(version: str) -> dict[str, str]: def _is_same_device( - gw_type: ConfGatewayType, user_input: dict[str, str], entry: ConfigEntry -): + gw_type: ConfGatewayType, user_input: dict[str, Any], entry: ConfigEntry +) -> bool: """Check if another ConfigDevice is actually the same as user_input. This function only compares addresses and tcp ports, so it is possible to fool it with tricks like port forwarding. @@ -91,7 +91,9 @@ def _is_same_device( if entry.data[CONF_DEVICE] != user_input[CONF_DEVICE]: return False if gw_type == CONF_GATEWAY_TYPE_TCP: - return entry.data[CONF_TCP_PORT] == user_input[CONF_TCP_PORT] + entry_tcp_port: int = entry.data[CONF_TCP_PORT] + input_tcp_port: int = user_input[CONF_TCP_PORT] + return entry_tcp_port == input_tcp_port if gw_type == CONF_GATEWAY_TYPE_MQTT: entry_topics = { entry.data[CONF_TOPIC_IN_PREFIX], diff --git a/homeassistant/components/mysensors/const.py b/homeassistant/components/mysensors/const.py index 1bd071be9a9..f8e157e3622 100644 --- a/homeassistant/components/mysensors/const.py +++ b/homeassistant/components/mysensors/const.py @@ -2,23 +2,23 @@ from __future__ import annotations from collections import defaultdict -from typing import Literal, Tuple +from typing import Final, Literal, Tuple, TypedDict -ATTR_DEVICES: str = "devices" -ATTR_GATEWAY_ID: str = "gateway_id" +ATTR_DEVICES: Final = "devices" +ATTR_GATEWAY_ID: Final = "gateway_id" -CONF_BAUD_RATE: str = "baud_rate" -CONF_DEVICE: str = "device" -CONF_GATEWAYS: str = "gateways" -CONF_NODES: str = "nodes" -CONF_PERSISTENCE: str = "persistence" -CONF_PERSISTENCE_FILE: str = "persistence_file" -CONF_RETAIN: str = "retain" -CONF_TCP_PORT: str = "tcp_port" -CONF_TOPIC_IN_PREFIX: str = "topic_in_prefix" -CONF_TOPIC_OUT_PREFIX: str = "topic_out_prefix" -CONF_VERSION: str = "version" -CONF_GATEWAY_TYPE: str = "gateway_type" +CONF_BAUD_RATE: Final = "baud_rate" +CONF_DEVICE: Final = "device" +CONF_GATEWAYS: Final = "gateways" +CONF_NODES: Final = "nodes" +CONF_PERSISTENCE: Final = "persistence" +CONF_PERSISTENCE_FILE: Final = "persistence_file" +CONF_RETAIN: Final = "retain" +CONF_TCP_PORT: Final = "tcp_port" +CONF_TOPIC_IN_PREFIX: Final = "topic_in_prefix" +CONF_TOPIC_OUT_PREFIX: Final = "topic_out_prefix" +CONF_VERSION: Final = "version" +CONF_GATEWAY_TYPE: Final = "gateway_type" ConfGatewayType = Literal["Serial", "TCP", "MQTT"] CONF_GATEWAY_TYPE_SERIAL: ConfGatewayType = "Serial" CONF_GATEWAY_TYPE_TCP: ConfGatewayType = "TCP" @@ -29,19 +29,28 @@ CONF_GATEWAY_TYPE_ALL: list[str] = [ CONF_GATEWAY_TYPE_TCP, ] -DOMAIN: str = "mysensors" +DOMAIN: Final = "mysensors" MYSENSORS_GATEWAY_START_TASK: str = "mysensors_gateway_start_task_{}" -MYSENSORS_GATEWAYS: str = "mysensors_gateways" -PLATFORM: str = "platform" -SCHEMA: str = "schema" +MYSENSORS_GATEWAYS: Final = "mysensors_gateways" +PLATFORM: Final = "platform" +SCHEMA: Final = "schema" CHILD_CALLBACK: str = "mysensors_child_callback_{}_{}_{}_{}" NODE_CALLBACK: str = "mysensors_node_callback_{}_{}" -MYSENSORS_DISCOVERY = "mysensors_discovery_{}_{}" -MYSENSORS_ON_UNLOAD = "mysensors_on_unload_{}" -TYPE: str = "type" +MYSENSORS_DISCOVERY: str = "mysensors_discovery_{}_{}" +MYSENSORS_ON_UNLOAD: str = "mysensors_on_unload_{}" +TYPE: Final = "type" UPDATE_DELAY: float = 0.1 -SERVICE_SEND_IR_CODE: str = "send_ir_code" + +class DiscoveryInfo(TypedDict): + """Represent the discovery info type for mysensors platforms.""" + + devices: list[DevId] + name: str # CONF_NAME is used in the notify base integration. + gateway_id: GatewayId + + +SERVICE_SEND_IR_CODE: Final = "send_ir_code" SensorType = str # S_DOOR, S_MOTION, S_SMOKE, ... diff --git a/homeassistant/components/mysensors/cover.py b/homeassistant/components/mysensors/cover.py index 2d852ef05b4..bab7a07a867 100644 --- a/homeassistant/components/mysensors/cover.py +++ b/homeassistant/components/mysensors/cover.py @@ -1,10 +1,13 @@ """Support for MySensors covers.""" +from __future__ import annotations + from enum import Enum, unique import logging +from typing import Any from homeassistant.components import mysensors from homeassistant.components.cover import ATTR_POSITION, DOMAIN, CoverEntity -from homeassistant.components.mysensors.const import MYSENSORS_DISCOVERY +from homeassistant.components.mysensors.const import MYSENSORS_DISCOVERY, DiscoveryInfo from homeassistant.config_entries import ConfigEntry from homeassistant.const import STATE_OFF, STATE_ON from homeassistant.core import HomeAssistant @@ -30,10 +33,10 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, -): +) -> None: """Set up this platform for a specific ConfigEntry(==Gateway).""" - async def async_discover(discovery_info): + async def async_discover(discovery_info: DiscoveryInfo) -> None: """Discover and add a MySensors cover.""" mysensors.setup_mysensors_platform( hass, @@ -57,7 +60,7 @@ async def async_setup_entry( class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): """Representation of the value of a MySensors Cover child node.""" - def get_cover_state(self): + def get_cover_state(self) -> CoverState: """Return a CoverState enum representing the state of the cover.""" set_req = self.gateway.const.SetReq v_up = self._values.get(set_req.V_UP) == STATE_ON @@ -69,7 +72,7 @@ class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): # or V_STATUS. amount = 100 if set_req.V_DIMMER in self._values: - amount = self._values.get(set_req.V_DIMMER) + amount = self._values[set_req.V_DIMMER] else: amount = 100 if self._values.get(set_req.V_LIGHT) == STATE_ON else 0 @@ -82,22 +85,22 @@ class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): return CoverState.OPEN @property - def is_closed(self): + def is_closed(self) -> bool: """Return True if the cover is closed.""" return self.get_cover_state() == CoverState.CLOSED @property - def is_closing(self): + def is_closing(self) -> bool: """Return True if the cover is closing.""" return self.get_cover_state() == CoverState.CLOSING @property - def is_opening(self): + def is_opening(self) -> bool: """Return True if the cover is opening.""" return self.get_cover_state() == CoverState.OPENING @property - def current_cover_position(self): + def current_cover_position(self) -> int | None: """Return current position of cover. None is unknown, 0 is closed, 100 is fully open. @@ -105,7 +108,7 @@ class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): set_req = self.gateway.const.SetReq return self._values.get(set_req.V_DIMMER) - async def async_open_cover(self, **kwargs): + async def async_open_cover(self, **kwargs: Any) -> None: """Move the cover up.""" set_req = self.gateway.const.SetReq self.gateway.set_child_value( @@ -119,7 +122,7 @@ class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): self._values[set_req.V_LIGHT] = STATE_ON self.async_write_ha_state() - async def async_close_cover(self, **kwargs): + async def async_close_cover(self, **kwargs: Any) -> None: """Move the cover down.""" set_req = self.gateway.const.SetReq self.gateway.set_child_value( @@ -133,7 +136,7 @@ class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): self._values[set_req.V_LIGHT] = STATE_OFF 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.""" position = kwargs.get(ATTR_POSITION) set_req = self.gateway.const.SetReq @@ -145,7 +148,7 @@ class MySensorsCover(mysensors.device.MySensorsEntity, CoverEntity): self._values[set_req.V_DIMMER] = position self.async_write_ha_state() - async def async_stop_cover(self, **kwargs): + async def async_stop_cover(self, **kwargs: Any) -> None: """Stop the device.""" set_req = self.gateway.const.SetReq self.gateway.set_child_value( diff --git a/homeassistant/components/mysensors/device.py b/homeassistant/components/mysensors/device.py index c066e633eaa..32305061ca7 100644 --- a/homeassistant/components/mysensors/device.py +++ b/homeassistant/components/mysensors/device.py @@ -66,10 +66,10 @@ class MySensorsDevice: return self.gateway_id, self.node_id, self.child_id, self.value_type @property - def _logger(self): + def _logger(self) -> logging.Logger: return logging.getLogger(f"{__name__}.{self.name}") - async def async_will_remove_from_hass(self): + async def async_will_remove_from_hass(self) -> None: """Remove this entity from home assistant.""" for platform in PLATFORM_TYPES: platform_str = MYSENSORS_PLATFORM_DEVICES.format(platform) @@ -91,17 +91,26 @@ class MySensorsDevice: @property def sketch_name(self) -> str: - """Return the name of the sketch running on the whole node (will be the same for several entities!).""" - return self._node.sketch_name + """Return the name of the sketch running on the whole node. + + The name will be the same for several entities. + """ + return self._node.sketch_name # type: ignore[no-any-return] @property def sketch_version(self) -> str: - """Return the version of the sketch running on the whole node (will be the same for several entities!).""" - return self._node.sketch_version + """Return the version of the sketch running on the whole node. + + The name will be the same for several entities. + """ + return self._node.sketch_version # type: ignore[no-any-return] @property def node_name(self) -> str: - """Name of the whole node (will be the same for several entities!).""" + """Name of the whole node. + + The name will be the same for several entities. + """ return f"{self.sketch_name} {self.node_id}" @property @@ -111,7 +120,7 @@ class MySensorsDevice: @property def device_info(self) -> DeviceInfo: - """Return a dict that allows home assistant to puzzle all entities belonging to a node together.""" + """Return the device info.""" return { "identifiers": {(DOMAIN, f"{self.gateway_id}-{self.node_id}")}, "name": self.node_name, @@ -120,13 +129,13 @@ class MySensorsDevice: } @property - def name(self): + def name(self) -> str: """Return the name of this entity.""" return f"{self.node_name} {self.child_id}" @property - def extra_state_attributes(self): - """Return device specific state attributes.""" + def _extra_attributes(self) -> dict[str, Any]: + """Return device specific attributes.""" node = self.gateway.sensors[self.node_id] child = node.children[self.child_id] attr = { @@ -136,10 +145,6 @@ class MySensorsDevice: ATTR_DESCRIPTION: child.description, ATTR_NODE_ID: self.node_id, } - # This works when we are actually an Entity (i.e. all platforms except device_tracker) - if hasattr(self, "platform"): - # pylint: disable=no-member - attr[ATTR_DEVICE] = self.platform.config_entry.data[CONF_DEVICE] set_req = self.gateway.const.SetReq @@ -148,7 +153,7 @@ class MySensorsDevice: return attr - async def async_update(self): + async def async_update(self) -> None: """Update the controller with the latest value from a sensor.""" node = self.gateway.sensors[self.node_id] child = node.children[self.child_id] @@ -175,17 +180,17 @@ class MySensorsDevice: else: self._values[value_type] = value - async def _async_update_callback(self): + async def _async_update_callback(self) -> None: """Update the device.""" raise NotImplementedError @callback - def async_update_callback(self): + def async_update_callback(self) -> None: """Update the device after delay.""" if self._update_scheduled: return - async def update(): + async def update() -> None: """Perform update.""" try: await self._async_update_callback() @@ -199,31 +204,47 @@ class MySensorsDevice: self.hass.loop.call_later(UPDATE_DELAY, delayed_update) -def get_mysensors_devices(hass, domain: str) -> dict[DevId, MySensorsDevice]: +def get_mysensors_devices( + hass: HomeAssistant, domain: str +) -> dict[DevId, MySensorsDevice]: """Return MySensors devices for a hass platform name.""" if MYSENSORS_PLATFORM_DEVICES.format(domain) not in hass.data[DOMAIN]: hass.data[DOMAIN][MYSENSORS_PLATFORM_DEVICES.format(domain)] = {} - return hass.data[DOMAIN][MYSENSORS_PLATFORM_DEVICES.format(domain)] + devices: dict[DevId, MySensorsDevice] = hass.data[DOMAIN][ + MYSENSORS_PLATFORM_DEVICES.format(domain) + ] + return devices class MySensorsEntity(MySensorsDevice, Entity): """Representation of a MySensors entity.""" @property - def should_poll(self): + def should_poll(self) -> bool: """Return the polling state. The gateway pushes its states.""" return False @property - def available(self): + def available(self) -> bool: """Return true if entity is available.""" return self.value_type in self._values - async def _async_update_callback(self): + @property + def extra_state_attributes(self) -> dict[str, Any]: + """Return entity specific state attributes.""" + attr = self._extra_attributes + + assert self.platform + assert self.platform.config_entry + attr[ATTR_DEVICE] = self.platform.config_entry.data[CONF_DEVICE] + + return attr + + async def _async_update_callback(self) -> None: """Update the entity.""" await self.async_update_ha_state(True) - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Register update callback.""" self.async_on_remove( async_dispatcher_connect( diff --git a/homeassistant/components/mysensors/device_tracker.py b/homeassistant/components/mysensors/device_tracker.py index 6297f8344fc..544fb8d6b09 100644 --- a/homeassistant/components/mysensors/device_tracker.py +++ b/homeassistant/components/mysensors/device_tracker.py @@ -1,8 +1,16 @@ """Support for tracking MySensors devices.""" +from __future__ import annotations + +from typing import Any, Callable + from homeassistant.components import mysensors from homeassistant.components.device_tracker import DOMAIN from homeassistant.components.mysensors import DevId -from homeassistant.components.mysensors.const import ATTR_GATEWAY_ID, GatewayId +from homeassistant.components.mysensors.const import ( + ATTR_GATEWAY_ID, + DiscoveryInfo, + GatewayId, +) from homeassistant.core import HomeAssistant from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.util import slugify @@ -11,8 +19,11 @@ from .helpers import on_unload async def async_setup_scanner( - hass: HomeAssistant, config, async_see, discovery_info=None -): + hass: HomeAssistant, + config: dict[str, Any], + async_see: Callable, + discovery_info: DiscoveryInfo | None = None, +) -> bool: """Set up the MySensors device scanner.""" if not discovery_info: return False @@ -55,13 +66,13 @@ async def async_setup_scanner( class MySensorsDeviceScanner(mysensors.device.MySensorsDevice): """Represent a MySensors scanner.""" - def __init__(self, hass: HomeAssistant, async_see, *args): + def __init__(self, hass: HomeAssistant, async_see: Callable, *args: Any) -> None: """Set up instance.""" super().__init__(*args) self.async_see = async_see self.hass = hass - async def _async_update_callback(self): + async def _async_update_callback(self) -> None: """Update the device.""" await self.async_update() node = self.gateway.sensors[self.node_id] @@ -74,5 +85,5 @@ class MySensorsDeviceScanner(mysensors.device.MySensorsDevice): host_name=self.name, gps=(latitude, longitude), battery=node.battery_level, - attributes=self.extra_state_attributes, + attributes=self._extra_attributes, ) diff --git a/homeassistant/components/mysensors/gateway.py b/homeassistant/components/mysensors/gateway.py index ffbcc47fc03..f1e2cd0a4e1 100644 --- a/homeassistant/components/mysensors/gateway.py +++ b/homeassistant/components/mysensors/gateway.py @@ -14,6 +14,10 @@ from mysensors import BaseAsyncGateway, Message, Sensor, mysensors import voluptuous as vol from homeassistant.components.mqtt import DOMAIN as MQTT_DOMAIN +from homeassistant.components.mqtt.models import ( + Message as MQTTMessage, + PublishPayloadType, +) from homeassistant.config_entries import ConfigEntry from homeassistant.const import EVENT_HOMEASSISTANT_STOP from homeassistant.core import Event, HomeAssistant, callback @@ -51,7 +55,7 @@ GATEWAY_READY_TIMEOUT = 20.0 MQTT_COMPONENT = "mqtt" -def is_serial_port(value): +def is_serial_port(value: str) -> str: """Validate that value is a windows serial port or a unix device.""" if sys.platform.startswith("win"): ports = (f"COM{idx + 1}" for idx in range(256)) @@ -61,7 +65,7 @@ def is_serial_port(value): return cv.isdevice(value) -def is_socket_address(value): +def is_socket_address(value: str) -> str: """Validate that value is a valid address.""" try: socket.getaddrinfo(value, None) @@ -179,15 +183,17 @@ async def _get_gateway( return None mqtt = hass.components.mqtt - def pub_callback(topic, payload, qos, retain): + def pub_callback(topic: str, payload: str, qos: int, retain: bool) -> None: """Call MQTT publish function.""" mqtt.async_publish(topic, payload, qos, retain) - def sub_callback(topic, sub_cb, qos): + def sub_callback( + topic: str, sub_cb: Callable[[str, PublishPayloadType, int], None], qos: int + ) -> None: """Call MQTT subscribe function.""" @callback - def internal_callback(msg): + def internal_callback(msg: MQTTMessage) -> None: """Call callback.""" sub_cb(msg.topic, msg.payload, msg.qos) @@ -234,7 +240,7 @@ async def _get_gateway( async def finish_setup( hass: HomeAssistant, entry: ConfigEntry, gateway: BaseAsyncGateway -): +) -> None: """Load any persistent devices and platforms and start gateway.""" discover_tasks = [] start_tasks = [] @@ -249,7 +255,7 @@ async def finish_setup( async def _discover_persistent_devices( hass: HomeAssistant, entry: ConfigEntry, gateway: BaseAsyncGateway -): +) -> None: """Discover platforms for devices loaded via persistence file.""" new_devices = defaultdict(list) for node_id in gateway.sensors: @@ -265,7 +271,9 @@ async def _discover_persistent_devices( discover_mysensors_platform(hass, entry.entry_id, platform, dev_ids) -async def gw_stop(hass, entry: ConfigEntry, gateway: BaseAsyncGateway): +async def gw_stop( + hass: HomeAssistant, entry: ConfigEntry, gateway: BaseAsyncGateway +) -> None: """Stop the gateway.""" connect_task = hass.data[DOMAIN].pop( MYSENSORS_GATEWAY_START_TASK.format(entry.entry_id), None @@ -275,11 +283,14 @@ async def gw_stop(hass, entry: ConfigEntry, gateway: BaseAsyncGateway): await gateway.stop() -async def _gw_start(hass: HomeAssistant, entry: ConfigEntry, gateway: BaseAsyncGateway): +async def _gw_start( + hass: HomeAssistant, entry: ConfigEntry, gateway: BaseAsyncGateway +) -> None: """Start the gateway.""" gateway_ready = asyncio.Event() - def gateway_connected(_: BaseAsyncGateway): + def gateway_connected(_: BaseAsyncGateway) -> None: + """Handle gateway connected.""" gateway_ready.set() gateway.on_conn_made = gateway_connected @@ -290,7 +301,8 @@ async def _gw_start(hass: HomeAssistant, entry: ConfigEntry, gateway: BaseAsyncG gateway.start() ) # store the connect task so it can be cancelled in gw_stop - async def stop_this_gw(_: Event): + async def stop_this_gw(_: Event) -> None: + """Stop the gateway.""" await gw_stop(hass, entry, gateway) on_unload( @@ -319,7 +331,7 @@ def _gw_callback_factory( """Return a new callback for the gateway.""" @callback - def mysensors_callback(msg: Message): + def mysensors_callback(msg: Message) -> None: """Handle messages from a MySensors gateway. All MySenors messages are received here. diff --git a/homeassistant/components/mysensors/handler.py b/homeassistant/components/mysensors/handler.py index 8558cd01f42..0fb86fd0eec 100644 --- a/homeassistant/components/mysensors/handler.py +++ b/homeassistant/components/mysensors/handler.py @@ -68,7 +68,7 @@ async def handle_sketch_version( @callback def _handle_child_update( hass: HomeAssistant, gateway_id: GatewayId, validated: dict[str, list[DevId]] -): +) -> None: """Handle a child update.""" signals: list[str] = [] @@ -91,7 +91,9 @@ def _handle_child_update( @callback -def _handle_node_update(hass: HomeAssistant, gateway_id: GatewayId, msg: Message): +def _handle_node_update( + hass: HomeAssistant, gateway_id: GatewayId, msg: Message +) -> None: """Handle a node update.""" signal = NODE_CALLBACK.format(gateway_id, msg.node_id) async_dispatcher_send(hass, signal) diff --git a/homeassistant/components/mysensors/helpers.py b/homeassistant/components/mysensors/helpers.py index 5bc0a0d0839..7c50526cd6e 100644 --- a/homeassistant/components/mysensors/helpers.py +++ b/homeassistant/components/mysensors/helpers.py @@ -117,7 +117,10 @@ def switch_ir_send_schema( def get_child_schema( - gateway: BaseAsyncGateway, child: ChildSensor, value_type_name: ValueType, schema + gateway: BaseAsyncGateway, + child: ChildSensor, + value_type_name: ValueType, + schema: dict, ) -> vol.Schema: """Return a child schema.""" set_req = gateway.const.SetReq @@ -136,7 +139,7 @@ def get_child_schema( def invalid_msg( gateway: BaseAsyncGateway, child: ChildSensor, value_type_name: ValueType -): +) -> str: """Return a message for an invalid child during schema validation.""" pres = gateway.const.Presentation set_req = gateway.const.SetReq diff --git a/homeassistant/components/mysensors/light.py b/homeassistant/components/mysensors/light.py index b9fbc139e4f..81089f052b6 100644 --- a/homeassistant/components/mysensors/light.py +++ b/homeassistant/components/mysensors/light.py @@ -1,4 +1,8 @@ """Support for MySensors lights.""" +from __future__ import annotations + +from typing import Any + from homeassistant.components import mysensors from homeassistant.components.light import ( ATTR_BRIGHTNESS, @@ -10,7 +14,6 @@ from homeassistant.components.light import ( SUPPORT_WHITE_VALUE, LightEntity, ) -from homeassistant.components.mysensors.const import MYSENSORS_DISCOVERY from homeassistant.config_entries import ConfigEntry from homeassistant.const import STATE_OFF, STATE_ON from homeassistant.core import HomeAssistant, callback @@ -19,6 +22,8 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback import homeassistant.util.color as color_util from homeassistant.util.color import rgb_hex_to_rgb_list +from .const import MYSENSORS_DISCOVERY, DiscoveryInfo, SensorType +from .device import MySensorsDevice from .helpers import on_unload SUPPORT_MYSENSORS_RGBW = SUPPORT_COLOR | SUPPORT_WHITE_VALUE @@ -28,15 +33,15 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, -): +) -> None: """Set up this platform for a specific ConfigEntry(==Gateway).""" - device_class_map = { + device_class_map: dict[SensorType, type[MySensorsDevice]] = { "S_DIMMER": MySensorsLightDimmer, "S_RGB_LIGHT": MySensorsLightRGB, "S_RGBW_LIGHT": MySensorsLightRGBW, } - async def async_discover(discovery_info): + async def async_discover(discovery_info: DiscoveryInfo) -> None: """Discover and add a MySensors light.""" mysensors.setup_mysensors_platform( hass, @@ -60,35 +65,35 @@ async def async_setup_entry( class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): """Representation of a MySensors Light child node.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Initialize a MySensors Light.""" super().__init__(*args) - self._state = None - self._brightness = None - self._hs = None - self._white = None + self._state: bool | None = None + self._brightness: int | None = None + self._hs: tuple[int, int] | None = None + self._white: int | None = None @property - def brightness(self): + def brightness(self) -> int | None: """Return the brightness of this light between 0..255.""" return self._brightness @property - def hs_color(self): + def hs_color(self) -> tuple[int, int] | None: """Return the hs color value [int, int].""" return self._hs @property - def white_value(self): + def white_value(self) -> int | None: """Return the white value of this light between 0..255.""" return self._white @property - def is_on(self): + def is_on(self) -> bool: """Return true if device is on.""" - return self._state + return bool(self._state) - def _turn_on_light(self): + def _turn_on_light(self) -> None: """Turn on light child device.""" set_req = self.gateway.const.SetReq @@ -103,10 +108,9 @@ class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): self._state = True self._values[set_req.V_LIGHT] = STATE_ON - def _turn_on_dimmer(self, **kwargs): + def _turn_on_dimmer(self, **kwargs: Any) -> None: """Turn on dimmer child device.""" set_req = self.gateway.const.SetReq - brightness = self._brightness if ( ATTR_BRIGHTNESS not in kwargs @@ -114,7 +118,7 @@ class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): or set_req.V_DIMMER not in self._values ): return - brightness = kwargs[ATTR_BRIGHTNESS] + brightness: int = kwargs[ATTR_BRIGHTNESS] percent = round(100 * brightness / 255) self.gateway.set_child_value( self.node_id, self.child_id, set_req.V_DIMMER, percent, ack=1 @@ -125,17 +129,20 @@ class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): self._brightness = brightness self._values[set_req.V_DIMMER] = percent - def _turn_on_rgb_and_w(self, hex_template, **kwargs): + def _turn_on_rgb_and_w(self, hex_template: str, **kwargs: Any) -> None: """Turn on RGB or RGBW child device.""" + assert self._hs + assert self._white is not None rgb = list(color_util.color_hs_to_RGB(*self._hs)) white = self._white hex_color = self._values.get(self.value_type) - hs_color = kwargs.get(ATTR_HS_COLOR) + hs_color: tuple[float, float] | None = kwargs.get(ATTR_HS_COLOR) + new_rgb: tuple[int, int, int] | None if hs_color is not None: new_rgb = color_util.color_hs_to_RGB(*hs_color) else: new_rgb = None - new_white = kwargs.get(ATTR_WHITE_VALUE) + new_white: int | None = kwargs.get(ATTR_WHITE_VALUE) if new_rgb is None and new_white is None: return @@ -155,11 +162,11 @@ class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): if self.assumed_state: # optimistically assume that light has changed state - self._hs = color_util.color_RGB_to_hs(*rgb) + self._hs = color_util.color_RGB_to_hs(*rgb) # type: ignore[assignment] self._white = white self._values[self.value_type] = hex_color - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the device off.""" value_type = self.gateway.const.SetReq.V_LIGHT self.gateway.set_child_value(self.node_id, self.child_id, value_type, 0, ack=1) @@ -170,13 +177,13 @@ class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): self.async_write_ha_state() @callback - def _async_update_light(self): + def _async_update_light(self) -> None: """Update the controller with values from light child.""" value_type = self.gateway.const.SetReq.V_LIGHT self._state = self._values[value_type] == STATE_ON @callback - def _async_update_dimmer(self): + def _async_update_dimmer(self) -> None: """Update the controller with values from dimmer child.""" value_type = self.gateway.const.SetReq.V_DIMMER if value_type in self._values: @@ -185,31 +192,31 @@ class MySensorsLight(mysensors.device.MySensorsEntity, LightEntity): self._state = False @callback - def _async_update_rgb_or_w(self): + def _async_update_rgb_or_w(self) -> None: """Update the controller with values from RGB or RGBW child.""" value = self._values[self.value_type] color_list = rgb_hex_to_rgb_list(value) if len(color_list) > 3: self._white = color_list.pop() - self._hs = color_util.color_RGB_to_hs(*color_list) + self._hs = color_util.color_RGB_to_hs(*color_list) # type: ignore[assignment] class MySensorsLightDimmer(MySensorsLight): """Dimmer child class to MySensorsLight.""" @property - def supported_features(self): + def supported_features(self) -> int: """Flag supported features.""" return SUPPORT_BRIGHTNESS - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the device on.""" self._turn_on_light() self._turn_on_dimmer(**kwargs) if self.assumed_state: self.async_write_ha_state() - async def async_update(self): + async def async_update(self) -> None: """Update the controller with the latest value from a sensor.""" await super().async_update() self._async_update_light() @@ -220,14 +227,14 @@ class MySensorsLightRGB(MySensorsLight): """RGB child class to MySensorsLight.""" @property - def supported_features(self): + def supported_features(self) -> int: """Flag supported features.""" set_req = self.gateway.const.SetReq if set_req.V_DIMMER in self._values: return SUPPORT_BRIGHTNESS | SUPPORT_COLOR return SUPPORT_COLOR - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the device on.""" self._turn_on_light() self._turn_on_dimmer(**kwargs) @@ -235,7 +242,7 @@ class MySensorsLightRGB(MySensorsLight): if self.assumed_state: self.async_write_ha_state() - async def async_update(self): + async def async_update(self) -> None: """Update the controller with the latest value from a sensor.""" await super().async_update() self._async_update_light() @@ -247,14 +254,14 @@ class MySensorsLightRGBW(MySensorsLightRGB): """RGBW child class to MySensorsLightRGB.""" @property - def supported_features(self): + def supported_features(self) -> int: """Flag supported features.""" set_req = self.gateway.const.SetReq if set_req.V_DIMMER in self._values: return SUPPORT_BRIGHTNESS | SUPPORT_MYSENSORS_RGBW return SUPPORT_MYSENSORS_RGBW - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the device on.""" self._turn_on_light() self._turn_on_dimmer(**kwargs) diff --git a/homeassistant/components/mysensors/notify.py b/homeassistant/components/mysensors/notify.py index 50fca55ab39..109b357dee7 100644 --- a/homeassistant/components/mysensors/notify.py +++ b/homeassistant/components/mysensors/notify.py @@ -1,9 +1,20 @@ """MySensors notification service.""" +from __future__ import annotations + +from typing import Any + from homeassistant.components import mysensors from homeassistant.components.notify import ATTR_TARGET, DOMAIN, BaseNotificationService +from homeassistant.core import HomeAssistant + +from .const import DevId, DiscoveryInfo -async def async_get_service(hass, config, discovery_info=None): +async def async_get_service( + hass: HomeAssistant, + config: dict[str, Any], + discovery_info: DiscoveryInfo | None = None, +) -> BaseNotificationService | None: """Get the MySensors notification service.""" if not discovery_info: return None @@ -19,7 +30,7 @@ async def async_get_service(hass, config, discovery_info=None): class MySensorsNotificationDevice(mysensors.device.MySensorsDevice): """Represent a MySensors Notification device.""" - def send_msg(self, msg): + def send_msg(self, msg: str) -> None: """Send a message.""" for sub_msg in [msg[i : i + 25] for i in range(0, len(msg), 25)]: # Max mysensors payload is 25 bytes. @@ -27,7 +38,7 @@ class MySensorsNotificationDevice(mysensors.device.MySensorsDevice): self.node_id, self.child_id, self.value_type, sub_msg ) - def __repr__(self): + def __repr__(self) -> str: """Return the representation.""" return f"" @@ -35,11 +46,15 @@ class MySensorsNotificationDevice(mysensors.device.MySensorsDevice): class MySensorsNotificationService(BaseNotificationService): """Implement a MySensors notification service.""" - def __init__(self, hass): + def __init__(self, hass: HomeAssistant) -> None: """Initialize the service.""" - self.devices = mysensors.get_mysensors_devices(hass, DOMAIN) + self.devices: dict[ + DevId, MySensorsNotificationDevice + ] = mysensors.get_mysensors_devices( + hass, DOMAIN + ) # type: ignore[assignment] - async def async_send_message(self, message="", **kwargs): + async def async_send_message(self, message: str = "", **kwargs: Any) -> None: """Send a message to a user.""" target_devices = kwargs.get(ATTR_TARGET) devices = [ diff --git a/homeassistant/components/mysensors/sensor.py b/homeassistant/components/mysensors/sensor.py index abbfa66ab8e..2ede5e38c6a 100644 --- a/homeassistant/components/mysensors/sensor.py +++ b/homeassistant/components/mysensors/sensor.py @@ -1,8 +1,9 @@ """Support for MySensors sensors.""" +from __future__ import annotations + from awesomeversion import AwesomeVersion from homeassistant.components import mysensors -from homeassistant.components.mysensors.const import MYSENSORS_DISCOVERY from homeassistant.components.sensor import DOMAIN, SensorEntity from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( @@ -26,9 +27,10 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddEntitiesCallback +from .const import MYSENSORS_DISCOVERY, DiscoveryInfo from .helpers import on_unload -SENSORS = { +SENSORS: dict[str, list[str | None] | dict[str, list[str | None]]] = { "V_TEMP": [None, "mdi:thermometer"], "V_HUM": [PERCENTAGE, "mdi:water-percent"], "V_DIMMER": [PERCENTAGE, "mdi:percent"], @@ -67,10 +69,10 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, -): +) -> None: """Set up this platform for a specific ConfigEntry(==Gateway).""" - async def async_discover(discovery_info): + async def async_discover(discovery_info: DiscoveryInfo) -> None: """Discover and add a MySensors sensor.""" mysensors.setup_mysensors_platform( hass, @@ -95,7 +97,7 @@ class MySensorsSensor(mysensors.device.MySensorsEntity, SensorEntity): """Representation of a MySensors Sensor child node.""" @property - def force_update(self): + def force_update(self) -> bool: """Return True if state updates should be forced. If True, a state change will be triggered anytime the state property is @@ -104,36 +106,43 @@ class MySensorsSensor(mysensors.device.MySensorsEntity, SensorEntity): return True @property - def state(self): + def state(self) -> str | None: """Return the state of the device.""" return self._values.get(self.value_type) @property - def icon(self): + def icon(self) -> str | None: """Return the icon to use in the frontend, if any.""" icon = self._get_sensor_type()[1] return icon @property - def unit_of_measurement(self): + def unit_of_measurement(self) -> str | None: """Return the unit of measurement of this entity.""" set_req = self.gateway.const.SetReq if ( AwesomeVersion(self.gateway.protocol_version) >= AwesomeVersion("1.5") and set_req.V_UNIT_PREFIX in self._values ): - return self._values[set_req.V_UNIT_PREFIX] + custom_unit: str = self._values[set_req.V_UNIT_PREFIX] + return custom_unit + + if set_req(self.value_type) == set_req.V_TEMP: + if self.hass.config.units.is_metric: + return TEMP_CELSIUS + return TEMP_FAHRENHEIT + unit = self._get_sensor_type()[0] return unit - def _get_sensor_type(self): + def _get_sensor_type(self) -> list[str | None]: """Return list with unit and icon of sensor type.""" pres = self.gateway.const.Presentation set_req = self.gateway.const.SetReq - SENSORS[set_req.V_TEMP.name][0] = ( - TEMP_CELSIUS if self.hass.config.units.is_metric else TEMP_FAHRENHEIT - ) - sensor_type = SENSORS.get(set_req(self.value_type).name, [None, None]) - if isinstance(sensor_type, dict): - sensor_type = sensor_type.get(pres(self.child_type).name, [None, None]) + + _sensor_type = SENSORS.get(set_req(self.value_type).name, [None, None]) + if isinstance(_sensor_type, dict): + sensor_type = _sensor_type.get(pres(self.child_type).name, [None, None]) + else: + sensor_type = _sensor_type return sensor_type diff --git a/homeassistant/components/mysensors/switch.py b/homeassistant/components/mysensors/switch.py index 3910df55eec..cdb4979d16b 100644 --- a/homeassistant/components/mysensors/switch.py +++ b/homeassistant/components/mysensors/switch.py @@ -1,16 +1,28 @@ """Support for MySensors switches.""" +from __future__ import annotations + +from contextlib import suppress +from typing import Any + import voluptuous as vol from homeassistant.components import mysensors from homeassistant.components.switch import DOMAIN, SwitchEntity from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON -from homeassistant.core import HomeAssistant +from homeassistant.core import HomeAssistant, ServiceCall import homeassistant.helpers.config_validation as cv from homeassistant.helpers.entity_platform import AddEntitiesCallback from ...config_entries import ConfigEntry from ...helpers.dispatcher import async_dispatcher_connect -from .const import DOMAIN as MYSENSORS_DOMAIN, MYSENSORS_DISCOVERY, SERVICE_SEND_IR_CODE +from .const import ( + DOMAIN as MYSENSORS_DOMAIN, + MYSENSORS_DISCOVERY, + SERVICE_SEND_IR_CODE, + DiscoveryInfo, + SensorType, +) +from .device import MySensorsDevice from .helpers import on_unload ATTR_IR_CODE = "V_IR_SEND" @@ -24,9 +36,9 @@ async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, -): +) -> None: """Set up this platform for a specific ConfigEntry(==Gateway).""" - device_class_map = { + device_class_map: dict[SensorType, type[MySensorsDevice]] = { "S_DOOR": MySensorsSwitch, "S_MOTION": MySensorsSwitch, "S_SMOKE": MySensorsSwitch, @@ -42,7 +54,7 @@ async def async_setup_entry( "S_WATER_QUALITY": MySensorsSwitch, } - async def async_discover(discovery_info): + async def async_discover(discovery_info: DiscoveryInfo) -> None: """Discover and add a MySensors switch.""" mysensors.setup_mysensors_platform( hass, @@ -52,7 +64,7 @@ async def async_setup_entry( async_add_entities=async_add_entities, ) - async def async_send_ir_code_service(service): + async def async_send_ir_code_service(service: ServiceCall) -> None: """Set IR code as device state attribute.""" entity_ids = service.data.get(ATTR_ENTITY_ID) ir_code = service.data.get(ATTR_IR_CODE) @@ -98,17 +110,23 @@ class MySensorsSwitch(mysensors.device.MySensorsEntity, SwitchEntity): """Representation of the value of a MySensors Switch child node.""" @property - def current_power_w(self): + def current_power_w(self) -> float | None: """Return the current power usage in W.""" set_req = self.gateway.const.SetReq - return self._values.get(set_req.V_WATT) + value = self._values.get(set_req.V_WATT) + float_value: float | None = None + if value is not None: + with suppress(ValueError): + float_value = float(value) + + return float_value @property - def is_on(self): + def is_on(self) -> bool: """Return True if switch is on.""" return self._values.get(self.value_type) == STATE_ON - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the switch on.""" self.gateway.set_child_value( self.node_id, self.child_id, self.value_type, 1, ack=1 @@ -118,7 +136,7 @@ class MySensorsSwitch(mysensors.device.MySensorsEntity, SwitchEntity): self._values[self.value_type] = STATE_ON self.async_write_ha_state() - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the switch off.""" self.gateway.set_child_value( self.node_id, self.child_id, self.value_type, 0, ack=1 @@ -132,18 +150,18 @@ class MySensorsSwitch(mysensors.device.MySensorsEntity, SwitchEntity): class MySensorsIRSwitch(MySensorsSwitch): """IR switch child class to MySensorsSwitch.""" - def __init__(self, *args): + def __init__(self, *args: Any) -> None: """Set up instance attributes.""" super().__init__(*args) - self._ir_code = None + self._ir_code: str | None = None @property - def is_on(self): + def is_on(self) -> bool: """Return True if switch is on.""" set_req = self.gateway.const.SetReq return self._values.get(set_req.V_LIGHT) == STATE_ON - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the IR switch on.""" set_req = self.gateway.const.SetReq if ATTR_IR_CODE in kwargs: @@ -162,7 +180,7 @@ class MySensorsIRSwitch(MySensorsSwitch): # Turn off switch after switch was turned on await self.async_turn_off() - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the IR switch off.""" set_req = self.gateway.const.SetReq self.gateway.set_child_value( @@ -173,7 +191,7 @@ class MySensorsIRSwitch(MySensorsSwitch): self._values[set_req.V_LIGHT] = STATE_OFF self.async_write_ha_state() - async def async_update(self): + async def async_update(self) -> None: """Update the controller with the latest value from a sensor.""" await super().async_update() self._ir_code = self._values.get(self.value_type) diff --git a/mypy.ini b/mypy.ini index 7c2dbd38ccd..bada042219a 100644 --- a/mypy.ini +++ b/mypy.ini @@ -517,6 +517,17 @@ no_implicit_optional = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.mysensors.*] +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.nam.*] check_untyped_defs = true disallow_incomplete_defs = true