Improve entity type hints [a] (#76986)

This commit is contained in:
epenet 2022-08-18 15:56:52 +02:00 committed by GitHub
parent 24f1287bf9
commit 65eb1584f7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
31 changed files with 152 additions and 124 deletions

View file

@ -80,7 +80,7 @@ class AdaxDevice(ClimateEntity):
manufacturer="Adax", manufacturer="Adax",
) )
async def async_set_hvac_mode(self, hvac_mode: str) -> None: async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set hvac mode.""" """Set hvac mode."""
if hvac_mode == HVACMode.HEAT: if hvac_mode == HVACMode.HEAT:
temperature = max(self.min_temp, self.target_temperature or self.min_temp) temperature = max(self.min_temp, self.target_temperature or self.min_temp)
@ -140,7 +140,7 @@ class LocalAdaxDevice(ClimateEntity):
manufacturer="Adax", manufacturer="Adax",
) )
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

@ -53,7 +53,7 @@ class AdsBinarySensor(AdsEntity, BinarySensorEntity):
super().__init__(ads_hub, name, ads_var) super().__init__(ads_hub, name, ads_var)
self._attr_device_class = device_class or BinarySensorDeviceClass.MOVING self._attr_device_class = device_class or BinarySensorDeviceClass.MOVING
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register device notification.""" """Register device notification."""
await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL) await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL)

View file

