Improve entity type hints [m] (#77816)

This commit is contained in:
epenet 2022-09-05 10:59:36 +02:00 committed by GitHub
parent ce6e57da5d
commit 6355e682fa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 103 additions and 85 deletions

View File

@ -157,7 +157,7 @@ class MagicSeaweedSensor(SensorEntity):
"""Return the unit system of this entity.""" """Return the unit system of this entity."""
return self._unit_system return self._unit_system
def update(self): def update(self) -> None:
"""Get the latest data from Magicseaweed and updates the states.""" """Get the latest data from Magicseaweed and updates the states."""
self.data.update() self.data.update()
if self.hour is None: if self.hour is None:

View File

@ -43,7 +43,7 @@ class MaxCubeBinarySensorBase(BinarySensorEntity):
self._device = device self._device = device
self._room = handler.cube.room_by_id(device.room_id) self._room = handler.cube.room_by_id(device.room_id)
def update(self): def update(self) -> None:
"""Get latest data from MAX! Cube.""" """Get latest data from MAX! Cube."""
self._cubehandle.update() self._cubehandle.update()

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import logging import logging
import socket import socket
from typing import Any
from maxcube.device import ( from maxcube.device import (
MAX_DEVICE_MODE_AUTOMATIC, MAX_DEVICE_MODE_AUTOMATIC,
@ -183,7 +184,7 @@ class MaxCubeClimate(ClimateEntity):
return None return None
return temp return temp
def set_temperature(self, **kwargs): def set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures.""" """Set new target temperatures."""
if (temp := kwargs.get(ATTR_TEMPERATURE)) is None: if (temp := kwargs.get(ATTR_TEMPERATURE)) is None:
raise ValueError( raise ValueError(
@ -207,7 +208,7 @@ class MaxCubeClimate(ClimateEntity):
return PRESET_AWAY return PRESET_AWAY
return PRESET_NONE return PRESET_NONE
def set_preset_mode(self, preset_mode): def set_preset_mode(self, preset_mode: str) -> None:
"""Set new operation mode.""" """Set new operation mode."""
if preset_mode == PRESET_COMFORT: if preset_mode == PRESET_COMFORT:
self._set_target(MAX_DEVICE_MODE_MANUAL, self._device.comfort_temperature) self._set_target(MAX_DEVICE_MODE_MANUAL, self._device.comfort_temperature)
@ -231,6 +232,6 @@ class MaxCubeClimate(ClimateEntity):
return {} return {}
return {ATTR_VALVE_POSITION: self._device.valve_position} return {ATTR_VALVE_POSITION: self._device.valve_position}
def update(self): def update(self) -> None:
"""Get latest data from MAX! Cube.""" """Get latest data from MAX! Cube."""
self._cubehandle.update() self._cubehandle.update()

View File

@ -1,4 +1,6 @@
"""Platform for Mazda switch integration.""" """Platform for Mazda switch integration."""
from typing import Any
from pymazda import Client as MazdaAPIClient from pymazda import Client as MazdaAPIClient
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
@ -57,13 +59,13 @@ class MazdaChargingSwitch(MazdaEntity, SwitchEntity):
self.async_write_ha_state() self.async_write_ha_state()
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Start charging the vehicle.""" """Start charging the vehicle."""
await self.client.start_charging(self.vehicle_id) await self.client.start_charging(self.vehicle_id)
await self.refresh_status_and_write_state() await self.refresh_status_and_write_state()
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Stop charging the vehicle.""" """Stop charging the vehicle."""
await self.client.stop_charging(self.vehicle_id) await self.client.stop_charging(self.vehicle_id)

View File

@ -215,7 +215,7 @@ class MeaterProbeTemperature(
return self.entity_description.value(device) return self.entity_description.value(device)
@property @property
def available(self): def available(self) -> bool:
"""Return if entity is available.""" """Return if entity is available."""
# See if the device was returned from the API. If not, it's offline # See if the device was returned from the API. If not, it's offline
return ( return (

View File

@ -170,7 +170,7 @@ class MediaroomDevice(MediaPlayerEntity):
"""Return True if entity is available.""" """Return True if entity is available."""
return self._available return self._available
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
async def async_notify_received(notify): async def async_notify_received(notify):
@ -247,7 +247,7 @@ class MediaroomDevice(MediaPlayerEntity):
"""Channel currently playing.""" """Channel currently playing."""
return self._channel return self._channel
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn on the receiver.""" """Turn on the receiver."""
try: try:
@ -259,7 +259,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn off the receiver.""" """Turn off the receiver."""
try: try:
@ -271,7 +271,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_media_play(self): async def async_media_play(self) -> None:
"""Send play command.""" """Send play command."""
try: try:
@ -284,7 +284,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_media_pause(self): async def async_media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
try: try:
@ -296,7 +296,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_media_stop(self): async def async_media_stop(self) -> None:
"""Send stop command.""" """Send stop command."""
try: try:
@ -308,7 +308,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_media_previous_track(self): async def async_media_previous_track(self) -> None:
"""Send Program Down command.""" """Send Program Down command."""
try: try:
@ -320,7 +320,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_media_next_track(self): async def async_media_next_track(self) -> None:
"""Send Program Up command.""" """Send Program Up command."""
try: try:
@ -332,7 +332,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_volume_up(self): async def async_volume_up(self) -> None:
"""Send volume up command.""" """Send volume up command."""
try: try:
@ -342,7 +342,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_volume_down(self): async def async_volume_down(self) -> None:
"""Send volume up command.""" """Send volume up command."""
try: try:
@ -351,7 +351,7 @@ class MediaroomDevice(MediaPlayerEntity):
self._available = False self._available = False
self.async_write_ha_state() self.async_write_ha_state()
async def async_mute_volume(self, mute): async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command.""" """Send mute command."""
try: try:

View File

@ -105,7 +105,7 @@ class MelCloudClimate(ClimateEntity):
self.api = device self.api = device
self._base_device = self.api.device self._base_device = self.api.device
async def async_update(self): async def async_update(self) -> None:
"""Update state from MELCloud.""" """Update state from MELCloud."""
await self.api.async_update() await self.api.async_update()
@ -257,7 +257,7 @@ class AtaDeviceClimate(MelCloudClimate):
"""Return vertical vane position or mode.""" """Return vertical vane position or mode."""
return self._device.vane_vertical return self._device.vane_vertical
async def async_set_swing_mode(self, swing_mode) -> None: async def async_set_swing_mode(self, swing_mode: str) -> None:
"""Set vertical vane position or mode.""" """Set vertical vane position or mode."""
await self.async_set_vane_vertical(swing_mode) await self.async_set_vane_vertical(swing_mode)
@ -362,7 +362,7 @@ class AtwDeviceZoneClimate(MelCloudClimate):
"""Return the temperature we try to reach.""" """Return the temperature we try to reach."""
return self._zone.target_temperature return self._zone.target_temperature
async def async_set_temperature(self, **kwargs) -> None: async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature.""" """Set new target temperature."""
await self._zone.set_target_temperature( await self._zone.set_target_temperature(
kwargs.get("temperature", self.target_temperature) kwargs.get("temperature", self.target_temperature)

View File

@ -165,7 +165,7 @@ class MelDeviceSensor(SensorEntity):
"""Return the state of the sensor.""" """Return the state of the sensor."""
return self.entity_description.value_fn(self._api) return self.entity_description.value_fn(self._api)
async def async_update(self): async def async_update(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
await self._api.async_update() await self._api.async_update()

View File

@ -1,6 +1,8 @@
"""Platform for water_heater integration.""" """Platform for water_heater integration."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from pymelcloud import DEVICE_TYPE_ATW, AtwDevice from pymelcloud import DEVICE_TYPE_ATW, AtwDevice
from pymelcloud.atw_device import ( from pymelcloud.atw_device import (
PROPERTY_OPERATION_MODE, PROPERTY_OPERATION_MODE,
@ -51,7 +53,7 @@ class AtwWaterHeater(WaterHeaterEntity):
self._device = device self._device = device
self._name = device.name self._name = device.name
async def async_update(self): async def async_update(self) -> None:
"""Update state from MELCloud.""" """Update state from MELCloud."""
await self._api.async_update() await self._api.async_update()
@ -109,7 +111,7 @@ class AtwWaterHeater(WaterHeaterEntity):
"""Return the temperature we try to reach.""" """Return the temperature we try to reach."""
return self._device.target_tank_temperature return self._device.target_tank_temperature
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature.""" """Set new target temperature."""
await self._device.set( await self._device.set(
{ {
@ -119,7 +121,7 @@ class AtwWaterHeater(WaterHeaterEntity):
} }
) )
async def async_set_operation_mode(self, operation_mode): async def async_set_operation_mode(self, operation_mode: str) -> None:
"""Set new target operation mode.""" """Set new target operation mode."""
await self._device.set({PROPERTY_OPERATION_MODE: operation_mode}) await self._device.set({PROPERTY_OPERATION_MODE: operation_mode})

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import ( from homeassistant.components.climate.const import (
@ -135,12 +136,12 @@ class MelissaClimate(ClimateEntity):
"""Return the maximum supported temperature for the thermostat.""" """Return the maximum supported temperature for the thermostat."""
return 30 return 30
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature.""" """Set new target temperature."""
temp = kwargs.get(ATTR_TEMPERATURE) temp = kwargs.get(ATTR_TEMPERATURE)
await self.async_send({self._api.TEMP: temp}) await self.async_send({self._api.TEMP: temp})
async def async_set_fan_mode(self, fan_mode): async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set fan mode.""" """Set fan mode."""
melissa_fan_mode = self.hass_fan_to_melissa(fan_mode) melissa_fan_mode = self.hass_fan_to_melissa(fan_mode)
await self.async_send({self._api.FAN: melissa_fan_mode}) await self.async_send({self._api.FAN: melissa_fan_mode})
@ -168,7 +169,7 @@ class MelissaClimate(ClimateEntity):
): ):
self._cur_settings = old_value self._cur_settings = old_value
async def async_update(self): async def async_update(self) -> None:
"""Get latest data from Melissa.""" """Get latest data from Melissa."""
try: try:
self._data = (await self._api.async_status(cached=True))[ self._data = (await self._api.async_status(cached=True))[

View File

@ -143,6 +143,6 @@ class MfiSensor(SensorEntity):
return "State" return "State"
return tag return tag
def update(self): def update(self) -> None:
"""Get the latest data.""" """Get the latest data."""
self._port.refresh() self._port.refresh()

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from mficlient.client import FailedToLogin, MFiClient from mficlient.client import FailedToLogin, MFiClient
import requests import requests
@ -94,19 +95,19 @@ class MfiSwitch(SwitchEntity):
"""Return true if the device is on.""" """Return true if the device is on."""
return self._port.output return self._port.output
def update(self): def update(self) -> None:
"""Get the latest state and update the state.""" """Get the latest state and update the state."""
self._port.refresh() self._port.refresh()
if self._target_state is not None: if self._target_state is not None:
self._port.data["output"] = float(self._target_state) self._port.data["output"] = float(self._target_state)
self._target_state = None self._target_state = None
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
self._port.control(True) self._port.control(True)
self._target_state = True self._target_state = True
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
self._port.control(False) self._port.control(False)
self._target_state = False self._target_state = False

View File

@ -1,4 +1,6 @@
"""Support for mill wifi-enabled home heaters.""" """Support for mill wifi-enabled home heaters."""
from typing import Any
import mill import mill
import voluptuous as vol import voluptuous as vol
@ -123,7 +125,7 @@ class MillHeater(CoordinatorEntity, ClimateEntity):
self._update_attr(heater) self._update_attr(heater)
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature.""" """Set new target temperature."""
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None: if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
return return
@ -132,7 +134,7 @@ class MillHeater(CoordinatorEntity, ClimateEntity):
) )
await self.coordinator.async_request_refresh() await self.coordinator.async_request_refresh()
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.""" """Set new target fan mode."""
fan_status = 1 if fan_mode == FAN_ON else 0 fan_status = 1 if fan_mode == FAN_ON else 0
await self.coordinator.mill_data_connection.heater_control( await self.coordinator.mill_data_connection.heater_control(
@ -221,7 +223,7 @@ class LocalMillHeater(CoordinatorEntity, ClimateEntity):
self._update_attr() self._update_attr()
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature.""" """Set new target temperature."""
if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None: if (temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
return return

View File

@ -184,7 +184,7 @@ class MinMaxSensor(SensorEntity):
self.count_sensors = len(self._entity_ids) self.count_sensors = len(self._entity_ids)
self.states = {} self.states = {}
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Handle added to Hass.""" """Handle added to Hass."""
self.async_on_remove( self.async_on_remove(
async_track_state_change_event( async_track_state_change_event(

View File

@ -112,7 +112,7 @@ class MobileAppEntity(TrackerEntity, RestoreEntity):
"""Return the device info.""" """Return the device info."""
return device_info(self._entry.data) return device_info(self._entry.data)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Call when entity about to be added to Home Assistant.""" """Call when entity about to be added to Home Assistant."""
await super().async_added_to_hass() await super().async_added_to_hass()
self._dispatch_unsub = async_dispatcher_connect( self._dispatch_unsub = async_dispatcher_connect(
@ -138,7 +138,7 @@ class MobileAppEntity(TrackerEntity, RestoreEntity):
data.update({key: attr[key] for key in attr if key in ATTR_KEYS}) data.update({key: attr[key] for key in attr if key in ATTR_KEYS})
self._data = data self._data = data
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Call when entity is being removed from hass.""" """Call when entity is being removed from hass."""
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()

View File

@ -1,5 +1,6 @@
"""Support for Alpha2 room control unit via Alpha2 base.""" """Support for Alpha2 room control unit via Alpha2 base."""
import logging import logging
from typing import Any
from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import ( from homeassistant.components.climate.const import (
@ -110,7 +111,7 @@ class Alpha2Climate(CoordinatorEntity[Alpha2BaseCoordinator], ClimateEntity):
self.coordinator.data["heat_areas"][self.heat_area_id].get("T_TARGET", 0.0) self.coordinator.data["heat_areas"][self.heat_area_id].get("T_TARGET", 0.0)
) )
async def async_set_temperature(self, **kwargs) -> None: async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperatures.""" """Set new target temperatures."""
if (target_temperature := kwargs.get(ATTR_TEMPERATURE)) is None: if (target_temperature := kwargs.get(ATTR_TEMPERATURE)) is None:
return return

View File

@ -112,7 +112,7 @@ class MoldIndicator(SensorEntity):
self._indoor_hum = None self._indoor_hum = None
self._crit_temp = None self._crit_temp = None
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
@callback @callback
@ -273,7 +273,7 @@ class MoldIndicator(SensorEntity):
return hum return hum
async def async_update(self): async def async_update(self) -> None:
"""Calculate latest state.""" """Calculate latest state."""
_LOGGER.debug("Update state for %s", self.entity_id) _LOGGER.debug("Update state for %s", self.entity_id)
# check all sensors # check all sensors

View File

@ -142,7 +142,7 @@ class MonopriceZone(MediaPlayerEntity):
self._mute = None self._mute = None
self._update_success = True self._update_success = True
def update(self): def update(self) -> None:
"""Retrieve latest state.""" """Retrieve latest state."""
try: try:
state = self._monoprice.zone_status(self._zone_id) state = self._monoprice.zone_status(self._zone_id)
@ -165,7 +165,7 @@ class MonopriceZone(MediaPlayerEntity):
self._source = None self._source = None
@property @property
def entity_registry_enabled_default(self): def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry.""" """Return if the entity should be enabled when first added to the entity registry."""
return self._zone_id < 20 or self._update_success return self._zone_id < 20 or self._update_success
@ -231,36 +231,36 @@ class MonopriceZone(MediaPlayerEntity):
self._monoprice.restore_zone(self._snapshot) self._monoprice.restore_zone(self._snapshot)
self.schedule_update_ha_state(True) self.schedule_update_ha_state(True)
def select_source(self, source): def select_source(self, source: str) -> None:
"""Set input source.""" """Set input source."""
if source not in self._source_name_id: if source not in self._source_name_id:
return return
idx = self._source_name_id[source] idx = self._source_name_id[source]
self._monoprice.set_source(self._zone_id, idx) self._monoprice.set_source(self._zone_id, idx)
def turn_on(self): def turn_on(self) -> None:
"""Turn the media player on.""" """Turn the media player on."""
self._monoprice.set_power(self._zone_id, True) self._monoprice.set_power(self._zone_id, True)
def turn_off(self): def turn_off(self) -> None:
"""Turn the media player off.""" """Turn the media player off."""
self._monoprice.set_power(self._zone_id, False) self._monoprice.set_power(self._zone_id, False)
def mute_volume(self, mute): def mute_volume(self, mute: bool) -> None:
"""Mute (true) or unmute (false) media player.""" """Mute (true) or unmute (false) media player."""
self._monoprice.set_mute(self._zone_id, mute) self._monoprice.set_mute(self._zone_id, mute)
def set_volume_level(self, volume): def set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.""" """Set volume level, range 0..1."""
self._monoprice.set_volume(self._zone_id, int(volume * 38)) self._monoprice.set_volume(self._zone_id, int(volume * 38))
def volume_up(self): def volume_up(self) -> None:
"""Volume up the media player.""" """Volume up the media player."""
if self._volume is None: if self._volume is None:
return return
self._monoprice.set_volume(self._zone_id, min(self._volume + 1, 38)) self._monoprice.set_volume(self._zone_id, min(self._volume + 1, 38))
def volume_down(self): def volume_down(self) -> None:
"""Volume down media player.""" """Volume down media player."""
if self._volume is None: if self._volume is None:
return return

View File

@ -61,7 +61,7 @@ class MotionBatterySensor(CoordinatorEntity, SensorEntity):
self._attr_unique_id = f"{blind.mac}-battery" self._attr_unique_id = f"{blind.mac}-battery"
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
if self.coordinator.data is None: if self.coordinator.data is None:
return False return False
@ -81,12 +81,12 @@ class MotionBatterySensor(CoordinatorEntity, SensorEntity):
"""Return device specific state attributes.""" """Return device specific state attributes."""
return {ATTR_BATTERY_VOLTAGE: self._blind.battery_voltage} return {ATTR_BATTERY_VOLTAGE: self._blind.battery_voltage}
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to multicast pushes.""" """Subscribe to multicast pushes."""
self._blind.Register_callback(self.unique_id, self.schedule_update_ha_state) self._blind.Register_callback(self.unique_id, self.schedule_update_ha_state)
await super().async_added_to_hass() await super().async_added_to_hass()
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe when removed.""" """Unsubscribe when removed."""
self._blind.Remove_callback(self.unique_id) self._blind.Remove_callback(self.unique_id)
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()
@ -145,7 +145,7 @@ class MotionSignalStrengthSensor(CoordinatorEntity, SensorEntity):
self._attr_name = name self._attr_name = name
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
if self.coordinator.data is None: if self.coordinator.data is None:
return False return False
@ -164,12 +164,12 @@ class MotionSignalStrengthSensor(CoordinatorEntity, SensorEntity):
"""Return the state of the sensor.""" """Return the state of the sensor."""
return self._device.RSSI return self._device.RSSI
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to multicast pushes.""" """Subscribe to multicast pushes."""
self._device.Register_callback(self.unique_id, self.schedule_update_ha_state) self._device.Register_callback(self.unique_id, self.schedule_update_ha_state)
await super().async_added_to_hass() await super().async_added_to_hass()
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe when removed.""" """Unsubscribe when removed."""
self._device.Remove_callback(self.unique_id) self._device.Remove_callback(self.unique_id)
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()

View File

@ -6,6 +6,7 @@ from datetime import timedelta
import hashlib import hashlib
import logging import logging
import os import os
from typing import Any
import mpd import mpd
from mpd.asyncio import MPDClient from mpd.asyncio import MPDClient
@ -18,6 +19,7 @@ from homeassistant.components.media_player import (
MediaPlayerEntityFeature, MediaPlayerEntityFeature,
) )
from homeassistant.components.media_player.browse_media import ( from homeassistant.components.media_player.browse_media import (
BrowseMedia,
async_process_play_media_url, async_process_play_media_url,
) )
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
@ -164,7 +166,7 @@ class MpdDevice(MediaPlayerEntity):
"""Return true if MPD is available and connected.""" """Return true if MPD is available and connected."""
return self._is_connected return self._is_connected
async def async_update(self): async def async_update(self) -> None:
"""Get the latest data and update the state.""" """Get the latest data and update the state."""
try: try:
if not self._is_connected: if not self._is_connected:
@ -273,7 +275,7 @@ class MpdDevice(MediaPlayerEntity):
"""Hash value for media image.""" """Hash value for media image."""
return self._media_image_hash return self._media_image_hash
async def async_get_media_image(self): async def async_get_media_image(self) -> tuple[bytes | None, str | None]:
"""Fetch media image of current playing track.""" """Fetch media image of current playing track."""
if not (file := self._currentsong.get("file")): if not (file := self._currentsong.get("file")):
return None, None return None, None
@ -380,12 +382,12 @@ class MpdDevice(MediaPlayerEntity):
"""Return the list of available input sources.""" """Return the list of available input sources."""
return self._playlists return self._playlists
async def async_select_source(self, source): async def async_select_source(self, source: str) -> None:
"""Choose a different available playlist and play it.""" """Choose a different available playlist and play it."""
await self.async_play_media(MEDIA_TYPE_PLAYLIST, source) await self.async_play_media(MEDIA_TYPE_PLAYLIST, source)
@Throttle(PLAYLIST_UPDATE_INTERVAL) @Throttle(PLAYLIST_UPDATE_INTERVAL)
async def _update_playlists(self, **kwargs): async def _update_playlists(self, **kwargs: Any) -> None:
"""Update available MPD playlists.""" """Update available MPD playlists."""
try: try:
self._playlists = [] self._playlists = []
@ -395,12 +397,12 @@ class MpdDevice(MediaPlayerEntity):
self._playlists = None self._playlists = None
_LOGGER.warning("Playlists could not be updated: %s:", error) _LOGGER.warning("Playlists could not be updated: %s:", error)
async def async_set_volume_level(self, volume): async def async_set_volume_level(self, volume: float) -> None:
"""Set volume of media player.""" """Set volume of media player."""
if "volume" in self._status: if "volume" in self._status:
await self._client.setvol(int(volume * 100)) await self._client.setvol(int(volume * 100))
async def async_volume_up(self): async def async_volume_up(self) -> None:
"""Service to send the MPD the command for volume up.""" """Service to send the MPD the command for volume up."""
if "volume" in self._status: if "volume" in self._status:
current_volume = int(self._status["volume"]) current_volume = int(self._status["volume"])
@ -408,7 +410,7 @@ class MpdDevice(MediaPlayerEntity):
if current_volume <= 100: if current_volume <= 100:
self._client.setvol(current_volume + 5) self._client.setvol(current_volume + 5)
async def async_volume_down(self): async def async_volume_down(self) -> None:
"""Service to send the MPD the command for volume down.""" """Service to send the MPD the command for volume down."""
if "volume" in self._status: if "volume" in self._status:
current_volume = int(self._status["volume"]) current_volume = int(self._status["volume"])
@ -416,30 +418,30 @@ class MpdDevice(MediaPlayerEntity):
if current_volume >= 0: if current_volume >= 0:
await self._client.setvol(current_volume - 5) await self._client.setvol(current_volume - 5)
async def async_media_play(self): async def async_media_play(self) -> None:
"""Service to send the MPD the command for play/pause.""" """Service to send the MPD the command for play/pause."""
if self._status["state"] == "pause": if self._status["state"] == "pause":
await self._client.pause(0) await self._client.pause(0)
else: else:
await self._client.play() await self._client.play()
async def async_media_pause(self): async def async_media_pause(self) -> None:
"""Service to send the MPD the command for play/pause.""" """Service to send the MPD the command for play/pause."""
await self._client.pause(1) await self._client.pause(1)
async def async_media_stop(self): async def async_media_stop(self) -> None:
"""Service to send the MPD the command for stop.""" """Service to send the MPD the command for stop."""
await self._client.stop() await self._client.stop()
async def async_media_next_track(self): async def async_media_next_track(self) -> None:
"""Service to send the MPD the command for next track.""" """Service to send the MPD the command for next track."""
await self._client.next() await self._client.next()
async def async_media_previous_track(self): async def async_media_previous_track(self) -> None:
"""Service to send the MPD the command for previous track.""" """Service to send the MPD the command for previous track."""
await self._client.previous() await self._client.previous()
async def async_mute_volume(self, mute): async def async_mute_volume(self, mute: bool) -> None:
"""Mute. Emulated with set_volume_level.""" """Mute. Emulated with set_volume_level."""
if "volume" in self._status: if "volume" in self._status:
if mute: if mute:
@ -449,7 +451,9 @@ class MpdDevice(MediaPlayerEntity):
await self.async_set_volume_level(self._muted_volume) await self.async_set_volume_level(self._muted_volume)
self._muted = mute self._muted = mute
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 media player the command for playing a playlist.""" """Send the media player the command for playing a playlist."""
if media_source.is_media_source_id(media_id): if media_source.is_media_source_id(media_id):
media_type = MEDIA_TYPE_MUSIC media_type = MEDIA_TYPE_MUSIC
@ -483,7 +487,7 @@ class MpdDevice(MediaPlayerEntity):
return REPEAT_MODE_ALL return REPEAT_MODE_ALL
return REPEAT_MODE_OFF return REPEAT_MODE_OFF
async def async_set_repeat(self, repeat): async def async_set_repeat(self, repeat: str) -> None:
"""Set repeat mode.""" """Set repeat mode."""
if repeat == REPEAT_MODE_OFF: if repeat == REPEAT_MODE_OFF:
await self._client.repeat(0) await self._client.repeat(0)
@ -500,28 +504,30 @@ class MpdDevice(MediaPlayerEntity):
"""Boolean if shuffle is enabled.""" """Boolean if shuffle is enabled."""
return bool(int(self._status["random"])) return bool(int(self._status["random"]))
async def async_set_shuffle(self, shuffle): async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode.""" """Enable/disable shuffle mode."""
await self._client.random(int(shuffle)) await self._client.random(int(shuffle))
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Service to send the MPD the command to stop playing.""" """Service to send the MPD the command to stop playing."""
await self._client.stop() await self._client.stop()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Service to send the MPD the command to start playing.""" """Service to send the MPD the command to start playing."""
await self._client.play() await self._client.play()
await self._update_playlists(no_throttle=True) await self._update_playlists(no_throttle=True)
async def async_clear_playlist(self): async def async_clear_playlist(self) -> None:
"""Clear players playlist.""" """Clear players playlist."""
await self._client.clear() await self._client.clear()
async def async_media_seek(self, position): async def async_media_seek(self, position: float) -> None:
"""Send seek command.""" """Send seek command."""
await self._client.seekcur(position) await self._client.seekcur(position)
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.""" """Implement the websocket media browsing helper."""
return await media_source.async_browse_media( return await media_source.async_browse_media(
self.hass, self.hass,

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import functools import functools
from typing import Any
import voluptuous as vol import voluptuous as vol
@ -125,7 +126,7 @@ class MqttScene(
async def _subscribe_topics(self): async def _subscribe_topics(self):
"""(Re)Subscribe to topics.""" """(Re)Subscribe to topics."""
async def async_activate(self, **kwargs): async def async_activate(self, **kwargs: Any) -> None:
"""Activate the scene. """Activate the scene.
This method is a coroutine. This method is a coroutine.

View File

@ -95,7 +95,7 @@ class MQTTRoomSensor(SensorEntity):
self._distance = None self._distance = None
self._updated = None self._updated = None
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events.""" """Subscribe to MQTT events."""
@callback @callback
@ -152,7 +152,7 @@ class MQTTRoomSensor(SensorEntity):
"""Return the current room of the entity.""" """Return the current room of the entity."""
return self._state return self._state
def update(self): def update(self) -> None:
"""Update the state for absent devices.""" """Update the state for absent devices."""
if ( if (
self._updated self._updated

View File

@ -141,7 +141,7 @@ class MVGLiveSensor(SensorEntity):
"""Return the unit this state is expressed in.""" """Return the unit this state is expressed in."""
return TIME_MINUTES return TIME_MINUTES
def update(self): def update(self) -> None:
"""Get the latest data and update the state.""" """Get the latest data and update the state."""
self.data.update() self.data.update()
if not self.data.departures: if not self.data.departures:

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from pymystrom.exceptions import MyStromConnectionError from pymystrom.exceptions import MyStromConnectionError
from pymystrom.switch import MyStromSwitch as _MyStromSwitch from pymystrom.switch import MyStromSwitch as _MyStromSwitch
@ -77,21 +78,21 @@ class MyStromSwitch(SwitchEntity):
"""Could the device be accessed during the last update call.""" """Could the device be accessed during the last update call."""
return self._available return self._available
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
try: try:
await self.plug.turn_on() await self.plug.turn_on()
except MyStromConnectionError: except MyStromConnectionError:
_LOGGER.error("No route to myStrom plug") _LOGGER.error("No route to myStrom plug")
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
try: try:
await self.plug.turn_off() await self.plug.turn_off()
except MyStromConnectionError: except MyStromConnectionError:
_LOGGER.error("No route to myStrom plug") _LOGGER.error("No route to myStrom plug")
async def async_update(self): async def async_update(self) -> None:
"""Get the latest data from the device and update the data.""" """Get the latest data from the device and update the data."""
try: try:
await self.plug.get_state() await self.plug.get_state()