Improve type hints in base entities (#138708)

This commit is contained in:
epenet 2025-02-17 14:45:00 +01:00 committed by GitHub
parent 4a385ed26c
commit d8d054e7dd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
31 changed files with 54 additions and 54 deletions

View File

@ -17,13 +17,13 @@ class BroadlinkEntity(Entity):
self._device = device self._device = device
self._coordinator = device.update_manager.coordinator self._coordinator = device.update_manager.coordinator
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Call when the entity is added to hass.""" """Call when the entity is added to hass."""
self.async_on_remove(self._coordinator.async_add_listener(self._recv_data)) self.async_on_remove(self._coordinator.async_add_listener(self._recv_data))
if self._coordinator.data: if self._coordinator.data:
self._update_state(self._coordinator.data) self._update_state(self._coordinator.data)
async def async_update(self): async def async_update(self) -> None:
"""Update the state of the entity.""" """Update the state of the entity."""
await self._coordinator.async_request_refresh() await self._coordinator.async_request_refresh()
@ -49,7 +49,7 @@ class BroadlinkEntity(Entity):
""" """
@property @property
def available(self): def available(self) -> bool:
"""Return True if the entity is available.""" """Return True if the entity is available."""
return self._device.available return self._device.available

View File

@ -16,7 +16,7 @@ class EnOceanEntity(Entity):
"""Initialize the device.""" """Initialize the device."""
self.dev_id = dev_id self.dev_id = dev_id
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

@ -45,10 +45,10 @@ class FloEntity(Entity):
"""Return True if device is available.""" """Return True if device is available."""
return self._device.available return self._device.available
async def async_update(self): async def async_update(self) -> None:
"""Update Flo entity.""" """Update Flo entity."""
await self._device.async_request_refresh() await self._device.async_request_refresh()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""When entity is added to hass.""" """When entity is added to hass."""
self.async_on_remove(self._device.async_add_listener(self.async_write_ha_state)) self.async_on_remove(self._device.async_add_listener(self.async_write_ha_state))

View File

@ -35,7 +35,7 @@ class SW16Entity(Entity):
self.async_write_ha_state() self.async_write_ha_state()
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return bool(self._client.is_connected) return bool(self._client.is_connected)
@ -44,7 +44,7 @@ class SW16Entity(Entity):
"""Update availability state.""" """Update availability state."""
self.async_write_ha_state() self.async_write_ha_state()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register update callback.""" """Register update callback."""
self._client.register_status_callback( self._client.register_status_callback(
self.handle_event_callback, self._device_port self.handle_event_callback, self._device_port

View File

@ -62,7 +62,7 @@ class HMDevice(Entity):
if self._state: if self._state:
self._state = self._state.upper() self._state = self._state.upper()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Load data init callbacks.""" """Load data init callbacks."""
self._subscribe_homematic_events() self._subscribe_homematic_events()
@ -77,7 +77,7 @@ class HMDevice(Entity):
return self._name return self._name
@property @property
def available(self): def available(self) -> bool:
"""Return true if device is available.""" """Return true if device is available."""
return self._available return self._available

View File

@ -54,7 +54,7 @@ class IHCEntity(Entity):
self.ihc_note = "" self.ihc_note = ""
self.ihc_position = "" self.ihc_position = ""
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Add callback for IHC changes.""" """Add callback for IHC changes."""
_LOGGER.debug("Adding IHC entity notify event: %s", self.ihc_id) _LOGGER.debug("Adding IHC entity notify event: %s", self.ihc_id)
self.ihc_controller.add_notify_event(self.ihc_id, self.on_ihc_change, True) self.ihc_controller.add_notify_event(self.ihc_id, self.on_ihc_change, True)

View File

@ -109,7 +109,7 @@ class InsteonEntity(Entity):
) )
self.async_write_ha_state() self.async_write_ha_state()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register INSTEON update events.""" """Register INSTEON update events."""
_LOGGER.debug( _LOGGER.debug(
"Tracking updates for device %s group %d name %s", "Tracking updates for device %s group %d name %s",
@ -137,7 +137,7 @@ class InsteonEntity(Entity):
) )
) )
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Unsubscribe to INSTEON update events.""" """Unsubscribe to INSTEON update events."""
_LOGGER.debug( _LOGGER.debug(
"Remove tracking updates for device %s group %d name %s", "Remove tracking updates for device %s group %d name %s",

View File

@ -18,7 +18,7 @@ class LupusecDevice(Entity):
self._device = device self._device = device
self._attr_unique_id = device.device_id self._attr_unique_id = device.device_id
def update(self): def update(self) -> None:
"""Update automation state.""" """Update automation state."""
self._device.refresh() self._device.refresh()

View File

@ -63,7 +63,7 @@ class LutronCasetaEntity(Entity):
info[ATTR_SUGGESTED_AREA] = area info[ATTR_SUGGESTED_AREA] = area
self._attr_device_info = info self._attr_device_info = info
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callbacks.""" """Register callbacks."""
self._smartbridge.add_subscriber(self.device_id, self.async_write_ha_state) self._smartbridge.add_subscriber(self.device_id, self.async_write_ha_state)