@ -70,7 +70,7 @@ class AdsSensor(AdsEntity, SensorEntity):
self._ads_type = ads_type self._ads_type = ads_type
self._factor = factor self._factor = factor
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register device notification.""" """Register device notification."""
await self.async_initialize_device( await self.async_initialize_device(
self._ads_var, self._ads_var,

View file

@ -1,6 +1,8 @@
"""Support for ADS switch platform.""" """Support for ADS switch platform."""
from __future__ import annotations from __future__ import annotations
from typing import Any
import pyads import pyads
import voluptuous as vol import voluptuous as vol
@ -41,7 +43,7 @@ def setup_platform(
class AdsSwitch(AdsEntity, SwitchEntity): class AdsSwitch(AdsEntity, SwitchEntity):
"""Representation of an ADS switch device.""" """Representation of an ADS switch device."""
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register device notification.""" """Register device notification."""
await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL) await self.async_initialize_device(self._ads_var, pyads.PLCTYPE_BOOL)
@ -50,10 +52,10 @@ class AdsSwitch(AdsEntity, SwitchEntity):
"""Return True if the entity is on.""" """Return True if the entity is on."""
return self._state_dict[STATE_KEY_STATE] return self._state_dict[STATE_KEY_STATE]
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
self._ads_hub.write_by_name(self._ads_var, True, pyads.PLCTYPE_BOOL) self._ads_hub.write_by_name(self._ads_var, True, pyads.PLCTYPE_BOOL)
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
self._ads_hub.write_by_name(self._ads_var, False, pyads.PLCTYPE_BOOL) self._ads_hub.write_by_name(self._ads_var, False, pyads.PLCTYPE_BOOL)

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 (
@ -114,7 +115,7 @@ class AdvantageAirAC(AdvantageAirAcEntity, ClimateEntity):
"""Return the current fan modes.""" """Return the current fan modes."""
return ADVANTAGE_AIR_FAN_MODES.get(self._ac["fan"]) return ADVANTAGE_AIR_FAN_MODES.get(self._ac["fan"])
async def async_set_hvac_mode(self, hvac_mode): async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the HVAC Mode and State.""" """Set the HVAC Mode and State."""
if hvac_mode == HVACMode.OFF: if hvac_mode == HVACMode.OFF:
await self.async_change( await self.async_change(
@ -132,13 +133,13 @@ class AdvantageAirAC(AdvantageAirAcEntity, ClimateEntity):
} }
) )
async def async_set_fan_mode(self, fan_mode): async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set the Fan Mode.""" """Set the Fan Mode."""
await self.async_change( await self.async_change(
{self.ac_key: {"info": {"fan": HASS_FAN_MODES.get(fan_mode)}}} {self.ac_key: {"info": {"fan": HASS_FAN_MODES.get(fan_mode)}}}
) )
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set the Temperature.""" """Set the Temperature."""
temp = kwargs.get(ATTR_TEMPERATURE) temp = kwargs.get(ATTR_TEMPERATURE)
await self.async_change({self.ac_key: {"info": {"setTemp": temp}}}) await self.async_change({self.ac_key: {"info": {"setTemp": temp}}})
@ -179,7 +180,7 @@ class AdvantageAirZone(AdvantageAirZoneEntity, ClimateEntity):
"""Return the target temperature.""" """Return the target temperature."""
return self._zone["setTemp"] return self._zone["setTemp"]
async def async_set_hvac_mode(self, hvac_mode): async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the HVAC Mode and State.""" """Set the HVAC Mode and State."""
if hvac_mode == HVACMode.OFF: if hvac_mode == HVACMode.OFF:
await self.async_change( await self.async_change(
@ -198,7 +199,7 @@ class AdvantageAirZone(AdvantageAirZoneEntity, ClimateEntity):
} }
) )
async def async_set_temperature(self, **kwargs): async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set the Temperature.""" """Set the Temperature."""
temp = kwargs.get(ATTR_TEMPERATURE) temp = kwargs.get(ATTR_TEMPERATURE)
await self.async_change( await self.async_change(

View file

@ -1,4 +1,6 @@
"""Switch platform for Advantage Air integration.""" """Switch platform for Advantage Air integration."""
from typing import Any
from homeassistant.components.switch import SwitchEntity from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -44,13 +46,13 @@ class AdvantageAirFreshAir(AdvantageAirAcEntity, SwitchEntity):
"""Return the fresh air status.""" """Return the fresh air status."""
return self._ac["freshAirStatus"] == ADVANTAGE_AIR_STATE_ON return self._ac["freshAirStatus"] == ADVANTAGE_AIR_STATE_ON
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn fresh air on.""" """Turn fresh air on."""
await self.async_change( await self.async_change(
{self.ac_key: {"info": {"freshAirStatus": ADVANTAGE_AIR_STATE_ON}}} {self.ac_key: {"info": {"freshAirStatus": ADVANTAGE_AIR_STATE_ON}}}
) )
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn fresh air off.""" """Turn fresh air off."""
await self.async_change( await self.async_change(
{self.ac_key: {"info": {"freshAirStatus": ADVANTAGE_AIR_STATE_OFF}}} {self.ac_key: {"info": {"freshAirStatus": ADVANTAGE_AIR_STATE_OFF}}}

View file

@ -91,7 +91,7 @@ class AgentCamera(MjpegCamera):
sw_version=device.client.version, sw_version=device.client.version,
) )
async def async_update(self): async def async_update(self) -> None:
"""Update our state from the Agent API.""" """Update our state from the Agent API."""
try: try:
await self.device.update() await self.device.update()
@ -148,7 +148,7 @@ class AgentCamera(MjpegCamera):
return self.device.online return self.device.online
@property @property
def motion_detection_enabled(self): def motion_detection_enabled(self) -> bool:
"""Return the camera motion detection status.""" """Return the camera motion detection status."""
return self.device.detector_active return self.device.detector_active
@ -160,11 +160,11 @@ class AgentCamera(MjpegCamera):
"""Disable alerts.""" """Disable alerts."""
await self.device.alerts_off() await self.device.alerts_off()
async def async_enable_motion_detection(self): async def async_enable_motion_detection(self) -> None:
"""Enable motion detection.""" """Enable motion detection."""
await self.device.detector_on() await self.device.detector_on()
async def async_disable_motion_detection(self): async def async_disable_motion_detection(self) -> None:
"""Disable motion detection.""" """Disable motion detection."""
await self.device.detector_off() await self.device.detector_off()
@ -176,7 +176,7 @@ class AgentCamera(MjpegCamera):
"""Stop recording.""" """Stop recording."""
await self.device.record_stop() await self.device.record_stop()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Enable the camera.""" """Enable the camera."""
await self.device.enable() await self.device.enable()
@ -184,6 +184,6 @@ class AgentCamera(MjpegCamera):
"""Take a snapshot.""" """Take a snapshot."""
await self.device.snapshot() await self.device.snapshot()
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Disable the camera.""" """Disable the camera."""
await self.device.disable() await self.device.disable()

View file

@ -154,7 +154,7 @@ class AirtouchAC(CoordinatorEntity, ClimateEntity):
modes.append(HVACMode.OFF) modes.append(HVACMode.OFF)
return modes return modes
async def async_set_hvac_mode(self, hvac_mode): async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new operation mode.""" """Set new operation mode."""
if hvac_mode not in HA_STATE_TO_AT: if hvac_mode not in HA_STATE_TO_AT:
raise ValueError(f"Unsupported HVAC mode: {hvac_mode}") raise ValueError(f"Unsupported HVAC mode: {hvac_mode}")
@ -170,7 +170,7 @@ class AirtouchAC(CoordinatorEntity, ClimateEntity):
_LOGGER.debug("Setting operation mode of %s to %s", self._ac_number, hvac_mode) _LOGGER.debug("Setting operation mode of %s to %s", self._ac_number, hvac_mode)
self.async_write_ha_state() self.async_write_ha_state()
async def async_set_fan_mode(self, fan_mode): async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new fan mode.""" """Set new fan mode."""
if fan_mode not in self.fan_modes: if fan_mode not in self.fan_modes:
raise ValueError(f"Unsupported fan mode: {fan_mode}") raise ValueError(f"Unsupported fan mode: {fan_mode}")
@ -182,14 +182,14 @@ class AirtouchAC(CoordinatorEntity, ClimateEntity):
self._unit = self._airtouch.GetAcs()[self._ac_number] self._unit = self._airtouch.GetAcs()[self._ac_number]
self.async_write_ha_state() self.async_write_ha_state()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn on.""" """Turn on."""
_LOGGER.debug("Turning %s on", self.unique_id) _LOGGER.debug("Turning %s on", self.unique_id)
# in case ac is not on. Airtouch turns itself off if no groups are turned on # in case ac is not on. Airtouch turns itself off if no groups are turned on
# (even if groups turned back on) # (even if groups turned back on)
await self._airtouch.TurnAcOn(self._ac_number) await self._airtouch.TurnAcOn(self._ac_number)
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn off.""" """Turn off."""
_LOGGER.debug("Turning %s off", self.unique_id) _LOGGER.debug("Turning %s off", self.unique_id)
await self._airtouch.TurnAcOff(self._ac_number) await self._airtouch.TurnAcOff(self._ac_number)
@ -266,7 +266,7 @@ class AirtouchGroup(CoordinatorEntity, ClimateEntity):
return HVACMode.FAN_ONLY return HVACMode.FAN_ONLY
async def async_set_hvac_mode(self, hvac_mode): async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new operation mode.""" """Set new operation mode."""
if hvac_mode not in HA_STATE_TO_AT: if hvac_mode not in HA_STATE_TO_AT:
raise ValueError(f"Unsupported HVAC mode: {hvac_mode}") raise ValueError(f"Unsupported HVAC mode: {hvac_mode}")
@ -304,7 +304,7 @@ class AirtouchGroup(CoordinatorEntity, ClimateEntity):
) )
self.async_write_ha_state() self.async_write_ha_state()
async def async_set_fan_mode(self, fan_mode): async def async_set_fan_mode(self, fan_mode: str) -> None:
"""Set new fan mode.""" """Set new fan mode."""
if fan_mode not in self.fan_modes: if fan_mode not in self.fan_modes:
raise ValueError(f"Unsupported fan mode: {fan_mode}") raise ValueError(f"Unsupported fan mode: {fan_mode}")
@ -315,7 +315,7 @@ class AirtouchGroup(CoordinatorEntity, ClimateEntity):
) )
self.async_write_ha_state() self.async_write_ha_state()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn on.""" """Turn on."""
_LOGGER.debug("Turning %s on", self.unique_id) _LOGGER.debug("Turning %s on", self.unique_id)
await self._airtouch.TurnGroupOn(self._group_number) await self._airtouch.TurnGroupOn(self._group_number)
@ -330,7 +330,7 @@ class AirtouchGroup(CoordinatorEntity, ClimateEntity):
await self.coordinator.async_request_refresh() await self.coordinator.async_request_refresh()
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.""" """Turn off."""
_LOGGER.debug("Turning %s off", self.unique_id) _LOGGER.debug("Turning %s off", self.unique_id)
await self._airtouch.TurnGroupOff(self._group_number) await self._airtouch.TurnGroupOff(self._group_number)

