Improve entity type hints [n] ()

pull/77841/head
epenet 2022-09-05 14:53:55 +02:00 committed by GitHub
parent 42393db9f3
commit 420733a064
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 69 additions and 59 deletions

View File

@ -133,34 +133,34 @@ class NAD(MediaPlayerEntity):
"""Boolean if volume is currently muted."""
return self._mute
def turn_off(self):
def turn_off(self) -> None:
"""Turn the media player off."""
self._nad_receiver.main_power("=", "Off")
def turn_on(self):
def turn_on(self) -> None:
"""Turn the media player on."""
self._nad_receiver.main_power("=", "On")
def volume_up(self):
def volume_up(self) -> None:
"""Volume up the media player."""
self._nad_receiver.main_volume("+")
def volume_down(self):
def volume_down(self) -> None:
"""Volume down the media player."""
self._nad_receiver.main_volume("-")
def set_volume_level(self, volume):
def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
self._nad_receiver.main_volume("=", self.calc_db(volume))
def mute_volume(self, mute):
def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
if mute:
self._nad_receiver.main_mute("=", "On")
else:
self._nad_receiver.main_mute("=", "Off")
def select_source(self, source):
def select_source(self, source: str) -> None:
"""Select input source."""
self._nad_receiver.main_source("=", self._reverse_mapping.get(source))
@ -175,7 +175,7 @@ class NAD(MediaPlayerEntity):
return sorted(self._reverse_mapping)
@property
def available(self):
def available(self) -> bool:
"""Return if device is available."""
return self._state is not None
@ -257,37 +257,37 @@ class NADtcp(MediaPlayerEntity):
"""Boolean if volume is currently muted."""
return self._mute
def turn_off(self):
def turn_off(self) -> None:
"""Turn the media player off."""
self._nad_receiver.power_off()
def turn_on(self):
def turn_on(self) -> None:
"""Turn the media player on."""
self._nad_receiver.power_on()
def volume_up(self):
def volume_up(self) -> None:
"""Step volume up in the configured increments."""
self._nad_receiver.set_volume(self._nad_volume + 2 * self._volume_step)
def volume_down(self):
def volume_down(self) -> None:
"""Step volume down in the configured increments."""
self._nad_receiver.set_volume(self._nad_volume - 2 * self._volume_step)
def set_volume_level(self, volume):
def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
nad_volume_to_set = int(
round(volume * (self._max_vol - self._min_vol) + self._min_vol)
)
self._nad_receiver.set_volume(nad_volume_to_set)
def mute_volume(self, mute):
def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
if mute:
self._nad_receiver.mute()
else:
self._nad_receiver.unmute()
def select_source(self, source):
def select_source(self, source: str) -> None:
"""Select input source."""
self._nad_receiver.select_source(source)
@ -301,7 +301,7 @@ class NADtcp(MediaPlayerEntity):
"""List of available input sources."""
return self._nad_receiver.available_sources()
def update(self):
def update(self) -> None:
"""Get the latest details from the device."""
try:
nad_status = self._nad_receiver.status()

View File

@ -219,7 +219,7 @@ class NSDepartureSensor(SensorEntity):
return attributes
@Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self):
def update(self) -> None:
"""Get the trip information."""
# If looking for a specific trip time, update around that trip time only.

View File

@ -55,7 +55,7 @@ class NessZoneBinarySensor(BinarySensorEntity):
self._type = zone_type
self._state = 0
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register callbacks."""
self.async_on_remove(
async_dispatcher_connect(

View File

@ -140,11 +140,11 @@ class NetdataSensor(SensorEntity):
return self._state
@property
def available(self):
def available(self) -> bool:
"""Could the resource be accessed during the last update call."""
return self.netdata.available
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest data from Netdata REST API."""
await self.netdata.async_update()
resource_data = self.netdata.api.metrics.get(self._sensor)
@ -186,11 +186,11 @@ class NetdataAlarms(SensorEntity):
return "mdi:crosshairs-question"
@property
def available(self):
def available(self) -> bool:
"""Could the resource be accessed during the last update call."""
return self.netdata.available
async def async_update(self):
async def async_update(self) -> None:
"""Get the latest alarms from Netdata REST API."""
await self.netdata.async_update()
alarms = self.netdata.api.alarms["alarms"]

View File