View File

@ -17,7 +17,7 @@ class ONVIFBaseEntity(Entity):
self.device: ONVIFDevice = device self.device: ONVIFDevice = device
@property @property
def available(self): def available(self) -> bool:
"""Return True if device is available.""" """Return True if device is available."""
return self.device.available return self.device.available

View File

@ -86,7 +86,7 @@ class PilightBaseDevice(RestoreEntity):
self._brightness = 255 self._brightness = 255
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Call when entity about to be added to hass.""" """Call when entity about to be added to hass."""
await super().async_added_to_hass() await super().async_added_to_hass()
if state := await self.async_get_last_state(): if state := await self.async_get_last_state():
@ -99,7 +99,7 @@ class PilightBaseDevice(RestoreEntity):
return self._name return self._name
@property @property
def assumed_state(self): def assumed_state(self) -> bool:
"""Return True if unable to access real state of the entity.""" """Return True if unable to access real state of the entity."""
return True return True

View File

@ -73,13 +73,13 @@ class PlaatoEntity(entity.Entity):
return None return None
@property @property
def available(self): def available(self) -> bool:
"""Return if sensor is available.""" """Return if sensor is available."""
if self._coordinator is not None: if self._coordinator is not None:
return self._coordinator.last_update_success return self._coordinator.last_update_success
return True return True
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""When entity is added to hass.""" """When entity is added to hass."""
if self._coordinator is not None: if self._coordinator is not None:
self.async_on_remove( self.async_on_remove(

View File

@ -52,7 +52,7 @@ class MinutPointEntity(Entity):
) )
await self._update_callback() await self._update_callback()
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Disconnect dispatcher listener when removed.""" """Disconnect dispatcher listener when removed."""
if self._async_unsub_dispatcher_connect: if self._async_unsub_dispatcher_connect:
self._async_unsub_dispatcher_connect() self._async_unsub_dispatcher_connect()
@ -61,7 +61,7 @@ class MinutPointEntity(Entity):
"""Update the value of the sensor.""" """Update the value of the sensor."""
@property @property
def available(self): def available(self) -> bool:
"""Return true if device is not offline.""" """Return true if device is not offline."""
return self._client.is_available(self.device_id) return self._client.is_available(self.device_id)

View File

@ -35,7 +35,7 @@ class QSEntity(Entity):
"""Receive update packet from QSUSB. Match dispather_send signature.""" """Receive update packet from QSUSB. Match dispather_send signature."""
self.async_write_ha_state() self.async_write_ha_state()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Listen for updates from QSUSb via dispatcher.""" """Listen for updates from QSUSb via dispatcher."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect(self.hass, self.qsid, self.update_packet) async_dispatcher_connect(self.hass, self.qsid, self.update_packet)

View File

@ -45,7 +45,7 @@ class RainCloudEntity(Entity):
"""Return the name of the sensor.""" """Return the name of the sensor."""
return self._name return self._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

@ -105,12 +105,12 @@ class RflinkDevice(Entity):
return self._state return self._state
@property @property
def assumed_state(self): def assumed_state(self) -> bool:
"""Assume device state until first device event sets state.""" """Assume device state until first device event sets state."""
return self._state is None return self._state is None
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return self._available return self._available
@ -120,7 +120,7 @@ class RflinkDevice(Entity):
self._available = availability self._available = availability
self.async_write_ha_state() self.async_write_ha_state()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register update callback.""" """Register update callback."""
await super().async_added_to_hass() await super().async_added_to_hass()
# Remove temporary bogus entity_id if added # Remove temporary bogus entity_id if added
@ -300,7 +300,7 @@ class RflinkCommand(RflinkDevice):
class SwitchableRflinkDevice(RflinkCommand, RestoreEntity): class SwitchableRflinkDevice(RflinkCommand, RestoreEntity):
"""Rflink entity which can switch on/off (eg: light, switch).""" """Rflink entity which can switch on/off (eg: light, switch)."""
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Restore RFLink device state (ON/OFF).""" """Restore RFLink device state (ON/OFF)."""
await super().async_added_to_hass() await super().async_added_to_hass()
if (old_state := await self.async_get_last_state()) is not None: if (old_state := await self.async_get_last_state()) is not None:

View File

@ -80,7 +80,7 @@ class IRobotEntity(Entity):
return None return None
return dt_util.utc_from_timestamp(ts) return dt_util.utc_from_timestamp(ts)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callback function.""" """Register callback function."""
self.vacuum.register_on_message_callback(self.on_message) self.vacuum.register_on_message_callback(self.on_message)

