Improve entity type hints [b] (#77012)

This commit is contained in:
epenet 2022-08-19 09:54:13 +02:00 committed by GitHub
parent dd109839b9
commit dedf063e43
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 73 additions and 59 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
from typing import Any
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
@ -121,7 +122,7 @@ class BalboaSpaClimate(BalboaEntity, ClimateEntity):
"""Return current preset mode."""
return self._client.get_heatmode(True)
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set a new target temperature."""
scale = self._client.get_tempscale()
newtemp = kwargs[ATTR_TEMPERATURE]
@ -133,7 +134,7 @@ class BalboaSpaClimate(BalboaEntity, ClimateEntity):
await asyncio.sleep(SET_TEMPERATURE_WAIT)
await self._client.send_temp_change(newtemp)
async def async_set_preset_mode(self, preset_mode) -> None:
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set new preset mode."""
modelist = self._client.get_heatmode_stringlist()
self._async_validate_mode_or_raise(preset_mode)
@ -141,7 +142,7 @@ class BalboaSpaClimate(BalboaEntity, ClimateEntity):
raise ValueError(f"{preset_mode} is not a valid preset mode")
await self._client.change_heatmode(modelist.index(preset_mode))
async def async_set_fan_mode(self, fan_mode):
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new fan mode."""
await self._client.change_blower(self._ha_to_balboa_blower_map[fan_mode])
@ -150,7 +151,7 @@ class BalboaSpaClimate(BalboaEntity, ClimateEntity):
if mode == self._client.HEATMODE_RNR:
raise ValueError(f"{mode} can only be reported but not set")
async def async_set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode.
OFF = Rest

View file

@ -162,7 +162,7 @@ class BayesianBinarySensor(BinarySensorEntity):
"state": self._process_state,
}
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""
Call when entity about to be added.
@ -397,7 +397,7 @@ class BayesianBinarySensor(BinarySensorEntity):
ATTR_PROBABILITY_THRESHOLD: self._probability_threshold,
}
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest data and update the states."""
if not self._callbacks:
self._recalculate_and_write_state()

View file

@ -136,7 +136,7 @@ class BboxUptimeSensor(SensorEntity):
self._attr_name = f"{name} {description.name}"
self.bbox_data = bbox_data
def update(self):
def update(self) -> None:
"""Get the latest data from Bbox and update the state."""
self.bbox_data.update()
self._attr_native_value = utcnow() - timedelta(
@ -155,7 +155,7 @@ class BboxSensor(SensorEntity):
self._attr_name = f"{name} {description.name}"
self.bbox_data = bbox_data
def update(self):
def update(self) -> None:
"""Get the latest data from Bbox and update the state."""
self.bbox_data.update()
sensor_type = self.entity_description.key

View file

@ -73,7 +73,7 @@ class BeewiSmartclimSensor(SensorEntity):
self._attr_device_class = self._device
self._attr_unique_id = f"{mac}_{device}"
def update(self):
def update(self) -> None:
"""Fetch new state data from the poller."""
self._poller.update_sensor()
self._attr_native_value = None

View file

@ -181,7 +181,7 @@ class BitcoinSensor(SensorEntity):
self.data = data
self._currency = currency
def update(self):
def update(self) -> None:
"""Get the latest data and updates the states."""
self.data.update()
stats = self.data.stats

View file

@ -54,7 +54,7 @@ class BizkaibusSensor(SensorEntity):
self.data = data
self._attr_name = name
def update(self):
def update(self) -> None:
"""Get the latest data from the webservice."""
self.data.update()
with suppress(TypeError):

View file

@ -158,7 +158,7 @@ class BlackbirdZone(MediaPlayerEntity):
self._zone_id = zone_id
self._attr_name = zone_name
def update(self):
def update(self) -> None:
"""Retrieve latest state."""
state = self._blackbird.zone_status(self._zone_id)
if not state:
@ -183,7 +183,7 @@ class BlackbirdZone(MediaPlayerEntity):
_LOGGER.debug("Setting all zones source to %s", idx)
self._blackbird.set_all_zone_source(idx)
def select_source(self, source):
def select_source(self, source: str) -> None:
"""Set input source."""
if source not in self._source_name_id:
return
@ -191,12 +191,12 @@ class BlackbirdZone(MediaPlayerEntity):
_LOGGER.debug("Setting zone %d source to %s", self._zone_id, idx)
self._blackbird.set_zone_source(self._zone_id, idx)
def turn_on(self):
def turn_on(self) -> None:
"""Turn the media player on."""
_LOGGER.debug("Turning zone %d on", self._zone_id)
self._blackbird.set_zone_power(self._zone_id, True)
def turn_off(self):
def turn_off(self) -> None:
"""Turn the media player off."""
_LOGGER.debug("Turning zone %d off", self._zone_id)
self._blackbird.set_zone_power(self._zone_id, False)

View file

@ -1,5 +1,6 @@
"""BleBox climate entity."""
from datetime import timedelta
from typing import Any
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
@ -73,7 +74,7 @@ class BleBoxClimateEntity(BleBoxEntity, ClimateEntity):
"""Return the desired thermostat temperature."""
return self._feature.desired
async def async_set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the climate entity mode."""
if hvac_mode == HVACMode.HEAT:
await self._feature.async_on()
@ -81,7 +82,7 @@ class BleBoxClimateEntity(BleBoxEntity, ClimateEntity):
await self._feature.async_off()
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set the thermostat temperature."""
value = kwargs[ATTR_TEMPERATURE]
await self._feature.async_set_temperature(value)

View file

@ -3,6 +3,7 @@ from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
import blebox_uniapi.light
from blebox_uniapi.light import BleboxColorMode
@ -175,6 +176,6 @@ class BleBoxLightEntity(BleBoxEntity, LightEntity):
f"Turning on with effect '{self.name}' failed: {effect} not in effect list."
) from exc
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the light off."""
await self._feature.async_off()

View file

@ -1,5 +1,6 @@
"""BleBox switch implementation."""
from datetime import timedelta
from typing import Any
from homeassistant.components.switch import SwitchDeviceClass, SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -35,10 +36,10 @@ class BleBoxSwitchEntity(BleBoxEntity, SwitchEntity):
"""Return whether switch is on."""
return self._feature.is_on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
await self._feature.async_turn_on()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
await self._feature.async_turn_off()

View file

@ -69,7 +69,7 @@ class BlinkBinarySensor(BinarySensorEntity):
model=self._camera.camera_type,
)
def update(self):
def update(self) -> None:
"""Update sensor state."""
self.data.refresh()
state = self._camera.attributes[self.entity_description.key]

View file

@ -58,18 +58,18 @@ class BlinkCamera(Camera):
"""Return the camera attributes."""
return self._camera.attributes
def enable_motion_detection(self):
def enable_motion_detection(self) -> None:
"""Enable motion detection for the camera."""
self._camera.arm = True
self.data.refresh()
def disable_motion_detection(self):
def disable_motion_detection(self) -> None:
"""Disable motion detection for the camera."""
self._camera.arm = False
self.data.refresh()
@property
def motion_detection_enabled(self):
def motion_detection_enabled(self) -> bool:
"""Return the state of the camera."""
return self._camera.arm

View file

@ -72,7 +72,7 @@ class BlinkSensor(SensorEntity):
model=self._camera.camera_type,
)
def update(self):
def update(self) -> None:
"""Retrieve sensor data from the camera."""
self.data.refresh()
try:

View file

@ -65,6 +65,6 @@ class BlockchainSensor(SensorEntity):
self._attr_name = name
self.addresses = addresses
def update(self):
def update(self) -> None:
"""Get the latest state of the sensor."""
self._attr_native_value = get_balance(self.addresses)

View file

@ -58,7 +58,7 @@ class BloomSkySensor(BinarySensorEntity):
self._attr_unique_id = f"{self._device_id}-{sensor_name}"
self._attr_device_class = SENSOR_TYPES.get(sensor_name)
def update(self):
def update(self) -> None:
"""Request an update from the BloomSky API."""
self._bloomsky.refresh_devices()

View file

@ -112,7 +112,7 @@ class BloomSkySensor(SensorEntity):
"""Return the class of this device, from component DEVICE_CLASSES."""
return SENSOR_DEVICE_CLASS.get(self._sensor_name)
def update(self):
def update(self) -> None:
"""Request an update from the BloomSky API."""
self._bloomsky.refresh_devices()
state = self._bloomsky.devices[self._device_id]["Data"][self._sensor_name]

View file

@ -6,6 +6,7 @@ from asyncio import CancelledError
from datetime import timedelta
from http import HTTPStatus
import logging
from typing import Any
from urllib import parse
import aiohttp
@ -22,6 +23,7 @@ from homeassistant.components.media_player import (
MediaPlayerEntityFeature,
)
from homeassistant.components.media_player.browse_media import (
BrowseMedia,
async_process_play_media_url,
)
from homeassistant.components.media_player.const import MEDIA_TYPE_MUSIC
@ -333,7 +335,7 @@ class BluesoundPlayer(MediaPlayerEntity):
)
raise
async def async_update(self):
async def async_update(self) -> None:
"""Update internal status of the entity."""
if not self._is_online:
return
@ -930,12 +932,12 @@ class BluesoundPlayer(MediaPlayerEntity):
while sleep > 0:
sleep = await self.async_increase_timer()
async def async_set_shuffle(self, shuffle):
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable or disable shuffle mode."""
value = "1" if shuffle else "0"
return await self.send_bluesound_command(f"/Shuffle?state={value}")
async def async_select_source(self, source):
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if self.is_grouped and not self.is_master:
return
@ -958,14 +960,14 @@ class BluesoundPlayer(MediaPlayerEntity):
return await self.send_bluesound_command(url)
async def async_clear_playlist(self):
async def async_clear_playlist(self) -> None:
"""Clear players playlist."""
if self.is_grouped and not self.is_master:
return
return await self.send_bluesound_command("Clear")
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Send media_next command to media player."""
if self.is_grouped and not self.is_master:
return
@ -978,7 +980,7 @@ class BluesoundPlayer(MediaPlayerEntity):
return await self.send_bluesound_command(cmd)
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Send media_previous command to media player."""
if self.is_grouped and not self.is_master:
return
@ -991,35 +993,37 @@ class BluesoundPlayer(MediaPlayerEntity):
return await self.send_bluesound_command(cmd)
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Send media_play command to media player."""
if self.is_grouped and not self.is_master:
return
return await self.send_bluesound_command("Play")
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Send media_pause command to media player."""
if self.is_grouped and not self.is_master:
return
return await self.send_bluesound_command("Pause")
async def async_media_stop(self):
async def async_media_stop(self) -> None:
"""Send stop command."""
if self.is_grouped and not self.is_master:
return
return await self.send_bluesound_command("Pause")
async def async_media_seek(self, position):
async def async_media_seek(self, position: float) -> None:
"""Send media_seek command to media player."""
if self.is_grouped and not self.is_master:
return
return await self.send_bluesound_command(f"Play?seek={float(position)}")
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:
"""Send the play_media command to the media player."""
if self.is_grouped and not self.is_master:
return
@ -1036,21 +1040,21 @@ class BluesoundPlayer(MediaPlayerEntity):
return await self.send_bluesound_command(url)
async def async_volume_up(self):
async def async_volume_up(self) -> None:
"""Volume up the media player."""
current_vol = self.volume_level
if not current_vol or current_vol >= 1:
return
return await self.async_set_volume_level(current_vol + 0.01)
async def async_volume_down(self):
async def async_volume_down(self) -> None:
"""Volume down the media player."""
current_vol = self.volume_level
if not current_vol or current_vol <= 0:
return
return await self.async_set_volume_level(current_vol - 0.01)
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: float) -> None:
"""Send volume_up command to media player."""
if volume < 0:
volume = 0
@ -1058,13 +1062,15 @@ class BluesoundPlayer(MediaPlayerEntity):
volume = 1
return await self.send_bluesound_command(f"Volume?level={float(volume) * 100}")
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command to media player."""
if mute:
return await self.send_bluesound_command("Volume?mute=1")
return await self.send_bluesound_command("Volume?mute=0")
async def async_browse_media(self, media_content_type=None, media_content_id=None):
async def async_browse_media(
self, media_content_type: str | None = None, media_content_id: str | None = None
) -> BrowseMedia:
"""Implement the websocket media browsing helper."""
return await media_source.async_browse_media(
self.hass,

View file

@ -2,6 +2,7 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from boschshcpy import (
SHCCamera360,
@ -183,11 +184,11 @@ class SHCSwitch(SHCEntity, SwitchEntity):
== self.entity_description.on_value
)
def turn_on(self, **kwargs) -> None:
def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
setattr(self._device, self.entity_description.on_key, True)
def turn_off(self, **kwargs) -> None:
def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
setattr(self._device, self.entity_description.on_key, False)
@ -218,10 +219,10 @@ class SHCRoutingSwitch(SHCEntity, SwitchEntity):
"""Return the state of the switch."""
return self._device.routing.name == "ENABLED"
def turn_on(self, **kwargs) -> None:
def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
self._device.routing = True
def turn_off(self, **kwargs) -> None:
def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
self._device.routing = False

View file

@ -2,9 +2,11 @@
import asyncio
from base64 import b64encode
from collections import defaultdict
from collections.abc import Iterable
from datetime import timedelta
from itertools import product
import logging
from typing import Any
from broadlink.exceptions import (
AuthorizationError,
@ -174,18 +176,18 @@ class BroadlinkRemote(BroadlinkEntity, RemoteEntity, RestoreEntity):
"""
return self._flags
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call when the remote is added to hass."""
state = await self.async_get_last_state()
self._attr_is_on = state is None or state.state != STATE_OFF
await super().async_added_to_hass()
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the remote."""
self._attr_is_on = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the remote."""
self._attr_is_on = False
self.async_write_ha_state()
@ -198,7 +200,7 @@ class BroadlinkRemote(BroadlinkEntity, RemoteEntity, RestoreEntity):
self._flags.update(await self._flag_storage.async_load() or {})
self._storage_loaded = True
async def async_send_command(self, command, **kwargs):
async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send a list of commands to a device."""
kwargs[ATTR_COMMAND] = command
kwargs = SERVICE_SEND_SCHEMA(kwargs)
@ -255,7 +257,7 @@ class BroadlinkRemote(BroadlinkEntity, RemoteEntity, RestoreEntity):
if at_least_one_sent:
self._flag_storage.async_delay_save(self._get_flags, FLAG_SAVE_DELAY)
async def async_learn_command(self, **kwargs):
async def async_learn_command(self, **kwargs: Any) -> None:
"""Learn a list of commands from a remote."""
kwargs = SERVICE_LEARN_SCHEMA(kwargs)
commands = kwargs[ATTR_COMMAND]
@ -419,7 +421,7 @@ class BroadlinkRemote(BroadlinkEntity, RemoteEntity, RestoreEntity):
self.hass, notification_id="learn_command"
)
async def async_delete_command(self, **kwargs):
async def async_delete_command(self, **kwargs: Any) -> None:
"""Delete a list of commands from a remote."""
kwargs = SERVICE_DELETE_SCHEMA(kwargs)
commands = kwargs[ATTR_COMMAND]

View file

@ -3,6 +3,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
import logging
from typing import Any
from broadlink.exceptions import BroadlinkException
import voluptuous as vol
@ -146,19 +147,19 @@ class BroadlinkSwitch(BroadlinkEntity, SwitchEntity, RestoreEntity, ABC):
self._command_off = command_off
self._attr_name = f"{device.name} Switch"
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Call when the switch is added to hass."""
state = await self.async_get_last_state()
self._attr_is_on = state is not None and state.state == STATE_ON
await super().async_added_to_hass()
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the switch."""
if await self._async_send_packet(self._command_on):
self._attr_is_on = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the switch."""
if await self._async_send_packet(self._command_off):
self._attr_is_on = False

View file

@ -106,14 +106,14 @@ class BSBLanClimate(ClimateEntity):
self._store_hvac_mode = self._attr_hvac_mode
await self.async_set_data(preset_mode=preset_mode)
async def async_set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set HVAC mode."""
_LOGGER.debug("Setting HVAC mode to: %s", hvac_mode)
# preset should be none when hvac mode is set
self._attr_preset_mode = PRESET_NONE
await self.async_set_data(hvac_mode=hvac_mode)
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures."""
await self.async_set_data(**kwargs)