View file

@ -88,7 +88,7 @@ class AlarmDecoderBinarySensor(BinarySensorEntity):
CONF_ZONE_NUMBER: self._zone_number, CONF_ZONE_NUMBER: self._zone_number,
} }
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect(self.hass, SIGNAL_ZONE_FAULT, self._fault_callback) async_dispatcher_connect(self.hass, SIGNAL_ZONE_FAULT, self._fault_callback)

View file

@ -24,7 +24,7 @@ class AlarmDecoderSensor(SensorEntity):
_attr_name = "Alarm Panel Display" _attr_name = "Alarm Panel Display"
_attr_should_poll = False _attr_should_poll = False
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect( async_dispatcher_connect(

View file

@ -170,7 +170,7 @@ class AmbiclimateEntity(ClimateEntity):
return return
await self._heater.set_target_temperature(temperature) await self._heater.set_target_temperature(temperature)
async def async_set_hvac_mode(self, hvac_mode: str) -> None: async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode.""" """Set new target hvac mode."""
if hvac_mode == HVACMode.HEAT: if hvac_mode == HVACMode.HEAT:
await self._heater.turn_on() await self._heater.turn_on()

View file

@ -298,7 +298,7 @@ class ADBDevice(MediaPlayerEntity):
self.turn_off_command = options.get(CONF_TURN_OFF_COMMAND) self.turn_off_command = options.get(CONF_TURN_OFF_COMMAND)
self.turn_on_command = options.get(CONF_TURN_ON_COMMAND) self.turn_on_command = options.get(CONF_TURN_ON_COMMAND)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Set config parameter when add to hass.""" """Set config parameter when add to hass."""
await super().async_added_to_hass() await super().async_added_to_hass()
self._process_config() self._process_config()
@ -321,7 +321,7 @@ class ADBDevice(MediaPlayerEntity):
"""Take a screen capture from the device.""" """Take a screen capture from the device."""
return await self.aftv.adb_screencap() return await self.aftv.adb_screencap()
async def async_get_media_image(self): async def async_get_media_image(self) -> tuple[bytes | None, str | None]:
"""Fetch current playing image.""" """Fetch current playing image."""
if not self._screencap or self.state in (STATE_OFF, None) or not self.available: if not self._screencap or self.state in (STATE_OFF, None) or not self.available:
return None, None return None, None
@ -337,22 +337,22 @@ class ADBDevice(MediaPlayerEntity):
return None, None return None, None
@adb_decorator() @adb_decorator()
async def async_media_play(self): async def async_media_play(self) -> None:
"""Send play command.""" """Send play command."""
await self.aftv.media_play() await self.aftv.media_play()
@adb_decorator() @adb_decorator()
async def async_media_pause(self): async def async_media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
await self.aftv.media_pause() await self.aftv.media_pause()
@adb_decorator() @adb_decorator()
async def async_media_play_pause(self): async def async_media_play_pause(self) -> None:
"""Send play/pause command.""" """Send play/pause command."""
await self.aftv.media_play_pause() await self.aftv.media_play_pause()
@adb_decorator() @adb_decorator()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn on the device.""" """Turn on the device."""
if self.turn_on_command: if self.turn_on_command:
await self.aftv.adb_shell(self.turn_on_command) await self.aftv.adb_shell(self.turn_on_command)
@ -360,7 +360,7 @@ class ADBDevice(MediaPlayerEntity):
await self.aftv.turn_on() await self.aftv.turn_on()
@adb_decorator() @adb_decorator()
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn off the device.""" """Turn off the device."""
if self.turn_off_command: if self.turn_off_command:
await self.aftv.adb_shell(self.turn_off_command) await self.aftv.adb_shell(self.turn_off_command)
@ -368,17 +368,17 @@ class ADBDevice(MediaPlayerEntity):
await self.aftv.turn_off() await self.aftv.turn_off()
@adb_decorator() @adb_decorator()
async def async_media_previous_track(self): async def async_media_previous_track(self) -> None:
"""Send previous track command (results in rewind).""" """Send previous track command (results in rewind)."""
await self.aftv.media_previous_track() await self.aftv.media_previous_track()
@adb_decorator() @adb_decorator()
async def async_media_next_track(self): async def async_media_next_track(self) -> None:
"""Send next track command (results in fast-forward).""" """Send next track command (results in fast-forward)."""
await self.aftv.media_next_track() await self.aftv.media_next_track()
@adb_decorator() @adb_decorator()
async def async_select_source(self, source): async def async_select_source(self, source: str) -> None:
"""Select input source. """Select input source.
If the source starts with a '!', then it will close the app instead of If the source starts with a '!', then it will close the app instead of
@ -469,7 +469,7 @@ class AndroidTVDevice(ADBDevice):
) )
@adb_decorator(override_available=True) @adb_decorator(override_available=True)
async def async_update(self): async def async_update(self) -> None:
"""Update the device state and, if necessary, re-connect.""" """Update the device state and, if necessary, re-connect."""
# Check if device is disconnected. # Check if device is disconnected.
if not self._attr_available: if not self._attr_available:
@ -514,12 +514,12 @@ class AndroidTVDevice(ADBDevice):
self._attr_source_list = None self._attr_source_list = None
@adb_decorator() @adb_decorator()
async def async_media_stop(self): async def async_media_stop(self) -> None:
"""Send stop command.""" """Send stop command."""
await self.aftv.media_stop() await self.aftv.media_stop()
@adb_decorator() @adb_decorator()
async def async_mute_volume(self, mute): async def async_mute_volume(self, mute: bool) -> None:
"""Mute the volume.""" """Mute the volume."""
is_muted = await self.aftv.is_volume_muted() is_muted = await self.aftv.is_volume_muted()
@ -528,17 +528,17 @@ class AndroidTVDevice(ADBDevice):
await self.aftv.mute_volume() await self.aftv.mute_volume()
@adb_decorator() @adb_decorator()
async def async_set_volume_level(self, volume): async def async_set_volume_level(self, volume: float) -> None:
"""Set the volume level.""" """Set the volume level."""
await self.aftv.set_volume_level(volume) await self.aftv.set_volume_level(volume)
@adb_decorator() @adb_decorator()
async def async_volume_down(self): async def async_volume_down(self) -> None:
"""Send volume down command.""" """Send volume down command."""
self._attr_volume_level = await self.aftv.volume_down(self._attr_volume_level) self._attr_volume_level = await self.aftv.volume_down(self._attr_volume_level)
@adb_decorator() @adb_decorator()
async def async_volume_up(self): async def async_volume_up(self) -> None:
"""Send volume up command.""" """Send volume up command."""
self._attr_volume_level = await self.aftv.volume_up(self._attr_volume_level) self._attr_volume_level = await self.aftv.volume_up(self._attr_volume_level)
@ -558,7 +558,7 @@ class FireTVDevice(ADBDevice):
) )
@adb_decorator(override_available=True) @adb_decorator(override_available=True)
async def async_update(self): async def async_update(self) -> None:
"""Update the device state and, if necessary, re-connect.""" """Update the device state and, if necessary, re-connect."""
# Check if device is disconnected. # Check if device is disconnected.
if not self._attr_available: if not self._attr_available:
@ -600,6 +600,6 @@ class FireTVDevice(ADBDevice):
self._attr_source_list = None self._attr_source_list = None
@adb_decorator() @adb_decorator()
async def async_media_stop(self): async def async_media_stop(self) -> None:
"""Send stop (back) command.""" """Send stop (back) command."""
await self.aftv.back() await self.aftv.back()

