Improve entity type hints [d] (#77031)

This commit is contained in:
epenet 2022-08-19 16:10:45 +02:00 committed by GitHub
parent 2d197fd59e
commit bf7239c25d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
31 changed files with 141 additions and 119 deletions

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
import voluptuous as vol import voluptuous as vol
@ -177,7 +178,7 @@ class DaikinClimate(ClimateEntity):
return self._api.device.mac return self._api.device.mac
@property @property
def temperature_unit(self): def temperature_unit(self) -> str:
"""Return the unit of measurement which this thermostat uses.""" """Return the unit of measurement which this thermostat uses."""
return TEMP_CELSIUS return TEMP_CELSIUS
@ -196,7 +197,7 @@ class DaikinClimate(ClimateEntity):
"""Return the supported step of target temperature.""" """Return the supported step of target temperature."""
return 1 return 1
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature.""" """Set new target temperature."""
await self._set(kwargs) await self._set(kwargs)
@ -232,7 +233,7 @@ class DaikinClimate(ClimateEntity):
"""Return the fan setting.""" """Return the fan setting."""
return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_FAN_MODE])[1].title() return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_FAN_MODE])[1].title()
async def async_set_fan_mode(self, fan_mode): async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set fan mode.""" """Set fan mode."""
await self._set({ATTR_FAN_MODE: fan_mode}) await self._set({ATTR_FAN_MODE: fan_mode})
@ -246,7 +247,7 @@ class DaikinClimate(ClimateEntity):
"""Return the fan setting.""" """Return the fan setting."""
return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_SWING_MODE])[1].title() return self._api.device.represent(HA_ATTR_TO_DAIKIN[ATTR_SWING_MODE])[1].title()
async def async_set_swing_mode(self, swing_mode): async def async_set_swing_mode(self, swing_mode: str) -> None:
"""Set new target temperature.""" """Set new target temperature."""
await self._set({ATTR_SWING_MODE: swing_mode}) await self._set({ATTR_SWING_MODE: swing_mode})
@ -275,7 +276,7 @@ class DaikinClimate(ClimateEntity):
return PRESET_ECO return PRESET_ECO
return PRESET_NONE return PRESET_NONE
async def async_set_preset_mode(self, preset_mode): async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set preset mode.""" """Set preset mode."""
if preset_mode == PRESET_AWAY: if preset_mode == PRESET_AWAY:
await self._api.device.set_holiday(ATTR_STATE_ON) await self._api.device.set_holiday(ATTR_STATE_ON)
@ -309,15 +310,15 @@ class DaikinClimate(ClimateEntity):
ret += [PRESET_ECO, PRESET_BOOST] ret += [PRESET_ECO, PRESET_BOOST]
return ret return ret
async def async_update(self): async def async_update(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
await self._api.async_update() await self._api.async_update()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn device on.""" """Turn device on."""
await self._api.device.set({}) await self._api.device.set({})
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn device off.""" """Turn device off."""
await self._api.device.set( await self._api.device.set(
{HA_ATTR_TO_DAIKIN[ATTR_HVAC_MODE]: HA_STATE_TO_DAIKIN[HVACMode.OFF]} {HA_ATTR_TO_DAIKIN[ATTR_HVAC_MODE]: HA_STATE_TO_DAIKIN[HVACMode.OFF]}

View File

@ -21,6 +21,7 @@ from homeassistant.const import (
TEMP_CELSIUS, TEMP_CELSIUS,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@ -181,7 +182,7 @@ class DaikinSensor(SensorEntity):
self._attr_name = f"{api.name} {description.name}" self._attr_name = f"{api.name} {description.name}"
@property @property
def unique_id(self): def unique_id(self) -> str:
"""Return a unique ID.""" """Return a unique ID."""
return f"{self._api.device.mac}-{self.entity_description.key}" return f"{self._api.device.mac}-{self.entity_description.key}"
@ -190,11 +191,11 @@ class DaikinSensor(SensorEntity):
"""Return the state of the sensor.""" """Return the state of the sensor."""
return self.entity_description.value_func(self._api.device) return self.entity_description.value_func(self._api.device)
async def async_update(self): async def async_update(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
await self._api.async_update() await self._api.async_update()
@property @property
def device_info(self): def device_info(self) -> DeviceInfo:
"""Return a device description for device registry.""" """Return a device description for device registry."""
return self._api.device_info return self._api.device_info

View File

@ -1,13 +1,16 @@
"""Support for Daikin AirBase zones.""" """Support for Daikin AirBase zones."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN as DAIKIN_DOMAIN from . import DOMAIN as DAIKIN_DOMAIN, DaikinApi
ZONE_ICON = "mdi:home-circle" ZONE_ICON = "mdi:home-circle"
STREAMER_ICON = "mdi:air-filter" STREAMER_ICON = "mdi:air-filter"
@ -32,7 +35,7 @@ async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None: ) -> None:
"""Set up Daikin climate based on config_entry.""" """Set up Daikin climate based on config_entry."""
daikin_api = hass.data[DAIKIN_DOMAIN][entry.entry_id] daikin_api: DaikinApi = hass.data[DAIKIN_DOMAIN][entry.entry_id]
switches: list[DaikinZoneSwitch | DaikinStreamerSwitch] = [] switches: list[DaikinZoneSwitch | DaikinStreamerSwitch] = []
if zones := daikin_api.device.zones: if zones := daikin_api.device.zones:
switches.extend( switches.extend(
@ -54,13 +57,13 @@ async def async_setup_entry(
class DaikinZoneSwitch(SwitchEntity): class DaikinZoneSwitch(SwitchEntity):
"""Representation of a zone.""" """Representation of a zone."""
def __init__(self, daikin_api, zone_id): def __init__(self, daikin_api: DaikinApi, zone_id):
"""Initialize the zone.""" """Initialize the zone."""
self._api = daikin_api self._api = daikin_api
self._zone_id = zone_id self._zone_id = zone_id
@property @property
def unique_id(self): def unique_id(self) -> str:
"""Return a unique ID.""" """Return a unique ID."""
return f"{self._api.device.mac}-zone{self._zone_id}" return f"{self._api.device.mac}-zone{self._zone_id}"
@ -70,29 +73,29 @@ class DaikinZoneSwitch(SwitchEntity):
return ZONE_ICON return ZONE_ICON
@property @property
def name(self): def name(self) -> str:
"""Return the name of the sensor.""" """Return the name of the sensor."""
return f"{self._api.name} {self._api.device.zones[self._zone_id][0]}" return f"{self._api.name} {self._api.device.zones[self._zone_id][0]}"
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return the state of the sensor.""" """Return the state of the sensor."""
return self._api.device.zones[self._zone_id][1] == "1" return self._api.device.zones[self._zone_id][1] == "1"
@property @property
def device_info(self): def device_info(self) -> DeviceInfo:
"""Return a device description for device registry.""" """Return a device description for device registry."""
return self._api.device_info return self._api.device_info
async def async_update(self): async def async_update(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
await self._api.async_update() await self._api.async_update()
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the zone on.""" """Turn the zone on."""
await self._api.device.set_zone(self._zone_id, "1") await self._api.device.set_zone(self._zone_id, "1")
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the zone off.""" """Turn the zone off."""
await self._api.device.set_zone(self._zone_id, "0") await self._api.device.set_zone(self._zone_id, "0")
@ -100,12 +103,12 @@ class DaikinZoneSwitch(SwitchEntity):
class DaikinStreamerSwitch(SwitchEntity): class DaikinStreamerSwitch(SwitchEntity):
"""Streamer state.""" """Streamer state."""
def __init__(self, daikin_api): def __init__(self, daikin_api: DaikinApi) -> None:
"""Initialize streamer switch.""" """Initialize streamer switch."""
self._api = daikin_api self._api = daikin_api
@property @property
def unique_id(self): def unique_id(self) -> str:
"""Return a unique ID.""" """Return a unique ID."""
return f"{self._api.device.mac}-streamer" return f"{self._api.device.mac}-streamer"
@ -115,30 +118,30 @@ class DaikinStreamerSwitch(SwitchEntity):
return STREAMER_ICON return STREAMER_ICON
@property @property
def name(self): def name(self) -> str:
"""Return the name of the sensor.""" """Return the name of the sensor."""
return f"{self._api.name} streamer" return f"{self._api.name} streamer"
@property @property
def is_on(self): def is_on(self) -> bool:
"""Return the state of the sensor.""" """Return the state of the sensor."""
return ( return (
DAIKIN_ATTR_STREAMER in self._api.device.represent(DAIKIN_ATTR_ADVANCED)[1] DAIKIN_ATTR_STREAMER in self._api.device.represent(DAIKIN_ATTR_ADVANCED)[1]
) )
@property @property
def device_info(self): def device_info(self) -> DeviceInfo:
"""Return a device description for device registry.""" """Return a device description for device registry."""
return self._api.device_info return self._api.device_info
async def async_update(self): async def async_update(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
await self._api.async_update() await self._api.async_update()
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the zone on.""" """Turn the zone on."""
await self._api.device.set_streamer("on") await self._api.device.set_streamer("on")
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the zone off.""" """Turn the zone off."""
await self._api.device.set_streamer("off") await self._api.device.set_streamer("off")

View File

@ -50,7 +50,7 @@ class DanfossAirBinarySensor(BinarySensorEntity):
self._type = sensor_type self._type = sensor_type
self._attr_device_class = device_class self._attr_device_class = device_class
def update(self): def update(self) -> None:
"""Fetch new state data for the sensor.""" """Fetch new state data for the sensor."""
self._data.update() self._data.update()

View File

@ -119,7 +119,7 @@ class DanfossAir(SensorEntity):
self._attr_device_class = device_class self._attr_device_class = device_class
self._attr_state_class = state_class self._attr_state_class = state_class
def update(self): def update(self) -> None:
"""Update the new state of the sensor. """Update the new state of the sensor.
This is done through the DanfossAir object that does the actual This is done through the DanfossAir object that does the actual

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from pydanfossair.commands import ReadCommand, UpdateCommand from pydanfossair.commands import ReadCommand, UpdateCommand
@ -75,17 +76,17 @@ class DanfossAir(SwitchEntity):
"""Return true if switch is on.""" """Return true if switch is on."""
return self._state return self._state
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
_LOGGER.debug("Turning on switch with command %s", self._on_command) _LOGGER.debug("Turning on switch with command %s", self._on_command)
self._data.update_state(self._on_command, self._state_command) self._data.update_state(self._on_command, self._state_command)
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
_LOGGER.debug("Turning off switch with command %s", self._off_command) _LOGGER.debug("Turning off switch with command %s", self._off_command)
self._data.update_state(self._off_command, self._state_command) self._data.update_state(self._off_command, self._state_command)
def update(self): def update(self) -> None:
"""Update the switch's state.""" """Update the switch's state."""
self._data.update() self._data.update()

View File

@ -825,7 +825,7 @@ class DarkSkyAlertSensor(SensorEntity):
"""Return the state attributes.""" """Return the state attributes."""
return self._alerts return self._alerts
def update(self): def update(self) -> None:
"""Get the latest data from Dark Sky and updates the states.""" """Get the latest data from Dark Sky and updates the states."""
# Call the API for new forecast data. Each sensor will re-trigger this # Call the API for new forecast data. Each sensor will re-trigger this
# same exact call, but that's fine. We cache results for a short period # same exact call, but that's fine. We cache results for a short period

View File

@ -131,7 +131,7 @@ class DarkSkyWeather(WeatherEntity):
self._ds_daily = None self._ds_daily = None
@property @property
def available(self): def available(self) -> bool:
"""Return if weather data is available from Dark Sky.""" """Return if weather data is available from Dark Sky."""
return self._ds_data is not None return self._ds_data is not None
@ -233,7 +233,7 @@ class DarkSkyWeather(WeatherEntity):
return data return data
def update(self): def update(self) -> None:
"""Get the latest data from Dark Sky.""" """Get the latest data from Dark Sky."""
self._dark_sky.update() self._dark_sky.update()

View File

@ -92,7 +92,7 @@ class DeLijnPublicTransportSensor(SensorEntity):
self.line = line self.line = line
self._attr_extra_state_attributes = {} self._attr_extra_state_attributes = {}
async def async_update(self): async def async_update(self) -> None:
"""Get the latest data from the De Lijn API.""" """Get the latest data from the De Lijn API."""
try: try:
await self.line.get_passages() await self.line.get_passages()

View File

@ -1,6 +1,8 @@
"""Demo implementation of the media player.""" """Demo implementation of the media player."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from homeassistant.components.media_player import ( from homeassistant.components.media_player import (
MediaPlayerDeviceClass, MediaPlayerDeviceClass,
MediaPlayerEntity, MediaPlayerEntity,
@ -164,57 +166,57 @@ class AbstractDemoPlayer(MediaPlayerEntity):
"""Return the device class of the media player.""" """Return the device class of the media player."""
return self._device_class return self._device_class
def turn_on(self): def turn_on(self) -> None:
"""Turn the media player on.""" """Turn the media player on."""
self._player_state = STATE_PLAYING self._player_state = STATE_PLAYING
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_off(self): def turn_off(self) -> None:
"""Turn the media player off.""" """Turn the media player off."""
self._player_state = STATE_OFF self._player_state = STATE_OFF
self.schedule_update_ha_state() self.schedule_update_ha_state()
def mute_volume(self, mute): def mute_volume(self, mute: bool) -> None:
"""Mute the volume.""" """Mute the volume."""
self._volume_muted = mute self._volume_muted = mute
self.schedule_update_ha_state() self.schedule_update_ha_state()
def volume_up(self): def volume_up(self) -> None:
"""Increase volume.""" """Increase volume."""
self._volume_level = min(1.0, self._volume_level + 0.1) self._volume_level = min(1.0, self._volume_level + 0.1)
self.schedule_update_ha_state() self.schedule_update_ha_state()
def volume_down(self): def volume_down(self) -> None:
"""Decrease volume.""" """Decrease volume."""
self._volume_level = max(0.0, self._volume_level - 0.1) self._volume_level = max(0.0, self._volume_level - 0.1)
self.schedule_update_ha_state() self.schedule_update_ha_state()
def set_volume_level(self, volume): def set_volume_level(self, volume: float) -> None:
"""Set the volume level, range 0..1.""" """Set the volume level, range 0..1."""
self._volume_level = volume self._volume_level = volume
self.schedule_update_ha_state() self.schedule_update_ha_state()
def media_play(self): def media_play(self) -> None:
"""Send play command.""" """Send play command."""
self._player_state = STATE_PLAYING self._player_state = STATE_PLAYING
self.schedule_update_ha_state() self.schedule_update_ha_state()
def media_pause(self): def media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
self._player_state = STATE_PAUSED self._player_state = STATE_PAUSED
self.schedule_update_ha_state() self.schedule_update_ha_state()
def media_stop(self): def media_stop(self) -> None:
"""Send stop command.""" """Send stop command."""
self._player_state = STATE_OFF self._player_state = STATE_OFF
self.schedule_update_ha_state() self.schedule_update_ha_state()
def set_shuffle(self, shuffle): def set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode.""" """Enable/disable shuffle mode."""
self._shuffle = shuffle self._shuffle = shuffle
self.schedule_update_ha_state() self.schedule_update_ha_state()
def select_sound_mode(self, sound_mode): def select_sound_mode(self, sound_mode: str) -> None:
"""Select sound mode.""" """Select sound mode."""
self._sound_mode = sound_mode self._sound_mode = sound_mode
self.schedule_update_ha_state() self.schedule_update_ha_state()
@ -291,12 +293,12 @@ class DemoYoutubePlayer(AbstractDemoPlayer):
if self._player_state == STATE_PLAYING: if self._player_state == STATE_PLAYING:
return self._progress_updated_at return self._progress_updated_at
def play_media(self, media_type, media_id, **kwargs): def play_media(self, media_type: str, media_id: str, **kwargs: Any) -> None:
"""Play a piece of media.""" """Play a piece of media."""
self.youtube_id = media_id self.youtube_id = media_id
self.schedule_update_ha_state() self.schedule_update_ha_state()
def media_pause(self): def media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
self._progress = self.media_position self._progress = self.media_position
self._progress_updated_at = dt_util.utcnow() self._progress_updated_at = dt_util.utcnow()
@ -393,38 +395,38 @@ class DemoMusicPlayer(AbstractDemoPlayer):
"""Flag media player features that are supported.""" """Flag media player features that are supported."""
return MUSIC_PLAYER_SUPPORT return MUSIC_PLAYER_SUPPORT
def media_previous_track(self): def media_previous_track(self) -> None:
"""Send previous track command.""" """Send previous track command."""
if self._cur_track > 0: if self._cur_track > 0:
self._cur_track -= 1 self._cur_track -= 1
self.schedule_update_ha_state() self.schedule_update_ha_state()
def media_next_track(self): def media_next_track(self) -> None:
"""Send next track command.""" """Send next track command."""
if self._cur_track < len(self.tracks) - 1: if self._cur_track < len(self.tracks) - 1:
self._cur_track += 1 self._cur_track += 1
self.schedule_update_ha_state() self.schedule_update_ha_state()
def clear_playlist(self): def clear_playlist(self) -> None:
"""Clear players playlist.""" """Clear players playlist."""
self.tracks = [] self.tracks = []
self._cur_track = 0 self._cur_track = 0
self._player_state = STATE_OFF self._player_state = STATE_OFF
self.schedule_update_ha_state() self.schedule_update_ha_state()
def set_repeat(self, repeat): def set_repeat(self, repeat: str) -> None:
"""Enable/disable repeat mode.""" """Enable/disable repeat mode."""
self._repeat = repeat self._repeat = repeat
self.schedule_update_ha_state() self.schedule_update_ha_state()
def join_players(self, group_members): def join_players(self, group_members: list[str]) -> None:
"""Join `group_members` as a player group with the current player.""" """Join `group_members` as a player group with the current player."""
self._group_members = [ self._group_members = [
self.entity_id, self.entity_id,
] + group_members ] + group_members
self.schedule_update_ha_state() self.schedule_update_ha_state()
def unjoin_player(self): def unjoin_player(self) -> None:
"""Remove this player from any group.""" """Remove this player from any group."""
self._group_members = [] self._group_members = []
self.schedule_update_ha_state() self.schedule_update_ha_state()
@ -505,19 +507,19 @@ class DemoTVShowPlayer(AbstractDemoPlayer):
"""Flag media player features that are supported.""" """Flag media player features that are supported."""
return NETFLIX_PLAYER_SUPPORT return NETFLIX_PLAYER_SUPPORT
def media_previous_track(self): def media_previous_track(self) -> None:
"""Send previous track command.""" """Send previous track command."""
if self._cur_episode > 1: if self._cur_episode > 1:
self._cur_episode -= 1 self._cur_episode -= 1
self.schedule_update_ha_state() self.schedule_update_ha_state()
def media_next_track(self): def media_next_track(self) -> None:
"""Send next track command.""" """Send next track command."""
if self._cur_episode < self._episode_count: if self._cur_episode < self._episode_count:
self._cur_episode += 1 self._cur_episode += 1
self.schedule_update_ha_state() self.schedule_update_ha_state()
def select_source(self, source): def select_source(self, source: str) -> None:
"""Set the input source.""" """Set the input source."""
self._source = source self._source = source
self.schedule_update_ha_state() self.schedule_update_ha_state()

View File

@ -240,7 +240,12 @@ class DemoVacuum(VacuumEntity):
self._battery_level += 5 self._battery_level += 5
self.schedule_update_ha_state() self.schedule_update_ha_state()
def send_command(self, command, params=None, **kwargs: Any) -> None: def send_command(
self,
command: str,
params: dict[str, Any] | list[Any] | None = None,
**kwargs: Any,
) -> None:
"""Send a command to the vacuum.""" """Send a command to the vacuum."""
if self.supported_features & VacuumEntityFeature.SEND_COMMAND == 0: if self.supported_features & VacuumEntityFeature.SEND_COMMAND == 0:
return return

View File

@ -1,6 +1,8 @@
"""Demo platform that offers a fake water heater device.""" """Demo platform that offers a fake water heater device."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from homeassistant.components.water_heater import ( from homeassistant.components.water_heater import (
WaterHeaterEntity, WaterHeaterEntity,
WaterHeaterEntityFeature, WaterHeaterEntityFeature,
@ -79,22 +81,22 @@ class DemoWaterHeater(WaterHeaterEntity):
"off", "off",
] ]
def set_temperature(self, **kwargs): def set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures.""" """Set new target temperatures."""
self._attr_target_temperature = kwargs.get(ATTR_TEMPERATURE) self._attr_target_temperature = kwargs.get(ATTR_TEMPERATURE)
self.schedule_update_ha_state() self.schedule_update_ha_state()
def set_operation_mode(self, operation_mode): def set_operation_mode(self, operation_mode: str) -> None:
"""Set new operation mode.""" """Set new operation mode."""
self._attr_current_operation = operation_mode self._attr_current_operation = operation_mode
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_away_mode_on(self): def turn_away_mode_on(self) -> None:
"""Turn away mode on.""" """Turn away mode on."""
self._attr_is_away_mode_on = True self._attr_is_away_mode_on = True
self.schedule_update_ha_state() self.schedule_update_ha_state()
def turn_away_mode_off(self): def turn_away_mode_off(self) -> None:
"""Turn away mode off.""" """Turn away mode off."""
self._attr_is_away_mode_on = False self._attr_is_away_mode_on = False
self.schedule_update_ha_state() self.schedule_update_ha_state()

View File

@ -253,51 +253,51 @@ class DenonDevice(MediaPlayerEntity):
if self._mediasource == name: if self._mediasource == name:
return pretty_name return pretty_name
def turn_off(self): def turn_off(self) -> None:
"""Turn off media player.""" """Turn off media player."""
self.telnet_command("PWSTANDBY") self.telnet_command("PWSTANDBY")
def volume_up(self): def volume_up(self) -> None:
"""Volume up media player.""" """Volume up media player."""
self.telnet_command("MVUP") self.telnet_command("MVUP")
def volume_down(self): def volume_down(self) -> None:
"""Volume down media player.""" """Volume down media player."""
self.telnet_command("MVDOWN") self.telnet_command("MVDOWN")
def set_volume_level(self, volume): def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.""" """Set volume level, range 0..1."""
self.telnet_command(f"MV{round(volume * self._volume_max):02}") self.telnet_command(f"MV{round(volume * self._volume_max):02}")
def mute_volume(self, mute): def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player.""" """Mute (true) or unmute (false) media player."""
mute_status = "ON" if mute else "OFF" mute_status = "ON" if mute else "OFF"
self.telnet_command(f"MU{mute_status})") self.telnet_command(f"MU{mute_status})")
def media_play(self): def media_play(self) -> None:
"""Play media player.""" """Play media player."""
self.telnet_command("NS9A") self.telnet_command("NS9A")
def media_pause(self): def media_pause(self) -> None:
"""Pause media player.""" """Pause media player."""
self.telnet_command("NS9B") self.telnet_command("NS9B")
def media_stop(self): def media_stop(self) -> None:
"""Pause media player.""" """Pause media player."""
self.telnet_command("NS9C") self.telnet_command("NS9C")
def media_next_track(self): def media_next_track(self) -> None:
"""Send the next track command.""" """Send the next track command."""
self.telnet_command("NS9D") self.telnet_command("NS9D")
def media_previous_track(self): def media_previous_track(self) -> None:
"""Send the previous track command.""" """Send the previous track command."""
self.telnet_command("NS9E") self.telnet_command("NS9E")
def turn_on(self): def turn_on(self) -> None:
"""Turn the media player on.""" """Turn the media player on."""
self.telnet_command("PWON") self.telnet_command("PWON")
def select_source(self, source): def select_source(self, source: str) -> None:
"""Select input source.""" """Select input source."""
self.telnet_command(f"SI{self._source_list.get(source)}") self.telnet_command(f"SI{self._source_list.get(source)}")

View File

@ -390,32 +390,32 @@ class DenonDevice(MediaPlayerEntity):
return self._receiver.dynamic_eq return self._receiver.dynamic_eq
@async_log_errors @async_log_errors
async def async_media_play_pause(self): async def async_media_play_pause(self) -> None:
"""Play or pause the media player.""" """Play or pause the media player."""
await self._receiver.async_toggle_play_pause() await self._receiver.async_toggle_play_pause()
@async_log_errors @async_log_errors
async def async_media_play(self): async def async_media_play(self) -> None:
"""Send play command.""" """Send play command."""
await self._receiver.async_play() await self._receiver.async_play()
@async_log_errors @async_log_errors
async def async_media_pause(self): async def async_media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
await self._receiver.async_pause() await self._receiver.async_pause()
@async_log_errors @async_log_errors
async def async_media_previous_track(self): async def async_media_previous_track(self) -> None:
"""Send previous track command.""" """Send previous track command."""
await self._receiver.async_previous_track() await self._receiver.async_previous_track()
@async_log_errors @async_log_errors
async def async_media_next_track(self): async def async_media_next_track(self) -> None:
"""Send next track command.""" """Send next track command."""
await self._receiver.async_next_track() await self._receiver.async_next_track()
@async_log_errors @async_log_errors
async def async_select_source(self, source: str): async def async_select_source(self, source: str) -> None:
"""Select input source.""" """Select input source."""
# Ensure that the AVR is turned on, which is necessary for input # Ensure that the AVR is turned on, which is necessary for input
# switch to work. # switch to work.
@ -423,27 +423,27 @@ class DenonDevice(MediaPlayerEntity):
await self._receiver.async_set_input_func(source) await self._receiver.async_set_input_func(source)
@async_log_errors @async_log_errors
async def async_select_sound_mode(self, sound_mode: str): async def async_select_sound_mode(self, sound_mode: str) -> None:
"""Select sound mode.""" """Select sound mode."""
await self._receiver.async_set_sound_mode(sound_mode) await self._receiver.async_set_sound_mode(sound_mode)
@async_log_errors @async_log_errors
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn on media player.""" """Turn on media player."""
await self._receiver.async_power_on() await self._receiver.async_power_on()
@async_log_errors @async_log_errors
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn off media player.""" """Turn off media player."""
await self._receiver.async_power_off() await self._receiver.async_power_off()
@async_log_errors @async_log_errors
async def async_volume_up(self): async def async_volume_up(self) -> None:
"""Volume up the media player.""" """Volume up the media player."""
await self._receiver.async_volume_up() await self._receiver.async_volume_up()
@async_log_errors @async_log_errors
async def async_volume_down(self): async def async_volume_down(self) -> None:
"""Volume down media player.""" """Volume down media player."""
await self._receiver.async_volume_down() await self._receiver.async_volume_down()
@ -458,7 +458,7 @@ class DenonDevice(MediaPlayerEntity):
await self._receiver.async_set_volume(volume_denon) await self._receiver.async_set_volume(volume_denon)
@async_log_errors @async_log_errors
async def async_mute_volume(self, mute: bool): async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command.""" """Send mute command."""
await self._receiver.async_mute(mute) await self._receiver.async_mute(mute)

View File

@ -168,7 +168,7 @@ class DerivativeSensor(RestoreEntity, SensorEntity):
self._unit_time = UNIT_TIME[unit_time] self._unit_time = UNIT_TIME[unit_time]
self._time_window = time_window.total_seconds() self._time_window = time_window.total_seconds()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Handle entity which will be added.""" """Handle entity which will be added."""
await super().async_added_to_hass() await super().async_added_to_hass()
if (state := await self.async_get_last_state()) is not None: if (state := await self.async_get_last_state()) is not None:

View File

@ -98,7 +98,7 @@ class DeutscheBahnSensor(SensorEntity):
connections["next_on"] = self.data.connections[2]["departure"] connections["next_on"] = self.data.connections[2]["departure"]
return connections return connections
def update(self): def update(self) -> None:
"""Get the latest delay from bahn.de and updates the state.""" """Get the latest delay from bahn.de and updates the state."""
self.data.update() self.data.update()
self._state = self.data.connections[0].get("departure", "Unknown") self._state = self.data.connections[0].get("departure", "Unknown")

View File

@ -88,7 +88,7 @@ class DevoloClimateDeviceEntity(DevoloMultiLevelSwitchDeviceEntity, ClimateEntit
"""Return the target temperature.""" """Return the target temperature."""
return self._value return self._value
def set_hvac_mode(self, hvac_mode: str) -> None: def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Do nothing as devolo devices do not support changing the hvac mode.""" """Do nothing as devolo devices do not support changing the hvac mode."""
def set_temperature(self, **kwargs: Any) -> None: def set_temperature(self, **kwargs: Any) -> None:

View File

@ -102,7 +102,7 @@ class DigitalOceanBinarySensor(BinarySensorEntity):
ATTR_VCPUS: self.data.vcpus, ATTR_VCPUS: self.data.vcpus,
} }
def update(self): def update(self) -> None:
"""Update state of sensor.""" """Update state of sensor."""
self._digital_ocean.update() self._digital_ocean.update()

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
import voluptuous as vol import voluptuous as vol
@ -94,17 +95,17 @@ class DigitalOceanSwitch(SwitchEntity):
ATTR_VCPUS: self.data.vcpus, ATTR_VCPUS: self.data.vcpus,
} }
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Boot-up the droplet.""" """Boot-up the droplet."""
if self.data.status != "active": if self.data.status != "active":
self.data.power_on() self.data.power_on()
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Shutdown the droplet.""" """Shutdown the droplet."""
if self.data.status == "active": if self.data.status == "active":
self.data.power_off() self.data.power_off()
def update(self): def update(self) -> None:
"""Get the latest data from the device and update the data.""" """Get the latest data from the device and update the data."""
self._digital_ocean.update() self._digital_ocean.update()

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from directv import DIRECTV from directv import DIRECTV
@ -278,7 +279,7 @@ class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerEntity):
return dt_util.as_local(self._program.start_time) return dt_util.as_local(self._program.start_time)
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn on the receiver.""" """Turn on the receiver."""
if self._is_client: if self._is_client:
raise NotImplementedError() raise NotImplementedError()
@ -286,7 +287,7 @@ class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerEntity):
_LOGGER.debug("Turn on %s", self.name) _LOGGER.debug("Turn on %s", self.name)
await self.dtv.remote("poweron", self._address) await self.dtv.remote("poweron", self._address)
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn off the receiver.""" """Turn off the receiver."""
if self._is_client: if self._is_client:
raise NotImplementedError() raise NotImplementedError()
@ -294,32 +295,34 @@ class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerEntity):
_LOGGER.debug("Turn off %s", self.name) _LOGGER.debug("Turn off %s", self.name)
await self.dtv.remote("poweroff", self._address) await self.dtv.remote("poweroff", self._address)
async def async_media_play(self): async def async_media_play(self) -> None:
"""Send play command.""" """Send play command."""
_LOGGER.debug("Play on %s", self.name) _LOGGER.debug("Play on %s", self.name)
await self.dtv.remote("play", self._address) await self.dtv.remote("play", self._address)
async def async_media_pause(self): async def async_media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
_LOGGER.debug("Pause on %s", self.name) _LOGGER.debug("Pause on %s", self.name)
await self.dtv.remote("pause", self._address) await self.dtv.remote("pause", self._address)
async def async_media_stop(self): async def async_media_stop(self) -> None:
"""Send stop command.""" """Send stop command."""
_LOGGER.debug("Stop on %s", self.name) _LOGGER.debug("Stop on %s", self.name)
await self.dtv.remote("stop", self._address) await self.dtv.remote("stop", self._address)
async def async_media_previous_track(self): async def async_media_previous_track(self) -> None:
"""Send rewind command.""" """Send rewind command."""
_LOGGER.debug("Rewind on %s", self.name) _LOGGER.debug("Rewind on %s", self.name)
await self.dtv.remote("rew", self._address) await self.dtv.remote("rew", self._address)
async def async_media_next_track(self): async def async_media_next_track(self) -> None:
"""Send fast forward command.""" """Send fast forward command."""
_LOGGER.debug("Fast forward on %s", self.name) _LOGGER.debug("Fast forward on %s", self.name)
await self.dtv.remote("ffwd", self._address) await self.dtv.remote("ffwd", self._address)
async def async_play_media(self, media_type, media_id, **kwargs): async def async_play_media(
self, media_type: str, media_id: str, **kwargs: Any
) -> None:
"""Select input source.""" """Select input source."""
if media_type != MEDIA_TYPE_CHANNEL: if media_type != MEDIA_TYPE_CHANNEL:
_LOGGER.error( _LOGGER.error(

View File

@ -157,7 +157,7 @@ class DiscogsSensor(SensorEntity):
return None return None
def update(self): def update(self) -> None:
"""Set state to the amount of records in user's collection.""" """Set state to the amount of records in user's collection."""
if self.entity_description.key == SENSOR_COLLECTION_TYPE: if self.entity_description.key == SENSOR_COLLECTION_TYPE:
self._attr_native_value = self._discogs_data["collection_count"] self._attr_native_value = self._discogs_data["collection_count"]

View File

@ -3,6 +3,7 @@ from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any
import urllib import urllib
from pyW215.pyW215 import SmartPlug from pyW215.pyW215 import SmartPlug
@ -106,15 +107,15 @@ class SmartPlugSwitch(SwitchEntity):
"""Return true if switch is on.""" """Return true if switch is on."""
return self.data.state == "ON" return self.data.state == "ON"
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
self.data.smartplug.state = "ON" self.data.smartplug.state = "ON"
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
self.data.smartplug.state = "OFF" self.data.smartplug.state = "OFF"
def update(self): def update(self) -> None:
"""Get the latest data from the smart plug and updates the states.""" """Get the latest data from the smart plug and updates the states."""
self.data.update() self.data.update()

View File

@ -578,7 +578,7 @@ class DlnaDmrEntity(MediaPlayerEntity):
await self._device.async_stop() await self._device.async_stop()
@catch_request_errors @catch_request_errors
async def async_media_seek(self, position: int | float) -> None: async def async_media_seek(self, position: float) -> None:
"""Send seek command.""" """Send seek command."""
assert self._device is not None assert self._device is not None
time = timedelta(seconds=position) time = timedelta(seconds=position)

View File

@ -134,7 +134,7 @@ class DoorBirdCamera(DoorBirdEntity, Camera):
) )
return self._last_image return self._last_image
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Add callback after being added to hass. """Add callback after being added to hass.
Registers entity_id map for the logbook Registers entity_id map for the logbook

View File

@ -133,7 +133,7 @@ class DovadoSensor(SensorEntity):
return round(float(state) / 1e6, 1) return round(float(state) / 1e6, 1)
return state return state
def update(self): def update(self) -> None:
"""Update sensor values.""" """Update sensor values."""
self._data.update() self._data.update()
self._attr_native_value = self._compute_state() self._attr_native_value = self._compute_state()

View File

@ -35,7 +35,7 @@ class DSMRSensor(SensorEntity):
slug = slugify(description.key.replace("/", "_")) slug = slugify(description.key.replace("/", "_"))
self.entity_id = f"sensor.{slug}" self.entity_id = f"sensor.{slug}"
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events.""" """Subscribe to MQTT events."""
@callback @callback