View File

@ -71,7 +71,7 @@ class SomaEntity(Entity):
self.api_is_available = True self.api_is_available = True
@property @property
def available(self): def available(self) -> bool:
"""Return true if the last API commands returned successfully.""" """Return true if the last API commands returned successfully."""
return self.is_available return self.is_available

View File

@ -27,20 +27,20 @@ class StarlineEntity(Entity):
self._unsubscribe_api: Callable | None = None self._unsubscribe_api: Callable | None = None
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return self._account.api.available return self._account.api.available
def update(self): def update(self) -> None:
"""Read new state data.""" """Read new state data."""
self.schedule_update_ha_state() self.schedule_update_ha_state()
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._unsubscribe_api = self._account.api.add_update_listener(self.update) self._unsubscribe_api = self._account.api.add_update_listener(self.update)
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Call when entity is being removed from Home Assistant.""" """Call when entity is being removed from Home Assistant."""
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()
if self._unsubscribe_api is not None: if self._unsubscribe_api is not None:

View File

@ -33,7 +33,7 @@ class TelldusLiveEntity(Entity):
self._id = device_id self._id = device_id
self._client = client self._client = client
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Call when entity is added to hass.""" """Call when entity is added to hass."""
_LOGGER.debug("Created device %s", self) _LOGGER.debug("Created device %s", self)
self.async_on_remove( self.async_on_remove(
@ -58,12 +58,12 @@ class TelldusLiveEntity(Entity):
return self.device.state return self.device.state
@property @property
def assumed_state(self): def assumed_state(self) -> bool:
"""Return true if unable to access real state of entity.""" """Return true if unable to access real state of entity."""
return True return True
@property @property
def available(self): def available(self) -> bool:
"""Return true if device is not offline.""" """Return true if device is not offline."""
return self._client.is_available(self.device_id) return self._client.is_available(self.device_id)

View File

@ -40,7 +40,7 @@ class TellstickDevice(Entity):
self._attr_name = tellcore_device.name self._attr_name = tellcore_device.name
self._attr_unique_id = tellcore_device.id self._attr_unique_id = tellcore_device.id
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(
@ -146,6 +146,6 @@ class TellstickDevice(Entity):
except TelldusError as err: except TelldusError as err:
_LOGGER.error(err) _LOGGER.error(err)
def update(self): def update(self) -> None:
"""Poll the current state of the device.""" """Poll the current state of the device."""
self._update_from_tellcore() self._update_from_tellcore()

View File

@ -30,7 +30,7 @@ class UpbEntity(Entity):
return self._element.as_dict() return self._element.as_dict()
@property @property
def available(self): def available(self) -> bool:
"""Is the entity available to be updated.""" """Is the entity available to be updated."""
return self._upb.is_connected() return self._upb.is_connected()
@ -43,7 +43,7 @@ class UpbEntity(Entity):
self._element_changed(element, changeset) self._element_changed(element, changeset)
self.async_write_ha_state() self.async_write_ha_state()
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Register callback for UPB changes and update entity state.""" """Register callback for UPB changes and update entity state."""
self._element.add_callback(self._element_callback) self._element.add_callback(self._element_callback)
self._element_callback(self._element, {}) self._element_callback(self._element, {})

View File

@ -31,6 +31,6 @@ class VeluxEntity(Entity):
self.node.register_device_updated_cb(after_update_callback) self.node.register_device_updated_cb(after_update_callback)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Store register state change callback.""" """Store register state change callback."""
self.async_register_callbacks() self.async_register_callbacks()

View File

@ -52,7 +52,7 @@ class VeraEntity[_DeviceTypeT: veraApi.VeraDevice](Entity):
"""Update the state.""" """Update the state."""
self.schedule_update_ha_state(True) self.schedule_update_ha_state(True)
def update(self): def update(self) -> None:
"""Force a refresh from the device if the device is unavailable.""" """Force a refresh from the device if the device is unavailable."""
refresh_needed = self.vera_device.should_poll or not self.available refresh_needed = self.vera_device.should_poll or not self.available
_LOGGER.debug("%s: update called (refresh=%s)", self._name, refresh_needed) _LOGGER.debug("%s: update called (refresh=%s)", self._name, refresh_needed)
@ -90,7 +90,7 @@ class VeraEntity[_DeviceTypeT: veraApi.VeraDevice](Entity):
return attr return attr
@property @property
def available(self): def available(self) -> bool:
"""If device communications have failed return false.""" """If device communications have failed return false."""
return not self.vera_device.comm_failure return not self.vera_device.comm_failure

View File

@ -57,7 +57,7 @@ class VolvoEntity(CoordinatorEntity[VolvoUpdateCoordinator]):
return f"{self._vehicle_name} {self._entity_name}" return f"{self._vehicle_name} {self._entity_name}"
@property @property
def assumed_state(self): def assumed_state(self) -> bool:
"""Return true if unable to access real state of entity.""" """Return true if unable to access real state of entity."""
return True return True

View File

@ -41,7 +41,7 @@ class WiffiEntity(Entity):
self._value = None self._value = None
self._timeout = options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT) self._timeout = options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT)
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Entity has been added to hass.""" """Entity has been added to hass."""
self.async_on_remove( self.async_on_remove(
async_dispatcher_connect( async_dispatcher_connect(

View File

@ -60,11 +60,11 @@ class WirelessTagBaseSensor(Entity):
return f"{value:.1f}" return f"{value:.1f}"
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return self._tag.is_alive return self._tag.is_alive
def update(self): def update(self) -> None:
"""Update state.""" """Update state."""
if not self.should_poll: if not self.should_poll:
return return

View File

@ -57,7 +57,7 @@ class XiaomiDevice(Entity):
self._is_gateway = False self._is_gateway = False
self._device_id = self._sid self._device_id = self._sid
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Start unavailability tracking.""" """Start unavailability tracking."""
self._xiaomi_hub.callbacks[self._sid].append(self.push_data) self._xiaomi_hub.callbacks[self._sid].append(self.push_data)
self._async_track_unavailable() self._async_track_unavailable()
@ -100,7 +100,7 @@ class XiaomiDevice(Entity):
return device_info return device_info
@property @property
def available(self): def available(self) -> bool:
"""Return True if entity is available.""" """Return True if entity is available."""
return self._is_available return self._is_available

View File

@ -185,7 +185,7 @@ class XiaomiGatewayDevice(CoordinatorEntity, Entity):
) )
@property @property
def available(self): def available(self) -> bool:
"""Return if entity is available.""" """Return if entity is available."""
if self.coordinator.data is None: if self.coordinator.data is None:
return False return False

