mirror of
https://github.com/home-assistant/core.git
synced 2025-07-08 13:57:10 +00:00
Small quality fixes for ESPHome (#143535)
This commit is contained in:
parent
e8c4d08b25
commit
4173ff5339
@ -73,7 +73,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ESPHomeConfigEntry) -> b
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ESPHomeConfigEntry) -> bool:
|
||||
"""Unload an esphome config entry."""
|
||||
entry_data = await cleanup_instance(hass, entry)
|
||||
entry_data = await cleanup_instance(entry)
|
||||
return await hass.config_entries.async_unload_platforms(
|
||||
entry, entry_data.loaded_platforms
|
||||
)
|
||||
|
@ -50,7 +50,7 @@ _ESPHOME_ACP_STATE_TO_HASS_STATE: EsphomeEnumMapper[
|
||||
|
||||
|
||||
class EspHomeACPFeatures(APIIntEnum):
|
||||
"""ESPHome AlarmCintolPanel feature numbers."""
|
||||
"""ESPHome AlarmControlPanel feature numbers."""
|
||||
|
||||
ARM_HOME = 1
|
||||
ARM_AWAY = 2
|
||||
|
@ -35,14 +35,13 @@ from homeassistant.components.intent import (
|
||||
async_register_timer_handler,
|
||||
)
|
||||
from homeassistant.components.media_player import async_process_play_media_url
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .const import DOMAIN
|
||||
from .entity import EsphomeAssistEntity
|
||||
from .entity import EsphomeAssistEntity, convert_api_error_ha_error
|
||||
from .entry_data import ESPHomeConfigEntry, RuntimeEntryData
|
||||
from .enum_mapper import EsphomeEnumMapper
|
||||
from .ffmpeg_proxy import async_create_proxy_url
|
||||
@ -111,7 +110,7 @@ class EsphomeAssistSatellite(
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config_entry: ConfigEntry,
|
||||
config_entry: ESPHomeConfigEntry,
|
||||
entry_data: RuntimeEntryData,
|
||||
) -> None:
|
||||
"""Initialize satellite."""
|
||||
@ -349,6 +348,7 @@ class EsphomeAssistSatellite(
|
||||
|
||||
self.cli.send_voice_assistant_event(event_type, data_to_send)
|
||||
|
||||
@convert_api_error_ha_error
|
||||
async def async_announce(
|
||||
self, announcement: assist_satellite.AssistSatelliteAnnouncement
|
||||
) -> None:
|
||||
@ -358,6 +358,7 @@ class EsphomeAssistSatellite(
|
||||
"""
|
||||
await self._do_announce(announcement, run_pipeline_after=False)
|
||||
|
||||
@convert_api_error_ha_error
|
||||
async def async_start_conversation(
|
||||
self, start_announcement: assist_satellite.AssistSatelliteAnnouncement
|
||||
) -> None:
|
||||
|
@ -180,13 +180,13 @@ class EsphomeClimateEntity(EsphomeEntity[ClimateInfo, ClimateState], ClimateEnti
|
||||
|
||||
def _get_precision(self) -> float:
|
||||
"""Return the precision of the climate device."""
|
||||
precicions = [PRECISION_WHOLE, PRECISION_HALVES, PRECISION_TENTHS]
|
||||
precisions = [PRECISION_WHOLE, PRECISION_HALVES, PRECISION_TENTHS]
|
||||
static_info = self._static_info
|
||||
if static_info.visual_current_temperature_step != 0:
|
||||
step = static_info.visual_current_temperature_step
|
||||
else:
|
||||
step = static_info.visual_target_temperature_step
|
||||
for prec in precicions:
|
||||
for prec in precisions:
|
||||
if step >= prec:
|
||||
return prec
|
||||
# Fall back to highest precision, tenths
|
||||
|
@ -17,15 +17,12 @@ STORAGE_VERSION = 1
|
||||
|
||||
@dataclass(slots=True)
|
||||
class DomainData:
|
||||
"""Define a class that stores global esphome data in hass.data[DOMAIN]."""
|
||||
"""Define a class that stores global esphome data."""
|
||||
|
||||
_stores: dict[str, ESPHomeStorage] = field(default_factory=dict)
|
||||
|
||||
def get_entry_data(self, entry: ESPHomeConfigEntry) -> RuntimeEntryData:
|
||||
"""Return the runtime entry data associated with this config entry.
|
||||
|
||||
Raises KeyError if the entry isn't loaded yet.
|
||||
"""
|
||||
"""Return the runtime entry data associated with this config entry."""
|
||||
return entry.runtime_data
|
||||
|
||||
def get_or_create_store(
|
||||
|
@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any, Concatenate, Generic, TypeVar, cast
|
||||
|
||||
from aioesphomeapi import (
|
||||
APIConnectionError,
|
||||
DeviceInfo as EsphomeDeviceInfo,
|
||||
EntityCategory as EsphomeEntityCategory,
|
||||
EntityInfo,
|
||||
EntityState,
|
||||
@ -155,7 +156,7 @@ def esphome_float_state_property[_EntityT: EsphomeEntity[Any, Any]](
|
||||
return _wrapper
|
||||
|
||||
|
||||
def convert_api_error_ha_error[**_P, _R, _EntityT: EsphomeEntity[Any, Any]](
|
||||
def convert_api_error_ha_error[**_P, _R, _EntityT: EsphomeBaseEntity](
|
||||
func: Callable[Concatenate[_EntityT, _P], Awaitable[None]],
|
||||
) -> Callable[Concatenate[_EntityT, _P], Coroutine[Any, Any, None]]:
|
||||
"""Decorate ESPHome command calls that send commands/make changes to the device.
|
||||
@ -194,15 +195,21 @@ ENTITY_CATEGORIES: EsphomeEnumMapper[EsphomeEntityCategory, EntityCategory | Non
|
||||
)
|
||||
|
||||
|
||||
class EsphomeEntity(Entity, Generic[_InfoT, _StateT]):
|
||||
class EsphomeBaseEntity(Entity):
|
||||
"""Define a base esphome entity."""
|
||||
|
||||
_attr_should_poll = False
|
||||
_attr_has_entity_name = True
|
||||
_attr_should_poll = False
|
||||
_device_info: EsphomeDeviceInfo
|
||||
device_entry: dr.DeviceEntry
|
||||
|
||||
|
||||
class EsphomeEntity(EsphomeBaseEntity, Generic[_InfoT, _StateT]):
|
||||
"""Define an esphome entity."""
|
||||
|
||||
_static_info: _InfoT
|
||||
_state: _StateT
|
||||
_has_state: bool
|
||||
device_entry: dr.DeviceEntry
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@ -325,15 +332,12 @@ class EsphomeEntity(Entity, Generic[_InfoT, _StateT]):
|
||||
self.async_write_ha_state()
|
||||
|
||||
|
||||
class EsphomeAssistEntity(Entity):
|
||||
class EsphomeAssistEntity(EsphomeBaseEntity):
|
||||
"""Define a base entity for Assist Pipeline entities."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
_attr_should_poll = False
|
||||
|
||||
def __init__(self, entry_data: RuntimeEntryData) -> None:
|
||||
"""Initialize the binary sensor."""
|
||||
self._entry_data: RuntimeEntryData = entry_data
|
||||
self._entry_data = entry_data
|
||||
assert entry_data.device_info is not None
|
||||
device_info = entry_data.device_info
|
||||
self._device_info = device_info
|
||||
|
@ -106,7 +106,7 @@ class EsphomeFan(EsphomeEntity[FanInfo, FanState], FanEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def is_on(self) -> bool | None:
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if the entity is on."""
|
||||
return self._state.state
|
||||
|
||||
@ -126,7 +126,7 @@ class EsphomeFan(EsphomeEntity[FanInfo, FanState], FanEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def oscillating(self) -> bool | None:
|
||||
def oscillating(self) -> bool:
|
||||
"""Return the oscillation state."""
|
||||
return self._state.oscillating
|
||||
|
||||
@ -138,7 +138,7 @@ class EsphomeFan(EsphomeEntity[FanInfo, FanState], FanEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def preset_mode(self) -> str | None:
|
||||
def preset_mode(self) -> str:
|
||||
"""Return the current fan preset mode."""
|
||||
return self._state.preset_mode
|
||||
|
||||
|
@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import lru_cache, partial
|
||||
from operator import methodcaller
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
from aioesphomeapi import (
|
||||
@ -108,7 +109,7 @@ def _mired_to_kelvin(mired_temperature: float) -> int:
|
||||
def _color_mode_to_ha(mode: int) -> str:
|
||||
"""Convert an esphome color mode to a HA color mode constant.
|
||||
|
||||
Choses the color mode that best matches the feature-set.
|
||||
Chose the color mode that best matches the feature-set.
|
||||
"""
|
||||
candidates = []
|
||||
for ha_mode, cap_lists in _COLOR_MODE_MAPPING.items():
|
||||
@ -148,7 +149,7 @@ def _least_complex_color_mode(color_modes: tuple[int, ...]) -> int:
|
||||
# popcount with bin() function because it appears
|
||||
# to be the best way: https://stackoverflow.com/a/9831671
|
||||
color_modes_list = list(color_modes)
|
||||
color_modes_list.sort(key=lambda mode: (mode).bit_count())
|
||||
color_modes_list.sort(key=methodcaller("bit_count"))
|
||||
return color_modes_list[0]
|
||||
|
||||
|
||||
@ -160,7 +161,7 @@ class EsphomeLight(EsphomeEntity[LightInfo, LightState], LightEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def is_on(self) -> bool | None:
|
||||
def is_on(self) -> bool:
|
||||
"""Return true if the light is on."""
|
||||
return self._state.state
|
||||
|
||||
@ -292,13 +293,13 @@ class EsphomeLight(EsphomeEntity[LightInfo, LightState], LightEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def brightness(self) -> int | None:
|
||||
def brightness(self) -> int:
|
||||
"""Return the brightness of this light between 0..255."""
|
||||
return round(self._state.brightness * 255)
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def color_mode(self) -> str | None:
|
||||
def color_mode(self) -> str:
|
||||
"""Return the color mode of the light."""
|
||||
if not self._supports_color_mode:
|
||||
supported_color_modes = self.supported_color_modes
|
||||
@ -310,7 +311,7 @@ class EsphomeLight(EsphomeEntity[LightInfo, LightState], LightEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def rgb_color(self) -> tuple[int, int, int] | None:
|
||||
def rgb_color(self) -> tuple[int, int, int]:
|
||||
"""Return the rgb color value [int, int, int]."""
|
||||
state = self._state
|
||||
if not self._supports_color_mode:
|
||||
@ -328,7 +329,7 @@ class EsphomeLight(EsphomeEntity[LightInfo, LightState], LightEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def rgbw_color(self) -> tuple[int, int, int, int] | None:
|
||||
def rgbw_color(self) -> tuple[int, int, int, int]:
|
||||
"""Return the rgbw color value [int, int, int, int]."""
|
||||
white = round(self._state.white * 255)
|
||||
rgb = cast("tuple[int, int, int]", self.rgb_color)
|
||||
@ -336,7 +337,7 @@ class EsphomeLight(EsphomeEntity[LightInfo, LightState], LightEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def rgbww_color(self) -> tuple[int, int, int, int, int] | None:
|
||||
def rgbww_color(self) -> tuple[int, int, int, int, int]:
|
||||
"""Return the rgbww color value [int, int, int, int, int]."""
|
||||
state = self._state
|
||||
rgb = cast("tuple[int, int, int]", self.rgb_color)
|
||||
@ -372,7 +373,7 @@ class EsphomeLight(EsphomeEntity[LightInfo, LightState], LightEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def effect(self) -> str | None:
|
||||
def effect(self) -> str:
|
||||
"""Return the current effect."""
|
||||
return self._state.effect
|
||||
|
||||
|
@ -40,25 +40,25 @@ class EsphomeLock(EsphomeEntity[LockInfo, LockEntityState], LockEntity):
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def is_locked(self) -> bool | None:
|
||||
def is_locked(self) -> bool:
|
||||
"""Return true if the lock is locked."""
|
||||
return self._state.state is LockState.LOCKED
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def is_locking(self) -> bool | None:
|
||||
def is_locking(self) -> bool:
|
||||
"""Return true if the lock is locking."""
|
||||
return self._state.state is LockState.LOCKING
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def is_unlocking(self) -> bool | None:
|
||||
def is_unlocking(self) -> bool:
|
||||
"""Return true if the lock is unlocking."""
|
||||
return self._state.state is LockState.UNLOCKING
|
||||
|
||||
@property
|
||||
@esphome_state_property
|
||||
def is_jammed(self) -> bool | None:
|
||||
def is_jammed(self) -> bool:
|
||||
"""Return true if the lock is jammed (incomplete locking)."""
|
||||
return self._state.state is LockState.JAMMED
|
||||
|
||||
|
@ -215,7 +215,7 @@ class ESPHomeManager:
|
||||
|
||||
async def on_stop(self, event: Event) -> None:
|
||||
"""Cleanup the socket client on HA close."""
|
||||
await cleanup_instance(self.hass, self.entry)
|
||||
await cleanup_instance(self.entry)
|
||||
|
||||
@property
|
||||
def services_issue(self) -> str:
|
||||
@ -376,7 +376,7 @@ class ESPHomeManager:
|
||||
async def on_connect(self) -> None:
|
||||
"""Subscribe to states and list entities on successful API login."""
|
||||
try:
|
||||
await self._on_connnect()
|
||||
await self._on_connect()
|
||||
except APIConnectionError as err:
|
||||
_LOGGER.warning(
|
||||
"Error getting setting up connection for %s: %s", self.host, err
|
||||
@ -412,7 +412,7 @@ class ESPHomeManager:
|
||||
self._async_on_log, self._log_level
|
||||
)
|
||||
|
||||
async def _on_connnect(self) -> None:
|
||||
async def _on_connect(self) -> None:
|
||||
"""Subscribe to states and list entities on successful API login."""
|
||||
entry = self.entry
|
||||
unique_id = entry.unique_id
|
||||
@ -939,9 +939,7 @@ def _setup_services(
|
||||
_async_register_service(hass, entry_data, device_info, service)
|
||||
|
||||
|
||||
async def cleanup_instance(
|
||||
hass: HomeAssistant, entry: ESPHomeConfigEntry
|
||||
) -> RuntimeEntryData:
|
||||
async def cleanup_instance(entry: ESPHomeConfigEntry) -> RuntimeEntryData:
|
||||
"""Cleanup the esphome client if it exists."""
|
||||
data = entry.runtime_data
|
||||
data.async_on_disconnect()
|
||||
|
Loading…
x
Reference in New Issue
Block a user