Compare commits

..

1 Commits

Author SHA1 Message Date
epenet
67c6d4011e Use external library wrapper in Tuya fan 2026-03-13 15:07:11 +00:00
198 changed files with 843 additions and 11580 deletions

View File

@@ -570,7 +570,6 @@ homeassistant.components.trafikverket_train.*
homeassistant.components.trafikverket_weatherstation.*
homeassistant.components.transmission.*
homeassistant.components.trend.*
homeassistant.components.trmnl.*
homeassistant.components.tts.*
homeassistant.components.twentemilieu.*
homeassistant.components.unifi.*

6
CODEOWNERS generated
View File

@@ -186,8 +186,6 @@ build.json @home-assistant/supervisor
/tests/components/auth/ @home-assistant/core
/homeassistant/components/automation/ @home-assistant/core
/tests/components/automation/ @home-assistant/core
/homeassistant/components/autoskope/ @mcisk
/tests/components/autoskope/ @mcisk
/homeassistant/components/avea/ @pattyland
/homeassistant/components/awair/ @ahayworth @ricohageman
/tests/components/awair/ @ahayworth @ricohageman
@@ -1770,8 +1768,6 @@ build.json @home-assistant/supervisor
/tests/components/trend/ @jpbede
/homeassistant/components/triggercmd/ @rvmey
/tests/components/triggercmd/ @rvmey
/homeassistant/components/trmnl/ @joostlek
/tests/components/trmnl/ @joostlek
/homeassistant/components/tts/ @home-assistant/core
/tests/components/tts/ @home-assistant/core
/homeassistant/components/tuya/ @Tuya @zlinoliver
@@ -1788,8 +1784,6 @@ build.json @home-assistant/supervisor
/tests/components/ukraine_alarm/ @PaulAnnekov
/homeassistant/components/unifi/ @Kane610
/tests/components/unifi/ @Kane610
/homeassistant/components/unifi_access/ @imhotep @RaHehl
/tests/components/unifi_access/ @imhotep @RaHehl
/homeassistant/components/unifi_direct/ @tofuSCHNITZEL
/homeassistant/components/unifiled/ @florisvdk
/homeassistant/components/unifiprotect/ @RaHehl

View File

@@ -1,12 +1,5 @@
{
"domain": "ubiquiti",
"name": "Ubiquiti",
"integrations": [
"airos",
"unifi",
"unifi_access",
"unifi_direct",
"unifiled",
"unifiprotect"
]
"integrations": ["airos", "unifi", "unifi_direct", "unifiled", "unifiprotect"]
}

View File

@@ -2,7 +2,6 @@
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.condition import (
Condition,
EntityStateConditionBase,
@@ -44,7 +43,7 @@ def make_entity_state_required_features_condition(
class CustomCondition(EntityStateRequiredFeaturesCondition):
"""Condition for entity state changes."""
_domain_specs = {domain: DomainSpec()}
_domain = domain
_states = {to_state}
_required_features = required_features

View File

@@ -2,7 +2,6 @@
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.entity import get_supported_features
from homeassistant.helpers.trigger import (
EntityTargetStateTriggerBase,
@@ -45,7 +44,7 @@ def make_entity_state_trigger_required_features(
class CustomTrigger(EntityStateTriggerRequiredFeatures):
"""Trigger for entity state changes."""
_domain_specs = {domain: DomainSpec()}
_domains = {domain}
_to_states = {to_state}
_required_features = required_features

View File

@@ -1,5 +1,6 @@
"""Defines a base Alexa Devices entity."""
from aioamazondevices.const.devices import SPEAKER_GROUP_DEVICE_TYPE
from aioamazondevices.structures import AmazonDevice
from homeassistant.helpers.device_registry import DeviceInfo
@@ -24,15 +25,20 @@ class AmazonEntity(CoordinatorEntity[AmazonDevicesCoordinator]):
"""Initialize the entity."""
super().__init__(coordinator)
self._serial_num = serial_num
model = self.device.model
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_num)},
name=self.device.account_name,
model=self.device.model,
model=model,
model_id=self.device.device_type,
manufacturer=self.device.manufacturer or "Amazon",
hw_version=self.device.hardware_version,
sw_version=self.device.software_version,
serial_number=serial_num,
sw_version=(
self.device.software_version
if model != SPEAKER_GROUP_DEVICE_TYPE
else None
),
serial_number=serial_num if model != SPEAKER_GROUP_DEVICE_TYPE else None,
)
self.entity_description = description
self._attr_unique_id = f"{serial_num}-{description.key}"

View File

@@ -8,5 +8,5 @@
"iot_class": "cloud_polling",
"loggers": ["aioamazondevices"],
"quality_scale": "platinum",
"requirements": ["aioamazondevices==13.0.1"]
"requirements": ["aioamazondevices==13.0.0"]
}

View File

@@ -17,7 +17,7 @@ from .coordinator import ArcamFmjConfigEntry, ArcamFmjCoordinator, ArcamFmjRunti
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.MEDIA_PLAYER, Platform.SENSOR]
PLATFORMS = [Platform.MEDIA_PLAYER]
async def async_setup_entry(hass: HomeAssistant, entry: ArcamFmjConfigEntry) -> bool:

View File

@@ -2,7 +2,6 @@
from __future__ import annotations
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import ArcamFmjCoordinator
@@ -13,16 +12,9 @@ class ArcamFmjEntity(CoordinatorEntity[ArcamFmjCoordinator]):
_attr_has_entity_name = True
def __init__(
self,
coordinator: ArcamFmjCoordinator,
description: EntityDescription | None = None,
) -> None:
def __init__(self, coordinator: ArcamFmjCoordinator) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._attr_device_info = coordinator.device_info
self._attr_entity_registry_enabled_default = coordinator.state.zn == 1
self._attr_unique_id = coordinator.zone_unique_id
if description is not None:
self._attr_unique_id = f"{self._attr_unique_id}-{description.key}"
self.entity_description = description

View File

@@ -1,162 +0,0 @@
"""Arcam sensors for incoming stream info."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from arcam.fmj import IncomingVideoAspectRatio, IncomingVideoColorspace
from arcam.fmj.state import IncomingAudioConfig, IncomingAudioFormat, State
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import EntityCategory, UnitOfFrequency
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import ArcamFmjConfigEntry
from .entity import ArcamFmjEntity
@dataclass(frozen=True, kw_only=True)
class ArcamFmjSensorEntityDescription(SensorEntityDescription):
"""Describes an Arcam FMJ sensor entity."""
value_fn: Callable[[State], int | float | str | None]
SENSORS: tuple[ArcamFmjSensorEntityDescription, ...] = (
ArcamFmjSensorEntityDescription(
key="incoming_video_horizontal_resolution",
translation_key="incoming_video_horizontal_resolution",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="px",
suggested_display_precision=0,
value_fn=lambda state: (
vp.horizontal_resolution
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_vertical_resolution",
translation_key="incoming_video_vertical_resolution",
entity_category=EntityCategory.DIAGNOSTIC,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement="px",
suggested_display_precision=0,
value_fn=lambda state: (
vp.vertical_resolution
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_refresh_rate",
translation_key="incoming_video_refresh_rate",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=0,
value_fn=lambda state: (
vp.refresh_rate
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_aspect_ratio",
translation_key="incoming_video_aspect_ratio",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingVideoAspectRatio],
value_fn=lambda state: (
vp.aspect_ratio.name.lower()
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_video_colorspace",
translation_key="incoming_video_colorspace",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingVideoColorspace],
value_fn=lambda state: (
vp.colorspace.name.lower()
if (vp := state.get_incoming_video_parameters()) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_audio_format",
translation_key="incoming_audio_format",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingAudioFormat],
value_fn=lambda state: (
result.name.lower()
if (result := state.get_incoming_audio_format()[0]) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_audio_config",
translation_key="incoming_audio_config",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.ENUM,
options=[member.name.lower() for member in IncomingAudioConfig],
value_fn=lambda state: (
result.name.lower()
if (result := state.get_incoming_audio_format()[1]) is not None
else None
),
),
ArcamFmjSensorEntityDescription(
key="incoming_audio_sample_rate",
translation_key="incoming_audio_sample_rate",
entity_category=EntityCategory.DIAGNOSTIC,
device_class=SensorDeviceClass.FREQUENCY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfFrequency.HERTZ,
suggested_display_precision=0,
value_fn=lambda state: (
None
if (sample_rate := state.get_incoming_audio_sample_rate()) == 0
else sample_rate
),
),
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ArcamFmjConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Arcam FMJ sensors from a config entry."""
coordinators = config_entry.runtime_data.coordinators
entities: list[ArcamFmjSensorEntity] = []
for coordinator in coordinators.values():
entities.extend(
ArcamFmjSensorEntity(coordinator, description) for description in SENSORS
)
async_add_entities(entities)
class ArcamFmjSensorEntity(ArcamFmjEntity, SensorEntity):
"""Representation of an Arcam FMJ sensor."""
entity_description: ArcamFmjSensorEntityDescription
@property
def native_value(self) -> int | float | str | None:
"""Return the sensor value."""
return self.entity_description.value_fn(self.coordinator.state)

View File

@@ -23,116 +23,5 @@
"trigger_type": {
"turn_on": "{entity_name} was requested to turn on"
}
},
"entity": {
"sensor": {
"incoming_audio_config": {
"name": "Incoming audio configuration",
"state": {
"auro_10_1": "Auro 10.1",
"auro_11_1": "Auro 11.1",
"auro_13_1": "Auro 13.1",
"auro_2_2_2": "Auro 2.2.2",
"auro_5_0": "Auro 5.0",
"auro_5_1": "Auro 5.1",
"auro_8_0": "Auro 8.0",
"auro_9_1": "Auro 9.1",
"auro_quad": "Auro quad",
"dual_mono": "Dual mono",
"dual_mono_lfe": "Dual mono + LFE",
"mono": "Mono",
"mono_lfe": "Mono + LFE",
"stereo_center": "Stereo center",
"stereo_center_lfe": "Stereo center + LFE",
"stereo_center_surr_lr": "Stereo center surround L/R",
"stereo_center_surr_lr_back_lr": "Stereo center surround L/R back L/R",
"stereo_center_surr_lr_back_lr_lfe": "Stereo center surround L/R back L/R + LFE",
"stereo_center_surr_lr_back_matrix": "Stereo center surround L/R back matrix",
"stereo_center_surr_lr_back_matrix_lfe": "Stereo center surround L/R back matrix + LFE",
"stereo_center_surr_lr_back_mono": "Stereo center surround L/R back mono",
"stereo_center_surr_lr_back_mono_lfe": "Stereo center surround L/R back mono + LFE",
"stereo_center_surr_lr_lfe": "Stereo center surround L/R + LFE",
"stereo_center_surr_mono": "Stereo center surround mono",
"stereo_center_surr_mono_lfe": "Stereo center surround mono + LFE",
"stereo_downmix": "Stereo downmix",
"stereo_downmix_lfe": "Stereo downmix + LFE",
"stereo_lfe": "Stereo + LFE",
"stereo_only": "Stereo only",
"stereo_only_lo_ro": "Stereo only Lo/Ro",
"stereo_only_lo_ro_lfe": "Stereo only Lo/Ro + LFE",
"stereo_surr_lr": "Stereo surround L/R",
"stereo_surr_lr_back_lr": "Stereo surround L/R back L/R",
"stereo_surr_lr_back_lr_lfe": "Stereo surround L/R back L/R + LFE",
"stereo_surr_lr_back_matrix": "Stereo surround L/R back matrix",
"stereo_surr_lr_back_matrix_lfe": "Stereo surround L/R back matrix + LFE",
"stereo_surr_lr_back_mono": "Stereo surround L/R back mono",
"stereo_surr_lr_back_mono_lfe": "Stereo surround L/R back mono + LFE",
"stereo_surr_lr_lfe": "Stereo surround L/R + LFE",
"stereo_surr_mono": "Stereo surround mono",
"stereo_surr_mono_lfe": "Stereo surround mono + LFE",
"undetected": "Undetected",
"unknown": "Unknown"
}
},
"incoming_audio_format": {
"name": "Incoming audio format",
"state": {
"analogue_direct": "Analogue direct",
"auro_3d": "Auro-3D",
"dolby_atmos": "Dolby Atmos",
"dolby_digital": "Dolby Digital",
"dolby_digital_ex": "Dolby Digital EX",
"dolby_digital_plus": "Dolby Digital Plus",
"dolby_digital_surround": "Dolby Digital Surround",
"dolby_digital_true_hd": "Dolby TrueHD",
"dts": "DTS",
"dts_96_24": "DTS 96/24",
"dts_core": "DTS Core",
"dts_es_discrete": "DTS-ES Discrete",
"dts_es_discrete_96_24": "DTS-ES Discrete 96/24",
"dts_es_matrix": "DTS-ES Matrix",
"dts_es_matrix_96_24": "DTS-ES Matrix 96/24",
"dts_hd_high_res_audio": "DTS-HD High Resolution Audio",
"dts_hd_master_audio": "DTS-HD Master Audio",
"dts_low_bit_rate": "DTS Low Bit Rate",
"dts_x": "DTS:X",
"imax_enhanced": "IMAX Enhanced",
"pcm": "PCM",
"pcm_zero": "PCM zero",
"undetected": "Undetected",
"unsupported": "Unsupported"
}
},
"incoming_audio_sample_rate": {
"name": "Incoming audio sample rate"
},
"incoming_video_aspect_ratio": {
"name": "Incoming video aspect ratio",
"state": {
"aspect_16_9": "16:9",
"aspect_4_3": "4:3",
"undefined": "Undefined"
}
},
"incoming_video_colorspace": {
"name": "Incoming video colorspace",
"state": {
"dolby_vision": "Dolby Vision",
"hdr10": "HDR10",
"hdr10_plus": "HDR10+",
"hlg": "HLG",
"normal": "Normal"
}
},
"incoming_video_horizontal_resolution": {
"name": "Incoming video horizontal resolution"
},
"incoming_video_refresh_rate": {
"name": "Incoming video refresh rate"
},
"incoming_video_vertical_resolution": {
"name": "Incoming video vertical resolution"
}
}
}
}

View File

