jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Type mysensors strictly (#51535)

Browse Source
This commit is contained in:
Martin Hjelmare 2021-06-07 16:04:04 +02:00 committed by GitHub
parent 564042ec67
commit 7560a77e0e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 329 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -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.*

11
homeassistant/components/mysensors/__init__.py
View File

@ -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

11
homeassistant/components/mysensors/binary_sensor.py
View File

@ -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)

55
homeassistant/components/mysensors/climate.py
View File

@ -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]]

8
homeassistant/components/mysensors/config_flow.py
View File

@ -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],

55
homeassistant/components/mysensors/const.py
View File

@ -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, ...

29
homeassistant/components/mysensors/cover.py
View File

@ -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(

71
homeassistant/components/mysensors/device.py
View File

@ -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(

23
homeassistant/components/mysensors/device_tracker.py
View File

@ -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,
)

36
homeassistant/components/mysensors/gateway.py
View File

@ -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.

6
homeassistant/components/mysensors/handler.py
View File

@ -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)

7
homeassistant/components/mysensors/helpers.py
View File

@ -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

77
homeassistant/components/mysensors/light.py
View File

@ -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)

27
homeassistant/components/mysensors/notify.py
View File

@ -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"<MySensorsNotificationDevice {self.name}>"
@ -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 = [

41
homeassistant/components/mysensors/sensor.py
View File

@ -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

52
homeassistant/components/mysensors/switch.py
View File

@ -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)

11
mypy.ini
View File

@ -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