Improve entity type hints [v] (#77885)

This commit is contained in:
epenet 2022-09-06 13:59:05 +02:00 committed by GitHub
parent 3798d28bec
commit 050cb275ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 86 additions and 62 deletions

View File

@ -125,7 +125,7 @@ class VasttrafikDepartureSensor(SensorEntity):
return self._state
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
def update(self) -> None:
"""Get the departure board."""
try:
self._departureboard = self._planner.departureboard(

View File

@ -124,7 +124,7 @@ class VenstarThermostat(VenstarEntity, ClimateEntity):
self._attr_name = self._client.name
@property
def supported_features(self):
def supported_features(self) -> int:
"""Return the list of supported features."""
features = (
ClimateEntityFeature.TARGET_TEMPERATURE
@ -141,7 +141,7 @@ class VenstarThermostat(VenstarEntity, ClimateEntity):
return features
@property
def temperature_unit(self):
def temperature_unit(self) -> str:
"""Return the unit of measurement, as defined by the API."""
if self._client.tempunits == self._client.TEMPUNITS_F:
return TEMP_FAHRENHEIT
@ -300,7 +300,7 @@ class VenstarThermostat(VenstarEntity, ClimateEntity):
_LOGGER.error("Failed to change the temperature")
self.schedule_update_ha_state()
def set_fan_mode(self, fan_mode):
def set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
if fan_mode == STATE_ON:
success = self._client.set_fan(self._client.FAN_ON)
@ -316,7 +316,7 @@ class VenstarThermostat(VenstarEntity, ClimateEntity):
self._set_operation_mode(hvac_mode)
self.schedule_update_ha_state()
def set_humidity(self, humidity):
def set_humidity(self, humidity: int) -> None:
"""Set new target humidity."""
success = self._client.set_hum_setpoint(humidity)
@ -324,7 +324,7 @@ class VenstarThermostat(VenstarEntity, ClimateEntity):
_LOGGER.error("Failed to change the target humidity level")
self.schedule_update_ha_state()
def set_preset_mode(self, preset_mode):
def set_preset_mode(self, preset_mode: str) -> None:
"""Set the hold mode."""
if preset_mode == PRESET_AWAY:
success = self._client.set_away(self._client.AWAY_AWAY)

View File

@ -88,7 +88,7 @@ class VeraThermostat(VeraDevice[veraApi.VeraThermostat], ClimateEntity):
"""Return a list of available fan modes."""
return FAN_OPERATION_LIST
def set_fan_mode(self, fan_mode) -> None:
def set_fan_mode(self, fan_mode: str) -> None:
"""Set new target temperature."""
if fan_mode == FAN_ON:
self.vera_device.fan_on()

View File

@ -2,6 +2,7 @@
from __future__ import annotations
from time import monotonic
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -76,14 +77,14 @@ class VerisureSmartplug(CoordinatorEntity[VerisureDataUpdateCoordinator], Switch
and self.serial_number in self.coordinator.data["smart_plugs"]
)
def turn_on(self, **kwargs) -> None:
def turn_on(self, **kwargs: Any) -> None:
"""Set smartplug status on."""
self.coordinator.verisure.set_smartplug_state(self.serial_number, True)
self._state = True
self._change_timestamp = monotonic()
self.schedule_update_ha_state()
def turn_off(self, **kwargs) -> None:
def turn_off(self, **kwargs: Any) -> None:
"""Set smartplug status off."""
self.coordinator.verisure.set_smartplug_state(self.serial_number, False)
self._state = False

View File

@ -89,7 +89,7 @@ class VSensor(SensorEntity):
"""Return if the sensor is available."""
return self._available
async def async_update(self):
async def async_update(self) -> None:
"""Fetch new state data for the sensor."""
samples = await self.consumer.fetchPeripheralSample(
None, self._identifier, self._parent_mac

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant
@ -86,11 +87,11 @@ class VActuator(SwitchEntity):
"""Return if the actuator is available."""
return self._available
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the actuator."""
await self.update_state(0)
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on the actuator."""
await self.update_state(1)
@ -102,7 +103,7 @@ class VActuator(SwitchEntity):
None, self._identifier, self._parent_mac, payload
)
async def async_update(self):
async def async_update(self) -> None:
"""Fetch state data from the actuator."""
samples = await self.consumer.fetchPeripheralSample(
None, self._identifier, self._parent_mac

View File

@ -1,5 +1,6 @@
"""Support for VeSync switches."""
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -53,7 +54,7 @@ def _setup_entities(devices, async_add_entities):
class VeSyncBaseSwitch(VeSyncDevice, SwitchEntity):
"""Base class for VeSync switch Device Representations."""
def turn_on(self, **kwargs):
def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
self.device.turn_on()
@ -66,7 +67,7 @@ class VeSyncSwitchHA(VeSyncBaseSwitch, SwitchEntity):
super().__init__(plug)
self.smartplug = plug
def update(self):
def update(self) -> None:
"""Update outlet details and energy usage."""
self.smartplug.update()
self.smartplug.update_energy()

View File

@ -161,7 +161,7 @@ class ViaggiaTrenoSensor(SensorEntity):
return True
return False
async def async_update(self):
async def async_update(self) -> None:
"""Update state."""
uri = self.uri
res = await async_http_request(self.hass, uri)

View File

@ -3,6 +3,7 @@ from __future__ import annotations
from contextlib import suppress
import logging
from typing import Any
from PyViCare.PyViCareUtils import (
PyViCareInvalidDataError,
@ -179,7 +180,7 @@ class ViCareClimate(ClimateEntity):
configuration_url="https://developer.viessmann.com/",
)
def update(self):
def update(self) -> None:
"""Let HA know there has been an update from the ViCare API."""
try:
_room_temperature = None
@ -326,7 +327,7 @@ class ViCareClimate(ClimateEntity):
"""Set target temperature step to wholes."""
return PRECISION_WHOLE
def set_temperature(self, **kwargs):
def set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures."""
if (temp := kwargs.get(ATTR_TEMPERATURE)) is not None:
self._circuit.setProgramTemperature(self._current_program, temp)
@ -342,7 +343,7 @@ class ViCareClimate(ClimateEntity):
"""Return the available preset mode."""
return list(HA_TO_VICARE_PRESET_HEATING)
def set_preset_mode(self, preset_mode):
def set_preset_mode(self, preset_mode: str) -> None:
"""Set new preset mode and deactivate any existing programs."""
vicare_program = HA_TO_VICARE_PRESET_HEATING.get(preset_mode)
if vicare_program is None:

View File

@ -1,6 +1,7 @@
"""Viessmann ViCare water_heater device."""
from contextlib import suppress
import logging
from typing import Any
from PyViCare.PyViCareUtils import (
PyViCareInvalidDataError,
@ -116,7 +117,7 @@ class ViCareWater(WaterHeaterEntity):
self._current_mode = None
self._heating_type = heating_type
def update(self):
def update(self) -> None:
"""Let HA know there has been an update from the ViCare API."""
try:
with suppress(PyViCareNotSupportedFeatureError):
@ -177,7 +178,7 @@ class ViCareWater(WaterHeaterEntity):
"""Return the temperature we try to reach."""
return self._target_temperature
def set_temperature(self, **kwargs):
def set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures."""
if (temp := kwargs.get(ATTR_TEMPERATURE)) is not None:
self._api.setDomesticHotWaterTemperature(temp)

View File

@ -46,7 +46,7 @@ class VilfoRouterSensor(SensorEntity):
self._attr_unique_id = f"{api.unique_id}_{description.key}"
@property
def available(self):
def available(self) -> bool:
"""Return whether the sensor is available or not."""
return self.api.available
@ -61,7 +61,7 @@ class VilfoRouterSensor(SensorEntity):
parent_device_name = self._device_info["name"]
return f"{parent_device_name} {self.entity_description.name}"
async def async_update(self):
async def async_update(self) -> None:
"""Update the router data."""
await self.api.async_update()
self._attr_native_value = self.api.data.get(self.entity_description.api_key)

View File

@ -114,12 +114,12 @@ class VivotekCam(Camera):
"""Return the camera motion detection status."""
return self._motion_detection_enabled
def disable_motion_detection(self):
def disable_motion_detection(self) -> None:
"""Disable motion detection in camera."""
response = self._cam.set_param(DEFAULT_EVENT_0_KEY, 0)
self._motion_detection_enabled = int(response) == 1
def enable_motion_detection(self):
def enable_motion_detection(self) -> None:
"""Enable motion detection in camera."""
response = self._cam.set_param(DEFAULT_EVENT_0_KEY, 1)
self._motion_detection_enabled = int(response) == 1
@ -134,6 +134,6 @@ class VivotekCam(Camera):
"""Return the camera model."""
return self._model_name
def update(self):
def update(self) -> None:
"""Update entity status."""
self._model_name = self._cam.model_name

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
import vlc
import voluptuous as vol
@ -134,37 +135,39 @@ class VlcDevice(MediaPlayerEntity):
"""When was the position of the current playing media valid."""
return self._media_position_updated_at
def media_seek(self, position):
def media_seek(self, position: float) -> None:
"""Seek the media to a specific location."""
track_length = self._vlc.get_length() / 1000
self._vlc.set_position(position / track_length)
def mute_volume(self, mute):
def mute_volume(self, mute: bool) -> None:
"""Mute the volume."""
self._vlc.audio_set_mute(mute)
self._muted = mute
def set_volume_level(self, volume):
def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
self._vlc.audio_set_volume(int(volume * 100))
self._volume = volume
def media_play(self):
def media_play(self) -> None:
"""Send play command."""
self._vlc.play()
self._state = STATE_PLAYING
def media_pause(self):
def media_pause(self) -> None:
"""Send pause command."""
self._vlc.pause()
self._state = STATE_PAUSED
def media_stop(self):
def media_stop(self) -> None:
"""Send stop command."""
self._vlc.stop()
self._state = STATE_IDLE
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 media from a URL or file."""
# Handle media_source
if media_source.is_media_source_id(media_id):
@ -191,7 +194,7 @@ class VlcDevice(MediaPlayerEntity):
self._state = STATE_PLAYING
async def async_browse_media(
self, media_content_type=None, media_content_id=None
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(

View File

@ -123,11 +123,11 @@ class VolkszaehlerSensor(SensorEntity):
self._attr_name = f"{name} {description.name}"
@property
def available(self):
def available(self) -> bool:
"""Could the device be accessed during the last update call."""
return self.vz_api.available
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest data from REST API."""
await self.vz_api.async_update()

View File

@ -3,13 +3,17 @@ Volumio Platform.
Volumio rest API: https://volumio.github.io/docs/API/REST_API.html
"""
from __future__ import annotations
from datetime import timedelta
import json
from typing import Any
from homeassistant.components.media_player import (
MediaPlayerEntity,
MediaPlayerEntityFeature,
)
from homeassistant.components.media_player.browse_media import BrowseMedia
from homeassistant.components.media_player.const import (
MEDIA_TYPE_MUSIC,
REPEAT_MODE_ALL,
@ -83,7 +87,7 @@ class Volumio(MediaPlayerEntity):
self._currentplaylist = None
self.thumbnail_cache = {}
async def async_update(self):
async def async_update(self) -> None:
"""Update state."""
self._state = await self._volumio.get_state()
await self._async_update_playlists()
@ -191,65 +195,65 @@ class Volumio(MediaPlayerEntity):
"""Name of the current input source."""
return self._currentplaylist
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Send media_next command to media player."""
await self._volumio.next()
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Send media_previous command to media player."""
await self._volumio.previous()
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Send media_play command to media player."""
await self._volumio.play()
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Send media_pause command to media player."""
if self._state.get("trackType") == "webradio":
await self._volumio.stop()
else:
await self._volumio.pause()
async def async_media_stop(self):
async def async_media_stop(self) -> None:
"""Send media_stop command to media player."""
await self._volumio.stop()
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."""
await self._volumio.set_volume_level(int(volume * 100))
async def async_volume_up(self):
async def async_volume_up(self) -> None:
"""Service to send the Volumio the command for volume up."""
await self._volumio.volume_up()
async def async_volume_down(self):
async def async_volume_down(self) -> None:
"""Service to send the Volumio the command for volume down."""
await self._volumio.volume_down()
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command to media player."""
if mute:
await self._volumio.mute()
else:
await self._volumio.unmute()
async def async_set_shuffle(self, shuffle):
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode."""
await self._volumio.set_shuffle(shuffle)
async def async_set_repeat(self, repeat):
async def async_set_repeat(self, repeat: str) -> None:
"""Set repeat mode."""
if repeat == REPEAT_MODE_OFF:
await self._volumio.repeatAll("false")
else:
await self._volumio.repeatAll("true")
async def async_select_source(self, source):
async def async_select_source(self, source: str) -> None:
"""Choose an available playlist and play it."""
await self._volumio.play_playlist(source)
self._currentplaylist = source
async def async_clear_playlist(self):
async def async_clear_playlist(self) -> None:
"""Clear players playlist."""
await self._volumio.clear_playlist()
self._currentplaylist = None
@ -259,11 +263,15 @@ class Volumio(MediaPlayerEntity):
"""Update available Volumio playlists."""
self._playlists = await self._volumio.get_playlists()
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."""
await self._volumio.replace_and_play(json.loads(media_id))
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."""
self.thumbnail_cache = {}
if media_content_type in (None, "library"):
@ -274,8 +282,11 @@ class Volumio(MediaPlayerEntity):
)
async def async_get_browse_image(
self, media_content_type, media_content_id, media_image_id=None
):
self,
media_content_type: str,
media_content_id: str,
media_image_id: str | None = None,
) -> tuple[bytes | None, str | None]:
"""Get album art from Volumio."""
cached_url = self.thumbnail_cache.get(media_content_id)
image_url = self._volumio.canonic_url(cached_url)

View File

@ -1,6 +1,8 @@
"""Support for Volvo heater."""
from __future__ import annotations
from typing import Any
from volvooncall.dashboard import Instrument
from homeassistant.components.switch import SwitchEntity
@ -67,12 +69,12 @@ class VolvoSwitch(VolvoEntity, SwitchEntity):
"""Determine if switch is on."""
return self.instrument.state
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self.instrument.turn_on()
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self.instrument.turn_off()
await self.coordinator.async_request_refresh()

View File

@ -78,7 +78,9 @@ class VulcanCalendarEntity(CalendarEntity):
"""Return the next upcoming event."""
return self._event
async def async_get_events(self, hass, start_date, end_date) -> list[CalendarEvent]:
async def async_get_events(
self, hass: HomeAssistant, start_date: datetime, end_date: datetime
) -> list[CalendarEvent]:
"""Get all events in a specific time frame."""
try:
events = await get_lessons(

View File

@ -111,7 +111,7 @@ class VultrBinarySensor(BinarySensorEntity):
ATTR_VCPUS: self.data.get("vcpu_count"),
}
def update(self):
def update(self) -> None:
"""Update state of sensor."""
self._vultr.update()
self.data = self._vultr.data[self.subscription]

View File

@ -112,7 +112,7 @@ class VultrSensor(SensorEntity):
except (TypeError, ValueError):
return self.data.get(self.entity_description.key)
def update(self):
def update(self) -> None:
"""Update state of sensor."""
self._vultr.update()
self.data = self._vultr.data[self.subscription]

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
@ -108,17 +109,17 @@ class VultrSwitch(SwitchEntity):
ATTR_VCPUS: self.data.get("vcpu_count"),
}
def turn_on(self, **kwargs):
def turn_on(self, **kwargs: Any) -> None:
"""Boot-up the subscription."""
if self.data["power_status"] != "running":
self._vultr.start(self.subscription)
def turn_off(self, **kwargs):
def turn_off(self, **kwargs: Any) -> None:
"""Halt the subscription."""
if self.data["power_status"] == "running":
self._vultr.halt(self.subscription)
def update(self):
def update(self) -> None:
"""Get the latest data from the device and update the data."""
self._vultr.update()
self.data = self._vultr.data[self.subscription]