@@ -1,53 +0,0 @@
"""The Autoskope integration."""
from __future__ import annotations
import aiohttp
from autoskope_client.api import AutoskopeApi
from autoskope_client.models import CannotConnect, InvalidAuth
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import DEFAULT_HOST
from .coordinator import AutoskopeConfigEntry, AutoskopeDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER]
async def async_setup_entry(hass: HomeAssistant, entry: AutoskopeConfigEntry) -> bool:
"""Set up Autoskope from a config entry."""
session = async_create_clientsession(hass, cookie_jar=aiohttp.CookieJar())
api = AutoskopeApi(
host=entry.data.get(CONF_HOST, DEFAULT_HOST),
username=entry.data[CONF_USERNAME],
password=entry.data[CONF_PASSWORD],
session=session,
)
try:
await api.connect()
except InvalidAuth as err:
# Raise ConfigEntryError until reauth flow is implemented (then ConfigEntryAuthFailed)
raise ConfigEntryError(
"Authentication failed, please check credentials"
) from err
except CannotConnect as err:
raise ConfigEntryNotReady("Could not connect to Autoskope API") from err
coordinator = AutoskopeDataUpdateCoordinator(hass, api, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: AutoskopeConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,89 +0,0 @@
"""Config flow for the Autoskope integration."""
from __future__ import annotations
from typing import Any
from autoskope_client.api import AutoskopeApi
from autoskope_client.models import CannotConnect, InvalidAuth
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.data_entry_flow import section
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.selector import (
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import DEFAULT_HOST, DOMAIN, SECTION_ADVANCED_SETTINGS
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
vol.Required(SECTION_ADVANCED_SETTINGS): section(
vol.Schema(
{
vol.Required(CONF_HOST, default=DEFAULT_HOST): TextSelector(
TextSelectorConfig(type=TextSelectorType.URL)
),
}
),
{"collapsed": True},
),
}
)
class AutoskopeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Autoskope."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
username = user_input[CONF_USERNAME].lower()
host = user_input[SECTION_ADVANCED_SETTINGS][CONF_HOST].lower()
try:
cv.url(host)
except vol.Invalid:
errors["base"] = "invalid_url"
if not errors:
await self.async_set_unique_id(f"{username}@{host}")
self._abort_if_unique_id_configured()
try:
async with AutoskopeApi(
host=host,
username=username,
password=user_input[CONF_PASSWORD],
):
pass
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
else:
return self.async_create_entry(
title=f"Autoskope ({username})",
data={
CONF_USERNAME: username,
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_HOST: host,
},
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

View File

@@ -1,9 +0,0 @@
"""Constants for the Autoskope integration."""
from datetime import timedelta
DOMAIN = "autoskope"
DEFAULT_HOST = "https://portal.autoskope.de"
SECTION_ADVANCED_SETTINGS = "advanced_settings"
UPDATE_INTERVAL = timedelta(seconds=60)

View File

@@ -1,60 +0,0 @@
"""Data update coordinator for the Autoskope integration."""
from __future__ import annotations
import logging
from autoskope_client.api import AutoskopeApi
from autoskope_client.models import CannotConnect, InvalidAuth, Vehicle
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, UPDATE_INTERVAL
_LOGGER = logging.getLogger(__name__)
type AutoskopeConfigEntry = ConfigEntry[AutoskopeDataUpdateCoordinator]
class AutoskopeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Vehicle]]):
"""Class to manage fetching Autoskope data."""
config_entry: AutoskopeConfigEntry
def __init__(
self, hass: HomeAssistant, api: AutoskopeApi, entry: AutoskopeConfigEntry
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=UPDATE_INTERVAL,
config_entry=entry,
)
self.api = api
async def _async_update_data(self) -> dict[str, Vehicle]:
"""Fetch data from API endpoint."""
try:
vehicles = await self.api.get_vehicles()
return {vehicle.id: vehicle for vehicle in vehicles}
except InvalidAuth:
# Attempt to re-authenticate using stored credentials
try:
await self.api.authenticate()
# Retry the request after successful re-authentication
vehicles = await self.api.get_vehicles()
return {vehicle.id: vehicle for vehicle in vehicles}
except InvalidAuth as reauth_err:
raise ConfigEntryAuthFailed(
f"Authentication failed: {reauth_err}"
) from reauth_err
except CannotConnect as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err

View File

@@ -1,145 +0,0 @@
"""Support for Autoskope device tracking."""
from __future__ import annotations
from autoskope_client.constants import MANUFACTURER
from autoskope_client.models import Vehicle
from homeassistant.components.device_tracker import SourceType, TrackerEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import AutoskopeConfigEntry, AutoskopeDataUpdateCoordinator
PARALLEL_UPDATES = 0
async def async_setup_entry(
hass: HomeAssistant,
entry: AutoskopeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Autoskope device tracker entities."""
coordinator: AutoskopeDataUpdateCoordinator = entry.runtime_data
tracked_vehicles: set[str] = set()
@callback
def update_entities() -> None:
"""Update entities based on coordinator data."""
current_vehicles = set(coordinator.data.keys())
vehicles_to_add = current_vehicles - tracked_vehicles
if vehicles_to_add:
new_entities = [
AutoskopeDeviceTracker(coordinator, vehicle_id)
for vehicle_id in vehicles_to_add
]
tracked_vehicles.update(vehicles_to_add)
async_add_entities(new_entities)
entry.async_on_unload(coordinator.async_add_listener(update_entities))
update_entities()
class AutoskopeDeviceTracker(
CoordinatorEntity[AutoskopeDataUpdateCoordinator], TrackerEntity
):
"""Representation of an Autoskope tracked device."""
_attr_has_entity_name = True
_attr_name: str | None = None
def __init__(
self, coordinator: AutoskopeDataUpdateCoordinator, vehicle_id: str
) -> None:
"""Initialize the TrackerEntity."""
super().__init__(coordinator)
self._vehicle_id = vehicle_id
self._attr_unique_id = vehicle_id
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if (
self._vehicle_id in self.coordinator.data
and (device_entry := self.device_entry) is not None
and device_entry.name != self._vehicle_data.name
):
device_registry = dr.async_get(self.hass)
device_registry.async_update_device(
device_entry.id, name=self._vehicle_data.name
)
super()._handle_coordinator_update()
@property
def device_info(self) -> DeviceInfo:
"""Return device info for the vehicle."""
vehicle = self.coordinator.data[self._vehicle_id]
return DeviceInfo(
identifiers={(DOMAIN, str(vehicle.id))},
name=vehicle.name,
manufacturer=MANUFACTURER,
model=vehicle.model,
serial_number=vehicle.imei,
)
@property
def available(self) -> bool:
"""Return if entity is available."""
return (
super().available
and self.coordinator.data is not None
and self._vehicle_id in self.coordinator.data
)
@property
def _vehicle_data(self) -> Vehicle:
"""Return the vehicle data for the current entity."""
return self.coordinator.data[self._vehicle_id]
@property
def latitude(self) -> float | None:
"""Return latitude value of the device."""
if (vehicle := self._vehicle_data) and vehicle.position:
return float(vehicle.position.latitude)
return None
@property
def longitude(self) -> float | None:
"""Return longitude value of the device."""
if (vehicle := self._vehicle_data) and vehicle.position:
return float(vehicle.position.longitude)
return None
@property
def source_type(self) -> SourceType:
"""Return the source type of the device."""
return SourceType.GPS
@property
def location_accuracy(self) -> float:
"""Return the location accuracy of the device in meters."""
if (vehicle := self._vehicle_data) and vehicle.gps_quality:
if vehicle.gps_quality > 0:
# HDOP to estimated accuracy in meters
# HDOP of 1-2 = good (5-10m), 2-5 = moderate (10-25m), >5 = poor (>25m)
return float(max(5, int(vehicle.gps_quality * 5.0)))
return 0.0
@property
def icon(self) -> str:
"""Return the icon based on the vehicle's activity."""
if self._vehicle_id not in self.coordinator.data:
return "mdi:car-clock"
vehicle = self._vehicle_data
if vehicle.position:
if vehicle.position.park_mode:
return "mdi:car-brake-parking"
if vehicle.position.speed > 5: # Moving threshold: 5 km/h
return "mdi:car-arrow-right"
return "mdi:car"
return "mdi:car-clock"

View File

@@ -1,11 +0,0 @@
{
"domain": "autoskope",
"name": "Autoskope",
"codeowners": ["@mcisk"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/autoskope",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["autoskope_client==1.4.1"]
}

View File

@@ -1,88 +0,0 @@
# + in comment indicates requirement for quality scale
# - in comment indicates issue to be fixed, not impacting quality scale
rules:
# Bronze
action-setup:
status: exempt
comment: |
Integration does not provide custom services.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: |
Integration does not provide custom services.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions:
status: exempt
comment: |
Integration does not provide custom services.
config-entry-unloading: done
docs-configuration-parameters: done
docs-installation-parameters: done
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow:
status: todo
comment: |
Reauthentication flow removed for initial PR, will be added in follow-up.
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: |
Integration does not use discovery. Autoskope devices use NB-IoT/LTE-M (via IoT SIMs) and LoRaWAN.
discovery:
status: exempt
comment: |
Integration does not use discovery. Autoskope devices use NB-IoT/LTE-M (via IoT SIMs) and LoRaWAN.
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices: done
entity-category: done
entity-device-class: done
entity-disabled-by-default:
status: exempt
comment: |
Only one entity type (device_tracker) is created, making this not applicable.
entity-translations: done
exception-translations: done
icon-translations: done
reconfiguration-flow:
status: todo
comment: |
Reconfiguration flow removed for initial PR, will be added in follow-up.
repair-issues: todo
stale-devices: done
# Platinum
async-dependency: done
inject-websession: done
strict-typing:
status: todo
comment: |
Integration needs to be added to .strict-typing file for full compliance.

View File

@@ -1,52 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_url": "Invalid URL",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
"username": "[%key:common::config_flow::data::username%]"
},
"data_description": {
"password": "The password for your Autoskope account.",
"username": "The username for your Autoskope account."
},
"description": "Enter your Autoskope credentials.",
"sections": {
"advanced_settings": {
"data": {
"host": "API endpoint"
},
"data_description": {
"host": "The URL of your Autoskope API endpoint. Only change this if you use a white-label portal."
},
"name": "Advanced settings"
}
},
"title": "Connect to Autoskope"
}
}
},
"issues": {
"cannot_connect": {
"description": "Home Assistant could not connect to the Autoskope API at {host}. Please check the connection details and ensure the API endpoint is reachable.\n\nError: {error}",
"title": "Failed to connect to Autoskope"
},
"invalid_auth": {
"description": "Authentication with Autoskope failed for user {username}. Please re-authenticate the integration with the correct password.",
"title": "Invalid Autoskope authentication"
},
"low_battery": {
"description": "The battery voltage for vehicle {vehicle_name} ({vehicle_id}) is low ({value}V). Consider checking or replacing the battery.",
"title": "Low vehicle battery ({vehicle_name})"
}
}
}

View File

@@ -2,7 +2,6 @@
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
@@ -15,7 +14,7 @@ from . import DOMAIN
class ButtonPressedTrigger(EntityTriggerBase):
"""Trigger for button entity presses."""
_domain_specs = {DOMAIN: DomainSpec()}
_domains = {DOMAIN}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:

View File

