Improve entity type hints [f] ()

pull/77155/head
epenet 2022-08-22 13:36:33 +02:00 committed by GitHub
parent b108ddbfd3
commit 58b9785485
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
25 changed files with 93 additions and 79 deletions
homeassistant/components

View File

@ -84,7 +84,7 @@ class BanSensor(SensorEntity):
"""Return the most recently banned IP Address."""
return self.last_ban
def update(self):
def update(self) -> None:
"""Update the list of banned ips."""
self.log_parser.read_log(self.jail)

View File

@ -1,6 +1,7 @@
"""Support for Cameras with FFmpeg as decoder."""
from __future__ import annotations
from aiohttp import web
from haffmpeg.camera import CameraMjpeg
from haffmpeg.tools import IMAGE_JPEG
import voluptuous as vol
@ -66,7 +67,9 @@ class FFmpegCamera(Camera):
extra_cmd=self._extra_arguments,
)
async def handle_async_mjpeg_stream(self, request):
async def handle_async_mjpeg_stream(
self, request: web.Request
) -> web.StreamResponse:
"""Generate an HTTP MJPEG stream from the camera."""
stream = CameraMjpeg(self._manager.binary)

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.climate import ENTITY_ID_FORMAT, ClimateEntity
from homeassistant.components.climate.const import (
@ -199,7 +200,7 @@ class FibaroThermostat(FibaroDevice, ClimateEntity):
if mode in OPMODES_PRESET:
self._preset_support.append(OPMODES_PRESET[mode])
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call when entity is added to hass."""
_LOGGER.debug(
"Climate %s\n"
@ -241,7 +242,7 @@ class FibaroThermostat(FibaroDevice, ClimateEntity):
mode = int(self._fan_mode_device.fibaro_device.properties.mode)
return FANMODES[mode]
def set_fan_mode(self, fan_mode):
def set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
if not self._fan_mode_device:
return
@ -270,7 +271,7 @@ class FibaroThermostat(FibaroDevice, ClimateEntity):
return [HVACMode.AUTO] # Default to this
return self._hvac_support
def set_hvac_mode(self, hvac_mode):
def set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target operation mode."""
if not self._op_mode_device:
return
@ -346,7 +347,7 @@ class FibaroThermostat(FibaroDevice, ClimateEntity):
return float(device.properties.targetLevel)
return None
def set_temperature(self, **kwargs):
def set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures."""
temperature = kwargs.get(ATTR_TEMPERATURE)
target = self._target_temp_device

View File

@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
from contextlib import suppress
from functools import partial
from typing import Any
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
@ -107,7 +108,7 @@ class FibaroLight(FibaroDevice, LightEntity):
super().__init__(fibaro_device)
self.entity_id = ENTITY_ID_FORMAT.format(self.ha_id)
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the light on."""
async with self._update_lock:
await self.hass.async_add_executor_job(partial(self._turn_on, **kwargs))
@ -134,7 +135,7 @@ class FibaroLight(FibaroDevice, LightEntity):
# The simplest case is left for last. No dimming, just switch on
self.call_turn_on()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the light off."""
async with self._update_lock:
await self.hass.async_add_executor_job(partial(self._turn_off, **kwargs))
@ -167,7 +168,7 @@ class FibaroLight(FibaroDevice, LightEntity):
return False
async def async_update(self):
async def async_update(self) -> None:
"""Update the state."""
async with self._update_lock:
await self.hass.async_add_executor_job(self._update)

View File

@ -142,7 +142,7 @@ class FibaroSensor(FibaroDevice, SensorEntity):
fibaro_device.properties.unit, fibaro_device.properties.unit
)
def update(self):
def update(self) -> None:
"""Update the state."""
with suppress(KeyError, ValueError):
self._attr_native_value = float(self.fibaro_device.properties.value)

View File

@ -224,7 +224,7 @@ class FidoSensor(SensorEntity):
"""Return the state attributes of the sensor."""
return {"number": self._number}
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest data from Fido and update the state."""
await self.fido_data.async_update()
if (sensor_type := self.entity_description.key) == "balance":

View File

@ -75,7 +75,7 @@ class FileSensor(SensorEntity):
self._attr_native_unit_of_measurement = unit_of_measurement
self._val_tpl = value_template
def update(self):
def update(self) -> None:
"""Get the latest entry from a file and updates the state."""
try:
with FileReadBackwards(self._file_path, encoding="utf-8") as file_data:

View File

@ -277,7 +277,7 @@ class SensorFilter(SensorEntity):
if update_ha:
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
if "recorder" in self.hass.config.components:

View File

@ -69,7 +69,7 @@ class ResponseSwitch(SwitchEntity):
return False
@property
def available(self):
def available(self) -> bool:
"""Return if switch is available."""
return self._client.on_duty
@ -99,11 +99,11 @@ class ResponseSwitch(SwitchEntity):
return attr
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Send Acknowledge response status."""
await self.async_set_response(True)
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Send Reject response status."""
await self.async_set_response(False)

View File

@ -1,5 +1,6 @@
"""Support for Firmata switch output."""
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -57,12 +58,12 @@ class FirmataSwitch(FirmataPinEntity, SwitchEntity):
"""Return true if switch is on."""
return self._api.is_on
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on switch."""
await self._api.turn_on()
self.async_write_ha_state()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off switch."""
await self._api.turn_off()
self.async_write_ha_state()

View File

@ -99,7 +99,7 @@ class ExchangeRateSensor(SensorEntity):
"""Return the icon to use in the frontend, if any."""
return ICON
def update(self):
def update(self) -> None:
"""Get the latest data and updates the states."""
self.data.update()
self._state = round(self.data.rate["rates"][self._target], 3)

View File

@ -210,7 +210,7 @@ class Flexit(ClimateEntity):
else:
_LOGGER.error("Modbus error setting target temperature to Flexit")
async def async_set_fan_mode(self, fan_mode):
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new fan mode."""
if await self._async_write_int16_to_register(
17, self.fan_modes.index(fan_mode)

View File

@ -1,6 +1,7 @@
"""Support for Flick Electric Pricing data."""
from datetime import timedelta
import logging
from typing import Any
import async_timeout
from pyflick import FlickAPI, FlickPrice
@ -45,7 +46,7 @@ class FlickPricingSensor(SensorEntity):
"""Entity object for Flick Electric sensor."""
self._api: FlickAPI = api
self._price: FlickPrice = None
self._attributes = {
self._attributes: dict[str, Any] = {
ATTR_ATTRIBUTION: ATTRIBUTION,
ATTR_FRIENDLY_NAME: FRIENDLY_NAME,
}
@ -65,7 +66,7 @@ class FlickPricingSensor(SensorEntity):
"""Return the state attributes."""
return self._attributes
async def async_update(self):
async def async_update(self) -> None:
"""Get the Flick Pricing data from the web service."""
if self._price and self._price.end_at >= utcnow():
return # Power price data is still valid

View File

@ -1,6 +1,8 @@
"""Switch representing the shutoff valve for the Flo by Moen integration."""
from __future__ import annotations
from typing import Any
from aioflo.location import SLEEP_MINUTE_OPTIONS, SYSTEM_MODE_HOME, SYSTEM_REVERT_MODES
import voluptuous as vol
@ -83,13 +85,13 @@ class FloSwitch(FloEntity, SwitchEntity):
return "mdi:valve-open"
return "mdi:valve-closed"
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Open the valve."""
await self._device.api_client.device.open_valve(self._device.id)
self._state = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Close the valve."""
await self._device.api_client.device.close_valve(self._device.id)
self._state = False
@ -101,7 +103,7 @@ class FloSwitch(FloEntity, SwitchEntity):
self._state = self._device.last_known_valve_state == "open"
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
self.async_on_remove(self._device.async_add_listener(self.async_update_state))

View File

@ -124,7 +124,7 @@ class FlumeSensor(CoordinatorEntity, SensorEntity):
return _format_state_value(self._flume_device.values[sensor_key])
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Request an update when added."""
await super().async_added_to_hass()
# We do not ask for an update with async_add_entities()

View File

@ -7,6 +7,7 @@ from __future__ import annotations
import datetime
import logging
from typing import Any
import voluptuous as vol
@ -220,13 +221,13 @@ class FluxSwitch(SwitchEntity, RestoreEntity):
"""Return true if switch is on."""
return self.unsub_tracker is not None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call when entity about to be added to hass."""
last_state = await self.async_get_last_state()
if last_state and last_state.state == STATE_ON:
await self.async_turn_on()
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on flux."""
if self.is_on:
return
@ -242,7 +243,7 @@ class FluxSwitch(SwitchEntity, RestoreEntity):
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off flux."""
if self.is_on:
self.unsub_tracker()

View File

@ -76,7 +76,7 @@ class Folder(SensorEntity):
self._unit_of_measurement = DATA_MEGABYTES
self._file_list = None
def update(self):
def update(self) -> None:
"""Update the sensor."""
files_list = get_files_list(self._folder_path, self._filter_term)
self._file_list = files_list

View File

@ -158,7 +158,7 @@ class FoobotSensor(SensorEntity):
data = None
return data
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest data."""
await self.foobot_data.async_update()

View File

@ -2,6 +2,7 @@
import asyncio
from collections import defaultdict
import logging
from typing import Any
from pyforked_daapd import ForkedDaapdAPI
from pylibrespot_java import LibrespotJavaAPI
@ -138,7 +139,7 @@ class ForkedDaapdZone(MediaPlayerEntity):
self._available = True
self._entry_id = entry_id
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Use lifecycle hooks."""
self.async_on_remove(
async_dispatcher_connect(
@ -168,7 +169,7 @@ class ForkedDaapdZone(MediaPlayerEntity):
"""Entity pushes its state to HA."""
return False
async def async_toggle(self):
async def async_toggle(self) -> None:
"""Toggle the power on the zone."""
if self.state == STATE_OFF:
await self.async_turn_on()
@ -180,21 +181,21 @@ class ForkedDaapdZone(MediaPlayerEntity):
"""Return whether the zone is available."""
return self._available
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Enable the output."""
await self._api.change_output(self._output_id, selected=True)
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Disable the output."""
await self._api.change_output(self._output_id, selected=False)
@property
def name(self):
def name(self) -> str:
"""Return the name of the zone."""
return f"{FD_NAME} output ({self._output['name']})"
@property
def state(self):
def state(self) -> str:
"""State of the zone."""
if self._output["selected"]:
return STATE_ON
@ -206,11 +207,11 @@ class ForkedDaapdZone(MediaPlayerEntity):
return self._output["volume"] / 100
@property
def is_volume_muted(self):
def is_volume_muted(self) -> bool:
"""Boolean if volume is currently muted."""
return self._output["volume"] == 0
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
if mute:
if self.volume_level == 0:
@ -221,7 +222,7 @@ class ForkedDaapdZone(MediaPlayerEntity):
target_volume = self._last_volume # restore volume level
await self.async_set_volume_level(volume=target_volume)
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume - input range [0,1]."""
await self._api.set_volume(volume=volume * 100, output_id=self._output_id)
@ -270,7 +271,7 @@ class ForkedDaapdMaster(MediaPlayerEntity):
self._source = SOURCE_NAME_DEFAULT
self._max_playlists = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Use lifecycle hooks."""
self.async_on_remove(
async_dispatcher_connect(
@ -421,7 +422,7 @@ class ForkedDaapdMaster(MediaPlayerEntity):
"""Return whether the master is available."""
return self._available
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Restore the last on outputs state."""
# restore state
await self._api.set_volume(volume=self._last_volume * 100)
@ -441,14 +442,14 @@ class ForkedDaapdMaster(MediaPlayerEntity):
[output["id"] for output in self._outputs]
)
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Pause player and store outputs state."""
await self.async_media_pause()
self._last_outputs = self._outputs
if any(output["selected"] for output in self._outputs):
await self._api.set_enabled_outputs([])
async def async_toggle(self):
async def async_toggle(self) -> None:
"""Toggle the power on the device.
Default media player component method counts idle as off.
@ -460,7 +461,7 @@ class ForkedDaapdMaster(MediaPlayerEntity):
await self.async_turn_off()
@property
def name(self):
def name(self) -> str:
"""Return the name of the device."""
return f"{FD_NAME} server"
@ -564,7 +565,7 @@ class ForkedDaapdMaster(MediaPlayerEntity):
"""List of available input sources."""
return [*self._sources_uris]
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
if mute:
if self.volume_level == 0:
@ -575,32 +576,32 @@ class ForkedDaapdMaster(MediaPlayerEntity):
target_volume = self._last_volume # restore volume level
await self._api.set_volume(volume=target_volume * 100)
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume - input range [0,1]."""
await self._api.set_volume(volume=volume * 100)
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Start playback."""
if self._use_pipe_control():
await self._pipe_call(self._use_pipe_control(), "async_media_play")
else:
await self._api.start_playback()
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Pause playback."""
if self._use_pipe_control():
await self._pipe_call(self._use_pipe_control(), "async_media_pause")
else:
await self._api.pause_playback()
async def async_media_stop(self):
async def async_media_stop(self) -> None:
"""Stop playback."""
if self._use_pipe_control():
await self._pipe_call(self._use_pipe_control(), "async_media_stop")
else:
await self._api.stop_playback()
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Skip to previous track."""
if self._use_pipe_control():
await self._pipe_call(
@ -609,22 +610,22 @@ class ForkedDaapdMaster(MediaPlayerEntity):
else:
await self._api.previous_track()
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Skip to next track."""
if self._use_pipe_control():
await self._pipe_call(self._use_pipe_control(), "async_media_next_track")
else:
await self._api.next_track()
async def async_media_seek(self, position):
async def async_media_seek(self, position: float) -> None:
"""Seek to position."""
await self._api.seek(position_ms=position * 1000)
async def async_clear_playlist(self):
async def async_clear_playlist(self) -> None:
"""Clear playlist."""
await self._api.clear_queue()
async def async_set_shuffle(self, shuffle):
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode."""
await self._api.shuffle(shuffle)
@ -662,7 +663,9 @@ class ForkedDaapdMaster(MediaPlayerEntity):
self._pause_requested = False
self._paused_event.clear()
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:
"""Play a URI."""
if media_source.is_media_source_id(media_id):
media_type = MEDIA_TYPE_MUSIC
@ -738,7 +741,7 @@ class ForkedDaapdMaster(MediaPlayerEntity):
else:
_LOGGER.debug("Media type '%s' not supported", media_type)
async def async_select_source(self, source):
async def async_select_source(self, source: str) -> None:
"""Change source.
Source name reflects whether in default mode or pipe mode.

View File

@ -109,7 +109,7 @@ class HassFoscamCamera(Camera):
if self._rtsp_port:
self._attr_supported_features = CameraEntityFeature.STREAM
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Handle entity addition to hass."""
# Get motion detection status
ret, response = await self.hass.async_add_executor_job(
@ -159,7 +159,7 @@ class HassFoscamCamera(Camera):
"""Camera Motion Detection Status."""
return self._motion_status
def enable_motion_detection(self):
def enable_motion_detection(self) -> None:
"""Enable motion detection in camera."""
try:
ret = self._foscam_session.enable_motion_detection()
@ -179,7 +179,7 @@ class HassFoscamCamera(Camera):
self._name,
)
def disable_motion_detection(self):
def disable_motion_detection(self) -> None:
"""Disable motion detection."""
try:
ret = self._foscam_session.disable_motion_detection()

View File

@ -93,7 +93,7 @@ class FreeboxDevice(ScannerEntity):
return self._name
@property
def is_connected(self):
def is_connected(self) -> bool:
"""Return true if the device is connected to the network."""
return self._active
@ -123,7 +123,7 @@ class FreeboxDevice(ScannerEntity):
self.async_update_state()
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register state update callback."""
self.async_update_state()
self.async_on_remove(

View File

@ -100,7 +100,7 @@ class FreeboxSensor(SensorEntity):
self.async_update_state()
self.async_write_ha_state()
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register state update callback."""
self.async_update_state()
self.async_on_remove(

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from freebox_api.exceptions import InsufficientPermissionsError
@ -65,15 +66,15 @@ class FreeboxWifiSwitch(SwitchEntity):
"Home Assistant does not have permissions to modify the Freebox settings. Please refer to documentation"
)
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._async_set_state(True)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self._async_set_state(False)
async def async_update(self):
async def async_update(self) -> None:
"""Get the state and update it."""
datas = await self._router.wifi.get_global_config()
active = datas["enabled"]

View File

@ -122,7 +122,7 @@ class FritzboxThermostat(FritzBoxEntity, ClimateEntity):
"""Return the list of available operation modes."""
return OPERATION_LIST
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new operation mode."""
if hvac_mode == HVACMode.OFF:
await self.async_set_temperature(temperature=OFF_REPORT_SET_TEMPERATURE)

View File

@ -207,59 +207,59 @@ class AFSAPIDevice(MediaPlayerEntity):
# Management actions
# power control
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn on the device."""
await self.fs_device.set_power(True)
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn off the device."""
await self.fs_device.set_power(False)
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Send play command."""
await self.fs_device.play()
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Send pause command."""
await self.fs_device.pause()
async def async_media_play_pause(self):
async def async_media_play_pause(self) -> None:
"""Send play/pause command."""
if self._attr_state == STATE_PLAYING:
await self.fs_device.pause()
else:
await self.fs_device.play()
async def async_media_stop(self):
async def async_media_stop(self) -> None:
"""Send play/pause command."""
await self.fs_device.pause()
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Send previous track command (results in rewind)."""
await self.fs_device.rewind()
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Send next track command (results in fast-forward)."""
await self.fs_device.forward()
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command."""
await self.fs_device.set_mute(mute)
# volume
async def async_volume_up(self):
async def async_volume_up(self) -> None:
"""Send volume up command."""
volume = await self.fs_device.get_volume()
volume = int(volume or 0) + 1
await self.fs_device.set_volume(min(volume, self._max_volume))
async def async_volume_down(self):
async def async_volume_down(self) -> None:
"""Send volume down command."""
volume = await self.fs_device.get_volume()
volume = int(volume or 0) - 1
await self.fs_device.set_volume(max(volume, 0))
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume command."""
if self._max_volume: # Can't do anything sensible if not set
volume = int(volume * self._max_volume)