Improve entity type hints [j-k] (#77594)

This commit is contained in:
epenet 2022-08-31 22:10:18 +02:00 committed by GitHub
parent a40c1401d3
commit 448f4ee755
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 60 additions and 54 deletions

View file

@ -1,4 +1,6 @@
"""Support for monitoring juicenet/juicepoint/juicebox based EVSE switches."""
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
@ -41,10 +43,10 @@ class JuiceNetChargeNowSwitch(JuiceNetDevice, SwitchEntity):
"""Return true if switch is on."""
return self.device.override_time != 0
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Charge now."""
await self.device.set_override(True)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Don't charge now."""
await self.device.set_override(False)

View file

@ -87,7 +87,7 @@ class KaiterraSensor(SensorEntity):
)
@property
def available(self):
def available(self) -> bool:
"""Return the availability of the sensor."""
return self._api.data.get(self._device_id) is not None
@ -110,7 +110,7 @@ class KaiterraSensor(SensorEntity):
return TEMP_CELSIUS
return value
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Register callback."""
self.async_on_remove(
async_dispatcher_connect(

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
import requests
import voluptuous as vol
@ -114,16 +115,16 @@ class KankunSwitch(SwitchEntity):
"""Return true if device is on."""
return self._state
def update(self):
def update(self) -> None:
"""Update device state."""
self._state = self._query_state()
def turn_on(self, **kwargs):
def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on."""
if self._switch("on"):
self._state = True
def turn_off(self, **kwargs):
def turn_off(self, **kwargs: Any) -> None:
"""Turn the device off."""
if self._switch("off"):
self._state = False

View file

@ -53,7 +53,7 @@ class RouterOnlineBinarySensor(BinarySensorEntity):
"""Return a client description for device registry."""
return self._router.device_info
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Client entity created."""
self.async_on_remove(
async_dispatcher_connect(

View file

@ -144,12 +144,12 @@ class KeeneticTracker(ScannerEntity):
}
return None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Client entity created."""
_LOGGER.debug("New network device tracker %s (%s)", self.name, self.unique_id)
@callback
def update_device():
def update_device() -> None:
_LOGGER.debug(
"Updating Keenetic tracked device %s (%s)",
self.entity_id,

View file

@ -253,7 +253,7 @@ class KefMediaPlayer(MediaPlayerEntity):
"""Return the state of the device."""
return self._state
async def async_update(self):
async def async_update(self) -> None:
"""Update latest state."""
_LOGGER.debug("Running async_update")
try:
@ -313,55 +313,55 @@ class KefMediaPlayer(MediaPlayerEntity):
"""Return the device's icon."""
return "mdi:speaker-wireless"
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn the media player off."""
await self._speaker.turn_off()
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn the media player on."""
if not self._supports_on:
raise NotImplementedError()
await self._speaker.turn_on()
async def async_volume_up(self):
async def async_volume_up(self) -> None:
"""Volume up the media player."""
await self._speaker.increase_volume()
async def async_volume_down(self):
async def async_volume_down(self) -> None:
"""Volume down the media player."""
await self._speaker.decrease_volume()
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self._speaker.set_volume(volume)
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Mute (True) or unmute (False) media player."""
if mute:
await self._speaker.mute()
else:
await self._speaker.unmute()
async def async_select_source(self, source: str):
async def async_select_source(self, source: str) -> None:
"""Select input source."""
if source in self.source_list:
await self._speaker.set_source(source)
else:
raise ValueError(f"Unknown input source: {source}.")
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Send play command."""
await self._speaker.set_play_pause()
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Send pause command."""
await self._speaker.set_play_pause()
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Send previous track command."""
await self._speaker.prev_track()
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._speaker.next_track()
@ -382,13 +382,13 @@ class KefMediaPlayer(MediaPlayerEntity):
**mode._asdict(),
)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Subscribe to DSP updates."""
self._update_dsp_task_remover = async_track_time_interval(
self.hass, self.update_dsp, DSP_SCAN_INTERVAL
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe to DSP updates."""
self._update_dsp_task_remover()
self._update_dsp_task_remover = None

View file

@ -45,7 +45,7 @@ class KiraRemote(Entity):
"""Return the Kira device's name."""
return self._name
def update(self):
def update(self) -> None:
"""No-op."""
def send_command(self, command, **kwargs):

View file

@ -1,4 +1,5 @@
"""KMtronic Switch integration."""
from typing import Any
import urllib.parse
from homeassistant.components.switch import SwitchEntity
@ -54,7 +55,7 @@ class KMtronicSwitch(CoordinatorEntity, SwitchEntity):
return not self._relay.is_energised
return self._relay.is_energised
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
if self._reverse:
await self._relay.de_energise()
@ -62,7 +63,7 @@ class KMtronicSwitch(CoordinatorEntity, SwitchEntity):
await self._relay.energise()
self.async_write_ha_state()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
if self._reverse:
await self._relay.energise()
@ -70,7 +71,7 @@ class KMtronicSwitch(CoordinatorEntity, SwitchEntity):
await self._relay.de_energise()
self.async_write_ha_state()
async def async_toggle(self, **kwargs) -> None:
async def async_toggle(self, **kwargs: Any) -> None:
"""Toggle the switch."""
await self._relay.toggle()
self.async_write_ha_state()

View file