@@ -5,14 +5,13 @@ import voluptuous as vol
from homeassistant.const import ATTR_TEMPERATURE, CONF_OPTIONS
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.automation import DomainSpec, NumericalDomainSpec
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST,
EntityTargetStateTriggerBase,
Trigger,
TriggerConfig,
make_entity_numerical_state_changed_trigger,
make_entity_numerical_state_crossed_threshold_trigger,
make_entity_numerical_state_attribute_changed_trigger,
make_entity_numerical_state_attribute_crossed_threshold_trigger,
make_entity_target_state_attribute_trigger,
make_entity_target_state_trigger,
make_entity_transition_trigger,
@@ -36,7 +35,7 @@ HVAC_MODE_CHANGED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST.extend
class HVACModeChangedTrigger(EntityTargetStateTriggerBase):
"""Trigger for entity state changes."""
_domain_specs = {DOMAIN: DomainSpec()}
_domains = {DOMAIN}
_schema = HVAC_MODE_CHANGED_TRIGGER_SCHEMA
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
@@ -53,17 +52,17 @@ TRIGGERS: dict[str, type[Trigger]] = {
"started_drying": make_entity_target_state_attribute_trigger(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.DRYING
),
"target_humidity_changed": make_entity_numerical_state_changed_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_HUMIDITY)}
"target_humidity_changed": make_entity_numerical_state_attribute_changed_trigger(
{DOMAIN}, {DOMAIN: ATTR_HUMIDITY}
),
"target_humidity_crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_HUMIDITY)}
"target_humidity_crossed_threshold": make_entity_numerical_state_attribute_crossed_threshold_trigger(
{DOMAIN}, {DOMAIN: ATTR_HUMIDITY}
),
"target_temperature_changed": make_entity_numerical_state_changed_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_TEMPERATURE)}
"target_temperature_changed": make_entity_numerical_state_attribute_changed_trigger(
{DOMAIN}, {DOMAIN: ATTR_TEMPERATURE}
),
"target_temperature_crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_TEMPERATURE)}
"target_temperature_crossed_threshold": make_entity_numerical_state_attribute_crossed_threshold_trigger(
{DOMAIN}, {DOMAIN: ATTR_TEMPERATURE}
),
"turned_off": make_entity_target_state_trigger(DOMAIN, HVACMode.OFF),
"turned_on": make_entity_transition_trigger(

View File

@@ -15,6 +15,7 @@ async def async_setup_intents(hass: HomeAssistant) -> None:
INTENT_OPEN_COVER,
DOMAIN,
SERVICE_OPEN_COVER,
"Opening {}",
description="Opens a cover",
platforms={DOMAIN},
device_classes={CoverDeviceClass},
@@ -26,6 +27,7 @@ async def async_setup_intents(hass: HomeAssistant) -> None:
INTENT_CLOSE_COVER,
DOMAIN,
SERVICE_CLOSE_COVER,
"Closing {}",
description="Closes a cover",
platforms={DOMAIN},
device_classes={CoverDeviceClass},

View File

@@ -1,82 +1,81 @@
"""Provides triggers for covers."""
from dataclasses import dataclass
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State, split_entity_id
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import EntityTriggerBase, Trigger
from homeassistant.helpers.trigger import (
EntityTriggerBase,
Trigger,
get_device_class_or_undefined,
)
from .const import ATTR_IS_CLOSED, DOMAIN, CoverDeviceClass
@dataclass(frozen=True, slots=True)
class CoverDomainSpec(DomainSpec):
"""DomainSpec with a target value for comparison."""
target_value: str | bool | None = None
class CoverTriggerBase(EntityTriggerBase[CoverDomainSpec]):
class CoverTriggerBase(EntityTriggerBase):
"""Base trigger for cover state changes."""
def _get_value(self, state: State) -> str | bool | None:
"""Extract the relevant value from state based on domain spec."""
domain_spec = self._domain_specs[split_entity_id(state.entity_id)[0]]
if domain_spec.value_source is not None:
return state.attributes.get(domain_spec.value_source)
return state.state
_binary_sensor_target_state: str
_cover_is_closed_target_value: bool
_device_classes: dict[str, str]
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities by cover device class."""
entities = super().entity_filter(entities)
return {
entity_id
for entity_id in entities
if get_device_class_or_undefined(self._hass, entity_id)
== self._device_classes[split_entity_id(entity_id)[0]]
}
def is_valid_state(self, state: State) -> bool:
"""Check if the state matches the target cover state."""
domain_spec = self._domain_specs[split_entity_id(state.entity_id)[0]]
return self._get_value(state) == domain_spec.target_value
if split_entity_id(state.entity_id)[0] == DOMAIN:
return (
state.attributes.get(ATTR_IS_CLOSED)
== self._cover_is_closed_target_value
)
return state.state == self._binary_sensor_target_state
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the transition is valid for a cover state change."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
return False
if (from_value := self._get_value(from_state)) is None:
return False
return from_value != self._get_value(to_state)
if split_entity_id(from_state.entity_id)[0] == DOMAIN:
if (from_is_closed := from_state.attributes.get(ATTR_IS_CLOSED)) is None:
return False
return from_is_closed != to_state.attributes.get(ATTR_IS_CLOSED) # type: ignore[no-any-return]
return from_state.state != to_state.state
def make_cover_opened_trigger(
*, device_classes: dict[str, str]
*, device_classes: dict[str, str], domains: set[str] | None = None
) -> type[CoverTriggerBase]:
"""Create a trigger cover_opened."""
class CoverOpenedTrigger(CoverTriggerBase):
"""Trigger for cover opened state changes."""
_domain_specs = {
domain: CoverDomainSpec(
device_class=dc,
value_source=ATTR_IS_CLOSED if domain == DOMAIN else None,
target_value=False if domain == DOMAIN else STATE_ON,
)
for domain, dc in device_classes.items()
}
_binary_sensor_target_state = STATE_ON
_cover_is_closed_target_value = False
_domains = domains or {DOMAIN}
_device_classes = device_classes
return CoverOpenedTrigger
def make_cover_closed_trigger(
*, device_classes: dict[str, str]
*, device_classes: dict[str, str], domains: set[str] | None = None
) -> type[CoverTriggerBase]:
"""Create a trigger cover_closed."""
class CoverClosedTrigger(CoverTriggerBase):
"""Trigger for cover closed state changes."""
_domain_specs = {
domain: CoverDomainSpec(
device_class=dc,
value_source=ATTR_IS_CLOSED if domain == DOMAIN else None,
target_value=True if domain == DOMAIN else STATE_OFF,
)
for domain, dc in device_classes.items()
}
_binary_sensor_target_state = STATE_OFF
_cover_is_closed_target_value = True
_domains = domains or {DOMAIN}
_device_classes = device_classes
return CoverClosedTrigger

View File

@@ -20,8 +20,14 @@ DEVICE_CLASSES_DOOR: dict[str, str] = {
TRIGGERS: dict[str, type[Trigger]] = {
"opened": make_cover_opened_trigger(device_classes=DEVICE_CLASSES_DOOR),
"closed": make_cover_closed_trigger(device_classes=DEVICE_CLASSES_DOOR),
"opened": make_cover_opened_trigger(
device_classes=DEVICE_CLASSES_DOOR,
domains={BINARY_SENSOR_DOMAIN, COVER_DOMAIN},
),
"closed": make_cover_closed_trigger(
device_classes=DEVICE_CLASSES_DOOR,
domains={BINARY_SENSOR_DOMAIN, COVER_DOMAIN},
),
}

View File

@@ -20,8 +20,14 @@ DEVICE_CLASSES_GARAGE_DOOR: dict[str, str] = {
TRIGGERS: dict[str, type[Trigger]] = {
"opened": make_cover_opened_trigger(device_classes=DEVICE_CLASSES_GARAGE_DOOR),
"closed": make_cover_closed_trigger(device_classes=DEVICE_CLASSES_GARAGE_DOOR),
"opened": make_cover_opened_trigger(
device_classes=DEVICE_CLASSES_GARAGE_DOOR,
domains={BINARY_SENSOR_DOMAIN, COVER_DOMAIN},
),
"closed": make_cover_closed_trigger(
device_classes=DEVICE_CLASSES_GARAGE_DOOR,
domains={BINARY_SENSOR_DOMAIN, COVER_DOMAIN},
),
}

View File

@@ -10,11 +10,7 @@ from functools import partial, wraps
import logging
from typing import Any, Concatenate
from aiohasupervisor import (
AddonNotSupportedError,
SupervisorError,
SupervisorNotFoundError,
)
from aiohasupervisor import SupervisorError
from aiohasupervisor.models import (
AddonsOptions,
AddonState as SupervisorAddonState,
@@ -169,7 +165,15 @@ class AddonManager:
)
addon_info = await self._supervisor_client.addons.addon_info(self.addon_slug)
return self._async_convert_installed_addon_info(addon_info)
addon_state = self.async_get_addon_state(addon_info)
return AddonInfo(
available=addon_info.available,
hostname=addon_info.hostname,
options=addon_info.options,
state=addon_state,
update_available=addon_info.update_available,
version=addon_info.version,
)
@callback
def async_get_addon_state(self, addon_info: InstalledAddonComplete) -> AddonState:
@@ -185,20 +189,6 @@ class AddonManager:
return addon_state
@callback
def _async_convert_installed_addon_info(
self, addon_info: InstalledAddonComplete
) -> AddonInfo:
"""Convert InstalledAddonComplete model to AddonInfo model."""
return AddonInfo(
available=addon_info.available,
hostname=addon_info.hostname,
options=addon_info.options,
state=self.async_get_addon_state(addon_info),
update_available=addon_info.update_available,
version=addon_info.version,
)
@api_error(
"Failed to set the {addon_name} app options",
expected_error_type=SupervisorError,
@@ -209,17 +199,21 @@ class AddonManager:
self.addon_slug, AddonsOptions(config=config)
)
def _check_addon_available(self, addon_info: AddonInfo) -> None:
"""Check if the managed add-on is available."""
if not addon_info.available:
raise AddonError(f"{self.addon_name} app is not available")
@api_error(
"Failed to install the {addon_name} app", expected_error_type=SupervisorError
)
async def async_install_addon(self) -> None:
"""Install the managed add-on."""
try:
await self._supervisor_client.store.install_addon(self.addon_slug)
except AddonNotSupportedError as err:
raise AddonError(
f"{self.addon_name} app is not available: {err!s}"
) from None
addon_info = await self.async_get_addon_info()
self._check_addon_available(addon_info)
await self._supervisor_client.store.install_addon(self.addon_slug)
@api_error(
"Failed to uninstall the {addon_name} app",
@@ -232,29 +226,17 @@ class AddonManager:
@api_error("Failed to update the {addon_name} app")
async def async_update_addon(self) -> None:
"""Update the managed add-on if needed."""
try:
# Not using async_get_addon_info here because it would make an unnecessary
# call to /store/addon/{slug}/info. This will raise if the addon is not
# installed so one call to /addon/{slug}/info is all that is needed
addon_info = await self._supervisor_client.addons.addon_info(
self.addon_slug
)
except SupervisorNotFoundError:
raise AddonError(f"{self.addon_name} app is not installed") from None
addon_info = await self.async_get_addon_info()
self._check_addon_available(addon_info)
if addon_info.state is AddonState.NOT_INSTALLED:
raise AddonError(f"{self.addon_name} app is not installed")
if not addon_info.update_available:
return
try:
await self._supervisor_client.store.addon_availability(self.addon_slug)
except AddonNotSupportedError as err:
raise AddonError(
f"{self.addon_name} app is not available: {err!s}"
) from None
await self.async_create_backup(
addon_info=self._async_convert_installed_addon_info(addon_info)
)
await self.async_create_backup()
await self._supervisor_client.store.update_addon(
self.addon_slug, StoreAddonUpdate(backup=False)
)
@@ -284,14 +266,10 @@ class AddonManager:
"Failed to create a backup of the {addon_name} app",
expected_error_type=SupervisorError,
)
async def async_create_backup(self, *, addon_info: AddonInfo | None = None) -> None:
async def async_create_backup(self) -> None:
"""Create a partial backup of the managed add-on."""
if addon_info:
addon_version = addon_info.version
else:
addon_version = (await self.async_get_addon_info()).version
name = f"addon_{self.addon_slug}_{addon_version}"
addon_info = await self.async_get_addon_info()
name = f"addon_{self.addon_slug}_{addon_info.version}"
self._logger.debug("Creating backup: %s", name)
await self._supervisor_client.backups.partial_backup(

View File

@@ -15,43 +15,50 @@ from homeassistant.components.weather import (
ATTR_WEATHER_HUMIDITY,
DOMAIN as WEATHER_DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.automation import NumericalDomainSpec
from homeassistant.core import HomeAssistant, split_entity_id
from homeassistant.helpers.trigger import (
EntityNumericalStateAttributeChangedTriggerBase,
EntityNumericalStateAttributeCrossedThresholdTriggerBase,
EntityTriggerBase,
Trigger,
get_device_class_or_undefined,
)
HUMIDITY_DOMAIN_SPECS: dict[str, NumericalDomainSpec] = {
CLIMATE_DOMAIN: NumericalDomainSpec(
value_source=CLIMATE_ATTR_CURRENT_HUMIDITY,
),
HUMIDIFIER_DOMAIN: NumericalDomainSpec(
value_source=HUMIDIFIER_ATTR_CURRENT_HUMIDITY,
),
SENSOR_DOMAIN: NumericalDomainSpec(
device_class=SensorDeviceClass.HUMIDITY,
),
WEATHER_DOMAIN: NumericalDomainSpec(
value_source=ATTR_WEATHER_HUMIDITY,
),
}
class _HumidityTriggerMixin(EntityTriggerBase):
"""Mixin for humidity triggers providing entity filtering and value extraction."""
_attributes = {
CLIMATE_DOMAIN: CLIMATE_ATTR_CURRENT_HUMIDITY,
HUMIDIFIER_DOMAIN: HUMIDIFIER_ATTR_CURRENT_HUMIDITY,
SENSOR_DOMAIN: None, # Use state.state
WEATHER_DOMAIN: ATTR_WEATHER_HUMIDITY,
}
_domains = {SENSOR_DOMAIN, CLIMATE_DOMAIN, HUMIDIFIER_DOMAIN, WEATHER_DOMAIN}
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities: all climate/humidifier/weather, sensor only with device_class humidity."""
entities = super().entity_filter(entities)
return {
entity_id
for entity_id in entities
if split_entity_id(entity_id)[0] != SENSOR_DOMAIN
or get_device_class_or_undefined(self._hass, entity_id)
== SensorDeviceClass.HUMIDITY
}
class HumidityChangedTrigger(EntityNumericalStateAttributeChangedTriggerBase):
class HumidityChangedTrigger(
_HumidityTriggerMixin, EntityNumericalStateAttributeChangedTriggerBase
):
"""Trigger for humidity value changes across multiple domains."""
_domain_specs = HUMIDITY_DOMAIN_SPECS
class HumidityCrossedThresholdTrigger(
EntityNumericalStateAttributeCrossedThresholdTriggerBase
_HumidityTriggerMixin, EntityNumericalStateAttributeCrossedThresholdTriggerBase
):
"""Trigger for humidity value crossing a threshold across multiple domains."""
_domain_specs = HUMIDITY_DOMAIN_SPECS
TRIGGERS: dict[str, type[Trigger]] = {
"changed": HumidityChangedTrigger,

View File

@@ -74,7 +74,7 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
"""Return the current speed percentage."""
device_data = self._device_data
if device_data.speed_set == FanSpeed.auto_get:
if device_data.speed_set == FanSpeed.auto:
return None
return ranged_value_to_percentage(self._speed_range, int(device_data.speed_set))
@@ -92,7 +92,7 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
if device_data.mode_set == FanMode.off:
return None
if (
device_data.speed_set == FanSpeed.auto_get
device_data.speed_set == FanSpeed.auto
and device_data.mode_set == FanMode.sensor
):
return "auto"
@@ -111,7 +111,7 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
infinitely.
"""
percentage = 25 if percentage == 0 else percentage
await self.async_set_mode_speed(preset_mode=preset_mode, percentage=percentage)
await self.async_set_mode_speed(fan_mode=preset_mode, percentage=percentage)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off the fan."""
@@ -124,10 +124,10 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set preset mode."""
await self.async_set_mode_speed(preset_mode=preset_mode)
await self.async_set_mode_speed(fan_mode=preset_mode)
async def async_set_mode_speed(
self, preset_mode: str | None = None, percentage: int | None = None
self, fan_mode: str | None = None, percentage: int | None = None
) -> None:
"""Set mode and speed.
@@ -137,7 +137,7 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
percentage = self.percentage if percentage is None else percentage
percentage = 25 if percentage is None else percentage
if preset_mode == "auto":
if fan_mode == "auto":
# auto is a special case with special mode and speed setting
await self.coordinator.api.ecocomfort.set_mode_speed_auto(self._device_sn)
await self.coordinator.async_request_refresh()
@@ -148,20 +148,21 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
return
# Determine the fan mode
if not self.is_on:
if fan_mode is not None:
# Set to requested fan_mode
mode = fan_mode
elif not self.is_on:
# Default to alternate fan mode if not turned on
mode = FanMode.alternate
else:
# Maintain current mode
mode = self._device_data.mode_set
speed = FanSpeed(
str(
math.ceil(
percentage_to_ranged_value(
self._speed_range,
percentage,
)
speed = str(
math.ceil(
percentage_to_ranged_value(
self._speed_range,
percentage,
)
)
)

View File

@@ -7,5 +7,5 @@
"integration_type": "device",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["pyintelliclima==0.3.1"]
"requirements": ["pyintelliclima==0.2.2"]
}

View File

@@ -68,12 +68,12 @@ class IntelliClimaVMCFanModeSelect(IntelliClimaECOEntity, SelectEntity):
# If in auto mode (sensor mode with auto speed), return None (handled by fan entity preset mode)
if (
device_data.speed_set == FanSpeed.auto_get
device_data.speed_set == FanSpeed.auto
and device_data.mode_set == FanMode.sensor
):
return None
return INTELLICLIMA_MODE_TO_FAN_MODE.get(device_data.mode_set)
return INTELLICLIMA_MODE_TO_FAN_MODE.get(FanMode(device_data.mode_set))
async def async_select_option(self, option: str) -> None:
"""Set the fan mode."""
@@ -83,7 +83,7 @@ class IntelliClimaVMCFanModeSelect(IntelliClimaECOEntity, SelectEntity):
# Determine speed: keep current speed if available, otherwise default to sleep
if (
device_data.speed_set == FanSpeed.auto_get
device_data.speed_set == FanSpeed.auto
or device_data.mode_set == FanMode.off
):
speed = FanSpeed.sleep

View File

@@ -6,7 +6,7 @@
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"no_devices": "No supported IntelliClima devices were found in your account",
"no_devices": "No IntelliClima devices found in your account",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {

View File

@@ -4,7 +4,6 @@ from typing import Any
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.automation import NumericalDomainSpec
from homeassistant.helpers.trigger import (
EntityNumericalStateAttributeChangedTriggerBase,
EntityNumericalStateAttributeCrossedThresholdTriggerBase,
@@ -21,18 +20,13 @@ def _convert_uint8_to_percentage(value: Any) -> float:
return (float(value) / 255.0) * 100.0
BRIGHTNESS_DOMAIN_SPECS = {
DOMAIN: NumericalDomainSpec(
value_source=ATTR_BRIGHTNESS,
value_converter=_convert_uint8_to_percentage,
),
}
class BrightnessChangedTrigger(EntityNumericalStateAttributeChangedTriggerBase):
"""Trigger for brightness changed."""
_domain_specs = BRIGHTNESS_DOMAIN_SPECS
_domains = {DOMAIN}
_attributes = {DOMAIN: ATTR_BRIGHTNESS}
_converter = staticmethod(_convert_uint8_to_percentage)
class BrightnessCrossedThresholdTrigger(
@@ -40,7 +34,9 @@ class BrightnessCrossedThresholdTrigger(
):
"""Trigger for brightness crossed threshold."""
_domain_specs = BRIGHTNESS_DOMAIN_SPECS
_domains = {DOMAIN}
_attributes = {DOMAIN: ATTR_BRIGHTNESS}
_converter = staticmethod(_convert_uint8_to_percentage)
TRIGGERS: dict[str, type[Trigger]] = {

View File

@@ -20,8 +20,6 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class RobotBinarySensorEntityDescription(

View File

@@ -14,9 +14,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
from .entity import LitterRobotEntity, _WhiskerEntityT
@dataclass(frozen=True, kw_only=True)
@@ -73,7 +71,6 @@ class LitterRobotButtonEntity(LitterRobotEntity[_WhiskerEntityT], ButtonEntity):
entity_description: RobotButtonEntityDescription[_WhiskerEntityT]
@whisker_command
async def async_press(self) -> None:
"""Press the button."""
await self.entity_description.press_fn(self.robot)

View File

@@ -10,7 +10,7 @@ from pylitterbot import Account
from pylitterbot.exceptions import LitterRobotException, LitterRobotLoginException
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -21,7 +21,6 @@ _LOGGER = logging.getLogger(__name__)
STEP_USER_DATA_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
)
STEP_REAUTH_RECONFIGURE_SCHEMA = vol.Schema({vol.Required(CONF_PASSWORD): str})
class LitterRobotConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -44,45 +43,24 @@ class LitterRobotConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Handle user's reauth credentials."""
errors: dict[str, str] = {}
errors = {}
if user_input:
reauth_entry = self._get_reauth_entry()
result, errors = await self._async_validate_and_update_entry(
reauth_entry, user_input
)
if result is not None:
return result
user_input = user_input | {CONF_USERNAME: self.username}
if not (error := await self._async_validate_input(user_input)):
await self.async_set_unique_id(self._account_user_id)
self._abort_if_unique_id_mismatch()
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data_updates=user_input
)
errors["base"] = error
return self.async_show_form(
step_id="reauth_confirm",
data_schema=STEP_REAUTH_RECONFIGURE_SCHEMA,
data_schema=vol.Schema({vol.Required(CONF_PASSWORD): str}),
description_placeholders={CONF_USERNAME: self.username},
errors=errors,
)
async def async_step_reconfigure(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a reconfiguration flow request."""
reconfigure_entry = self._get_reconfigure_entry()
self.username = reconfigure_entry.data[CONF_USERNAME]
self._async_abort_entries_match({CONF_USERNAME: self.username})
errors: dict[str, str] = {}
if user_input:
result, errors = await self._async_validate_and_update_entry(
reconfigure_entry, user_input
)
if result is not None:
return result
return self.async_show_form(
step_id="reconfigure",
data_schema=STEP_REAUTH_RECONFIGURE_SCHEMA,
errors=errors,
)
async def async_step_user(
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
@@ -103,25 +81,6 @@ class LitterRobotConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)
async def _async_validate_and_update_entry(
self, entry: ConfigEntry, user_input: dict[str, Any]
) -> tuple[ConfigFlowResult | None, dict[str, str]]:
"""Validate credentials and update an existing entry if valid."""
errors: dict[str, str] = {}
full_input: dict[str, Any] = user_input | {CONF_USERNAME: self.username}
if not (error := await self._async_validate_input(full_input)):
await self.async_set_unique_id(self._account_user_id)
self._abort_if_unique_id_mismatch()
return (
self.async_update_reload_and_abort(
entry,
data_updates=full_input,
),
errors,
)
errors["base"] = error
return None, errors
async def _async_validate_input(self, user_input: Mapping[str, Any]) -> str:
"""Validate login credentials."""
account = Account(websession=async_get_clientsession(self.hass))

View File

@@ -46,22 +46,11 @@ class LitterRobotDataUpdateCoordinator(DataUpdateCoordinator[None]):
async def _async_update_data(self) -> None:
"""Update all device states from the Litter-Robot API."""
try:
await self.account.refresh_robots()
await self.account.load_pets()
for pet in self.account.pets:
# Need to fetch weight history for `get_visits_since`
await pet.fetch_weight_history()
except LitterRobotLoginException as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="invalid_credentials"
) from ex
except LitterRobotException as ex:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": str(ex)},
) from ex
await self.account.refresh_robots()
await self.account.load_pets()
for pet in self.account.pets:
# Need to fetch weight history for `get_visits_since`
await pet.fetch_weight_history()
async def _async_setup(self) -> None:
"""Set up the coordinator."""
@@ -74,15 +63,9 @@ class LitterRobotDataUpdateCoordinator(DataUpdateCoordinator[None]):
load_pets=True,
)
except LitterRobotLoginException as ex:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN, translation_key="invalid_credentials"
) from ex
raise ConfigEntryAuthFailed("Invalid credentials") from ex
except LitterRobotException as ex:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="cannot_connect",
translation_placeholders={"error": str(ex)},
) from ex
raise UpdateFailed("Unable to connect to Whisker API") from ex
def litter_robots(self) -> Generator[LitterRobot]:
"""Get Litter-Robots from the account."""

View File

@@ -1,24 +0,0 @@
"""Diagnostics support for Litter-Robot."""
from __future__ import annotations
from typing import Any
from pylitterbot.utils import REDACT_FIELDS
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.core import HomeAssistant
from .coordinator import LitterRobotConfigEntry
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: LitterRobotConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
account = entry.runtime_data.account
data = {
"robots": [robot.to_dict() for robot in account.robots],
"pets": [pet.to_dict() for pet in account.pets],
}
return async_redact_data(data, REDACT_FIELDS)

View File

@@ -2,14 +2,11 @@
from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from typing import Any, Concatenate, Generic, TypeVar
from typing import Generic, TypeVar
from pylitterbot import Pet, Robot
from pylitterbot.exceptions import LitterRobotException
from pylitterbot.robot import EVENT_UPDATE
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import EntityDescription
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@@ -20,26 +17,6 @@ from .coordinator import LitterRobotDataUpdateCoordinator
_WhiskerEntityT = TypeVar("_WhiskerEntityT", bound=Robot | Pet)
def whisker_command[_WhiskerEntityT2: LitterRobotEntity, **_P](
func: Callable[Concatenate[_WhiskerEntityT2, _P], Awaitable[None]],
) -> Callable[Concatenate[_WhiskerEntityT2, _P], Coroutine[Any, Any, None]]:
"""Wrap a Whisker command to handle exceptions."""
async def handler(
self: _WhiskerEntityT2, *args: _P.args, **kwargs: _P.kwargs
) -> None:
try:
await func(self, *args, **kwargs)
except LitterRobotException as ex:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="command_failed",
translation_placeholders={"error": str(ex)},
) from ex
return handler
def get_device_info(whisker_entity: Robot | Pet) -> DeviceInfo:
"""Get device info for a robot or pet."""
if isinstance(whisker_entity, Robot):

View File

@@ -23,16 +23,16 @@ rules:
unique-config-entry: done
# Silver
action-exceptions: done
action-exceptions: todo
config-entry-unloading: done
docs-configuration-parameters:
status: done
comment: No options to configure
docs-installation-parameters: done
entity-unavailable: done
entity-unavailable: todo
integration-owner: done
log-when-unavailable: done
parallel-updates: done
log-when-unavailable: todo
parallel-updates: todo
reauthentication-flow: done
test-coverage:
status: todo
@@ -42,28 +42,30 @@ rules:
# Gold
devices: done
diagnostics: done
diagnostics: todo
discovery-update-info:
status: done
comment: The integration is cloud-based
discovery:
status: todo
comment: Need to validate discovery
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: done
docs-troubleshooting: done
docs-use-cases: done
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: done
entity-device-class: done
entity-disabled-by-default: done
entity-translations: done
exception-translations: done
entity-translations:
status: todo
comment: Make sure all translated states are in sentence case
exception-translations: todo
icon-translations: done
reconfiguration-flow: done
reconfiguration-flow: todo
repair-issues:
status: done
comment: |

View File

@@ -15,9 +15,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import LitterRobotConfigEntry, LitterRobotDataUpdateCoordinator
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
from .entity import LitterRobotEntity, _WhiskerEntityT
_CastTypeT = TypeVar("_CastTypeT", int, float, str)
@@ -156,7 +154,6 @@ class LitterRobotSelectEntity(
"""Return the selected entity option to represent the entity state."""
return str(self.entity_description.current_fn(self.robot))
@whisker_command
async def async_select_option(self, option: str) -> None:
"""Change the selected option."""
await self.entity_description.select_fn(self.robot, option)

View File

@@ -23,8 +23,6 @@ from homeassistant.util import dt as dt_util
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT
PARALLEL_UPDATES = 0
def icon_for_gauge_level(gauge_level: int | None = None, offset: int = 0) -> str:
"""Return a gauge icon valid identifier."""

View File

@@ -3,7 +3,6 @@
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]",
"unique_id_mismatch": "The Whisker account does not match the previously configured account. Please re-authenticate using the same account, or remove this integration and set it up again if you want to use a different account."
},
"error": {
@@ -22,14 +21,6 @@
"description": "Please update your password for {username}",
"title": "[%key:common::config_flow::title::reauth%]"
},
"reconfigure": {
"data": {
"password": "[%key:common::config_flow::data::password%]"
},
"data_description": {
"password": "[%key:component::litterrobot::config::step::user::data_description::password%]"
}
},
"user": {
"data": {
"password": "[%key:common::config_flow::data::password%]",
@@ -204,20 +195,6 @@
}
}
},
"exceptions": {
"cannot_connect": {
"message": "Unable to fetch data from the Whisker API: {error}"
},
"command_failed": {
"message": "An error occurred while communicating with the device: {error}"
},
"firmware_update_failed": {
"message": "Unable to start firmware update on {name}"
},
"invalid_credentials": {
"message": "Invalid credentials. Please check your username and password, then try again"
}
},
"issues": {
"deprecated_entity": {
"description": "The Litter-Robot entity `{entity}` is deprecated and will be removed in a future release.\nPlease update your dashboards, automations and scripts, disable `{entity}` and reload the integration/restart Home Assistant to fix this issue.",

View File

@@ -25,9 +25,7 @@ from homeassistant.helpers.issue_registry import (
from .const import DOMAIN
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
from .entity import LitterRobotEntity, _WhiskerEntityT
@dataclass(frozen=True, kw_only=True)
@@ -137,12 +135,10 @@ class RobotSwitchEntity(LitterRobotEntity[_WhiskerEntityT], SwitchEntity):
"""Return true if switch is on."""
return self.entity_description.value_fn(self.robot)
@whisker_command
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self.entity_description.set_fn(self.robot, True)
@whisker_command
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self.entity_description.set_fn(self.robot, False)

View File

@@ -16,9 +16,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
from .entity import LitterRobotEntity, _WhiskerEntityT
@dataclass(frozen=True, kw_only=True)
@@ -76,7 +74,6 @@ class LitterRobotTimeEntity(LitterRobotEntity[_WhiskerEntityT], TimeEntity):
"""Return the value reported by the time."""
return self.entity_description.value_fn(self.robot)
@whisker_command
async def async_set_value(self, value: time) -> None:
"""Update the current value."""
await self.entity_description.set_fn(self.robot, value)

View File

@@ -17,11 +17,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, whisker_command
PARALLEL_UPDATES = 1
from .entity import LitterRobotEntity
SCAN_INTERVAL = timedelta(days=1)
@@ -83,15 +80,11 @@ class RobotUpdateEntity(LitterRobotEntity[LitterRobot4], UpdateEntity):
latest_version = self.robot.firmware
self._attr_latest_version = latest_version
@whisker_command
async def async_install(
self, version: str | None, backup: bool, **kwargs: Any
) -> None:
"""Install an update."""
if await self.robot.has_firmware_update(True):
if not await self.robot.update_firmware():
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="firmware_update_failed",
translation_placeholders={"name": self.robot.name},
)
message = f"Unable to start firmware update on {self.robot.name}"
raise HomeAssistantError(message)

View File

@@ -19,9 +19,7 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, whisker_command
PARALLEL_UPDATES = 1
from .entity import LitterRobotEntity
LITTER_BOX_STATUS_STATE_MAP = {
LitterBoxStatus.CLEAN_CYCLE: VacuumActivity.CLEANING,
@@ -68,18 +66,15 @@ class LitterRobotCleaner(LitterRobotEntity[LitterRobot], StateVacuumEntity):
"""Return the state of the cleaner."""
return LITTER_BOX_STATUS_STATE_MAP.get(self.robot.status, VacuumActivity.ERROR)
@whisker_command
async def async_start(self) -> None:
"""Start a clean cycle."""
await self.robot.set_power_status(True)
await self.robot.start_cleaning()
@whisker_command
async def async_stop(self, **kwargs: Any) -> None:
"""Stop the vacuum cleaner."""
await self.robot.set_power_status(False)
@whisker_command
async def async_set_sleep_mode(
self, enabled: bool, start_time: str | None = None
) -> None:

View File

@@ -187,27 +187,6 @@ DISCOVERY_SCHEMAS = [
# allow None value to account for 'default' value
allow_none_value=True,
),
MatterDiscoverySchema(
platform=Platform.NUMBER,
entity_description=MatterNumberEntityDescription(
key="power_on_level",
entity_category=EntityCategory.CONFIG,
translation_key="power_on_level",
native_max_value=255,
native_min_value=0,
mode=NumberMode.BOX,
# use 255 to indicate that the value should revert to the default
device_to_ha=lambda x: 255 if x is None else x,
ha_to_device=lambda x: None if x == 255 else int(x),
native_step=1,
native_unit_of_measurement=None,
),
entity_class=MatterNumber,
required_attributes=(clusters.LevelControl.Attributes.StartUpCurrentLevel,),
not_device_type=(device_types.Speaker,),
# allow None value to account for 'default' value
allow_none_value=True,
),
MatterDiscoverySchema(
platform=Platform.NUMBER,
entity_description=MatterNumberEntityDescription(

View File

@@ -238,9 +238,6 @@
"on_transition_time": {
"name": "On transition time"
},
"power_on_level": {
"name": "Power-on level"
},
"pump_setpoint": {
"name": "Setpoint"
},
@@ -325,11 +322,11 @@
}
},
"startup_on_off": {
"name": "Power-on behavior",
"name": "Power-on behavior on startup",
"state": {
"off": "[%key:common::state::off%]",
"on": "[%key:common::state::on%]",
"previous": "Previous state",
"previous": "Previous",
"toggle": "[%key:common::action::toggle%]"
}
},

View File

@@ -500,7 +500,6 @@ class DishWasherProgramId(MieleEnum, missing_to_none=True):
pasta_paela = 14
tall_items = 17, 42
glasses_warm = 19
quick_intense = 21
normal = 30
power_wash = 44, 204
comfort_wash = 203

View File

@@ -759,7 +759,6 @@
"pyrolytic": "Pyrolytic",
"quiche_lorraine": "Quiche Lorraine",
"quick_hygiene": "QuickHygiene",
"quick_intense": "QuickIntense",
"quick_mw": "Quick MW",
"quick_power_dry": "QuickPowerDry",
"quick_power_wash": "QuickPowerWash",

View File

@@ -6,20 +6,28 @@ from homeassistant.components.binary_sensor import (
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import (
EntityTargetStateTriggerBase,
EntityTriggerBase,
Trigger,
get_device_class_or_undefined,
)
class _MotionBinaryTriggerBase(EntityTriggerBase):
"""Base trigger for motion binary sensor state changes."""
_domain_specs = {
BINARY_SENSOR_DOMAIN: DomainSpec(device_class=BinarySensorDeviceClass.MOTION)
}
_domains = {BINARY_SENSOR_DOMAIN}
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities by motion device class."""
entities = super().entity_filter(entities)
return {
entity_id
for entity_id in entities
if get_device_class_or_undefined(self._hass, entity_id)
== BinarySensorDeviceClass.MOTION
}
class MotionDetectedTrigger(_MotionBinaryTriggerBase, EntityTargetStateTriggerBase):

View File

@@ -6,20 +6,28 @@ from homeassistant.components.binary_sensor import (
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import (
EntityTargetStateTriggerBase,
EntityTriggerBase,
Trigger,
get_device_class_or_undefined,
)
class _OccupancyBinaryTriggerBase(EntityTriggerBase):
"""Base trigger for occupancy binary sensor state changes."""
_domain_specs = {
BINARY_SENSOR_DOMAIN: DomainSpec(device_class=BinarySensorDeviceClass.OCCUPANCY)
}
_domains = {BINARY_SENSOR_DOMAIN}
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities by occupancy device class."""
entities = super().entity_filter(entities)
return {
entity_id
for entity_id in entities
if get_device_class_or_undefined(self._hass, entity_id)
== BinarySensorDeviceClass.OCCUPANCY
}
class OccupancyDetectedTrigger(

View File

@@ -12,5 +12,5 @@
"integration_type": "device",
"iot_class": "local_polling",
"quality_scale": "platinum",
"requirements": ["python-pooldose==0.8.5"]
"requirements": ["python-pooldose==0.8.2"]
}

View File

@@ -14,11 +14,13 @@ from .coordinator import PranaConfigEntry, PranaCoordinator
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.FAN, Platform.SWITCH]
# Keep platforms sorted alphabetically to satisfy lint rule
PLATFORMS = [Platform.SWITCH]
async def async_setup_entry(hass: HomeAssistant, entry: PranaConfigEntry) -> bool:
"""Set up Prana from a config entry."""
coordinator = PranaCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator

View File

@@ -1,186 +0,0 @@
"""Fan platform for Prana integration."""
from collections.abc import Callable
from dataclasses import dataclass
import math
from typing import Any
from prana_local_api_client.models.prana_state import FanState
from homeassistant.components.fan import (
FanEntity,
FanEntityDescription,
FanEntityFeature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.percentage import (
percentage_to_ranged_value,
ranged_value_to_percentage,
)
from homeassistant.util.scaling import int_states_in_range
from . import PranaConfigEntry
from .entity import PranaBaseEntity, PranaCoordinator, PranaEntityDescription, StrEnum
PARALLEL_UPDATES = 1
# The Prana device API expects fan speed values in scaled units (tenths of a speed step)
# rather than the raw step value used internally by this integration. This factor is
# applied when sending speeds to the API to match its expected units.
PRANA_SPEED_MULTIPLIER = 10
class PranaFanType(StrEnum):
"""Enumerates Prana fan types exposed by the device API."""
SUPPLY = "supply"
EXTRACT = "extract"
BOUNDED = "bounded"
@dataclass(frozen=True, kw_only=True)
class PranaFanEntityDescription(FanEntityDescription, PranaEntityDescription):
"""Description of a Prana fan entity."""
value_fn: Callable[[PranaCoordinator], FanState]
speed_range: Callable[[PranaCoordinator], tuple[int, int]]
ENTITIES: tuple[PranaEntityDescription, ...] = (
PranaFanEntityDescription(
key=PranaFanType.SUPPLY,
translation_key="supply",
value_fn=lambda coord: (
coord.data.supply if not coord.data.bound else coord.data.bounded
),
speed_range=lambda coord: (
1,
coord.data.supply.max_speed
if not coord.data.bound
else coord.data.bounded.max_speed,
),
),
PranaFanEntityDescription(
key=PranaFanType.EXTRACT,
translation_key="extract",
value_fn=lambda coord: (
coord.data.extract if not coord.data.bound else coord.data.bounded
),
speed_range=lambda coord: (
1,
coord.data.extract.max_speed
if not coord.data.bound
else coord.data.bounded.max_speed,
),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: PranaConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Prana fan entities from a config entry."""
async_add_entities(
PranaFan(entry.runtime_data, entity_description)
for entity_description in ENTITIES
)
class PranaFan(PranaBaseEntity, FanEntity):
"""Representation of a Prana fan entity."""
entity_description: PranaFanEntityDescription
_attr_preset_modes = ["night", "boost"]
_attr_supported_features = (
FanEntityFeature.SET_SPEED
| FanEntityFeature.TURN_ON
| FanEntityFeature.TURN_OFF
| FanEntityFeature.PRESET_MODE
)
@property
def _api_target_key(self) -> str:
"""Return the correct target key for API commands based on bounded state."""
# If the device is in bound mode, both supply and extract fans control the same bounded fan speeds.
if self.coordinator.data.bound:
return PranaFanType.BOUNDED
# Otherwise, return the specific fan type (supply or extract) for API commands.
return self.entity_description.key
@property
def speed_count(self) -> int:
"""Return the number of speeds the fan supports."""
return int_states_in_range(
self.entity_description.speed_range(self.coordinator)
)
@property
def percentage(self) -> int | None:
"""Return the current fan speed percentage."""
current_speed = self.entity_description.value_fn(self.coordinator).speed
return ranged_value_to_percentage(
self.entity_description.speed_range(self.coordinator), current_speed
)
async def async_set_percentage(self, percentage: int) -> None:
"""Set fan speed (0-100%) by converting to device-specific speed steps."""
if percentage == 0:
await self.async_turn_off()
return
await self.coordinator.api_client.set_speed(
math.ceil(
percentage_to_ranged_value(
self.entity_description.speed_range(self.coordinator),
percentage,
)
)
* PRANA_SPEED_MULTIPLIER,
self._api_target_key,
)
await self.coordinator.async_refresh()
@property
def is_on(self) -> bool:
"""Return true if the fan is on."""
return self.entity_description.value_fn(self.coordinator).is_on
async def async_turn_on(
self,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn the fan on and optionally set speed or preset mode."""
if percentage == 0:
await self.async_turn_off()
return
await self.coordinator.api_client.set_speed_is_on(True, self._api_target_key)
if percentage is not None:
await self.async_set_percentage(percentage)
if preset_mode is not None:
await self.async_set_preset_mode(preset_mode)
if percentage is None and preset_mode is None:
await self.coordinator.async_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the fan off."""
await self.coordinator.api_client.set_speed_is_on(False, self._api_target_key)
await self.coordinator.async_refresh()
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set the preset mode (e.g., night or boost)."""
await self.coordinator.api_client.set_switch(preset_mode, True)
await self.coordinator.async_refresh()
@property
def preset_mode(self) -> str | None:
"""Return the current preset mode."""
if self.coordinator.data.night:
return "night"
if self.coordinator.data.boost:
return "boost"
return None

View File

@@ -1,13 +1,5 @@
{
"entity": {
"fan": {
"extract": {
"default": "mdi:arrow-expand-right"
},
"supply": {
"default": "mdi:arrow-expand-left"
}
},
"switch": {
"auto": {
"default": "mdi:fan-auto"

View File

@@ -25,30 +25,6 @@
}
},
"entity": {
"fan": {
"extract": {
"name": "Extract fan",
"state_attributes": {
"preset_mode": {
"state": {
"boost": "Boost",
"night": "Night"
}
}
}
},
"supply": {
"name": "Supply fan",
"state_attributes": {
"preset_mode": {
"state": {
"boost": "[%key:component::prana::entity::fan::extract::state_attributes::preset_mode::state::boost%]",
"night": "[%key:component::prana::entity::fan::extract::state_attributes::preset_mode::state::night%]"
}
}
}
}
},
"switch": {
"auto": {
"name": "Auto"

View File

@@ -2,7 +2,6 @@
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
@@ -15,7 +14,7 @@ from . import DOMAIN
class SceneActivatedTrigger(EntityTriggerBase):
"""Trigger for scene entity activations."""
_domain_specs = {DOMAIN: DomainSpec()}
_domains = {DOMAIN}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:

View File

@@ -2,7 +2,6 @@
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import (
EntityTransitionTriggerBase,
Trigger,
@@ -15,7 +14,7 @@ from .const import ATTR_NEXT_EVENT, DOMAIN
class ScheduleBackToBackTrigger(EntityTransitionTriggerBase):
"""Trigger for back-to-back schedule blocks."""
_domain_specs = {DOMAIN: DomainSpec()}
_domains = {DOMAIN}
_from_states = {STATE_OFF, STATE_ON}
_to_states = {STATE_ON}

View File

@@ -74,11 +74,6 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
def format_zigbee_address(address: str) -> str:
"""Format a zigbee address to be more readable."""
return ":".join(address.lower()[i : i + 2] for i in range(0, 16, 2))
@dataclass
class SmartThingsData:
"""Define an object to hold SmartThings data."""
@@ -495,14 +490,6 @@ def create_devices(
kwargs[ATTR_CONNECTIONS] = {
(dr.CONNECTION_NETWORK_MAC, device.device.hub.mac_address)
}
if device.device.hub.hub_eui:
connections = kwargs.setdefault(ATTR_CONNECTIONS, set())
connections.add(
(
dr.CONNECTION_ZIGBEE,
format_zigbee_address(device.device.hub.hub_eui),
)
)
if device.device.parent_device_id and device.device.parent_device_id in devices:
kwargs[ATTR_VIA_DEVICE] = (DOMAIN, device.device.parent_device_id)
if (ocf := device.device.ocf) is not None:
@@ -526,10 +513,6 @@ def create_devices(
ATTR_SW_VERSION: viper.software_version,
}
)
if (zigbee := device.device.zigbee) is not None:
kwargs[ATTR_CONNECTIONS] = {
(dr.CONNECTION_ZIGBEE, format_zigbee_address(zigbee.eui))
}
if (matter := device.device.matter) is not None:
kwargs.update(
{

View File

@@ -200,14 +200,6 @@ CAPABILITY_TO_SENSORS: dict[
supported_states_attributes=Attribute.SUPPORTED_STATUS,
)
},
Capability.CUSTOM_COOKTOP_OPERATING_STATE: {
Attribute.COOKTOP_OPERATING_STATE: SmartThingsBinarySensorEntityDescription(
key=Attribute.COOKTOP_OPERATING_STATE,
translation_key="cooktop_operating_state",
is_on_key="run",
supported_states_attributes=Attribute.SUPPORTED_COOKTOP_OPERATING_STATE,
)
},
}

View File

@@ -25,10 +25,6 @@
"hostname": "hub*",
"macaddress": "286D97*"
},
{
"hostname": "smarthub",
"macaddress": "683A48*"
},
{
"hostname": "samsung-*"
}
@@ -38,5 +34,5 @@
"iot_class": "cloud_push",
"loggers": ["pysmartthings"],
"quality_scale": "bronze",
"requirements": ["pysmartthings==3.7.0"]
"requirements": ["pysmartthings==3.6.0"]
}

View File

@@ -49,9 +49,6 @@
"child_lock": {
"name": "Child lock"
},
"cooktop_operating_state": {
"name": "[%key:component::smartthings::entity::sensor::cooktop_operating_state::name%]"
},
"cool_select_plus_door": {
"name": "CoolSelect+ door"
},

View File

@@ -42,7 +42,7 @@
"username": "[%key:common::config_flow::data::username%]"
},
"description": "Please enter your MySubaru credentials\nNOTE: Initial setup may take up to 30 seconds",
"title": "MySubaru Connected Services configuration"
"title": "Subaru Starlink configuration"
}
}
},
@@ -95,7 +95,7 @@
"update_enabled": "Enable vehicle polling"
},
"description": "When enabled, vehicle polling will send a remote command to your vehicle every 2 hours to obtain new sensor data. Without vehicle polling, new sensor data is only received when the vehicle automatically pushes data (normally after engine shutdown).",
"title": "MySubaru Connected Services options"
"title": "Subaru Starlink options"
}
}
},

View File

@@ -7,6 +7,7 @@ import io
import logging
import os
from pathlib import Path
from ssl import SSLContext
from types import MappingProxyType
from typing import Any, cast
@@ -47,8 +48,8 @@ from homeassistant.const import (
from homeassistant.core import Context, HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.util.json import JsonValueType
from homeassistant.util.ssl import get_default_context, get_default_no_verify_context
from .const import (
ATTR_ARGS,
@@ -565,7 +566,11 @@ class TelegramNotificationService:
username=kwargs.get(ATTR_USERNAME, ""),
password=kwargs.get(ATTR_PASSWORD, ""),
authentication=kwargs.get(ATTR_AUTHENTICATION),
verify_ssl=kwargs.get(ATTR_VERIFY_SSL, False),
verify_ssl=(
get_default_context()
if kwargs.get(ATTR_VERIFY_SSL, False)
else get_default_no_verify_context()
),
)
media: InputMedia
@@ -733,7 +738,11 @@ class TelegramNotificationService:
username=kwargs.get(ATTR_USERNAME, ""),
password=kwargs.get(ATTR_PASSWORD, ""),
authentication=kwargs.get(ATTR_AUTHENTICATION),
verify_ssl=kwargs.get(ATTR_VERIFY_SSL, False),
verify_ssl=(
get_default_context()
if kwargs.get(ATTR_VERIFY_SSL, False)
else get_default_no_verify_context()
),
)
if file_type == SERVICE_SEND_PHOTO:
@@ -1046,7 +1055,7 @@ async def load_data(
username: str,
password: str,
authentication: str | None,
verify_ssl: bool,
verify_ssl: SSLContext,
num_retries: int = 5,
) -> io.BytesIO:
"""Load data into ByteIO/File container from a source."""
@@ -1062,13 +1071,16 @@ async def load_data(
elif authentication == HTTP_BASIC_AUTHENTICATION:
params["auth"] = httpx.BasicAuth(username, password)
if verify_ssl is not None:
params["verify"] = verify_ssl
retry_num = 0
async with get_async_client(hass, verify_ssl) as client:
async with httpx.AsyncClient(
timeout=DEFAULT_TIMEOUT_SECONDS, headers=headers, **params
) as client:
while retry_num < num_retries:
try:
response = await client.get(
url, headers=headers, timeout=DEFAULT_TIMEOUT_SECONDS, **params
)
req = await client.get(url)
except (httpx.HTTPError, httpx.InvalidURL) as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
@@ -1076,15 +1088,15 @@ async def load_data(
translation_placeholders={"error": str(err)},
) from err
if response.status_code != 200:
if req.status_code != 200:
_LOGGER.warning(
"Status code %s (retry #%s) loading %s",
response.status_code,
req.status_code,
retry_num + 1,
url,
)
else:
data = io.BytesIO(response.content)
data = io.BytesIO(req.content)
if data.read():
data.seek(0)
data.name = url
@@ -1099,7 +1111,7 @@ async def load_data(
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="failed_to_load_url",
translation_placeholders={"error": str(response.status_code)},
translation_placeholders={"error": str(req.status_code)},
)
elif filepath is not None:
if hass.config.is_allowed_path(filepath):

View File

@@ -2,7 +2,6 @@
from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import (
ENTITY_STATE_TRIGGER_SCHEMA,
EntityTriggerBase,
@@ -15,7 +14,7 @@ from .const import DOMAIN
class TextChangedTrigger(EntityTriggerBase):
"""Trigger for text entity when its content changes."""
_domain_specs = {DOMAIN: DomainSpec()}
_domains = {DOMAIN}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_state(self, state: State) -> bool:

View File

@@ -1,29 +0,0 @@
"""The TRMNL integration."""
from __future__ import annotations
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from .coordinator import TRMNLConfigEntry, TRMNLCoordinator
PLATFORMS: list[Platform] = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: TRMNLConfigEntry) -> bool:
"""Set up TRMNL from a config entry."""
coordinator = TRMNLCoordinator(hass, entry)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: TRMNLConfigEntry) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,49 +0,0 @@
"""Config flow for TRMNL."""
from __future__ import annotations
from typing import Any
from trmnl import TRMNLClient
from trmnl.exceptions import TRMNLAuthenticationError, TRMNLError
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_KEY
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, LOGGER
class TRMNLConfigFlow(ConfigFlow, domain=DOMAIN):
"""TRMNL config flow."""
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow initialized by the user."""
errors: dict[str, str] = {}
if user_input:
session = async_get_clientsession(self.hass)
client = TRMNLClient(token=user_input[CONF_API_KEY], session=session)
try:
user = await client.get_me()
except TRMNLAuthenticationError:
errors["base"] = "invalid_auth"
except TRMNLError:
errors["base"] = "cannot_connect"
except Exception: # noqa: BLE001
LOGGER.exception("Unexpected error")
errors["base"] = "unknown"
else:
await self.async_set_unique_id(str(user.identifier))
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user.name,
data={CONF_API_KEY: user_input[CONF_API_KEY]},
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
errors=errors,
)

View File

@@ -1,7 +0,0 @@
"""Constants for the TRMNL integration."""
import logging
DOMAIN = "trmnl"
LOGGER = logging.getLogger(__package__)

View File

@@ -1,57 +0,0 @@
"""Define an object to manage fetching TRMNL data."""
from __future__ import annotations
from datetime import timedelta
from trmnl import TRMNLClient
from trmnl.exceptions import TRMNLAuthenticationError, TRMNLError
from trmnl.models import Device
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, LOGGER
type TRMNLConfigEntry = ConfigEntry[TRMNLCoordinator]
class TRMNLCoordinator(DataUpdateCoordinator[dict[int, Device]]):
"""Class to manage fetching TRMNL data."""
config_entry: TRMNLConfigEntry
def __init__(self, hass: HomeAssistant, config_entry: TRMNLConfigEntry) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
logger=LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=timedelta(hours=1),
)
self.client = TRMNLClient(
token=config_entry.data[CONF_API_KEY],
session=async_get_clientsession(hass),
)
async def _async_update_data(self) -> dict[int, Device]:
"""Fetch data from TRMNL."""
try:
devices = await self.client.get_devices()
except TRMNLAuthenticationError as err:
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from err
except TRMNLError as err:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="update_error",
translation_placeholders={"error": str(err)},
) from err
return {device.identifier: device for device in devices}

View File

@@ -1,37 +0,0 @@
"""Base class for TRMNL entities."""
from __future__ import annotations
from trmnl.models import Device
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import TRMNLCoordinator
class TRMNLEntity(CoordinatorEntity[TRMNLCoordinator]):
"""Defines a base TRMNL entity."""
_attr_has_entity_name = True
def __init__(self, coordinator: TRMNLCoordinator, device_id: int) -> None:
"""Initialize TRMNL entity."""
super().__init__(coordinator)
self._device_id = device_id
device = self._device
self._attr_device_info = DeviceInfo(
connections={(CONNECTION_NETWORK_MAC, device.mac_address)},
name=device.name,
manufacturer="TRMNL",
)
@property
def _device(self) -> Device:
"""Return the device from coordinator data."""
return self.coordinator.data[self._device_id]
@property
def available(self) -> bool:
"""Return if the device is available."""
return super().available and self._device_id in self.coordinator.data

View File

@@ -1,11 +0,0 @@
{
"domain": "trmnl",
"name": "TRMNL",
"codeowners": ["@joostlek"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/trmnl",
"integration_type": "hub",
"iot_class": "cloud_polling",
"quality_scale": "bronze",
"requirements": ["trmnl==0.1.0"]
}

View File

@@ -1,74 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: This integration does not provide additional actions.
appropriate-polling: done
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: This integration does not provide additional actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup:
status: exempt
comment: Entities of this integration do not explicitly subscribe to events.
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: todo
config-entry-unloading: done
docs-configuration-parameters:
status: exempt
comment: There are no configuration parameters
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage: todo
# Gold
devices: done
diagnostics: todo
discovery-update-info:
status: exempt
comment: Uses the cloud API
discovery:
status: exempt
comment: Can't be discovered
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: done
exception-translations: todo
icon-translations: todo
reconfiguration-flow: todo
repair-issues:
status: exempt
comment: There are no repairable issues
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: done

View File

@@ -1,92 +0,0 @@
"""Support for TRMNL sensors."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from trmnl.models import Device
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.const import (
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
EntityCategory,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import TRMNLConfigEntry
from .coordinator import TRMNLCoordinator
from .entity import TRMNLEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class TRMNLSensorEntityDescription(SensorEntityDescription):
"""Describes a TRMNL sensor entity."""
value_fn: Callable[[Device], int | float | None]
SENSOR_DESCRIPTIONS: tuple[TRMNLSensorEntityDescription, ...] = (
TRMNLSensorEntityDescription(
key="battery",
device_class=SensorDeviceClass.BATTERY,
native_unit_of_measurement=PERCENTAGE,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda device: device.percent_charged,
),
TRMNLSensorEntityDescription(
key="rssi",
device_class=SensorDeviceClass.SIGNAL_STRENGTH,
native_unit_of_measurement=SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
state_class=SensorStateClass.MEASUREMENT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda device: device.rssi,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: TRMNLConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up TRMNL sensor entities based on a config entry."""
coordinator = entry.runtime_data
async_add_entities(
TRMNLSensor(coordinator, device_id, description)
for device_id in coordinator.data
for description in SENSOR_DESCRIPTIONS
)
class TRMNLSensor(TRMNLEntity, SensorEntity):
"""Defines a TRMNL sensor."""
entity_description: TRMNLSensorEntityDescription
def __init__(
self,
coordinator: TRMNLCoordinator,
device_id: int,
description: TRMNLSensorEntityDescription,
) -> None:
"""Initialize TRMNL sensor."""
super().__init__(coordinator, device_id)
self.entity_description = description
self._attr_unique_id = f"{device_id}_{description.key}"
@property
def native_value(self) -> int | float | None:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self._device)

View File

@@ -1,30 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
},
"data_description": {
"api_key": "The API key for your TRMNL account."
}
}
}
},
"exceptions": {
"authentication_error": {
"message": "Authentication failed. Please check your API key."
},
"update_error": {
"message": "An error occurred while communicating with TRMNL: {error}"
}
}
}

View File

@@ -5,11 +5,11 @@ from __future__ import annotations
from dataclasses import dataclass
from tuya_device_handlers.device_wrapper.base import DeviceWrapper
from tuya_device_handlers.device_wrapper.binary_sensor import (
DPCodeBitmapBitWrapper,
DPCodeInSetWrapper,
from tuya_device_handlers.device_wrapper.binary_sensor import DPCodeBitmapBitWrapper
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeWrapper,
)
from tuya_device_handlers.device_wrapper.common import DPCodeBooleanWrapper
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.binary_sensor import (
@@ -376,10 +376,29 @@ BINARY_SENSORS: dict[DeviceCategory, tuple[TuyaBinarySensorEntityDescription, ..
}
class _CustomDPCodeWrapper(DPCodeWrapper[bool]):
"""Custom DPCode Wrapper to check for values in a set."""
_valid_values: set[bool | float | int | str]
def __init__(
self, dpcode: str, valid_values: set[bool | float | int | str]
) -> None:
"""Init CustomDPCodeBooleanWrapper."""
super().__init__(dpcode)
self._valid_values = valid_values
def read_device_status(self, device: CustomerDevice) -> bool | None:
"""Read the device value for the dpcode."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
return raw_value in self._valid_values
def _get_dpcode_wrapper(
device: CustomerDevice,
description: TuyaBinarySensorEntityDescription,
) -> DeviceWrapper[bool] | None:
) -> DPCodeWrapper | None:
"""Get DPCode wrapper for an entity description."""
dpcode = description.dpcode or description.key
if description.bitmap_key is not None:
@@ -393,7 +412,7 @@ def _get_dpcode_wrapper(
# Legacy / compatibility
if dpcode not in device.status:
return None
return DPCodeInSetWrapper(
return _CustomDPCodeWrapper(
dpcode,
description.on_value
if isinstance(description.on_value, set)

View File

@@ -205,7 +205,7 @@ class _PresetWrapper(DPCodeEnumWrapper):
def read_device_status(self, device: CustomerDevice) -> str | None:
"""Read the device status."""
if (raw := self._read_dpcode_value(device)) not in self.options:
if (raw := self._read_dpcode_value(device)) in TUYA_HVAC_TO_HA:
return None
return raw

View File

@@ -8,25 +8,17 @@ from tuya_device_handlers.device_wrapper.base import DeviceWrapper
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeEnumWrapper,
DPCodeIntegerWrapper,
)
from tuya_device_handlers.type_information import IntegerTypeInformation
from tuya_device_handlers.utils import RemapHelper
from tuya_device_handlers.device_wrapper.fan import (
FanSpeedEnumWrapper,
FanSpeedIntegerWrapper,
)
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.fan import (
DIRECTION_FORWARD,
DIRECTION_REVERSE,
FanEntity,
FanEntityFeature,
)
from homeassistant.components.fan import FanEntity, FanEntityFeature
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
)
from . import TuyaConfigEntry
from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
@@ -54,19 +46,6 @@ TUYA_SUPPORT_TYPE: set[DeviceCategory] = {
}
class _DirectionEnumWrapper(DPCodeEnumWrapper):
"""Wrapper for fan direction DP code."""
def read_device_status(self, device: CustomerDevice) -> str | None:
"""Read the device status and return the direction string."""
if (value := self._read_dpcode_value(device)) and value in {
DIRECTION_FORWARD,
DIRECTION_REVERSE,
}:
return value
return None
def _has_a_valid_dpcode(device: CustomerDevice) -> bool:
"""Check if the device has at least one valid DP code."""
properties_to_check: list[DPCode | tuple[DPCode, ...] | None] = [
@@ -80,50 +59,15 @@ def _has_a_valid_dpcode(device: CustomerDevice) -> bool:
return any(get_dpcode(device, code) for code in properties_to_check)
class _FanSpeedEnumWrapper(DPCodeEnumWrapper[int]):
"""Wrapper for fan speed DP code (from an enum)."""
def read_device_status(self, device: CustomerDevice) -> int | None:
"""Get the current speed as a percentage."""
if (value := self._read_dpcode_value(device)) is None:
return None
return ordered_list_item_to_percentage(self.options, value)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value."""
return percentage_to_ordered_list_item(self.options, value)
class _FanSpeedIntegerWrapper(DPCodeIntegerWrapper[int]):
"""Wrapper for fan speed DP code (from an integer)."""
def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
"""Init DPCodeIntegerWrapper."""
super().__init__(dpcode, type_information)
self._remap_helper = RemapHelper.from_type_information(type_information, 1, 100)
def read_device_status(self, device: CustomerDevice) -> int | None:
"""Get the current speed as a percentage."""
if (value := self._read_dpcode_value(device)) is None:
return None
return round(self._remap_helper.remap_value_to(value))
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value."""
return round(self._remap_helper.remap_value_from(value))
def _get_speed_wrapper(
device: CustomerDevice,
) -> _FanSpeedEnumWrapper | _FanSpeedIntegerWrapper | None:
) -> DeviceWrapper[int] | None:
"""Get the speed wrapper for the device."""
if int_wrapper := _FanSpeedIntegerWrapper.find_dpcode(
if int_wrapper := FanSpeedIntegerWrapper.find_dpcode(
device, _SPEED_DPCODES, prefer_function=True
):
return int_wrapper
return _FanSpeedEnumWrapper.find_dpcode(
device, _SPEED_DPCODES, prefer_function=True
)
return FanSpeedEnumWrapper.find_dpcode(device, _SPEED_DPCODES, prefer_function=True)
async def async_setup_entry(
@@ -145,7 +89,7 @@ async def async_setup_entry(
TuyaFanEntity(
device,
manager,
direction_wrapper=_DirectionEnumWrapper.find_dpcode(
direction_wrapper=DPCodeEnumWrapper.find_dpcode(
device, _DIRECTION_DPCODES, prefer_function=True
),
mode_wrapper=DPCodeEnumWrapper.find_dpcode(

View File

@@ -490,9 +490,9 @@
}
},
"relay_status": {
"name": "Power-on behavior",
"name": "Power on behavior",
"state": {
"last": "Previous state",
"last": "Remember last state",
"memory": "[%key:component::tuya::entity::select::relay_status::state::last%]",
"off": "[%key:common::state::off%]",
"on": "[%key:common::state::on%]",

View File

@@ -76,7 +76,9 @@ async def async_remove_config_entry_device(
"""Remove config entry from a device."""
hub = config_entry.runtime_data
return not any(
identifier in hub.api.devices for _, identifier in device_entry.connections
identifier
for _, identifier in device_entry.connections
if identifier in hub.api.clients or identifier in hub.api.devices
)

View File

@@ -1,54 +0,0 @@
"""The UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import ApiAuthError, ApiConnectionError, UnifiAccessApiClient
from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_VERIFY_SSL, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .coordinator import UnifiAccessConfigEntry, UnifiAccessCoordinator
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.EVENT, Platform.SWITCH]
async def async_setup_entry(hass: HomeAssistant, entry: UnifiAccessConfigEntry) -> bool:
"""Set up UniFi Access from a config entry."""
session = async_get_clientsession(hass, verify_ssl=entry.data[CONF_VERIFY_SSL])
client = UnifiAccessApiClient(
host=entry.data[CONF_HOST],
api_token=entry.data[CONF_API_TOKEN],
session=session,
verify_ssl=entry.data[CONF_VERIFY_SSL],
)
try:
await client.authenticate()
except ApiAuthError as err:
raise ConfigEntryNotReady(
f"Authentication failed for UniFi Access at {entry.data[CONF_HOST]}"
) from err
except ApiConnectionError as err:
raise ConfigEntryNotReady(
f"Unable to connect to UniFi Access at {entry.data[CONF_HOST]}"
) from err
coordinator = UnifiAccessCoordinator(hass, entry, client)
await coordinator.async_config_entry_first_refresh()
entry.runtime_data = coordinator
entry.async_on_unload(client.close)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(
hass: HomeAssistant, entry: UnifiAccessConfigEntry
) -> bool:
"""Unload a config entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@@ -1,53 +0,0 @@
"""Button platform for the UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import Door, UnifiAccessError
from homeassistant.components.button import ButtonEntity
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import UnifiAccessConfigEntry, UnifiAccessCoordinator
from .entity import UnifiAccessEntity
PARALLEL_UPDATES = 1
async def async_setup_entry(
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up UniFi Access button entities."""
coordinator = entry.runtime_data
async_add_entities(
UnifiAccessUnlockButton(coordinator, door)
for door in coordinator.data.doors.values()
)
class UnifiAccessUnlockButton(UnifiAccessEntity, ButtonEntity):
"""Representation of a UniFi Access door unlock button."""
_attr_translation_key = "unlock"
def __init__(
self,
coordinator: UnifiAccessCoordinator,
door: Door,
) -> None:
"""Initialize the button entity."""
super().__init__(coordinator, door, "unlock")
async def async_press(self) -> None:
"""Unlock the door."""
try:
await self.coordinator.client.unlock_door(self._door_id)
except UnifiAccessError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unlock_failed",
) from err

View File

@@ -1,68 +0,0 @@
"""Config flow for UniFi Access integration."""
from __future__ import annotations
import logging
from typing import Any
from unifi_access_api import ApiAuthError, ApiConnectionError, UnifiAccessApiClient
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_API_TOKEN, CONF_HOST, CONF_VERIFY_SSL
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class UnifiAccessConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for UniFi Access."""
VERSION = 1
MINOR_VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
session = async_get_clientsession(
self.hass, verify_ssl=user_input[CONF_VERIFY_SSL]
)
client = UnifiAccessApiClient(
host=user_input[CONF_HOST],
api_token=user_input[CONF_API_TOKEN],
session=session,
verify_ssl=user_input[CONF_VERIFY_SSL],
)
try:
await client.authenticate()
except ApiAuthError:
errors["base"] = "invalid_auth"
except ApiConnectionError:
errors["base"] = "cannot_connect"
except Exception:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unknown"
else:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
return self.async_create_entry(
title="UniFi Access",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_API_TOKEN): str,
vol.Required(CONF_VERIFY_SSL, default=False): bool,
}
),
errors=errors,
)

View File

@@ -1,3 +0,0 @@
"""Constants for the UniFi Access integration."""
DOMAIN = "unifi_access"

View File

@@ -1,240 +0,0 @@
"""Data update coordinator for the UniFi Access integration."""
from __future__ import annotations
import asyncio
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import Any, cast
from unifi_access_api import (
ApiAuthError,
ApiConnectionError,
ApiError,
Door,
EmergencyStatus,
UnifiAccessApiClient,
WsMessageHandler,
)
from unifi_access_api.models.websocket import (
HwDoorbell,
InsightsAdd,
LocationUpdateState,
LocationUpdateV2,
SettingUpdate,
V2LocationState,
V2LocationUpdate,
WebsocketMessage,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
type UnifiAccessConfigEntry = ConfigEntry[UnifiAccessCoordinator]
@dataclass(frozen=True)
class DoorEvent:
"""Represent a door event from WebSocket."""
door_id: str
category: str
event_type: str
event_data: dict[str, Any]
@dataclass(frozen=True)
class UnifiAccessData:
"""Data provided by the UniFi Access coordinator."""
doors: dict[str, Door]
emergency: EmergencyStatus
class UnifiAccessCoordinator(DataUpdateCoordinator[UnifiAccessData]):
"""Coordinator for fetching UniFi Access door data."""
config_entry: UnifiAccessConfigEntry
def __init__(
self,
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
client: UnifiAccessApiClient,
) -> None:
"""Initialize the coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=entry,
name=DOMAIN,
update_interval=None,
)
self.client = client
self._event_listeners: list[Callable[[DoorEvent], None]] = []
@callback
def async_subscribe_door_events(
self,
event_callback: Callable[[DoorEvent], None],
) -> CALLBACK_TYPE:
"""Subscribe to door events (doorbell, access)."""
def _unsubscribe() -> None:
self._event_listeners.remove(event_callback)
self._event_listeners.append(event_callback)
return _unsubscribe
async def _async_setup(self) -> None:
"""Set up the WebSocket connection for push updates."""
handlers: dict[str, WsMessageHandler] = {
"access.data.device.location_update_v2": self._handle_location_update,
"access.data.v2.location.update": self._handle_v2_location_update,
"access.hw.door_bell": self._handle_doorbell,
"access.logs.insights.add": self._handle_insights_add,
"access.data.setting.update": self._handle_setting_update,
}
self.client.start_websocket(
handlers,
on_connect=self._on_ws_connect,
on_disconnect=self._on_ws_disconnect,
)
async def _async_update_data(self) -> UnifiAccessData:
"""Fetch all doors and emergency status from the API."""
try:
async with asyncio.timeout(10):
doors, emergency = await asyncio.gather(
self.client.get_doors(),
self.client.get_emergency_status(),
)
except ApiAuthError as err:
raise UpdateFailed(f"Authentication failed: {err}") from err
except ApiConnectionError as err:
raise UpdateFailed(f"Error connecting to API: {err}") from err
except ApiError as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
except TimeoutError as err:
raise UpdateFailed("Timeout communicating with UniFi Access API") from err
return UnifiAccessData(
doors={door.id: door for door in doors},
emergency=emergency,
)
def _on_ws_connect(self) -> None:
"""Handle WebSocket connection established."""
_LOGGER.debug("WebSocket connected to UniFi Access")
if not self.last_update_success:
self.config_entry.async_create_background_task(
self.hass,
self.async_request_refresh(),
"unifi_access_reconnect_refresh",
)
def _on_ws_disconnect(self) -> None:
"""Handle WebSocket disconnection."""
_LOGGER.warning("WebSocket disconnected from UniFi Access")
self.async_set_update_error(
UpdateFailed("WebSocket disconnected from UniFi Access")
)
async def _handle_location_update(self, msg: WebsocketMessage) -> None:
"""Handle location_update_v2 messages."""
update = cast(LocationUpdateV2, msg)
self._process_door_update(update.data.id, update.data.state)
async def _handle_v2_location_update(self, msg: WebsocketMessage) -> None:
"""Handle V2 location update messages."""
update = cast(V2LocationUpdate, msg)
self._process_door_update(update.data.id, update.data.state)
def _process_door_update(
self, door_id: str, ws_state: LocationUpdateState | V2LocationState | None
) -> None:
"""Process a door state update from WebSocket."""
if self.data is None or door_id not in self.data.doors:
return
if ws_state is None:
return
current_door = self.data.doors[door_id]
updates: dict[str, object] = {}
if ws_state.dps is not None:
updates["door_position_status"] = ws_state.dps
if ws_state.lock == "locked":
updates["door_lock_relay_status"] = "lock"
elif ws_state.lock == "unlocked":
updates["door_lock_relay_status"] = "unlock"
if not updates:
return
updated_door = current_door.with_updates(**updates)
self.async_set_updated_data(
UnifiAccessData(
doors={**self.data.doors, door_id: updated_door},
emergency=self.data.emergency,
)
)
async def _handle_setting_update(self, msg: WebsocketMessage) -> None:
"""Handle settings update messages (evacuation/lockdown)."""
if self.data is None:
return
update = cast(SettingUpdate, msg)
self.async_set_updated_data(
UnifiAccessData(
doors=self.data.doors,
emergency=EmergencyStatus(
evacuation=update.data.evacuation,
lockdown=update.data.lockdown,
),
)
)
async def _handle_doorbell(self, msg: WebsocketMessage) -> None:
"""Handle doorbell press events."""
doorbell = cast(HwDoorbell, msg)
self._dispatch_door_event(
doorbell.data.door_id,
"doorbell",
"ring",
{},
)
async def _handle_insights_add(self, msg: WebsocketMessage) -> None:
"""Handle access insights events (entry/exit)."""
insights = cast(InsightsAdd, msg)
door = insights.data.metadata.door
if not door.id:
return
event_type = (
"access_granted" if insights.data.result == "ACCESS" else "access_denied"
)
attrs: dict[str, Any] = {}
if insights.data.metadata.actor.display_name:
attrs["actor"] = insights.data.metadata.actor.display_name
if insights.data.metadata.authentication.display_name:
attrs["authentication"] = insights.data.metadata.authentication.display_name
if insights.data.result:
attrs["result"] = insights.data.result
self._dispatch_door_event(door.id, "access", event_type, attrs)
@callback
def _dispatch_door_event(
self,
door_id: str,
category: str,
event_type: str,
event_data: dict[str, Any],
) -> None:
"""Dispatch a door event to all subscribed listeners."""
event = DoorEvent(door_id, category, event_type, event_data)
for listener in self._event_listeners:
listener(event)

View File

@@ -1,58 +0,0 @@
"""Base entity for the UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import Door
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import UnifiAccessCoordinator
class UnifiAccessEntity(CoordinatorEntity[UnifiAccessCoordinator]):
"""Base entity for UniFi Access doors."""
_attr_has_entity_name = True
def __init__(
self,
coordinator: UnifiAccessCoordinator,
door: Door,
key: str,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self._door_id = door.id
self._attr_unique_id = f"{door.id}-{key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, door.id)},
name=door.name,
manufacturer="Ubiquiti",
)
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self._door_id in self.coordinator.data.doors
@property
def _door(self) -> Door:
"""Return the current door state from coordinator data."""
return self.coordinator.data.doors[self._door_id]
class UnifiAccessHubEntity(CoordinatorEntity[UnifiAccessCoordinator]):
"""Base entity for hub-level (controller-wide) UniFi Access entities."""
_attr_has_entity_name = True
def __init__(self, coordinator: UnifiAccessCoordinator) -> None:
"""Initialize the hub entity."""
super().__init__(coordinator)
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, coordinator.config_entry.entry_id)},
name="UniFi Access",
manufacturer="Ubiquiti",
)

