Improve entity type hints [i] (#77529)

This commit is contained in:
epenet 2022-08-30 19:21:08 +02:00 committed by GitHub
parent 50663bbc5d
commit 23090cb8a2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 74 additions and 58 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from iaqualink.const import (
AQUALINK_TEMP_CELSIUS_HIGH,
@ -105,7 +106,7 @@ class HassAqualinkThermostat(AqualinkEntity, ClimateEntity):
return float(self.dev.state)
@refresh_system
async def async_set_temperature(self, **kwargs) -> None:
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
await await_or_reraise(self.dev.set_temperature(int(kwargs[ATTR_TEMPERATURE])))

View file

@ -1,6 +1,8 @@
"""Support for Aqualink pool feature switches."""
from __future__ import annotations
from typing import Any
from homeassistant.components.switch import DOMAIN, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@ -52,11 +54,11 @@ class HassAqualinkSwitch(AqualinkEntity, SwitchEntity):
return self.dev.is_on
@refresh_system
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await await_or_reraise(self.dev.turn_on())
@refresh_system
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await await_or_reraise(self.dev.turn_off())

View file

@ -1,6 +1,8 @@
"""Support for IHC switches."""
from __future__ import annotations
from typing import Any
from ihcsdk.ihccontroller import IHCController
from homeassistant.components.switch import SwitchEntity
@ -64,14 +66,14 @@ class IHCSwitch(IHCDevice, SwitchEntity):
"""Return true if switch is on."""
return self._state
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
if self._ihc_on_id:
await async_pulse(self.hass, self.ihc_controller, self._ihc_on_id)
else:
await async_set_bool(self.hass, self.ihc_controller, self.ihc_id, True)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
if self._ihc_off_id:
await async_pulse(self.hass, self.ihc_controller, self._ihc_off_id)

View file

@ -89,7 +89,7 @@ class ImapSensor(SensorEntity):
self._does_push = None
self._idle_loop_task = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle when an entity is about to be added to Home Assistant."""
if not self.should_poll:
self._idle_loop_task = self.hass.loop.create_task(self.idle_loop())
@ -110,12 +110,12 @@ class ImapSensor(SensorEntity):
return self._email_count
@property
def available(self):
def available(self) -> bool:
"""Return the availability of the device."""
return self._connection is not None
@property
def should_poll(self):
def should_poll(self) -> bool:
"""Return if polling is needed."""
return not self._does_push
@ -151,7 +151,7 @@ class ImapSensor(SensorEntity):
except (AioImapException, asyncio.TimeoutError):
self.disconnected()
async def async_update(self):
async def async_update(self) -> None:
"""Periodic polling of state."""
try:
if await self.connection():

View file

@ -252,7 +252,7 @@ class EmailContentSensor(SensorEntity):
return email_message.get_payload()
def update(self):
def update(self) -> None:
"""Read emails and publish state change."""
email_message = self._email_reader.read_next()

View file

@ -75,7 +75,7 @@ class InComfortClimate(IncomfortChild, ClimateEntity):
"""Return max valid temperature that can be set."""
return 30.0
async def async_set_temperature(self, **kwargs) -> None:
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set a new target temperature for this zone."""
temperature = kwargs.get(ATTR_TEMPERATURE)
await self._room.set_override(temperature)

View file

@ -242,7 +242,7 @@ class InfluxSensor(SensorEntity):
"""Return the unit of measurement of this entity, if any."""
return self._unit_of_measurement
def update(self):
def update(self) -> None:
"""Get the latest data from Influxdb and updates the states."""
self.data.update()
if (value := self.data.value) is None:

View file

@ -1,6 +1,8 @@
"""Support for Insteon thermostat."""
from __future__ import annotations
from typing import Any
from pyinsteon.config import CELSIUS
from pyinsteon.constants import ThermostatMode
@ -182,7 +184,7 @@ class InsteonClimateEntity(InsteonEntity, ClimateEntity):
attr["humidifier"] = humidifier
return attr
async def async_set_temperature(self, **kwargs) -> None:
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
target_temp = kwargs.get(ATTR_TEMPERATURE)
target_temp_low = kwargs.get(ATTR_TARGET_TEMP_LOW)
@ -214,7 +216,7 @@ class InsteonClimateEntity(InsteonEntity, ClimateEntity):
await self._insteon_device.async_set_humidity_low_set_point(low)
await self._insteon_device.async_set_humidity_high_set_point(high)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register INSTEON update events."""
await super().async_added_to_hass()
await self._insteon_device.async_read_op_flags()

View file

@ -1,4 +1,6 @@
"""Support for INSTEON dimmers via PowerLinc Modem."""
from typing import Any
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
@ -37,10 +39,10 @@ class InsteonSwitchEntity(InsteonEntity, SwitchEntity):
"""Return the boolean response if the node is on."""
return bool(self._insteon_device_group.value)
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn switch on."""
await self._insteon_device.async_on(group=self._insteon_device_group.group)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn switch off."""
await self._insteon_device.async_off(group=self._insteon_device_group.group)

View file

@ -150,7 +150,7 @@ class IntegrationSensor(RestoreEntity, SensorEntity):
self._attr_unique_id = unique_id
self._sensor_source_id = source_entity
self._round_digits = round_digits
self._state = None
self._state: Decimal | None = None
self._method = integration_method
self._attr_name = name if name is not None else f"{source_entity} integral"
@ -174,7 +174,7 @@ class IntegrationSensor(RestoreEntity, SensorEntity):
return self._unit_template.format(integral_unit)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if state := await self.async_get_last_state():

View file

@ -1,6 +1,8 @@
"""Intellifire Climate Entities."""
from __future__ import annotations
from typing import Any
from homeassistant.components.climate import (
ClimateEntity,
ClimateEntityDescription,
@ -70,7 +72,7 @@ class IntellifireClimate(IntellifireEntity, ClimateEntity):
return HVACMode.HEAT
return HVACMode.OFF
async def async_set_temperature(self, **kwargs) -> None:
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Turn on thermostat by setting a target temperature."""
raw_target_temp = kwargs[ATTR_TEMPERATURE]
self.last_temp = int(raw_target_temp)
@ -93,7 +95,7 @@ class IntellifireClimate(IntellifireEntity, ClimateEntity):
"""Return target temperature."""
return float(self.coordinator.read_api.data.thermostat_setpoint_c)
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set HVAC mode to normal or thermostat control."""
LOGGER.debug(
"Setting mode to [%s] - using last temp: %s", hvac_mode, self.last_temp

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import logging
from random import randrange
from typing import NamedTuple
from typing import Any, NamedTuple
from pyintesishome import IHAuthenticationError, IHConnectionError, IntesisHome
import voluptuous as vol
@ -204,7 +204,7 @@ class IntesisAC(ClimateEntity):
self._attr_hvac_modes.extend(mode_list)
self._attr_hvac_modes.append(HVACMode.OFF)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Subscribe to event updates."""
_LOGGER.debug("Added climate device with state: %s", repr(self._ih_device))
await self._controller.add_update_callback(self.async_update_callback)
@ -256,7 +256,7 @@ class IntesisAC(ClimateEntity):
"""Return the current preset mode."""
return self._preset
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
if hvac_mode := kwargs.get(ATTR_HVAC_MODE):
await self.async_set_hvac_mode(hvac_mode)
@ -295,7 +295,7 @@ class IntesisAC(ClimateEntity):
self._hvac_mode = hvac_mode
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 fan mode (from quiet, low, medium, high, auto)."""
await self._controller.set_fan_speed(self._device_id, fan_mode)
@ -303,12 +303,12 @@ class IntesisAC(ClimateEntity):
self._fan_speed = fan_mode
self.async_write_ha_state()
async def async_set_preset_mode(self, preset_mode):
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set preset mode."""
ih_preset_mode = MAP_PRESET_MODE_TO_IH.get(preset_mode)
await self._controller.set_preset_mode(self._device_id, ih_preset_mode)
async def async_set_swing_mode(self, swing_mode):
async def async_set_swing_mode(self, swing_mode: str) -> None:
"""Set the vertical vane."""
if swing_settings := MAP_SWING_TO_IH.get(swing_mode):
await self._controller.set_vertical_vane(
@ -318,7 +318,7 @@ class IntesisAC(ClimateEntity):
self._device_id, swing_settings.hvane
)
async def async_update(self):
async def async_update(self) -> None:
"""Copy values from controller dictionary to climate device."""
# Update values from controller's device dictionary
self._connected = self._controller.is_connected
@ -353,7 +353,7 @@ class IntesisAC(ClimateEntity):
self._device_id
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Shutdown the controller when the device is being removed."""
await self._controller.stop()

View file

@ -62,7 +62,7 @@ class Iperf3Sensor(RestoreEntity, SensorEntity):
ATTR_VERSION: self._iperf3_data.data[ATTR_VERSION],
}
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
@ -76,7 +76,7 @@ class Iperf3Sensor(RestoreEntity, SensorEntity):
return
self._attr_native_value = state.state
def update(self):
def update(self) -> None:
"""Get the latest data and update the states."""
data = self._iperf3_data.data.get(self.entity_description.key)
if data is not None:

View file

@ -190,7 +190,7 @@ class IPMAWeather(WeatherEntity):
self._forecast = None
@Throttle(MIN_TIME_BETWEEN_UPDATES)
async def async_update(self):
async def async_update(self) -> None:
"""Update Condition and Forecast."""
async with async_timeout.timeout(10):
new_observation = await self._location.observation(self._api)

View file

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

View file

@ -54,7 +54,7 @@ class IslamicPrayerTimeSensor(SensorEntity):
dt_util.UTC
)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
self.async_on_remove(
async_dispatcher_connect(self.hass, DATA_UPDATED, self.async_write_ha_state)

View file

@ -1,7 +1,9 @@
"""Support for iTach IR devices."""
from __future__ import annotations
from collections.abc import Iterable
import logging
from typing import Any
import pyitachip2ir
import voluptuous as vol
@ -123,19 +125,19 @@ class ITachIP2IRRemote(remote.RemoteEntity):
"""Return true if device is on."""
return self._power
def turn_on(self, **kwargs):
def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
self._power = True
self.itachip2ir.send(self._name, "ON", self._ir_count)
self.schedule_update_ha_state()
def turn_off(self, **kwargs):
def turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
self._power = False
self.itachip2ir.send(self._name, "OFF", self._ir_count)
self.schedule_update_ha_state()
def send_command(self, command, **kwargs):
def send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send a command to one device."""
num_repeats = kwargs.get(ATTR_NUM_REPEATS, DEFAULT_NUM_REPEATS)
for single_command in command:
@ -143,6 +145,6 @@ class ITachIP2IRRemote(remote.RemoteEntity):
self._name, single_command, self._ir_count * num_repeats
)
def update(self):
def update(self) -> None:
"""Update the device."""
self.itachip2ir.update()

View file

@ -1,6 +1,8 @@
"""Support for interfacing to iTunes API."""
from __future__ import annotations
from typing import Any
import requests
import voluptuous as vol
@ -268,7 +270,7 @@ class ItunesDevice(MediaPlayerEntity):
return STATE_PLAYING
def update(self):
def update(self) -> None:
"""Retrieve latest state."""
now_playing = self.client.now_playing()
self.update_state(now_playing)
@ -354,48 +356,48 @@ class ItunesDevice(MediaPlayerEntity):
"""Boolean if shuffle is enabled."""
return self.shuffled
def set_volume_level(self, volume):
def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
response = self.client.set_volume(int(volume * 100))
self.update_state(response)
def mute_volume(self, mute):
def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
response = self.client.set_muted(mute)
self.update_state(response)
def set_shuffle(self, shuffle):
def set_shuffle(self, shuffle: bool) -> None:
"""Shuffle (true) or no shuffle (false) media player."""
response = self.client.set_shuffle(shuffle)
self.update_state(response)
def media_play(self):
def media_play(self) -> None:
"""Send media_play command to media player."""
response = self.client.play()
self.update_state(response)
def media_pause(self):
def media_pause(self) -> None:
"""Send media_pause command to media player."""
response = self.client.pause()
self.update_state(response)
def media_next_track(self):
def media_next_track(self) -> None:
"""Send media_next command to media player."""
response = self.client.next()
self.update_state(response)
def media_previous_track(self):
def media_previous_track(self) -> None:
"""Send media_previous command media player."""
response = self.client.previous()
self.update_state(response)
def play_media(self, media_type, media_id, **kwargs):
def play_media(self, media_type: str, media_id: str, **kwargs: Any) -> None:
"""Send the play_media command to the media player."""
if media_type == MEDIA_TYPE_PLAYLIST:
response = self.client.play_playlist(media_id)
self.update_state(response)
def turn_off(self):
def turn_off(self) -> None:
"""Turn the media player off."""
response = self.client.stop()
self.update_state(response)
@ -471,7 +473,7 @@ class AirPlayDevice(MediaPlayerEntity):
return STATE_OFF
def update(self):
def update(self) -> None:
"""Retrieve latest state."""
@property
@ -484,20 +486,20 @@ class AirPlayDevice(MediaPlayerEntity):
"""Flag of media content that is supported."""
return MEDIA_TYPE_MUSIC
def set_volume_level(self, volume):
def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
volume = int(volume * 100)
response = self.client.set_volume_airplay_device(self._id, volume)
self.update_state(response)
def turn_on(self):
def turn_on(self) -> None:
"""Select AirPlay."""
self.update_state({"selected": True})
self.schedule_update_ha_state()
response = self.client.toggle_airplay_device(self._id, True)
self.update_state(response)
def turn_off(self):
def turn_off(self) -> None:
"""Deselect AirPlay."""
self.update_state({"selected": False})
self.schedule_update_ha_state()

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from pizone import Controller, Zone
import voluptuous as vol
@ -170,7 +171,7 @@ class ControllerDevice(ClimateEntity):
for zone in controller.zones:
self.zones[zone] = ZoneDevice(self, zone)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call on adding to hass."""
# Register for connect/disconnect/update events
@callback
@ -395,7 +396,7 @@ class ControllerDevice(ClimateEntity):
else:
self.set_available(True)
async def async_set_temperature(self, **kwargs) -> None:
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
if not self.supported_features & ClimateEntityFeature.TARGET_TEMPERATURE:
self.async_schedule_update_ha_state(True)
@ -468,7 +469,7 @@ class ZoneDevice(ClimateEntity):
via_device=(IZONE, controller.unique_id),
)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call on adding to hass."""
@callback
@ -517,7 +518,7 @@ class ZoneDevice(ClimateEntity):
@property # type: ignore[misc]
@_return_on_connection_error(0)
def supported_features(self):
def supported_features(self) -> int:
"""Return the list of supported features."""
if self._zone.mode == Zone.Mode.AUTO:
return self._attr_supported_features
@ -588,7 +589,7 @@ class ZoneDevice(ClimateEntity):
)
self.async_write_ha_state()
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
if self._zone.mode != Zone.Mode.AUTO:
return
@ -606,7 +607,7 @@ class ZoneDevice(ClimateEntity):
"""Return true if on."""
return self._zone.mode != Zone.Mode.CLOSE
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn device on (open zone)."""
if self._zone.type == Zone.Type.AUTO:
await self._controller.wrap_and_catch(self._zone.set_mode(Zone.Mode.AUTO))
@ -614,7 +615,7 @@ class ZoneDevice(ClimateEntity):
await self._controller.wrap_and_catch(self._zone.set_mode(Zone.Mode.OPEN))
self.async_write_ha_state()
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn device off (close zone)."""
await self._controller.wrap_and_catch(self._zone.set_mode(Zone.Mode.CLOSE))
self.async_write_ha_state()