@ -393,7 +393,7 @@ class KodiEntity(MediaPlayerEntity):
return STATE_PLAYING
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Connect the websocket if needed."""
if not self._connection.can_subscribe:
return
@ -481,7 +481,7 @@ class KodiEntity(MediaPlayerEntity):
self._connection.server.System.OnSleep = self.async_on_quit
@cmd
async def async_update(self):
async def async_update(self) -> None:
"""Retrieve latest state."""
if not self._connection.connected:
self._reset_state()
@ -526,7 +526,7 @@ class KodiEntity(MediaPlayerEntity):
self._reset_state([])
@property
def should_poll(self):
def should_poll(self) -> bool:
"""Return True if entity has to be polled for state."""
return not self._connection.can_subscribe
@ -636,68 +636,68 @@ class KodiEntity(MediaPlayerEntity):
return None
async def async_turn_on(self):
async def async_turn_on(self) -> None:
"""Turn the media player on."""
_LOGGER.debug("Firing event to turn on device")
self.hass.bus.async_fire(EVENT_TURN_ON, {ATTR_ENTITY_ID: self.entity_id})
async def async_turn_off(self):
async def async_turn_off(self) -> None:
"""Turn the media player off."""
_LOGGER.debug("Firing event to turn off device")
self.hass.bus.async_fire(EVENT_TURN_OFF, {ATTR_ENTITY_ID: self.entity_id})
@cmd
async def async_volume_up(self):
async def async_volume_up(self) -> None:
"""Volume up the media player."""
await self._kodi.volume_up()
@cmd
async def async_volume_down(self):
async def async_volume_down(self) -> None:
"""Volume down the media player."""
await self._kodi.volume_down()
@cmd
async def async_set_volume_level(self, volume):
async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1."""
await self._kodi.set_volume_level(int(volume * 100))
@cmd
async def async_mute_volume(self, mute):
async def async_mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player."""
await self._kodi.mute(mute)
@cmd
async def async_media_play_pause(self):
async def async_media_play_pause(self) -> None:
"""Pause media on media player."""
await self._kodi.play_pause()
@cmd
async def async_media_play(self):
async def async_media_play(self) -> None:
"""Play media."""
await self._kodi.play()
@cmd
async def async_media_pause(self):
async def async_media_pause(self) -> None:
"""Pause the media player."""
await self._kodi.pause()
@cmd
async def async_media_stop(self):
async def async_media_stop(self) -> None:
"""Stop the media player."""
await self._kodi.stop()
@cmd
async def async_media_next_track(self):
async def async_media_next_track(self) -> None:
"""Send next track command."""
await self._kodi.next_track()
@cmd
async def async_media_previous_track(self):
async def async_media_previous_track(self) -> None:
"""Send next track command."""
await self._kodi.previous_track()
@cmd
async def async_media_seek(self, position):
async def async_media_seek(self, position: float) -> None:
"""Send seek command."""
await self._kodi.media_seek(position)
@ -746,7 +746,7 @@ class KodiEntity(MediaPlayerEntity):
await self._kodi.play_file(media_id)
@cmd
async def async_set_shuffle(self, shuffle):
async def async_set_shuffle(self, shuffle: bool) -> None:
"""Set shuffle mode, for the first player."""
if self._no_active_players:
raise RuntimeError("Error: No active player.")
@ -790,7 +790,7 @@ class KodiEntity(MediaPlayerEntity):
)
return result
async def async_clear_playlist(self):
async def async_clear_playlist(self) -> None:
"""Clear default playlist (i.e. playlistid=0)."""
await self._kodi.clear_playlist()

View file

@ -76,7 +76,7 @@ class KonnectedBinarySensor(BinarySensorEntity):
identifiers={(KONNECTED_DOMAIN, self._device_id)},
)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Store entity_id and register state change callback."""
self._data[ATTR_ENTITY_ID] = self.entity_id
self.async_on_remove(

View file

@ -127,7 +127,7 @@ class KonnectedSensor(SensorEntity):
"""Return the state of the sensor."""
return self._state
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Store entity_id and register state change callback."""
entity_id_key = self._addr or self.entity_description.key
self._data[entity_id_key] = self.entity_id

View file

@ -1,5 +1,6 @@
"""Support for wired switches attached to a Konnected device."""
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -89,11 +90,11 @@ class KonnectedSwitch(SwitchEntity):
return DeviceInfo(identifiers={(KONNECTED_DOMAIN, self._device_id)})
@property
def available(self):
def available(self) -> bool:
"""Return whether the panel is available."""
return self.panel.available
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Send a command to turn on the switch."""
resp = await self.panel.update_switch(
self._zone_num,
@ -110,7 +111,7 @@ class KonnectedSwitch(SwitchEntity):
# Immediately set the state back off for momentary switches
self._set_state(False)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Send a command to turn off the switch."""
resp = await self.panel.update_switch(
self._zone_num, int(self._activation == STATE_LOW)
@ -142,7 +143,7 @@ class KonnectedSwitch(SwitchEntity):
"""Update the switch state."""
self._set_state(state)
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Store entity_id and register state change callback."""
self._data["entity_id"] = self.entity_id
self.async_on_remove(

View file

@ -4,6 +4,7 @@ from __future__ import annotations
from abc import ABC
from datetime import timedelta
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
@ -126,7 +127,7 @@ class PlenticoreDataSwitch(CoordinatorEntity, SwitchEntity, ABC):
self.coordinator.stop_fetch_data(self.module_id, self.data_id)
await super().async_will_remove_from_hass()
async def async_turn_on(self, **kwargs) -> None:
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn device on."""
if await self.coordinator.async_write_data(
self.module_id, {self.data_id: self.on_value}
@ -134,7 +135,7 @@ class PlenticoreDataSwitch(CoordinatorEntity, SwitchEntity, ABC):
self.coordinator.name = f"{self.platform_name} {self._name} {self.on_label}"
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs) -> None:
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn device off."""
if await self.coordinator.async_write_data(
self.module_id, {self.data_id: self.off_value}