@ -1,5 +1,6 @@
"""Support for Netgear switches."""
import logging
from typing import Any
from pynetgear import ALLOW, BLOCK
@ -87,12 +88,12 @@ class NetgearAllowBlock(NetgearDeviceEntity, 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."""
await self._router.async_allow_block_device(self._mac, ALLOW)
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._router.async_allow_block_device(self._mac, BLOCK)
await self.coordinator.async_request_refresh()

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections import namedtuple
from datetime import timedelta
import logging
from typing import Any
from pynetio import Netio
import voluptuous as vol
@ -148,15 +149,15 @@ class NetioSwitch(SwitchEntity):
return self._name
@property
def available(self):
def available(self) -> bool:
"""Return true if entity is available."""
return not hasattr(self, "telnet")
def turn_on(self, **kwargs):
def turn_on(self, **kwargs: Any) -> None:
"""Turn switch on."""
self._set(True)
def turn_off(self, **kwargs):
def turn_off(self, **kwargs: Any) -> None:
"""Turn switch off."""
self._set(False)
@ -172,6 +173,6 @@ class NetioSwitch(SwitchEntity):
"""Return the switch's status."""
return self.netio.states[int(self.outlet) - 1]
def update(self):
def update(self) -> None:
"""Update the state."""
self.netio.update()

View File

@ -177,7 +177,7 @@ class NeurioEnergy(SensorEntity):
"""Icon to use in the frontend, if any."""
return ICON
def update(self):
def update(self) -> None:
"""Get the latest data, update state."""
self.update_sensor()

View File

@ -1,6 +1,8 @@
"""Support for Nexia / Trane XL thermostats."""
from __future__ import annotations
from typing import Any
from nexia.const import (
HOLD_PERMANENT,
HOLD_RESUME_SCHEDULE,
@ -195,7 +197,7 @@ class NexiaZone(NexiaThermostatZoneEntity, ClimateEntity):
"""Return the fan setting."""
return self._thermostat.get_fan_mode()
async def async_set_fan_mode(self, fan_mode):
async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new target fan mode."""
await self._thermostat.set_fan_mode(fan_mode)
self._signal_thermostat_update()
@ -216,7 +218,7 @@ class NexiaZone(NexiaThermostatZoneEntity, ClimateEntity):
"""Preset that is active."""
return self._zone.get_preset()
async def async_set_humidity(self, humidity):
async def async_set_humidity(self, humidity: int) -> None:
"""Dehumidify target."""
if self._thermostat.has_dehumidify_support():
await self.async_set_dehumidify_setpoint(humidity)
@ -303,7 +305,7 @@ class NexiaZone(NexiaThermostatZoneEntity, ClimateEntity):
return NEXIA_TO_HA_HVAC_MODE_MAP[mode]
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set target temperature."""
new_heat_temp = kwargs.get(ATTR_TARGET_TEMP_LOW)
new_cool_temp = kwargs.get(ATTR_TARGET_TEMP_HIGH)
@ -364,27 +366,27 @@ class NexiaZone(NexiaThermostatZoneEntity, ClimateEntity):
attrs[ATTR_HUMIDIFY_SETPOINT] = humdify_setpoint
return attrs
async def async_set_preset_mode(self, preset_mode: str):
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set the preset mode."""
await self._zone.set_preset(preset_mode)
self._signal_zone_update()
async def async_turn_aux_heat_off(self):
async def async_turn_aux_heat_off(self) -> None:
"""Turn Aux Heat off."""
await self._thermostat.set_emergency_heat(False)
self._signal_thermostat_update()
async def async_turn_aux_heat_on(self):
async def async_turn_aux_heat_on(self) -> None:
"""Turn Aux Heat on."""
self._thermostat.set_emergency_heat(True)
self._signal_thermostat_update()
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn off the zone."""
await self.async_set_hvac_mode(OPERATION_MODE_OFF)
self._signal_zone_update()
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn on the zone."""
await self.async_set_hvac_mode(OPERATION_MODE_AUTO)
self._signal_zone_update()

View File

@ -169,7 +169,7 @@ class NextBusDepartureSensor(SensorEntity):
"""Return additional state attributes."""
return self._attributes
def update(self):
def update(self) -> None:
"""Update sensor with new departures times."""
# Note: using Multi because there is a bug with the single stop impl
results = self._client.get_predictions_for_multi_stops(

View File

@ -53,6 +53,6 @@ class NextcloudBinarySensor(BinarySensorEntity):
"""Return the unique ID for this binary sensor."""
return f"{self.hass.data[DOMAIN]['instance']}#{self._name}"
def update(self):
def update(self) -> None:
"""Update the binary sensor."""
self._is_on = self.hass.data[DOMAIN][self._name]

View File

@ -53,6 +53,6 @@ class NextcloudSensor(SensorEntity):
"""Return the unique ID for this sensor."""
return f"{self.hass.data[DOMAIN]['instance']}#{self._name}"
def update(self):
def update(self) -> None:
"""Update the sensor."""
self._state = self.hass.data[DOMAIN][self._name]

View File

@ -153,7 +153,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)

View File

@ -158,7 +158,7 @@ class NMBSLiveBoard(SensorEntity):
return attrs
def update(self):
def update(self) -> None:
"""Set the state equal to the next departure."""
liveboard = self._api_client.get_liveboard(self._station)
@ -278,7 +278,7 @@ class NMBSSensor(SensorEntity):
return "vias" in self._attrs and int(self._attrs["vias"]["number"]) > 0
def update(self):
def update(self) -> None:
"""Set the state to the duration of a connection."""
connections = self._api_client.get_connections(
self._station_from, self._station_to

View File

@ -130,7 +130,7 @@ class NOAATidesAndCurrentsSensor(SensorEntity):
return f"Low tide at {tidetime}"
return None
def update(self):
def update(self) -> None:
"""Get the latest data from NOAA Tides and Currents API."""
begin = datetime.now()
delta = timedelta(days=2)

View File

@ -2,6 +2,7 @@
from datetime import datetime
import logging
import time
from typing import Any
from nuheat.config import SCHEDULE_HOLD, SCHEDULE_RUN, SCHEDULE_TEMPORARY_HOLD
from nuheat.util import (
@ -100,7 +101,7 @@ class NuHeatThermostat(CoordinatorEntity, ClimateEntity):
return self._thermostat.room
@property
def temperature_unit(self):
def temperature_unit(self) -> str:
"""Return the unit of measurement."""
if self._temperature_unit == "C":
return TEMP_CELSIUS
@ -121,7 +122,7 @@ class NuHeatThermostat(CoordinatorEntity, ClimateEntity):
return self._thermostat.serial_number
@property
def available(self):
def available(self) -> bool:
"""Return the unique id."""
return self.coordinator.last_update_success and self._thermostat.online
@ -178,7 +179,7 @@ class NuHeatThermostat(CoordinatorEntity, ClimateEntity):
"""Return available preset modes."""
return PRESET_MODES
def set_preset_mode(self, preset_mode):
def set_preset_mode(self, preset_mode: str) -> None:
"""Update the hold mode of the thermostat."""
self._set_schedule_mode(
PRESET_MODE_TO_SCHEDULE_MODE_MAP.get(preset_mode, SCHEDULE_RUN)
@ -191,7 +192,7 @@ class NuHeatThermostat(CoordinatorEntity, ClimateEntity):
self._thermostat.schedule_mode = schedule_mode
self._schedule_update()
def set_temperature(self, **kwargs):
def set_temperature(self, **kwargs: Any) -> None:
"""Set a new target temperature."""
self._set_temperature_and_mode(
kwargs.get(ATTR_TEMPERATURE), hvac_mode=kwargs.get(ATTR_HVAC_MODE)

View File

@ -54,7 +54,7 @@ class NukiDoorsensorEntity(NukiEntity, BinarySensorEntity):
return data
@property
def available(self):
def available(self) -> bool:
"""Return true if door sensor is present and activated."""
return super().available and self._nuki_device.is_door_sensor_activated

View File

@ -92,7 +92,7 @@ class NumatoGpioBinarySensor(BinarySensorEntity):
self._state = None
self._api = api
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Connect state update callback."""
self.async_on_remove(
async_dispatcher_connect(
@ -118,7 +118,7 @@ class NumatoGpioBinarySensor(BinarySensorEntity):
"""Return the state of the entity."""
return self._state != self._invert_logic
def update(self):
def update(self) -> None:
"""Update the GPIO state."""
try:
self._state = self._api.read_input(self._device_id, self._port)

View File

@ -102,7 +102,7 @@ class NumatoGpioAdc(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 state."""
try:
adc_val = self._api.read_adc_input(self._device_id, self._port)

View File

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from numato_gpio import NumatoGpioError
@ -88,7 +89,7 @@ class NumatoGpioSwitch(SwitchEntity):
"""Return true if port is turned on."""
return self._state
def turn_on(self, **kwargs):
def turn_on(self, **kwargs: Any) -> None:
"""Turn the port on."""
try:
self._api.write_output(
@ -104,7 +105,7 @@ class NumatoGpioSwitch(SwitchEntity):
err,
)
def turn_off(self, **kwargs):
def turn_off(self, **kwargs: Any) -> None:
"""Turn the port off."""
try:
self._api.write_output(

View File

@ -109,7 +109,7 @@ class NWSSensor(CoordinatorEntity, SensorEntity):
return f"{base_unique_id(self._latitude, self._longitude)}_{self.entity_description.key}"
@property
def available(self):
def available(self) -> bool:
"""Return if state is available."""
if self.coordinator.last_update_success_time:
last_success_time = (

View File

@ -269,7 +269,7 @@ class NWSWeather(WeatherEntity):
return f"{base_unique_id(self.latitude, self.longitude)}_{self.mode}"
@property
def available(self):
def available(self) -> bool:
"""Return if state is available."""
last_success = (
self.coordinator_observation.last_update_success
@ -289,7 +289,7 @@ class NWSWeather(WeatherEntity):
last_success_time = False
return last_success or last_success_time
async def async_update(self):
async def async_update(self) -> None:
"""Update the entity.
Only used by the generic entity update service.

View File

@ -1,6 +1,8 @@
"""Support for NZBGet switches."""
from __future__ import annotations
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME
@ -61,12 +63,12 @@ class NZBGetDownloadSwitch(NZBGetEntity, SwitchEntity):
"""Return the state of the switch."""
return not self.coordinator.data["status"].get("DownloadPaused", False)
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Set downloads to enabled."""
await self.hass.async_add_executor_job(self.coordinator.nzbget.resumedownload)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Set downloads to paused."""
await self.hass.async_add_executor_job(self.coordinator.nzbget.pausedownload)
await self.coordinator.async_request_refresh()