View file

@ -3,6 +3,7 @@ from __future__ import annotations
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any
from anel_pwrctrl import DeviceMaster from anel_pwrctrl import DeviceMaster
import voluptuous as vol import voluptuous as vol
@ -78,16 +79,16 @@ class PwrCtrlSwitch(SwitchEntity):
self._attr_unique_id = f"{port.device.host}-{port.get_index()}" self._attr_unique_id = f"{port.device.host}-{port.get_index()}"
self._attr_name = port.label self._attr_name = port.label
def update(self): def update(self) -> None:
"""Trigger update for all switches on the parent device.""" """Trigger update for all switches on the parent device."""
self._parent_device.update() self._parent_device.update()
self._attr_is_on = self._port.get_state() self._attr_is_on = self._port.get_state()
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
self._port.on() self._port.on()
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
self._port.off() self._port.off()

View file

@ -38,6 +38,6 @@ class OnlineStatus(BinarySensorEntity):
self._data = data self._data = data
self._attr_name = config[CONF_NAME] self._attr_name = config[CONF_NAME]
def update(self): def update(self) -> None:
"""Get the status report from APCUPSd and set this entity's state.""" """Get the status report from APCUPSd and set this entity's state."""
self._attr_is_on = int(self._data.status[KEY_STATUS], 16) & VALUE_ONLINE > 0 self._attr_is_on = int(self._data.status[KEY_STATUS], 16) & VALUE_ONLINE > 0

View file

@ -460,7 +460,7 @@ class APCUPSdSensor(SensorEntity):
self._data = data self._data = data
self._attr_name = f"{SENSOR_PREFIX}{description.name}" self._attr_name = f"{SENSOR_PREFIX}{description.name}"
def update(self): def update(self) -> None:
"""Get the latest status and use it to update our sensor state.""" """Get the latest status and use it to update our sensor state."""
key = self.entity_description.key.upper() key = self.entity_description.key.upper()
if key not in self._data.status: if key not in self._data.status:

View file