View File

@@ -1,96 +0,0 @@
"""Event platform for the UniFi Access integration."""
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import DoorEvent, UnifiAccessConfigEntry, UnifiAccessCoordinator
from .entity import UnifiAccessEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class UnifiAccessEventEntityDescription(EventEntityDescription):
"""Describes a UniFi Access event entity."""
category: str
DOORBELL_EVENT_DESCRIPTION = UnifiAccessEventEntityDescription(
key="doorbell",
translation_key="doorbell",
device_class=EventDeviceClass.DOORBELL,
event_types=["ring"],
category="doorbell",
)
ACCESS_EVENT_DESCRIPTION = UnifiAccessEventEntityDescription(
key="access",
translation_key="access",
event_types=["access_granted", "access_denied"],
category="access",
)
EVENT_DESCRIPTIONS: list[UnifiAccessEventEntityDescription] = [
DOORBELL_EVENT_DESCRIPTION,
ACCESS_EVENT_DESCRIPTION,
]
async def async_setup_entry(
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up UniFi Access event entities."""
coordinator = entry.runtime_data
async_add_entities(
UnifiAccessEventEntity(coordinator, door_id, description)
for door_id in coordinator.data.doors
for description in EVENT_DESCRIPTIONS
)
class UnifiAccessEventEntity(UnifiAccessEntity, EventEntity):
"""Representation of a UniFi Access event entity."""
entity_description: UnifiAccessEventEntityDescription
def __init__(
self,
coordinator: UnifiAccessCoordinator,
door_id: str,
description: UnifiAccessEventEntityDescription,
) -> None:
"""Initialize the event entity."""
door = coordinator.data.doors[door_id]
super().__init__(coordinator, door, description.key)
self.entity_description = description
async def async_added_to_hass(self) -> None:
"""Subscribe to door events when added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self.coordinator.async_subscribe_door_events(self._async_handle_event)
)
@callback
def _async_handle_event(self, event: DoorEvent) -> None:
"""Handle incoming event from coordinator."""
if (
event.door_id != self._door_id
or event.category != self.entity_description.category
or event.event_type not in self.event_types
):
return
self._trigger_event(event.event_type, event.event_data)
self.async_write_ha_state()