View File

@ -71,7 +71,7 @@ class DteEnergyBridgeSensor(SensorEntity):
self._attr_name = name self._attr_name = name
def update(self): def update(self) -> None:
"""Get the energy usage data from the DTE energy bridge.""" """Get the energy usage data from the DTE energy bridge."""
try: try:
response = requests.get(self._url, timeout=5) response = requests.get(self._url, timeout=5)

View File

@ -125,7 +125,7 @@ class DublinPublicTransportSensor(SensorEntity):
"""Icon to use in the frontend, if any.""" """Icon to use in the frontend, if any."""
return ICON return ICON
def update(self): def update(self) -> None:
"""Get the latest data from opendata.ch and update the states.""" """Get the latest data from opendata.ch and update the states."""
self.data.update() self.data.update()
self._times = self.data.info self._times = self.data.info

View File

@ -164,11 +164,11 @@ class DwdWeatherWarningsSensor(SensorEntity):
return data return data
@property @property
def available(self): def available(self) -> bool:
"""Could the device be accessed during the last update call.""" """Could the device be accessed during the last update call."""
return self._api.api.data_valid return self._api.api.data_valid
def update(self): def update(self) -> None:
"""Get the latest data from the DWD-Weather-Warnings API.""" """Get the latest data from the DWD-Weather-Warnings API."""
_LOGGER.debug( _LOGGER.debug(
"Update requested for %s (%s) by %s", "Update requested for %s (%s) by %s",

View File

@ -92,7 +92,7 @@ class DweetSensor(SensorEntity):
"""Return the state.""" """Return the state."""
return self._state return self._state
def update(self): def update(self) -> None:
"""Get the latest data from REST API.""" """Get the latest data from REST API."""
self.dweet.update() self.dweet.update()

View File

@ -1,5 +1,7 @@
"""Support for the Dynalite channels and presets as switches.""" """Support for the Dynalite channels and presets as switches."""
from typing import Any
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -27,10 +29,10 @@ class DynaliteSwitch(DynaliteBase, SwitchEntity):
"""Return true if switch is on.""" """Return true if switch is on."""
return self._device.is_on return self._device.is_on
async def async_turn_on(self, **kwargs) -> None: async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
await self._device.async_turn_on() await self._device.async_turn_on()
async def async_turn_off(self, **kwargs) -> None: async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
await self._device.async_turn_off() await self._device.async_turn_off()