View File

@ -17,7 +17,7 @@ class XS1DeviceEntity(Entity):
"""Initialize the XS1 device.""" """Initialize the XS1 device."""
self.device = device self.device = device
async def async_update(self): async def async_update(self) -> None:
"""Retrieve latest device state.""" """Retrieve latest device state."""
async with UPDATE_LOCK: async with UPDATE_LOCK:
await self.hass.async_add_executor_job(self.device.update) await self.hass.async_add_executor_job(self.device.update)

View File

@ -78,13 +78,13 @@ class MusicCastDeviceEntity(MusicCastEntity):
return device_info return device_info
async def async_added_to_hass(self): async def async_added_to_hass(self) -> None:
"""Run when this Entity has been added to HA.""" """Run when this Entity has been added to HA."""
await super().async_added_to_hass() await super().async_added_to_hass()
# All entities should register callbacks to update HA when their state changes # All entities should register callbacks to update HA when their state changes
self.coordinator.musiccast.register_callback(self.async_write_ha_state) self.coordinator.musiccast.register_callback(self.async_write_ha_state)
async def async_will_remove_from_hass(self): async def async_will_remove_from_hass(self) -> None:
"""Entity being removed from hass.""" """Entity being removed from hass."""
await super().async_will_remove_from_hass() await super().async_will_remove_from_hass()
self.coordinator.musiccast.remove_callback(self.async_write_ha_state) self.coordinator.musiccast.remove_callback(self.async_write_ha_state)