View File

@@ -1,22 +0,0 @@
{
"entity": {
"button": {
"unlock": {
"default": "mdi:lock-open"
}
},
"event": {
"access": {
"default": "mdi:door"
}
},
"switch": {
"evacuation": {
"default": "mdi:exit-run"
},
"lockdown": {
"default": "mdi:lock-alert"
}
}
}
}

View File

@@ -1,12 +0,0 @@
{
"domain": "unifi_access",
"name": "UniFi Access",
"codeowners": ["@imhotep", "@RaHehl"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/unifi_access",
"integration_type": "hub",
"iot_class": "local_push",
"loggers": ["unifi_access_api"],
"quality_scale": "bronze",
"requirements": ["py-unifi-access==1.0.0"]
}

View File

@@ -1,66 +0,0 @@
rules:
# Bronze
action-setup:
status: exempt
comment: Integration does not register custom actions.
appropriate-polling:
status: exempt
comment: Integration uses WebSocket push updates, no polling.
brands: done
common-modules: done
config-flow-test-coverage: done
config-flow: done
dependency-transparency: done
docs-actions:
status: exempt
comment: Integration does not register custom actions.
docs-high-level-description: done
docs-installation-instructions: done
docs-removal-instructions: done
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done
# Silver
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters: todo
docs-installation-parameters: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: todo
test-coverage: done
# Gold
devices: done
diagnostics: todo
discovery-update-info: todo
discovery: todo
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-supported-functions: todo
docs-troubleshooting: todo
docs-use-cases: todo
dynamic-devices: todo
entity-category: todo
entity-device-class: todo
entity-disabled-by-default: todo
entity-translations: todo
exception-translations: done
icon-translations: done
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -1,72 +0,0 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"data": {
"api_token": "[%key:common::config_flow::data::api_token%]",
"host": "[%key:common::config_flow::data::host%]",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"data_description": {
"api_token": "API token generated in the UniFi Access settings.",
"host": "Hostname or IP address of the UniFi Access controller.",
"verify_ssl": "Verify the SSL certificate of the controller."
}
}
}
},
"entity": {
"button": {
"unlock": {
"name": "Unlock"
}
},
"event": {
"access": {
"name": "Access",
"state_attributes": {
"event_type": {
"state": {
"access_denied": "Access denied",
"access_granted": "Access granted"
}
}
}
},
"doorbell": {
"name": "Doorbell",
"state_attributes": {
"event_type": {
"state": {
"ring": "Ring"
}
}
}
}
},
"switch": {
"evacuation": {
"name": "Evacuation"
},
"lockdown": {
"name": "Lockdown"
}
}
},
"exceptions": {
"emergency_failed": {
"message": "Failed to set emergency status."
},
"unlock_failed": {
"message": "Failed to unlock the door."
}
}
}