@ -1,5 +1,8 @@
"""Support for Apple TV media player.""" """Support for Apple TV media player."""
from __future__ import annotations
import logging import logging
from typing import Any
from pyatv import exceptions from pyatv import exceptions
from pyatv.const import ( from pyatv.const import (
@ -276,7 +279,9 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity):
return dt_util.utcnow() return dt_util.utcnow()
return None return None
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.""" """Send the play_media command to the media player."""
# If input (file) has a file format supported by pyatv, then stream it with # If input (file) has a file format supported by pyatv, then stream it with
# RAOP. Otherwise try to play it with regular AirPlay. # RAOP. Otherwise try to play it with regular AirPlay.
@ -314,7 +319,7 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity):
return self.atv.metadata.artwork_id return self.atv.metadata.artwork_id
return None return None
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 image.""" """Fetch media image of current playing image."""
state = self.state state = self.state
if self._playing and state not in [STATE_OFF, STATE_IDLE]: if self._playing and state not in [STATE_OFF, STATE_IDLE]:
@ -391,8 +396,8 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity):
async def async_browse_media( async def async_browse_media(
self, self,
media_content_type=None, media_content_type: str | None = None,
media_content_id=None, media_content_id: str | None = None,
) -> BrowseMedia: ) -> BrowseMedia:
"""Implement the websocket media browsing helper.""" """Implement the websocket media browsing helper."""
if media_content_id == "apps" or ( if media_content_id == "apps" or (
@ -427,12 +432,12 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity):
return cur_item return cur_item
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn the media player on.""" """Turn the media player on."""
if self._is_feature_available(FeatureName.TurnOn): if self._is_feature_available(FeatureName.TurnOn):
await self.atv.power.turn_on() await self.atv.power.turn_on()
async def async_turn_off(self): async def async_turn_off(self) -> None:
"""Turn the media player off.""" """Turn the media player off."""
if (self._is_feature_available(FeatureName.TurnOff)) and ( if (self._is_feature_available(FeatureName.TurnOff)) and (
not self._is_feature_available(FeatureName.PowerState) not self._is_feature_available(FeatureName.PowerState)
@ -440,58 +445,58 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity):
): ):
await self.atv.power.turn_off() await self.atv.power.turn_off()
async def async_media_play_pause(self): async def async_media_play_pause(self) -> None:
"""Pause media on media player.""" """Pause media on media player."""
if self._playing: if self._playing:
await self.atv.remote_control.play_pause() await self.atv.remote_control.play_pause()
async def async_media_play(self): async def async_media_play(self) -> None:
"""Play media.""" """Play media."""
if self.atv: if self.atv:
await self.atv.remote_control.play() await self.atv.remote_control.play()
async def async_media_stop(self): async def async_media_stop(self) -> None:
"""Stop the media player.""" """Stop the media player."""
if self.atv: if self.atv:
await self.atv.remote_control.stop() await self.atv.remote_control.stop()
async def async_media_pause(self): async def async_media_pause(self) -> None:
"""Pause the media player.""" """Pause the media player."""
if self.atv: if self.atv:
await self.atv.remote_control.pause() await self.atv.remote_control.pause()
async def async_media_next_track(self): async def async_media_next_track(self) -> None:
"""Send next track command.""" """Send next track command."""
if self.atv: if self.atv:
await self.atv.remote_control.next() await self.atv.remote_control.next()
async def async_media_previous_track(self): async def async_media_previous_track(self) -> None:
"""Send previous track command.""" """Send previous track command."""
if self.atv: if self.atv:
await self.atv.remote_control.previous() await self.atv.remote_control.previous()
async def async_media_seek(self, position): async def async_media_seek(self, position: float) -> None:
"""Send seek command.""" """Send seek command."""
if self.atv: if self.atv:
await self.atv.remote_control.set_position(position) await self.atv.remote_control.set_position(position)
async def async_volume_up(self): async def async_volume_up(self) -> None:
"""Turn volume up for media player.""" """Turn volume up for media player."""
if self.atv: if self.atv:
await self.atv.audio.volume_up() await self.atv.audio.volume_up()
async def async_volume_down(self): async def async_volume_down(self) -> None:
"""Turn volume down for media player.""" """Turn volume down for media player."""
if self.atv: if self.atv:
await self.atv.audio.volume_down() await self.atv.audio.volume_down()
async def async_set_volume_level(self, volume): async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.""" """Set volume level, range 0..1."""
if self.atv: if self.atv:
# pyatv expects volume in percent # pyatv expects volume in percent
await self.atv.audio.set_volume(volume * 100.0) await self.atv.audio.set_volume(volume * 100.0)
async def async_set_repeat(self, repeat): async def async_set_repeat(self, repeat: str) -> None:
"""Set repeat mode.""" """Set repeat mode."""
if self.atv: if self.atv:
mode = { mode = {
@ -500,7 +505,7 @@ class AppleTvMediaPlayer(AppleTVEntity, MediaPlayerEntity):
}.get(repeat, RepeatState.Off) }.get(repeat, RepeatState.Off)
await self.atv.remote_control.set_repeat(mode) await self.atv.remote_control.set_repeat(mode)
async def async_set_shuffle(self, shuffle): async def async_set_shuffle(self, shuffle: bool) -> None:
"""Enable/disable shuffle mode.""" """Enable/disable shuffle mode."""
if self.atv: if self.atv:
await self.atv.remote_control.set_shuffle( await self.atv.remote_control.set_shuffle(

View file

@ -1,6 +1,8 @@
"""Remote control support for Apple TV.""" """Remote control support for Apple TV."""
import asyncio import asyncio
from collections.abc import Iterable
import logging import logging
from typing import Any
from homeassistant.components.remote import ( from homeassistant.components.remote import (
ATTR_DELAY_SECS, ATTR_DELAY_SECS,
@ -40,15 +42,15 @@ class AppleTVRemote(AppleTVEntity, RemoteEntity):
"""Return true if device is on.""" """Return true if device is on."""
return self.atv is not None return self.atv is not None
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the device on.""" """Turn the device on."""
await self.manager.connect() await self.manager.connect()
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the device off.""" """Turn the device off."""
await self.manager.disconnect() await self.manager.disconnect()
async def async_send_command(self, command, **kwargs): async def async_send_command(self, command: Iterable[str], **kwargs: Any) -> None:
"""Send a command to one device.""" """Send a command to one device."""
num_repeats = kwargs[ATTR_NUM_REPEATS] num_repeats = kwargs[ATTR_NUM_REPEATS]
delay = kwargs.get(ATTR_DELAY_SECS, DEFAULT_DELAY_SECS) delay = kwargs.get(ATTR_DELAY_SECS, DEFAULT_DELAY_SECS)

View file

@ -144,7 +144,7 @@ class AquaLogicSensor(SensorEntity):
self._processor = processor self._processor = processor
self._attr_name = f"AquaLogic {description.name}" self._attr_name = f"AquaLogic {description.name}"
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect( async_dispatcher_connect(

View file

@ -1,6 +1,8 @@
"""Support for AquaLogic switches.""" """Support for AquaLogic switches."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from aqualogic.core import States from aqualogic.core import States
import voluptuous as vol import voluptuous as vol
@ -82,19 +84,19 @@ class AquaLogicSwitch(SwitchEntity):
state = panel.get_state(self._state_name) state = panel.get_state(self._state_name)
return state return state
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on.""" """Turn the device on."""
if (panel := self._processor.panel) is None: if (panel := self._processor.panel) is None:
return return
panel.set_state(self._state_name, True) panel.set_state(self._state_name, True)
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the device off.""" """Turn the device off."""
if (panel := self._processor.panel) is None: if (panel := self._processor.panel) is None:
return return
panel.set_state(self._state_name, False) panel.set_state(self._state_name, False)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect(self.hass, UPDATE_TOPIC, self.async_write_ha_state) async_dispatcher_connect(self.hass, UPDATE_TOPIC, self.async_write_ha_state)

View file

@ -131,7 +131,7 @@ class SharpAquosTVDevice(MediaPlayerEntity):
self._attr_state = state self._attr_state = state
@_retry @_retry
def update(self): def update(self) -> None:
"""Retrieve the latest data.""" """Retrieve the latest data."""
if self._remote.power() == 1: if self._remote.power() == 1:
self._attr_state = STATE_ON self._attr_state = STATE_ON
@ -153,7 +153,7 @@ class SharpAquosTVDevice(MediaPlayerEntity):
self._attr_volume_level = self._remote.volume() / 60 self._attr_volume_level = self._remote.volume() / 60
@_retry @_retry
def turn_off(self): def turn_off(self) -> None:
"""Turn off tvplayer.""" """Turn off tvplayer."""
self._remote.power(0) self._remote.power(0)
@ -168,46 +168,46 @@ class SharpAquosTVDevice(MediaPlayerEntity):
self._remote.volume(int(self.volume_level * 60) - 2) self._remote.volume(int(self.volume_level * 60) - 2)
@_retry @_retry
def set_volume_level(self, volume): def set_volume_level(self, volume: float) -> None:
"""Set Volume media player.""" """Set Volume media player."""
self._remote.volume(int(volume * 60)) self._remote.volume(int(volume * 60))
@_retry @_retry
def mute_volume(self, mute): def mute_volume(self, mute: bool) -> None:
"""Send mute command.""" """Send mute command."""
self._remote.mute(0) self._remote.mute(0)
@_retry @_retry
def turn_on(self): def turn_on(self) -> None:
"""Turn the media player on.""" """Turn the media player on."""
self._remote.power(1) self._remote.power(1)
@_retry @_retry
def media_play_pause(self): def media_play_pause(self) -> None:
"""Simulate play pause media player.""" """Simulate play pause media player."""
self._remote.remote_button(40) self._remote.remote_button(40)
@_retry @_retry
def media_play(self): def media_play(self) -> None:
"""Send play command.""" """Send play command."""
self._remote.remote_button(16) self._remote.remote_button(16)
@_retry @_retry
def media_pause(self): def media_pause(self) -> None:
"""Send pause command.""" """Send pause command."""
self._remote.remote_button(16) self._remote.remote_button(16)
@_retry @_retry
def media_next_track(self): def media_next_track(self) -> None:
"""Send next track command.""" """Send next track command."""
self._remote.remote_button(21) self._remote.remote_button(21)
@_retry @_retry
def media_previous_track(self): def media_previous_track(self) -> None:
"""Send the previous track command.""" """Send the previous track command."""
self._remote.remote_button(19) self._remote.remote_button(19)
def select_source(self, source): def select_source(self, source: str) -> None:
"""Set the input source.""" """Set the input source."""
for key, value in SOURCES.items(): for key, value in SOURCES.items():
if source == value: if source == value:

View file

@ -1,5 +1,8 @@
"""Arcam media player.""" """Arcam media player."""
from __future__ import annotations
import logging import logging
from typing import Any
from arcam.fmj import SourceCodes from arcam.fmj import SourceCodes
from arcam.fmj.state import State from arcam.fmj.state import State
@ -107,7 +110,7 @@ class ArcamFmj(MediaPlayerEntity):
name=self._device_name, name=self._device_name,
) )
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Once registered, add listener for events.""" """Once registered, add listener for events."""
await self._state.start() await self._state.start()
await self._state.update() await self._state.update()
@ -139,17 +142,17 @@ class ArcamFmj(MediaPlayerEntity):
async_dispatcher_connect(self.hass, SIGNAL_CLIENT_STOPPED, _stopped) async_dispatcher_connect(self.hass, SIGNAL_CLIENT_STOPPED, _stopped)
) )
async def async_update(self): async def async_update(self) -> None:
"""Force update of state.""" """Force update of state."""
_LOGGER.debug("Update state %s", self.name) _LOGGER.debug("Update state %s", self.name)
await self._state.update() await self._state.update()
async def async_mute_volume(self, mute): async def async_mute_volume(self, mute: bool) -> None:
"""Send mute command.""" """Send mute command."""
await self._state.set_mute(mute) await self._state.set_mute(mute)
self.async_write_ha_state() self.async_write_ha_state()
async def async_select_source(self, source): async def async_select_source(self, source: str) -> None:
"""Select a specific source.""" """Select a specific source."""
try: try:
value = SourceCodes[source] value = SourceCodes[source]
@ -160,7 +163,7 @@ class ArcamFmj(MediaPlayerEntity):
await self._state.set_source(value) await self._state.set_source(value)
self.async_write_ha_state() self.async_write_ha_state()
async def async_select_sound_mode(self, sound_mode): async def async_select_sound_mode(self, sound_mode: str) -> None:
"""Select a specific source.""" """Select a specific source."""
try: try:
await self._state.set_decode_mode(sound_mode) await self._state.set_decode_mode(sound_mode)
@ -170,22 +173,22 @@ class ArcamFmj(MediaPlayerEntity):
self.async_write_ha_state() self.async_write_ha_state()
async def async_set_volume_level(self, volume): async def async_set_volume_level(self, volume: float) -> None:
"""Set volume level, range 0..1.""" """Set volume level, range 0..1."""
await self._state.set_volume(round(volume * 99.0)) await self._state.set_volume(round(volume * 99.0))
self.async_write_ha_state() self.async_write_ha_state()
async def async_volume_up(self): async def async_volume_up(self) -> None:
"""Turn volume up for media player.""" """Turn volume up for media player."""
await self._state.inc_volume() await self._state.inc_volume()
self.async_write_ha_state() self.async_write_ha_state()
async def async_volume_down(self): async def async_volume_down(self) -> None:
"""Turn volume up for media player.""" """Turn volume up for media player."""
await self._state.dec_volume() await self._state.dec_volume()
self.async_write_ha_state() self.async_write_ha_state()
async def async_turn_on(self): async def async_turn_on(self) -> None:
"""Turn the media player on.""" """Turn the media player on."""
if self._state.get_power() is not None: if self._state.get_power() is not None:
_LOGGER.debug("Turning on device using connection") _LOGGER.debug("Turning on device using connection")
@ -194,11 +197,13 @@ class ArcamFmj(MediaPlayerEntity):
_LOGGER.debug("Firing event to turn on device") _LOGGER.debug("Firing event to turn on device")
self.hass.bus.async_fire(EVENT_TURN_ON, {ATTR_ENTITY_ID: self.entity_id}) 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.""" """Turn the media player off."""
await self._state.set_power(False) await self._state.set_power(False)
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."""
if media_content_id not in (None, "root"): if media_content_id not in (None, "root"):
raise BrowseError( raise BrowseError(
@ -231,7 +236,9 @@ class ArcamFmj(MediaPlayerEntity):
return root return root
async def async_play_media(self, media_type: str, media_id: str, **kwargs) -> None: async def async_play_media(
self, media_type: str, media_id: str, **kwargs: Any
) -> None:
"""Play media.""" """Play media."""
if media_id.startswith("preset:"): if media_id.startswith("preset:"):

View file

@ -86,7 +86,7 @@ class ArestBinarySensor(BinarySensorEntity):
if request.status_code != HTTPStatus.OK: if request.status_code != HTTPStatus.OK:
_LOGGER.error("Can't set mode of %s", resource) _LOGGER.error("Can't set mode of %s", resource)
def update(self): def update(self) -> None:
"""Get the latest data from aREST API.""" """Get the latest data from aREST API."""
self.arest.update() self.arest.update()
self._attr_is_on = bool(self.arest.data.get("state")) self._attr_is_on = bool(self.arest.data.get("state"))
@ -102,7 +102,7 @@ class ArestData:
self.data = {} self.data = {}
@Throttle(MIN_TIME_BETWEEN_UPDATES) @Throttle(MIN_TIME_BETWEEN_UPDATES)
def update(self): def update(self) -> None:
"""Get the latest data from aREST device.""" """Get the latest data from aREST device."""
try: try:
response = requests.get(f"{self._resource}/digital/{self._pin}", timeout=10) response = requests.get(f"{self._resource}/digital/{self._pin}", timeout=10)

View file

@ -159,7 +159,7 @@ class ArestSensor(SensorEntity):
if request.status_code != HTTPStatus.OK: if request.status_code != HTTPStatus.OK:
_LOGGER.error("Can't set mode of %s", resource) _LOGGER.error("Can't set mode of %s", resource)
def update(self): def update(self) -> None:
"""Get the latest data from aREST API.""" """Get the latest data from aREST API."""
self.arest.update() self.arest.update()
self._attr_available = self.arest.available self._attr_available = self.arest.available

View file

@ -3,6 +3,7 @@ from __future__ import annotations
from http import HTTPStatus from http import HTTPStatus
import logging import logging
from typing import Any
import requests import requests
import voluptuous as vol import voluptuous as vol
@ -122,7 +123,7 @@ class ArestSwitchFunction(ArestSwitchBase):
except ValueError: except ValueError:
_LOGGER.error("Response invalid") _LOGGER.error("Response invalid")
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on.""" """Turn the device on."""
request = requests.get( request = requests.get(
f"{self._resource}/{self._func}", timeout=10, params={"params": "1"} f"{self._resource}/{self._func}", timeout=10, params={"params": "1"}
@ -133,7 +134,7 @@ class ArestSwitchFunction(ArestSwitchBase):
else: else:
_LOGGER.error("Can't turn on function %s at %s", self._func, self._resource) _LOGGER.error("Can't turn on function %s at %s", self._func, self._resource)
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the device off.""" """Turn the device off."""
request = requests.get( request = requests.get(
f"{self._resource}/{self._func}", timeout=10, params={"params": "0"} f"{self._resource}/{self._func}", timeout=10, params={"params": "0"}
@ -146,7 +147,7 @@ class ArestSwitchFunction(ArestSwitchBase):
"Can't turn off function %s at %s", self._func, self._resource "Can't turn off function %s at %s", self._func, self._resource
) )
def update(self): def update(self) -> None:
"""Get the latest data from aREST API and update the state.""" """Get the latest data from aREST API and update the state."""
try: try:
request = requests.get(f"{self._resource}/{self._func}", timeout=10) request = requests.get(f"{self._resource}/{self._func}", timeout=10)
@ -171,7 +172,7 @@ class ArestSwitchPin(ArestSwitchBase):
_LOGGER.error("Can't set mode") _LOGGER.error("Can't set mode")
self._attr_available = False self._attr_available = False
def turn_on(self, **kwargs): def turn_on(self, **kwargs: Any) -> None:
"""Turn the device on.""" """Turn the device on."""
turn_on_payload = int(not self.invert) turn_on_payload = int(not self.invert)
request = requests.get( request = requests.get(
@ -182,7 +183,7 @@ class ArestSwitchPin(ArestSwitchBase):
else: else:
_LOGGER.error("Can't turn on pin %s at %s", self._pin, self._resource) _LOGGER.error("Can't turn on pin %s at %s", self._pin, self._resource)
def turn_off(self, **kwargs): def turn_off(self, **kwargs: Any) -> None:
"""Turn the device off.""" """Turn the device off."""
turn_off_payload = int(self.invert) turn_off_payload = int(self.invert)
request = requests.get( request = requests.get(
@ -193,7 +194,7 @@ class ArestSwitchPin(ArestSwitchBase):
else: else:
_LOGGER.error("Can't turn off pin %s at %s", self._pin, self._resource) _LOGGER.error("Can't turn off pin %s at %s", self._pin, self._resource)
def update(self): def update(self) -> None:
"""Get the latest data from aREST API and update the state.""" """Get the latest data from aREST API and update the state."""
try: try:
request = requests.get(f"{self._resource}/digital/{self._pin}", timeout=10) request = requests.get(f"{self._resource}/digital/{self._pin}", timeout=10)

View file

@ -1,6 +1,8 @@
"""Initialization of ATAG One climate platform.""" """Initialization of ATAG One climate platform."""
from __future__ import annotations from __future__ import annotations
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 (
PRESET_AWAY, PRESET_AWAY,
@ -78,12 +80,12 @@ class AtagThermostat(AtagEntity, ClimateEntity):
preset = self.coordinator.data.climate.preset_mode preset = self.coordinator.data.climate.preset_mode
return PRESET_INVERTED.get(preset) return PRESET_INVERTED.get(preset)
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.coordinator.data.climate.set_temp(kwargs.get(ATTR_TEMPERATURE)) await self.coordinator.data.climate.set_temp(kwargs.get(ATTR_TEMPERATURE))
self.async_write_ha_state() self.async_write_ha_state()
async def async_set_hvac_mode(self, hvac_mode: str) -> None: async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set new target hvac mode.""" """Set new target hvac mode."""
await self.coordinator.data.climate.set_hvac_mode(hvac_mode) await self.coordinator.data.climate.set_hvac_mode(hvac_mode)
self.async_write_ha_state() self.async_write_ha_state()

View file

@ -1,4 +1,6 @@
"""ATAG water heater component.""" """ATAG water heater component."""
from typing import Any
from homeassistant.components.water_heater import ( from homeassistant.components.water_heater import (
STATE_ECO, STATE_ECO,
STATE_PERFORMANCE, STATE_PERFORMANCE,
@ -42,7 +44,7 @@ class AtagWaterHeater(AtagEntity, WaterHeaterEntity):
operation = self.coordinator.data.dhw.current_operation operation = self.coordinator.data.dhw.current_operation
return operation if operation in self.operation_list else STATE_OFF return operation if operation in self.operation_list else STATE_OFF
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 await self.coordinator.data.dhw.set_temp(kwargs.get(ATTR_TEMPERATURE)): if await self.coordinator.data.dhw.set_temp(kwargs.get(ATTR_TEMPERATURE)):
self.async_write_ha_state() self.async_write_ha_state()
@ -53,11 +55,11 @@ class AtagWaterHeater(AtagEntity, WaterHeaterEntity):
return self.coordinator.data.dhw.target_temperature return self.coordinator.data.dhw.target_temperature
@property @property
def max_temp(self): def max_temp(self) -> float:
"""Return the maximum temperature.""" """Return the maximum temperature."""
return self.coordinator.data.dhw.max_temp return self.coordinator.data.dhw.max_temp
@property @property
def min_temp(self): def min_temp(self) -> float:
"""Return the minimum temperature.""" """Return the minimum temperature."""
return self.coordinator.data.dhw.min_temp return self.coordinator.data.dhw.min_temp

View file

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from atenpdu import AtenPE, AtenPEError from atenpdu import AtenPE, AtenPEError
import voluptuous as vol import voluptuous as vol
@ -101,17 +102,17 @@ class AtenSwitch(SwitchEntity):
self._attr_unique_id = f"{mac}-{outlet}" self._attr_unique_id = f"{mac}-{outlet}"
self._attr_name = name or f"Outlet {outlet}" self._attr_name = name or f"Outlet {outlet}"
async def async_turn_on(self, **kwargs): async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on.""" """Turn the switch on."""
await self._device.setOutletStatus(self._outlet, "on") await self._device.setOutletStatus(self._outlet, "on")
self._attr_is_on = True self._attr_is_on = True
async def async_turn_off(self, **kwargs): async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off.""" """Turn the switch off."""
await self._device.setOutletStatus(self._outlet, "off") await self._device.setOutletStatus(self._outlet, "off")
self._attr_is_on = False self._attr_is_on = False
async def async_update(self): async def async_update(self) -> None:
"""Process update from entity.""" """Process update from entity."""
status = await self._device.displayOutletStatus(self._outlet) status = await self._device.displayOutletStatus(self._outlet)
if status == "on": if status == "on":

View file

@ -269,7 +269,7 @@ class AtomeSensor(SensorEntity):
self._attr_native_unit_of_measurement = ENERGY_KILO_WATT_HOUR self._attr_native_unit_of_measurement = ENERGY_KILO_WATT_HOUR
self._attr_state_class = SensorStateClass.TOTAL_INCREASING self._attr_state_class = SensorStateClass.TOTAL_INCREASING
def update(self): def update(self) -> None:
"""Update device state.""" """Update device state."""
update_function = getattr(self._data, f"update_{self._sensor_type}_usage") update_function = getattr(self._data, f"update_{self._sensor_type}_usage")
update_function() update_function()

View file

@ -288,7 +288,7 @@ class AugustDoorbellBinarySensor(AugustEntityMixin, BinarySensorEntity):
self._check_for_off_update_listener() self._check_for_off_update_listener()
self._check_for_off_update_listener = None self._check_for_off_update_listener = None
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Call the mixin to subscribe and setup an async_track_point_in_utc_time to turn off the sensor if needed.""" """Call the mixin to subscribe and setup an async_track_point_in_utc_time to turn off the sensor if needed."""
self._schedule_update_to_recheck_turn_off_sensor() self._schedule_update_to_recheck_turn_off_sensor()
await super().async_added_to_hass() await super().async_added_to_hass()

View file

@ -43,12 +43,12 @@ class AugustCamera(AugustEntityMixin, Camera):
self._attr_unique_id = f"{self._device_id:s}_camera" self._attr_unique_id = f"{self._device_id:s}_camera"
@property @property
def is_recording(self): def is_recording(self) -> bool:
"""Return true if the device is recording.""" """Return true if the device is recording."""
return self._device.has_subscription return self._device.has_subscription
@property @property
def motion_detection_enabled(self): def motion_detection_enabled(self) -> bool:
"""Return the camera motion detection status.""" """Return the camera motion detection status."""
return True return True

View file

@ -82,7 +82,7 @@ class AuroraSensor(AuroraEntity, SensorEntity):
self.entity_description = entity_description self.entity_description = entity_description
self.available_prev = True self.available_prev = True
def update(self): def update(self) -> None:
"""Fetch new state data for the sensor. """Fetch new state data for the sensor.
This is the only method that should fetch new data for Home Assistant. This is the only method that should fetch new data for Home Assistant.