Fix typing for wemo (#62157)

This commit is contained in:
Eric Severance 2021-12-19 16:09:30 -08:00 committed by GitHub
parent 7919960570
commit 1318597370
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 207 additions and 122 deletions

View File

@ -151,6 +151,7 @@ homeassistant.components.water_heater.*
homeassistant.components.watttime.* homeassistant.components.watttime.*
homeassistant.components.weather.* homeassistant.components.weather.*
homeassistant.components.websocket_api.* homeassistant.components.websocket_api.*
homeassistant.components.wemo.*
homeassistant.components.zodiac.* homeassistant.components.zodiac.*
homeassistant.components.zeroconf.* homeassistant.components.zeroconf.*
homeassistant.components.zone.* homeassistant.components.zone.*

View File

@ -1,7 +1,9 @@
"""Support for WeMo device discovery.""" """Support for WeMo device discovery."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence
import logging import logging
from typing import Any, Optional, Tuple
import pywemo import pywemo
import voluptuous as vol import voluptuous as vol
@ -14,10 +16,11 @@ from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DISCOVERY, EVENT_HOMEASSISTANT_STOP from homeassistant.const import CONF_DISCOVERY, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant, callback from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.event import async_call_later from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.async_ import gather_with_concurrency from homeassistant.util.async_ import gather_with_concurrency
from .const import DOMAIN from .const import DOMAIN
@ -44,21 +47,20 @@ WEMO_MODEL_DISPATCH = {
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
HostPortTuple = Tuple[str, Optional[int]]
def coerce_host_port(value):
def coerce_host_port(value: str) -> HostPortTuple:
"""Validate that provided value is either just host or host:port. """Validate that provided value is either just host or host:port.
Returns (host, None) or (host, port) respectively. Returns (host, None) or (host, port) respectively.
""" """
host, _, port = value.partition(":") host, _, port_str = value.partition(":")
if not host: if not host:
raise vol.Invalid("host cannot be empty") raise vol.Invalid("host cannot be empty")
if port: port = cv.port(port_str) if port_str else None
port = cv.port(port)
else:
port = None
return host, port return host, port
@ -82,7 +84,7 @@ CONFIG_SCHEMA = vol.Schema(
) )
async def async_setup(hass, config): async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up for WeMo devices.""" """Set up for WeMo devices."""
hass.data[DOMAIN] = { hass.data[DOMAIN] = {
"config": config.get(DOMAIN, {}), "config": config.get(DOMAIN, {}),
@ -112,11 +114,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
discovery_responder = pywemo.ssdp.DiscoveryResponder(registry.port) discovery_responder = pywemo.ssdp.DiscoveryResponder(registry.port)
await hass.async_add_executor_job(discovery_responder.start) await hass.async_add_executor_job(discovery_responder.start)
static_conf = config.get(CONF_STATIC, []) static_conf: Sequence[HostPortTuple] = config.get(CONF_STATIC, [])
wemo_dispatcher = WemoDispatcher(entry) wemo_dispatcher = WemoDispatcher(entry)
wemo_discovery = WemoDiscovery(hass, wemo_dispatcher, static_conf) wemo_discovery = WemoDiscovery(hass, wemo_dispatcher, static_conf)
async def async_stop_wemo(event): async def async_stop_wemo(event: Event) -> None:
"""Shutdown Wemo subscriptions and subscription thread on exit.""" """Shutdown Wemo subscriptions and subscription thread on exit."""
_LOGGER.debug("Shutting down WeMo event subscriptions") _LOGGER.debug("Shutting down WeMo event subscriptions")
await hass.async_add_executor_job(registry.stop) await hass.async_add_executor_job(registry.stop)
@ -142,8 +144,8 @@ class WemoDispatcher:
def __init__(self, config_entry: ConfigEntry) -> None: def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize the WemoDispatcher.""" """Initialize the WemoDispatcher."""
self._config_entry = config_entry self._config_entry = config_entry
self._added_serial_numbers = set() self._added_serial_numbers: set[str] = set()
self._loaded_components = set() self._loaded_components: set[str] = set()
async def async_add_unique_device( async def async_add_unique_device(
self, hass: HomeAssistant, wemo: pywemo.WeMoDevice self, hass: HomeAssistant, wemo: pywemo.WeMoDevice
@ -191,16 +193,16 @@ class WemoDiscovery:
self, self,
hass: HomeAssistant, hass: HomeAssistant,
wemo_dispatcher: WemoDispatcher, wemo_dispatcher: WemoDispatcher,
static_config: list[tuple[[str, str | None]]], static_config: Sequence[HostPortTuple],
) -> None: ) -> None:
"""Initialize the WemoDiscovery.""" """Initialize the WemoDiscovery."""
self._hass = hass self._hass = hass
self._wemo_dispatcher = wemo_dispatcher self._wemo_dispatcher = wemo_dispatcher
self._stop = None self._stop: CALLBACK_TYPE | None = None
self._scan_delay = 0 self._scan_delay = 0
self._static_config = static_config self._static_config = static_config
async def async_discover_and_schedule(self, *_) -> None: async def async_discover_and_schedule(self, *_: tuple[Any]) -> None:
"""Periodically scan the network looking for WeMo devices.""" """Periodically scan the network looking for WeMo devices."""
_LOGGER.debug("Scanning network for WeMo devices") _LOGGER.debug("Scanning network for WeMo devices")
try: try:
@ -229,26 +231,23 @@ class WemoDiscovery:
self._stop() self._stop()
self._stop = None self._stop = None
async def discover_statics(self): async def discover_statics(self) -> None:
"""Initialize or Re-Initialize connections to statically configured devices.""" """Initialize or Re-Initialize connections to statically configured devices."""
if self._static_config: if not self._static_config:
return
_LOGGER.debug("Adding statically configured WeMo devices") _LOGGER.debug("Adding statically configured WeMo devices")
for device in await gather_with_concurrency( for device in await gather_with_concurrency(
MAX_CONCURRENCY, MAX_CONCURRENCY,
*( *(
self._hass.async_add_executor_job( self._hass.async_add_executor_job(validate_static_config, host, port)
validate_static_config, host, port
)
for host, port in self._static_config for host, port in self._static_config
), ),
): ):
if device: if device:
await self._wemo_dispatcher.async_add_unique_device( await self._wemo_dispatcher.async_add_unique_device(self._hass, device)
self._hass, device
)
def validate_static_config(host, port): def validate_static_config(host: str, port: int | None) -> pywemo.WeMoDevice | None:
"""Handle a static config.""" """Handle a static config."""
url = pywemo.setup_url_for_address(host, port) url = pywemo.setup_url_for_address(host, port)

View File

@ -4,16 +4,24 @@ import asyncio
from pywemo import Insight, Maker from pywemo import Insight, Maker
from homeassistant.components.binary_sensor import BinarySensorEntity from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN as WEMO_DOMAIN from .const import DOMAIN as WEMO_DOMAIN
from .entity import WemoEntity from .entity import WemoEntity
from .wemo_device import DeviceCoordinator
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up WeMo binary sensors.""" """Set up WeMo binary sensors."""
async def _discovered_wemo(coordinator): async def _discovered_wemo(coordinator: DeviceCoordinator) -> None:
"""Handle a discovered Wemo device.""" """Handle a discovered Wemo device."""
if isinstance(coordinator.wemo, Insight): if isinstance(coordinator.wemo, Insight):
async_add_entities([InsightBinarySensor(coordinator)]) async_add_entities([InsightBinarySensor(coordinator)])
@ -38,7 +46,7 @@ class WemoBinarySensor(WemoEntity, BinarySensorEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return true if the state is on. Standby is on.""" """Return true if the state is on. Standby is on."""
return self.wemo.get_state() return bool(self.wemo.get_state())
class MakerBinarySensor(WemoEntity, BinarySensorEntity): class MakerBinarySensor(WemoEntity, BinarySensorEntity):
@ -49,7 +57,7 @@ class MakerBinarySensor(WemoEntity, BinarySensorEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return true if the Maker's sensor is pulled low.""" """Return true if the Maker's sensor is pulled low."""
return self.wemo.has_sensor and self.wemo.sensor_state == 0 return bool(self.wemo.has_sensor) and self.wemo.sensor_state == 0
class InsightBinarySensor(WemoBinarySensor): class InsightBinarySensor(WemoBinarySensor):

View File

@ -2,12 +2,13 @@
import pywemo import pywemo
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_flow from homeassistant.helpers import config_entry_flow
from . import DOMAIN from . import DOMAIN
async def _async_has_devices(hass): async def _async_has_devices(hass: HomeAssistant) -> bool:
"""Return if there are devices that can be discovered.""" """Return if there are devices that can be discovered."""
return bool(await hass.async_add_executor_job(pywemo.discover_devices)) return bool(await hass.async_add_executor_job(pywemo.discover_devices))

View File

@ -1,10 +1,20 @@
"""Triggers for WeMo devices.""" """Triggers for WeMo devices."""
from __future__ import annotations
from typing import Any
from pywemo.subscribe import EVENT_TYPE_LONG_PRESS from pywemo.subscribe import EVENT_TYPE_LONG_PRESS
import voluptuous as vol import voluptuous as vol
from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.homeassistant.triggers import event as event_trigger from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN as WEMO_DOMAIN, WEMO_SUBSCRIPTION_EVENT from .const import DOMAIN as WEMO_DOMAIN, WEMO_SUBSCRIPTION_EVENT
from .wemo_device import async_get_coordinator from .wemo_device import async_get_coordinator
@ -18,7 +28,9 @@ TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
) )
async def async_get_triggers(hass, device_id): async def async_get_triggers(
hass: HomeAssistant, device_id: str
) -> list[dict[str, Any]]:
"""Return a list of triggers.""" """Return a list of triggers."""
wemo_trigger = { wemo_trigger = {
@ -44,7 +56,12 @@ async def async_get_triggers(hass, device_id):
return triggers return triggers
async def async_attach_trigger(hass, config, action, automation_info): async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
action: AutomationActionType,
automation_info: AutomationTriggerInfo,
) -> CALLBACK_TYPE:
"""Attach a trigger.""" """Attach a trigger."""
event_config = event_trigger.TRIGGER_SCHEMA( event_config = event_trigger.TRIGGER_SCHEMA(
{ {

View File

@ -33,7 +33,7 @@ class WemoEntity(CoordinatorEntity):
self._available = True self._available = True
@property @property
def name_suffix(self): def name_suffix(self) -> str | None:
"""Suffix to append to the WeMo device name.""" """Suffix to append to the WeMo device name."""
return self._name_suffix return self._name_suffix
@ -42,7 +42,7 @@ class WemoEntity(CoordinatorEntity):
"""Return the name of the device if any.""" """Return the name of the device if any."""
if suffix := self.name_suffix: if suffix := self.name_suffix:
return f"{self.wemo.name} {suffix}" return f"{self.wemo.name} {suffix}"
return self.wemo.name return str(self.wemo.name)
@property @property
def available(self) -> bool: def available(self) -> bool:
@ -50,10 +50,10 @@ class WemoEntity(CoordinatorEntity):
return super().available and self._available return super().available and self._available
@property @property
def unique_id_suffix(self): def unique_id_suffix(self) -> str | None:
"""Suffix to append to the WeMo device's unique ID.""" """Suffix to append to the WeMo device's unique ID."""
if self._unique_id_suffix is None and self.name_suffix is not None: if self._unique_id_suffix is None and self.name_suffix is not None:
return self._name_suffix.lower() return self.name_suffix.lower()
return self._unique_id_suffix return self._unique_id_suffix
@property @property
@ -61,7 +61,7 @@ class WemoEntity(CoordinatorEntity):
"""Return the id of this WeMo device.""" """Return the id of this WeMo device."""
if suffix := self.unique_id_suffix: if suffix := self.unique_id_suffix:
return f"{self.wemo.serialnumber}_{suffix}" return f"{self.wemo.serialnumber}_{suffix}"
return self.wemo.serialnumber return str(self.wemo.serialnumber)
@property @property
def device_info(self) -> DeviceInfo: def device_info(self) -> DeviceInfo:

View File

@ -1,14 +1,19 @@
"""Support for WeMo humidifier.""" """Support for WeMo humidifier."""
from __future__ import annotations
import asyncio import asyncio
from datetime import timedelta from datetime import timedelta
import math import math
from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.fan import SUPPORT_SET_SPEED, FanEntity from homeassistant.components.fan import SUPPORT_SET_SPEED, FanEntity
from homeassistant.core import callback from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_platform from homeassistant.helpers import entity_platform
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util.percentage import ( from homeassistant.util.percentage import (
int_states_in_range, int_states_in_range,
percentage_to_ranged_value, percentage_to_ranged_value,
@ -21,6 +26,7 @@ from .const import (
SERVICE_SET_HUMIDITY, SERVICE_SET_HUMIDITY,
) )
from .entity import WemoEntity from .entity import WemoEntity
from .wemo_device import DeviceCoordinator
SCAN_INTERVAL = timedelta(seconds=10) SCAN_INTERVAL = timedelta(seconds=10)
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
@ -63,10 +69,14 @@ SET_HUMIDITY_SCHEMA = {
} }
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up WeMo binary sensors.""" """Set up WeMo binary sensors."""
async def _discovered_wemo(coordinator): async def _discovered_wemo(coordinator: DeviceCoordinator) -> None:
"""Handle a discovered Wemo device.""" """Handle a discovered Wemo device."""
async_add_entities([WemoHumidifier(coordinator)]) async_add_entities([WemoHumidifier(coordinator)])
@ -95,7 +105,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class WemoHumidifier(WemoEntity, FanEntity): class WemoHumidifier(WemoEntity, FanEntity):
"""Representation of a WeMo humidifier.""" """Representation of a WeMo humidifier."""
def __init__(self, coordinator): def __init__(self, coordinator: DeviceCoordinator) -> None:
"""Initialize the WeMo switch.""" """Initialize the WeMo switch."""
super().__init__(coordinator) super().__init__(coordinator)
if self.wemo.fan_mode != WEMO_FAN_OFF: if self.wemo.fan_mode != WEMO_FAN_OFF:
@ -104,12 +114,12 @@ class WemoHumidifier(WemoEntity, FanEntity):
self._last_fan_on_mode = WEMO_FAN_MEDIUM self._last_fan_on_mode = WEMO_FAN_MEDIUM
@property @property
def icon(self): def icon(self) -> str:
"""Return the icon of device based on its type.""" """Return the icon of device based on its type."""
return "mdi:water-percent" return "mdi:water-percent"
@property @property
def extra_state_attributes(self): def extra_state_attributes(self) -> dict[str, Any]:
"""Return device specific state attributes.""" """Return device specific state attributes."""
return { return {
ATTR_CURRENT_HUMIDITY: self.wemo.current_humidity_percent, ATTR_CURRENT_HUMIDITY: self.wemo.current_humidity_percent,
@ -145,26 +155,26 @@ class WemoHumidifier(WemoEntity, FanEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return true if the state is on.""" """Return true if the state is on."""
return self.wemo.get_state() return bool(self.wemo.get_state())
def turn_on( def turn_on(
self, self,
speed: str = None, speed: str | None = None,
percentage: int = None, percentage: int | None = None,
preset_mode: str = None, preset_mode: str | None = None,
**kwargs, **kwargs: Any,
) -> None: ) -> None:
"""Turn the fan on.""" """Turn the fan on."""
self.set_percentage(percentage) self.set_percentage(percentage)
def turn_off(self, **kwargs) -> None: def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
with self._wemo_exception_handler("turn off"): with self._wemo_exception_handler("turn off"):
self.wemo.set_state(WEMO_FAN_OFF) self.wemo.set_state(WEMO_FAN_OFF)
self.schedule_update_ha_state() self.schedule_update_ha_state()
def set_percentage(self, percentage: int) -> None: def set_percentage(self, percentage: int | None) -> None:
"""Set the fan_mode of the Humidifier.""" """Set the fan_mode of the Humidifier."""
if percentage is None: if percentage is None:
named_speed = self._last_fan_on_mode named_speed = self._last_fan_on_mode

View File

@ -1,5 +1,8 @@
"""Support for Belkin WeMo lights.""" """Support for Belkin WeMo lights."""
from __future__ import annotations
import asyncio import asyncio
from typing import Any
from pywemo.ouimeaux_device import bridge from pywemo.ouimeaux_device import bridge
@ -14,10 +17,12 @@ from homeassistant.components.light import (
SUPPORT_TRANSITION, SUPPORT_TRANSITION,
LightEntity, LightEntity,
) )
from homeassistant.core import callback from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import CONNECTION_ZIGBEE from homeassistant.helpers.device_registry import CONNECTION_ZIGBEE
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
import homeassistant.util.color as color_util import homeassistant.util.color as color_util
from .const import DOMAIN as WEMO_DOMAIN from .const import DOMAIN as WEMO_DOMAIN
@ -32,10 +37,14 @@ SUPPORT_WEMO = (
WEMO_OFF = 0 WEMO_OFF = 0
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up WeMo lights.""" """Set up WeMo lights."""
async def _discovered_wemo(coordinator: DeviceCoordinator): async def _discovered_wemo(coordinator: DeviceCoordinator) -> None:
"""Handle a discovered Wemo device.""" """Handle a discovered Wemo device."""
if isinstance(coordinator.wemo, bridge.Bridge): if isinstance(coordinator.wemo, bridge.Bridge):
async_setup_bridge(hass, config_entry, async_add_entities, coordinator) async_setup_bridge(hass, config_entry, async_add_entities, coordinator)
@ -53,12 +62,17 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
@callback @callback
def async_setup_bridge(hass, config_entry, async_add_entities, coordinator): def async_setup_bridge(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
coordinator: DeviceCoordinator,
) -> None:
"""Set up a WeMo link.""" """Set up a WeMo link."""
known_light_ids = set() known_light_ids = set()
@callback @callback
def async_update_lights(): def async_update_lights() -> None:
"""Check to see if the bridge has any new lights.""" """Check to see if the bridge has any new lights."""
new_lights = [] new_lights = []
@ -87,7 +101,7 @@ class WemoLight(WemoEntity, LightEntity):
@property @property
def name(self) -> str: def name(self) -> str:
"""Return the name of the device if any.""" """Return the name of the device if any."""
return self.light.name return str(self.light.name)
@property @property
def available(self) -> bool: def available(self) -> bool:
@ -95,9 +109,9 @@ class WemoLight(WemoEntity, LightEntity):
return super().available and self.light.state.get("available") return super().available and self.light.state.get("available")
@property @property
def unique_id(self): def unique_id(self) -> str:
"""Return the ID of this light.""" """Return the ID of this light."""
return self.light.uniqueID return str(self.light.uniqueID)
@property @property
def device_info(self) -> DeviceInfo: def device_info(self) -> DeviceInfo:
@ -111,33 +125,35 @@ class WemoLight(WemoEntity, LightEntity):
) )
@property @property
def brightness(self): def brightness(self) -> int:
"""Return the brightness of this light between 0..255.""" """Return the brightness of this light between 0..255."""
return self.light.state.get("level", 255) return int(self.light.state.get("level", 255))
@property @property
def hs_color(self): def hs_color(self) -> tuple[float, float] | None:
"""Return the hs color values of this light.""" """Return the hs color values of this light."""
if xy_color := self.light.state.get("color_xy"): if xy_color := self.light.state.get("color_xy"):
return color_util.color_xy_to_hs(*xy_color) return color_util.color_xy_to_hs(*xy_color)
return None return None
@property @property
def color_temp(self): def color_temp(self) -> int | None:
"""Return the color temperature of this light in mireds.""" """Return the color temperature of this light in mireds."""
return self.light.state.get("temperature_mireds") if (temp := self.light.state.get("temperature_mireds")) is not None:
return int(temp)
return None
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return true if device is on.""" """Return true if device is on."""
return self.light.state.get("onoff") != WEMO_OFF return bool(self.light.state.get("onoff") != WEMO_OFF)
@property @property
def supported_features(self): def supported_features(self) -> int:
"""Flag supported features.""" """Flag supported features."""
return SUPPORT_WEMO return SUPPORT_WEMO
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the light on.""" """Turn the light on."""
xy_color = None xy_color = None
@ -168,7 +184,7 @@ class WemoLight(WemoEntity, LightEntity):
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the light off.""" """Turn the light off."""
transition_time = int(kwargs.get(ATTR_TRANSITION, 0)) transition_time = int(kwargs.get(ATTR_TRANSITION, 0))
@ -182,12 +198,12 @@ class WemoDimmer(WemoEntity, LightEntity):
"""Representation of a WeMo dimmer.""" """Representation of a WeMo dimmer."""
@property @property
def supported_features(self): def supported_features(self) -> int:
"""Flag supported features.""" """Flag supported features."""
return SUPPORT_BRIGHTNESS return SUPPORT_BRIGHTNESS
@property @property
def brightness(self): def brightness(self) -> int:
"""Return the brightness of this light between 1 and 100.""" """Return the brightness of this light between 1 and 100."""
wemo_brightness = int(self.wemo.get_brightness()) wemo_brightness = int(self.wemo.get_brightness())
return int((wemo_brightness * 255) / 100) return int((wemo_brightness * 255) / 100)
@ -195,9 +211,9 @@ class WemoDimmer(WemoEntity, LightEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return true if the state is on.""" """Return true if the state is on."""
return self.wemo.get_state() return bool(self.wemo.get_state())
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the dimmer on.""" """Turn the dimmer on."""
# Wemo dimmer switches use a range of [0, 100] to control # Wemo dimmer switches use a range of [0, 100] to control
# brightness. Level 255 might mean to set it to previous value # brightness. Level 255 might mean to set it to previous value
@ -212,7 +228,7 @@ class WemoDimmer(WemoEntity, LightEntity):
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the dimmer off.""" """Turn the dimmer off."""
with self._wemo_exception_handler("turn off"): with self._wemo_exception_handler("turn off"):
self.wemo.off() self.wemo.off()

View File

@ -7,8 +7,11 @@ from homeassistant.components.sensor import (
SensorEntityDescription, SensorEntityDescription,
SensorStateClass, SensorStateClass,
) )
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ENERGY_KILO_WATT_HOUR, POWER_WATT from homeassistant.const import ENERGY_KILO_WATT_HOUR, POWER_WATT
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType from homeassistant.helpers.typing import StateType
from homeassistant.util import convert from homeassistant.util import convert
@ -17,10 +20,14 @@ from .entity import WemoEntity
from .wemo_device import DeviceCoordinator from .wemo_device import DeviceCoordinator
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up WeMo sensors.""" """Set up WeMo sensors."""
async def _discovered_wemo(coordinator: DeviceCoordinator): async def _discovered_wemo(coordinator: DeviceCoordinator) -> None:
"""Handle a discovered Wemo device.""" """Handle a discovered Wemo device."""
async_add_entities( async_add_entities(
[InsightCurrentPower(coordinator), InsightTodayEnergy(coordinator)] [InsightCurrentPower(coordinator), InsightTodayEnergy(coordinator)]
@ -42,7 +49,7 @@ class InsightSensor(WemoEntity, SensorEntity):
@property @property
def name_suffix(self) -> str: def name_suffix(self) -> str:
"""Return the name of the entity if any.""" """Return the name of the entity if any."""
return self.entity_description.name return str(self.entity_description.name)
@property @property
def unique_id_suffix(self) -> str: def unique_id_suffix(self) -> str:
@ -50,7 +57,7 @@ class InsightSensor(WemoEntity, SensorEntity):
return self.entity_description.key return self.entity_description.key
@property @property
def available(self) -> str: def available(self) -> bool:
"""Return true if sensor is available.""" """Return true if sensor is available."""
return ( return (
self.entity_description.key in self.wemo.insight_params self.entity_description.key in self.wemo.insight_params
@ -72,12 +79,11 @@ class InsightCurrentPower(InsightSensor):
@property @property
def native_value(self) -> StateType: def native_value(self) -> StateType:
"""Return the current power consumption.""" """Return the current power consumption."""
return ( milliwatts = convert(
convert(
self.wemo.insight_params.get(self.entity_description.key), float, 0.0 self.wemo.insight_params.get(self.entity_description.key), float, 0.0
) )
/ 1000.0 assert isinstance(milliwatts, float)
) return milliwatts / 1000.0
class InsightTodayEnergy(InsightSensor): class InsightTodayEnergy(InsightSensor):
@ -94,7 +100,8 @@ class InsightTodayEnergy(InsightSensor):
@property @property
def native_value(self) -> StateType: def native_value(self) -> StateType:
"""Return the current energy use today.""" """Return the current energy use today."""
miliwatts = convert( milliwatt_seconds = convert(
self.wemo.insight_params.get(self.entity_description.key), float, 0.0 self.wemo.insight_params.get(self.entity_description.key), float, 0.0
) )
return round(miliwatts / (1000.0 * 1000.0 * 60), 2) assert isinstance(milliwatt_seconds, float)
return round(milliwatt_seconds / (1000.0 * 1000.0 * 60), 2)

View File

@ -1,16 +1,23 @@
"""Support for WeMo switches.""" """Support for WeMo switches."""
from __future__ import annotations
import asyncio import asyncio
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Any
from pywemo import CoffeeMaker, Insight, Maker from pywemo import CoffeeMaker, Insight, Maker
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_OFF, STATE_ON, STATE_STANDBY, STATE_UNKNOWN from homeassistant.const import STATE_OFF, STATE_ON, STATE_STANDBY, STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import convert from homeassistant.util import convert
from .const import DOMAIN as WEMO_DOMAIN from .const import DOMAIN as WEMO_DOMAIN
from .entity import WemoEntity from .entity import WemoEntity
from .wemo_device import DeviceCoordinator
SCAN_INTERVAL = timedelta(seconds=10) SCAN_INTERVAL = timedelta(seconds=10)
PARALLEL_UPDATES = 0 PARALLEL_UPDATES = 0
@ -29,10 +36,14 @@ WEMO_OFF = 0
WEMO_STANDBY = 8 WEMO_STANDBY = 8
async def async_setup_entry(hass, config_entry, async_add_entities): async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up WeMo switches.""" """Set up WeMo switches."""
async def _discovered_wemo(coordinator): async def _discovered_wemo(coordinator: DeviceCoordinator) -> None:
"""Handle a discovered Wemo device.""" """Handle a discovered Wemo device."""
async_add_entities([WemoSwitch(coordinator)]) async_add_entities([WemoSwitch(coordinator)])
@ -50,9 +61,9 @@ class WemoSwitch(WemoEntity, SwitchEntity):
"""Representation of a WeMo switch.""" """Representation of a WeMo switch."""
@property @property
def extra_state_attributes(self): def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the device.""" """Return the state attributes of the device."""
attr = {} attr: dict[str, Any] = {}
if isinstance(self.wemo, Maker): if isinstance(self.wemo, Maker):
# Is the maker sensor on or off. # Is the maker sensor on or off.
if self.wemo.maker_params["hassensor"]: if self.wemo.maker_params["hassensor"]:
@ -81,10 +92,11 @@ class WemoSwitch(WemoEntity, SwitchEntity):
attr["on_total_time"] = WemoSwitch.as_uptime( attr["on_total_time"] = WemoSwitch.as_uptime(
self.wemo.insight_params.get("ontotal", 0) self.wemo.insight_params.get("ontotal", 0)
) )
attr["power_threshold_w"] = ( threshold = convert(
convert(self.wemo.insight_params.get("powerthreshold"), float, 0.0) self.wemo.insight_params.get("powerthreshold"), float, 0.0
/ 1000.0
) )
assert isinstance(threshold, float)
attr["power_threshold_w"] = threshold / 1000.0
if isinstance(self.wemo, CoffeeMaker): if isinstance(self.wemo, CoffeeMaker):
attr[ATTR_COFFEMAKER_MODE] = self.wemo.mode attr[ATTR_COFFEMAKER_MODE] = self.wemo.mode
@ -92,7 +104,7 @@ class WemoSwitch(WemoEntity, SwitchEntity):
return attr return attr
@staticmethod @staticmethod
def as_uptime(_seconds): def as_uptime(_seconds: int) -> str:
"""Format seconds into uptime string in the format: 00d 00h 00m 00s.""" """Format seconds into uptime string in the format: 00d 00h 00m 00s."""
uptime = datetime(1, 1, 1) + timedelta(seconds=_seconds) uptime = datetime(1, 1, 1) + timedelta(seconds=_seconds)
return "{:0>2d}d {:0>2d}h {:0>2d}m {:0>2d}s".format( return "{:0>2d}d {:0>2d}h {:0>2d}m {:0>2d}s".format(
@ -100,26 +112,28 @@ class WemoSwitch(WemoEntity, SwitchEntity):
) )
@property @property
def current_power_w(self): def current_power_w(self) -> float | None:
"""Return the current power usage in W.""" """Return the current power usage in W."""
if isinstance(self.wemo, Insight): if not isinstance(self.wemo, Insight):
return ( return None
convert(self.wemo.insight_params.get("currentpower"), float, 0.0) milliwatts = convert(self.wemo.insight_params.get("currentpower"), float, 0.0)
/ 1000.0 assert isinstance(milliwatts, float)
) return milliwatts / 1000.0
@property @property
def today_energy_kwh(self): def today_energy_kwh(self) -> float | None:
"""Return the today total energy usage in kWh.""" """Return the today total energy usage in kWh."""
if isinstance(self.wemo, Insight): if not isinstance(self.wemo, Insight):
miliwatts = convert(self.wemo.insight_params.get("todaymw"), float, 0.0) return None
return round(miliwatts / (1000.0 * 1000.0 * 60), 2) milliwatt_seconds = convert(self.wemo.insight_params.get("todaymw"), float, 0.0)
assert isinstance(milliwatt_seconds, float)
return round(milliwatt_seconds / (1000.0 * 1000.0 * 60), 2)
@property @property
def detail_state(self): def detail_state(self) -> str:
"""Return the state of the device.""" """Return the state of the device."""
if isinstance(self.wemo, CoffeeMaker): if isinstance(self.wemo, CoffeeMaker):
return self.wemo.mode_string return str(self.wemo.mode_string)
if isinstance(self.wemo, Insight): if isinstance(self.wemo, Insight):
standby_state = int(self.wemo.insight_params.get("state", 0)) standby_state = int(self.wemo.insight_params.get("state", 0))
if standby_state == WEMO_ON: if standby_state == WEMO_ON:
@ -129,9 +143,10 @@ class WemoSwitch(WemoEntity, SwitchEntity):
if standby_state == WEMO_STANDBY: if standby_state == WEMO_STANDBY:
return STATE_STANDBY return STATE_STANDBY
return STATE_UNKNOWN return STATE_UNKNOWN
assert False # Unreachable code statement.
@property @property
def icon(self): def icon(self) -> str | None:
"""Return the icon of device based on its type.""" """Return the icon of device based on its type."""
if isinstance(self.wemo, CoffeeMaker): if isinstance(self.wemo, CoffeeMaker):
return "mdi:coffee" return "mdi:coffee"
@ -140,16 +155,16 @@ class WemoSwitch(WemoEntity, SwitchEntity):
@property @property
def is_on(self) -> bool: def is_on(self) -> bool:
"""Return true if the state is on. Standby is on.""" """Return true if the state is on. Standby is on."""
return self.wemo.get_state() return bool(self.wemo.get_state())
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
with self._wemo_exception_handler("turn on"): with self._wemo_exception_handler("turn on"):
self.wemo.on() self.wemo.on()
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
with self._wemo_exception_handler("turn off"): with self._wemo_exception_handler("turn off"):
self.wemo.off() self.wemo.off()

View File

@ -165,4 +165,5 @@ async def async_register_device(
@callback @callback
def async_get_coordinator(hass: HomeAssistant, device_id: str) -> DeviceCoordinator: def async_get_coordinator(hass: HomeAssistant, device_id: str) -> DeviceCoordinator:
"""Return DeviceCoordinator for device_id.""" """Return DeviceCoordinator for device_id."""
return hass.data[DOMAIN]["devices"][device_id] coordinator: DeviceCoordinator = hass.data[DOMAIN]["devices"][device_id]
return coordinator

View File

@ -1672,6 +1672,17 @@ no_implicit_optional = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.wemo.*]
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.zodiac.*] [mypy-homeassistant.components.zodiac.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true
@ -2066,9 +2077,6 @@ ignore_errors = true
[mypy-homeassistant.components.vizio.*] [mypy-homeassistant.components.vizio.*]
ignore_errors = true ignore_errors = true
[mypy-homeassistant.components.wemo.*]
ignore_errors = true
[mypy-homeassistant.components.withings.*] [mypy-homeassistant.components.withings.*]
ignore_errors = true ignore_errors = true

View File

@ -127,7 +127,6 @@ IGNORED_MODULES: Final[list[str]] = [
"homeassistant.components.vera.*", "homeassistant.components.vera.*",
"homeassistant.components.verisure.*", "homeassistant.components.verisure.*",
"homeassistant.components.vizio.*", "homeassistant.components.vizio.*",
"homeassistant.components.wemo.*",
"homeassistant.components.withings.*", "homeassistant.components.withings.*",
"homeassistant.components.xbox.*", "homeassistant.components.xbox.*",
"homeassistant.components.xiaomi_aqara.*", "homeassistant.components.xiaomi_aqara.*",

View File

@ -8,7 +8,7 @@ from homeassistant.components.homeassistant import (
DOMAIN as HA_DOMAIN, DOMAIN as HA_DOMAIN,
SERVICE_UPDATE_ENTITY, SERVICE_UPDATE_ENTITY,
) )
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN from homeassistant.components.light import ATTR_COLOR_TEMP, DOMAIN as LIGHT_DOMAIN
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
@ -88,13 +88,16 @@ async def test_light_update_entity(
# On state. # On state.
pywemo_bridge_light.state["onoff"] = 1 pywemo_bridge_light.state["onoff"] = 1
pywemo_bridge_light.state["temperature_mireds"] = 432
await hass.services.async_call( await hass.services.async_call(
HA_DOMAIN, HA_DOMAIN,
SERVICE_UPDATE_ENTITY, SERVICE_UPDATE_ENTITY,
{ATTR_ENTITY_ID: [wemo_entity.entity_id]}, {ATTR_ENTITY_ID: [wemo_entity.entity_id]},
blocking=True, blocking=True,
) )
assert hass.states.get(wemo_entity.entity_id).state == STATE_ON state = hass.states.get(wemo_entity.entity_id)
assert state.attributes.get(ATTR_COLOR_TEMP) == 432
assert state.state == STATE_ON
# Off state. # Off state.
pywemo_bridge_light.state["onoff"] = 0 pywemo_bridge_light.state["onoff"] = 0