View File

@@ -1,110 +0,0 @@
"""Switch platform for the UniFi Access integration."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from unifi_access_api import EmergencyStatus, UnifiAccessError
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN
from .coordinator import UnifiAccessConfigEntry, UnifiAccessCoordinator, UnifiAccessData
from .entity import UnifiAccessHubEntity
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class UnifiAccessSwitchEntityDescription(SwitchEntityDescription):
"""Describes a UniFi Access switch entity."""
value_fn: Callable[[EmergencyStatus], bool]
set_fn: Callable[[EmergencyStatus, bool], EmergencyStatus]
SWITCH_DESCRIPTIONS: tuple[UnifiAccessSwitchEntityDescription, ...] = (
UnifiAccessSwitchEntityDescription(
key="evacuation",
translation_key="evacuation",
value_fn=lambda s: s.evacuation,
set_fn=lambda s, v: EmergencyStatus(evacuation=v, lockdown=s.lockdown),
),
UnifiAccessSwitchEntityDescription(
key="lockdown",
translation_key="lockdown",
value_fn=lambda s: s.lockdown,
set_fn=lambda s, v: EmergencyStatus(evacuation=s.evacuation, lockdown=v),
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: UnifiAccessConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up UniFi Access switch entities."""
coordinator = entry.runtime_data
async_add_entities(
UnifiAccessEmergencySwitch(coordinator, description)
for description in SWITCH_DESCRIPTIONS
)
class UnifiAccessEmergencySwitch(UnifiAccessHubEntity, SwitchEntity):
"""Representation of a UniFi Access emergency switch."""
entity_description: UnifiAccessSwitchEntityDescription
def __init__(
self,
coordinator: UnifiAccessCoordinator,
description: UnifiAccessSwitchEntityDescription,
) -> None:
"""Initialize the switch entity."""
super().__init__(coordinator)
self._attr_unique_id = f"{coordinator.config_entry.entry_id}-{description.key}"
self.entity_description = description
@property
def is_on(self) -> bool:
"""Return True if the switch is on."""
return self.entity_description.value_fn(self.coordinator.data.emergency)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._async_set_emergency(True)
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self._async_set_emergency(False)
async def _async_set_emergency(self, value: bool) -> None:
"""Set emergency status."""
new_status = self.entity_description.set_fn(
self.coordinator.data.emergency, value
)
try:
await self.coordinator.client.set_emergency_status(new_status)
except UnifiAccessError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="emergency_failed",
) from err
# Optimistically update state; the WebSocket confirmation via
# access.data.setting.update typically arrives ~200ms later.
# Guard against flipping coordinator.last_update_success back to True
# while the WebSocket is disconnected and all entities are unavailable.
if self.coordinator.last_update_success:
self.coordinator.async_set_updated_data(
UnifiAccessData(
doors=self.coordinator.data.doors,
emergency=new_status,
)
)

View File

@@ -116,6 +116,13 @@ SENSOR_DESCRIPTIONS: tuple[UptimeKumaSensorEntityDescription, ...] = (
value_fn=lambda m: m.monitor_port,
create_entity=lambda t: t in HAS_PORT,
),
UptimeKumaSensorEntityDescription(
key=UptimeKumaSensor.PORT,
translation_key=UptimeKumaSensor.PORT,
entity_category=EntityCategory.DIAGNOSTIC,
value_fn=lambda m: m.monitor_port,
create_entity=lambda t: t in HAS_PORT,
),
UptimeKumaSensorEntityDescription(
key=UptimeKumaSensor.UPTIME_RATIO_1D,
translation_key=UptimeKumaSensor.UPTIME_RATIO_1D,

View File

@@ -398,7 +398,7 @@ SENSOR_DESCRIPTIONS = {
Keys.WARNING: VictronBLESensorEntityDescription(
key=Keys.WARNING,
device_class=SensorDeviceClass.ENUM,
translation_key="warning",
translation_key="alarm",
options=ALARM_OPTIONS,
),
Keys.YIELD_TODAY: VictronBLESensorEntityDescription(

View File

@@ -248,24 +248,7 @@
"name": "[%key:component::victron_ble::common::starter_voltage%]"
},
"warning": {
"name": "Warning",
"state": {
"bms_lockout": "[%key:component::victron_ble::entity::sensor::alarm::state::bms_lockout%]",
"dc_ripple": "[%key:component::victron_ble::entity::sensor::alarm::state::dc_ripple%]",
"high_starter_voltage": "[%key:component::victron_ble::entity::sensor::alarm::state::high_starter_voltage%]",
"high_temperature": "[%key:component::victron_ble::entity::sensor::alarm::state::high_temperature%]",
"high_v_ac_out": "[%key:component::victron_ble::entity::sensor::alarm::state::high_v_ac_out%]",
"high_voltage": "[%key:component::victron_ble::entity::sensor::alarm::state::high_voltage%]",
"low_soc": "[%key:component::victron_ble::entity::sensor::alarm::state::low_soc%]",
"low_starter_voltage": "[%key:component::victron_ble::entity::sensor::alarm::state::low_starter_voltage%]",
"low_temperature": "[%key:component::victron_ble::entity::sensor::alarm::state::low_temperature%]",
"low_v_ac_out": "[%key:component::victron_ble::entity::sensor::alarm::state::low_v_ac_out%]",
"low_voltage": "[%key:component::victron_ble::entity::sensor::alarm::state::low_voltage%]",
"mid_voltage": "[%key:component::victron_ble::common::midpoint_voltage%]",
"no_alarm": "[%key:component::victron_ble::entity::sensor::alarm::state::no_alarm%]",
"overload": "[%key:component::victron_ble::entity::sensor::alarm::state::overload%]",
"short_circuit": "[%key:component::victron_ble::entity::sensor::alarm::state::short_circuit%]"
}
"name": "Warning"
},
"yield_today": {
"name": "Yield today"

View File

@@ -7,5 +7,5 @@
"integration_type": "service",
"iot_class": "cloud_polling",
"loggers": ["pywaze", "homeassistant.helpers.location"],
"requirements": ["pywaze==1.2.0"]
"requirements": ["pywaze==1.1.1"]
}

View File

@@ -156,7 +156,6 @@ class WebDavBackupAgent(BackupAgent):
f"{self._backup_path}/{filename_tar}",
timeout=BACKUP_TIMEOUT,
content_length=backup.size,
progress=lambda current, total: on_progress(bytes_uploaded=current),
)
_LOGGER.debug(

View File

@@ -20,8 +20,14 @@ DEVICE_CLASSES_WINDOW: dict[str, str] = {
TRIGGERS: dict[str, type[Trigger]] = {
"opened": make_cover_opened_trigger(device_classes=DEVICE_CLASSES_WINDOW),
"closed": make_cover_closed_trigger(device_classes=DEVICE_CLASSES_WINDOW),
"opened": make_cover_opened_trigger(
device_classes=DEVICE_CLASSES_WINDOW,
domains={BINARY_SENSOR_DOMAIN, COVER_DOMAIN},
),
"closed": make_cover_closed_trigger(
device_classes=DEVICE_CLASSES_WINDOW,
domains={BINARY_SENSOR_DOMAIN, COVER_DOMAIN},
),
}

View File

@@ -103,12 +103,6 @@ ZEROCONF_PROPERTIES_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
# USB devices to ignore in serial port selection (non-Zigbee devices)
# Format: (manufacturer, description)
IGNORED_USB_DEVICES = {
("Nabu Casa", "ZWA-2"),
}
class OptionsMigrationIntent(StrEnum):
"""Zigbee options flow intents."""
@@ -182,12 +176,7 @@ async def list_serial_ports(hass: HomeAssistant) -> list[USBDevice]:
ports.append(addon_port)
# Filter out ignored USB devices
return [
port
for port in ports
if (port.manufacturer, port.description) not in IGNORED_USB_DEVICES
]
return ports
class BaseZhaFlow(ConfigEntryBaseFlow):

View File

@@ -23,7 +23,7 @@
"universal_silabs_flasher",
"serialx"
],
"requirements": ["zha==1.0.2", "serialx==0.6.2"],
"requirements": ["zha==1.0.1", "serialx==0.6.2"],
"usb": [
{
"description": "*2652*",

Some files were not shown because too many files have changed in this diff Show More