Compare commits

...

30 Commits

Author SHA1 Message Date
jbouwh
33af94fd14 Remove duplicatie statement 2026-03-14 11:19:19 +00:00
jbouwh
b662948113 Fix MQTT device tracker overrides via JSON state attributes without reset 2026-03-14 11:10:38 +00:00
Nathan Spencer
4fbb22e861 Update Whisker quality scale docs rules (#165510) 2026-03-14 11:38:29 +01:00
hanwg
45199a341f Pass web session to download files for Telegram bot (#165424) 2026-03-14 09:57:39 +01:00
Jan-Philipp Benecke
de5f42d7a0 Add progress reporting to WebDAV upload (#165398) 2026-03-14 08:35:47 +01:00
Artur Pragacz
4459dce73a Reorder code to group intent errors (#165431) 2026-03-13 18:58:19 -05:00
Artur Pragacz
a465905467 Remove speech parameter from service intent handler (#165225) 2026-03-13 18:57:16 -05:00
Raphael Hehl
a47faa3ced Add UniFi Access integration (#165404)
Co-authored-by: RaHehl <rahehl@users.noreply.github.com>
2026-03-14 00:00:18 +01:00
Josh
7276403ab9 Allow deleting UniFi client devices (#165505) 2026-03-13 23:06:58 +01:00
Raj Laud
018717af4f Fix victron_ble warning sensor using duplicate alarm translation key (#165502)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-03-13 22:23:54 +01:00
Norbert Rittel
274c2b8092 Shorten "Power-on behavior" name in matter to be consistent (#165490) 2026-03-13 21:22:49 +01:00
David Bishop
bfe15a55c9 Add entity-unavailable and log-when-unavailable (#165486)
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-13 20:20:55 +00:00
dvdinth
54ad67b810 Bump pyintelliclima dependency for IntelliClima integration (#165478)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Abílio Costa <abmantis@users.noreply.github.com>
2026-03-13 20:16:27 +00:00
Nathan Spencer
4d2732df6f Add diagnostics to Whisker (#165487) 2026-03-13 20:38:57 +01:00
Andres Ruiz
2be3291d8e Update brand name for Subaru integration (#165485) 2026-03-13 20:26:44 +01:00
Joost Lekkerkerker
4326cb96ea Add zigbee address to SmartThings devices (#165474)
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
2026-03-13 20:14:58 +01:00
Norbert Rittel
278894d4b4 Make "power-on behavior" states more consistent in tuya (#165344) 2026-03-13 18:53:32 +00:00
Ariel Ebersberger
eb17367229 Add DomainSpec to trigger and condition helpers (#165392) 2026-03-13 19:50:19 +01:00
Mike Degatano
d96191723f Improve error handling when addon unavailable for install/update (#165352) 2026-03-13 19:28:19 +01:00
mcisk
b6c7b2952e Add autoskope integration (#146772)
Co-authored-by: Franck Nijhof <git@frenck.dev>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-13 19:19:00 +01:00
David Bishop
356de12bce Add parallel-updates and action-exceptions for Whisker (#165433)
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-13 18:33:42 +01:00
epenet
57c49d0c48 Fix missing Tuya climate preset_mode (#165460) 2026-03-13 17:49:10 +01:00
Joost Lekkerkerker
af22b5fdbb Bump pySmartThings to 3.7.0 (#165468) 2026-03-13 17:12:15 +01:00
Joost Lekkerkerker
9c710961f0 Add Matter fixtures to SmartThings (#165466) 2026-03-13 17:09:38 +01:00
epenet
2a2da83173 Use external library wrapper in Tuya binary_sensor (#165465) 2026-03-13 17:05:52 +01:00
jvmahon
00a52245e3 Add Matter start-up Power-on level entity (#164775) 2026-03-13 17:04:12 +01:00
TheJulianJES
adb30e1ec1 Hide ZWA-2 adapter in Zigbee serial port selector (#155526) 2026-03-13 16:56:12 +01:00
TheJulianJES
34a7fcf8d3 Bump ZHA to 1.0.2 (#165423) 2026-03-13 16:15:51 +01:00
prana-dev-official
95a57a2984 Add fan platform for Prana Integration (#163379)
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
2026-03-13 16:05:37 +01:00
epenet
7f39cc0aeb Bump tuya-device-handlers to 0.0.12 (#165462) 2026-03-13 15:58:12 +01:00
153 changed files with 7391 additions and 785 deletions

4
CODEOWNERS generated
View File

@@ -186,6 +186,8 @@ 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
@@ -1784,6 +1786,8 @@ 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

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

View File

@@ -2,6 +2,7 @@
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,
@@ -44,7 +45,7 @@ def make_entity_state_trigger_required_features(
class CustomTrigger(EntityStateTriggerRequiredFeatures):
"""Trigger for entity state changes."""
_domains = {domain}
_domain_specs = {domain: DomainSpec()}
_to_states = {to_state}
_required_features = required_features

View File

@@ -0,0 +1,53 @@
"""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

@@ -0,0 +1,89 @@
"""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

@@ -0,0 +1,9 @@
"""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

@@ -0,0 +1,60 @@
"""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

@@ -0,0 +1,145 @@
"""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

@@ -0,0 +1,11 @@
{
"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

@@ -0,0 +1,88 @@
# + 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

@@ -0,0 +1,52 @@
{
"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,6 +2,7 @@
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,
@@ -14,7 +15,7 @@ from . import DOMAIN
class ButtonPressedTrigger(EntityTriggerBase):
"""Trigger for button entity presses."""
_domains = {DOMAIN}
_domain_specs = {DOMAIN: DomainSpec()}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:

View File

@@ -5,13 +5,14 @@ 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_attribute_changed_trigger,
make_entity_numerical_state_attribute_crossed_threshold_trigger,
make_entity_numerical_state_changed_trigger,
make_entity_numerical_state_crossed_threshold_trigger,
make_entity_target_state_attribute_trigger,
make_entity_target_state_trigger,
make_entity_transition_trigger,
@@ -35,7 +36,7 @@ HVAC_MODE_CHANGED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST.extend
class HVACModeChangedTrigger(EntityTargetStateTriggerBase):
"""Trigger for entity state changes."""
_domains = {DOMAIN}
_domain_specs = {DOMAIN: DomainSpec()}
_schema = HVAC_MODE_CHANGED_TRIGGER_SCHEMA
def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
@@ -52,17 +53,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_attribute_changed_trigger(
{DOMAIN}, {DOMAIN: ATTR_HUMIDITY}
"target_humidity_changed": make_entity_numerical_state_changed_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_HUMIDITY)}
),
"target_humidity_crossed_threshold": make_entity_numerical_state_attribute_crossed_threshold_trigger(
{DOMAIN}, {DOMAIN: ATTR_HUMIDITY}
"target_humidity_crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_HUMIDITY)}
),
"target_temperature_changed": make_entity_numerical_state_attribute_changed_trigger(
{DOMAIN}, {DOMAIN: ATTR_TEMPERATURE}
"target_temperature_changed": make_entity_numerical_state_changed_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_TEMPERATURE)}
),
"target_temperature_crossed_threshold": make_entity_numerical_state_attribute_crossed_threshold_trigger(
{DOMAIN}, {DOMAIN: ATTR_TEMPERATURE}
"target_temperature_crossed_threshold": make_entity_numerical_state_crossed_threshold_trigger(
{DOMAIN: NumericalDomainSpec(value_source=ATTR_TEMPERATURE)}
),
"turned_off": make_entity_target_state_trigger(DOMAIN, HVACMode.OFF),
"turned_on": make_entity_transition_trigger(

View File

@@ -15,7 +15,6 @@ 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},
@@ -27,7 +26,6 @@ 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,81 +1,82 @@
"""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.trigger import (
EntityTriggerBase,
Trigger,
get_device_class_or_undefined,
)
from homeassistant.helpers.automation import DomainSpec
from homeassistant.helpers.trigger import EntityTriggerBase, Trigger
from .const import ATTR_IS_CLOSED, DOMAIN, CoverDeviceClass
class CoverTriggerBase(EntityTriggerBase):
@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]):
"""Base trigger for cover state changes."""
_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 _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
def is_valid_state(self, state: State) -> bool:
"""Check if the state matches the target cover state."""
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
domain_spec = self._domain_specs[split_entity_id(state.entity_id)[0]]
return self._get_value(state) == domain_spec.target_value
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 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
if (from_value := self._get_value(from_state)) is None:
return False
return from_value != self._get_value(to_state)
def make_cover_opened_trigger(
*, device_classes: dict[str, str], domains: set[str] | None = None
*, device_classes: dict[str, str]
) -> type[CoverTriggerBase]:
"""Create a trigger cover_opened."""
class CoverOpenedTrigger(CoverTriggerBase):
"""Trigger for cover opened state changes."""
_binary_sensor_target_state = STATE_ON
_cover_is_closed_target_value = False
_domains = domains or {DOMAIN}
_device_classes = device_classes
_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()
}
return CoverOpenedTrigger
def make_cover_closed_trigger(
*, device_classes: dict[str, str], domains: set[str] | None = None
*, device_classes: dict[str, str]
) -> type[CoverTriggerBase]:
"""Create a trigger cover_closed."""
class CoverClosedTrigger(CoverTriggerBase):
"""Trigger for cover closed state changes."""
_binary_sensor_target_state = STATE_OFF
_cover_is_closed_target_value = True
_domains = domains or {DOMAIN}
_device_classes = device_classes
_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()
}
return CoverClosedTrigger

View File

@@ -20,14 +20,8 @@ DEVICE_CLASSES_DOOR: dict[str, str] = {
TRIGGERS: dict[str, type[Trigger]] = {
"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},
),
"opened": make_cover_opened_trigger(device_classes=DEVICE_CLASSES_DOOR),
"closed": make_cover_closed_trigger(device_classes=DEVICE_CLASSES_DOOR),
}

View File

@@ -20,14 +20,8 @@ DEVICE_CLASSES_GARAGE_DOOR: dict[str, str] = {
TRIGGERS: dict[str, type[Trigger]] = {
"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},
),
"opened": make_cover_opened_trigger(device_classes=DEVICE_CLASSES_GARAGE_DOOR),
"closed": make_cover_closed_trigger(device_classes=DEVICE_CLASSES_GARAGE_DOOR),
}

View File

@@ -10,7 +10,11 @@ from functools import partial, wraps
import logging
from typing import Any, Concatenate
from aiohasupervisor import SupervisorError
from aiohasupervisor import (
AddonNotSupportedError,
SupervisorError,
SupervisorNotFoundError,
)
from aiohasupervisor.models import (
AddonsOptions,
AddonState as SupervisorAddonState,
@@ -165,15 +169,7 @@ class AddonManager:
)
addon_info = await self._supervisor_client.addons.addon_info(self.addon_slug)
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,
)
return self._async_convert_installed_addon_info(addon_info)
@callback
def async_get_addon_state(self, addon_info: InstalledAddonComplete) -> AddonState:
@@ -189,6 +185,20 @@ 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,
@@ -199,21 +209,17 @@ 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."""
addon_info = await self.async_get_addon_info()
self._check_addon_available(addon_info)
await self._supervisor_client.store.install_addon(self.addon_slug)
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
@api_error(
"Failed to uninstall the {addon_name} app",
@@ -226,17 +232,29 @@ 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."""
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")
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
if not addon_info.update_available:
return
await self.async_create_backup()
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._supervisor_client.store.update_addon(
self.addon_slug, StoreAddonUpdate(backup=False)
)
@@ -266,10 +284,14 @@ class AddonManager:
"Failed to create a backup of the {addon_name} app",
expected_error_type=SupervisorError,
)
async def async_create_backup(self) -> None:
async def async_create_backup(self, *, addon_info: AddonInfo | None = None) -> None:
"""Create a partial backup of the managed add-on."""
addon_info = await self.async_get_addon_info()
name = f"addon_{self.addon_slug}_{addon_info.version}"
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}"
self._logger.debug("Creating backup: %s", name)
await self._supervisor_client.backups.partial_backup(

View File

@@ -15,50 +15,43 @@ from homeassistant.components.weather import (
ATTR_WEATHER_HUMIDITY,
DOMAIN as WEATHER_DOMAIN,
)
from homeassistant.core import HomeAssistant, split_entity_id
from homeassistant.core import HomeAssistant
from homeassistant.helpers.automation import NumericalDomainSpec
from homeassistant.helpers.trigger import (
EntityNumericalStateAttributeChangedTriggerBase,
EntityNumericalStateAttributeCrossedThresholdTriggerBase,
EntityTriggerBase,
Trigger,
get_device_class_or_undefined,
)
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
}
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 HumidityChangedTrigger(
_HumidityTriggerMixin, EntityNumericalStateAttributeChangedTriggerBase
):
class HumidityChangedTrigger(EntityNumericalStateAttributeChangedTriggerBase):
"""Trigger for humidity value changes across multiple domains."""
_domain_specs = HUMIDITY_DOMAIN_SPECS
class HumidityCrossedThresholdTrigger(
_HumidityTriggerMixin, EntityNumericalStateAttributeCrossedThresholdTriggerBase
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:
if device_data.speed_set == FanSpeed.auto_get:
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
device_data.speed_set == FanSpeed.auto_get
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(fan_mode=preset_mode, percentage=percentage)
await self.async_set_mode_speed(preset_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(fan_mode=preset_mode)
await self.async_set_mode_speed(preset_mode=preset_mode)
async def async_set_mode_speed(
self, fan_mode: str | None = None, percentage: int | None = None
self, preset_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 fan_mode == "auto":
if preset_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,21 +148,20 @@ class IntelliClimaVMCFan(IntelliClimaECOEntity, FanEntity):
return
# Determine the fan mode
if fan_mode is not None:
# Set to requested fan_mode
mode = fan_mode
elif not self.is_on:
if 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 = str(
math.ceil(
percentage_to_ranged_value(
self._speed_range,
percentage,
speed = FanSpeed(
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.2.2"]
"requirements": ["pyintelliclima==0.3.1"]
}

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
device_data.speed_set == FanSpeed.auto_get
and device_data.mode_set == FanMode.sensor
):
return None
return INTELLICLIMA_MODE_TO_FAN_MODE.get(FanMode(device_data.mode_set))
return INTELLICLIMA_MODE_TO_FAN_MODE.get(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
device_data.speed_set == FanSpeed.auto_get
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 IntelliClima devices found in your account",
"no_devices": "No supported IntelliClima devices were found in your account",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {

View File

@@ -4,6 +4,7 @@ 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,
@@ -20,13 +21,18 @@ 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."""
_domains = {DOMAIN}
_attributes = {DOMAIN: ATTR_BRIGHTNESS}
_converter = staticmethod(_convert_uint8_to_percentage)
_domain_specs = BRIGHTNESS_DOMAIN_SPECS
class BrightnessCrossedThresholdTrigger(
@@ -34,9 +40,7 @@ class BrightnessCrossedThresholdTrigger(
):
"""Trigger for brightness crossed threshold."""
_domains = {DOMAIN}
_attributes = {DOMAIN: ATTR_BRIGHTNESS}
_converter = staticmethod(_convert_uint8_to_percentage)
_domain_specs = BRIGHTNESS_DOMAIN_SPECS
TRIGGERS: dict[str, type[Trigger]] = {

View File

@@ -20,6 +20,8 @@ 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,7 +14,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
@@ -71,6 +73,7 @@ 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

@@ -46,11 +46,18 @@ class LitterRobotDataUpdateCoordinator(DataUpdateCoordinator[None]):
async def _async_update_data(self) -> None:
"""Update all device states from the Litter-Robot API."""
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()
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("Invalid credentials") from ex
except LitterRobotException as ex:
raise UpdateFailed(
f"Unable to fetch data from the Whisker API: {ex}"
) from ex
async def _async_setup(self) -> None:
"""Set up the coordinator."""

View File

@@ -0,0 +1,24 @@
"""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,11 +2,14 @@
from __future__ import annotations
from typing import Generic, TypeVar
from collections.abc import Awaitable, Callable, Coroutine
from typing import Any, Concatenate, 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
@@ -17,6 +20,26 @@ 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: todo
action-exceptions: done
config-entry-unloading: done
docs-configuration-parameters:
status: done
comment: No options to configure
docs-installation-parameters: done
entity-unavailable: todo
entity-unavailable: done
integration-owner: done
log-when-unavailable: todo
parallel-updates: todo
log-when-unavailable: done
parallel-updates: done
reauthentication-flow: done
test-coverage:
status: todo
@@ -42,20 +42,20 @@ rules:
# Gold
devices: done
diagnostics: todo
diagnostics: done
discovery-update-info:
status: done
comment: The integration is cloud-based
discovery:
status: todo
comment: Need to validate discovery
docs-data-update: todo
docs-examples: todo
docs-known-limitations: todo
docs-supported-devices: todo
docs-data-update: done
docs-examples: done
docs-known-limitations: done
docs-supported-devices: done
docs-supported-functions: done
docs-troubleshooting: todo
docs-use-cases: todo
docs-troubleshooting: done
docs-use-cases: done
dynamic-devices: todo
entity-category: done
entity-device-class: done

View File

@@ -15,7 +15,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import LitterRobotConfigEntry, LitterRobotDataUpdateCoordinator
from .entity import LitterRobotEntity, _WhiskerEntityT
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
_CastTypeT = TypeVar("_CastTypeT", int, float, str)
@@ -154,6 +156,7 @@ 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,6 +23,8 @@ 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

@@ -195,6 +195,14 @@
}
}
},
"exceptions": {
"command_failed": {
"message": "An error occurred while communicating with the device: {error}"
},
"firmware_update_failed": {
"message": "Unable to start firmware update on {name}"
}
},
"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,7 +25,9 @@ from homeassistant.helpers.issue_registry import (
from .const import DOMAIN
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
@@ -135,10 +137,12 @@ 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,7 +16,9 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity, _WhiskerEntityT
from .entity import LitterRobotEntity, _WhiskerEntityT, whisker_command
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
@@ -74,6 +76,7 @@ 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,8 +17,11 @@ 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
from .entity import LitterRobotEntity, whisker_command
PARALLEL_UPDATES = 1
SCAN_INTERVAL = timedelta(days=1)
@@ -80,11 +83,15 @@ 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():
message = f"Unable to start firmware update on {self.robot.name}"
raise HomeAssistantError(message)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="firmware_update_failed",
translation_placeholders={"name": self.robot.name},
)

View File

@@ -19,7 +19,9 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util import dt as dt_util
from .coordinator import LitterRobotConfigEntry
from .entity import LitterRobotEntity
from .entity import LitterRobotEntity, whisker_command
PARALLEL_UPDATES = 1
LITTER_BOX_STATUS_STATE_MAP = {
LitterBoxStatus.CLEAN_CYCLE: VacuumActivity.CLEANING,
@@ -66,15 +68,18 @@ 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,6 +187,27 @@ 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,6 +238,9 @@
"on_transition_time": {
"name": "On transition time"
},
"power_on_level": {
"name": "Power-on level"
},
"pump_setpoint": {
"name": "Setpoint"
},
@@ -322,11 +325,11 @@
}
},
"startup_on_off": {
"name": "Power-on behavior on startup",
"name": "Power-on behavior",
"state": {
"off": "[%key:common::state::off%]",
"on": "[%key:common::state::on%]",
"previous": "Previous",
"previous": "Previous state",
"toggle": "[%key:common::action::toggle%]"
}
},

View File

@@ -6,28 +6,20 @@ 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."""
_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
}
_domain_specs = {
BINARY_SENSOR_DOMAIN: DomainSpec(device_class=BinarySensorDeviceClass.MOTION)
}
class MotionDetectedTrigger(_MotionBinaryTriggerBase, EntityTargetStateTriggerBase):

View File

@@ -163,8 +163,6 @@ class MqttDeviceTracker(MqttEntity, TrackerEntity):
latitude: float | None
longitude: float | None
gps_accuracy: float
# Reset manually set location to allow automatic zone detection
self._attr_location_name = None
if isinstance(
latitude := extra_state_attributes.get(ATTR_LATITUDE), (int, float)
) and isinstance(

View File

@@ -6,28 +6,20 @@ 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."""
_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
}
_domain_specs = {
BINARY_SENSOR_DOMAIN: DomainSpec(device_class=BinarySensorDeviceClass.OCCUPANCY)
}
class OccupancyDetectedTrigger(

View File

@@ -14,13 +14,11 @@ from .coordinator import PranaConfigEntry, PranaCoordinator
_LOGGER = logging.getLogger(__name__)
# Keep platforms sorted alphabetically to satisfy lint rule
PLATFORMS = [Platform.SWITCH]
PLATFORMS = [Platform.FAN, 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

@@ -0,0 +1,186 @@
"""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,5 +1,13 @@
{
"entity": {
"fan": {
"extract": {
"default": "mdi:arrow-expand-right"
},
"supply": {
"default": "mdi:arrow-expand-left"
}
},
"switch": {
"auto": {
"default": "mdi:fan-auto"

View File

@@ -25,6 +25,30 @@
}
},
"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,6 +2,7 @@
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,
@@ -14,7 +15,7 @@ from . import DOMAIN
class SceneActivatedTrigger(EntityTriggerBase):
"""Trigger for scene entity activations."""
_domains = {DOMAIN}
_domain_specs = {DOMAIN: DomainSpec()}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_transition(self, from_state: State, to_state: State) -> bool:

View File

@@ -2,6 +2,7 @@
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,
@@ -14,7 +15,7 @@ from .const import ATTR_NEXT_EVENT, DOMAIN
class ScheduleBackToBackTrigger(EntityTransitionTriggerBase):
"""Trigger for back-to-back schedule blocks."""
_domains = {DOMAIN}
_domain_specs = {DOMAIN: DomainSpec()}
_from_states = {STATE_OFF, STATE_ON}
_to_states = {STATE_ON}

View File

@@ -74,6 +74,11 @@ 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."""
@@ -490,6 +495,14 @@ 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:
@@ -513,6 +526,10 @@ 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

@@ -34,5 +34,5 @@
"iot_class": "cloud_push",
"loggers": ["pysmartthings"],
"quality_scale": "bronze",
"requirements": ["pysmartthings==3.6.0"]
"requirements": ["pysmartthings==3.7.0"]
}

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": "Subaru Starlink configuration"
"title": "MySubaru Connected Services 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": "Subaru Starlink options"
"title": "MySubaru Connected Services options"
}
}
},

View File

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

View File

@@ -2,6 +2,7 @@
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,
@@ -14,7 +15,7 @@ from .const import DOMAIN
class TextChangedTrigger(EntityTriggerBase):
"""Trigger for text entity when its content changes."""
_domains = {DOMAIN}
_domain_specs = {DOMAIN: DomainSpec()}
_schema = ENTITY_STATE_TRIGGER_SCHEMA
def is_valid_state(self, state: State) -> bool:

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
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeWrapper,
from tuya_device_handlers.device_wrapper.binary_sensor import (
DPCodeBitmapBitWrapper,
DPCodeInSetWrapper,
)
from tuya_device_handlers.device_wrapper.common import DPCodeBooleanWrapper
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.binary_sensor import (
@@ -376,29 +376,10 @@ 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,
) -> DPCodeWrapper | None:
) -> DeviceWrapper[bool] | None:
"""Get DPCode wrapper for an entity description."""
dpcode = description.dpcode or description.key
if description.bitmap_key is not None:
@@ -412,7 +393,7 @@ def _get_dpcode_wrapper(
# Legacy / compatibility
if dpcode not in device.status:
return None
return _CustomDPCodeWrapper(
return DPCodeInSetWrapper(
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)) in TUYA_HVAC_TO_HA:
if (raw := self._read_dpcode_value(device)) not in self.options:
return None
return raw

View File

@@ -44,7 +44,7 @@
"iot_class": "cloud_push",
"loggers": ["tuya_sharing"],
"requirements": [
"tuya-device-handlers==0.0.11",
"tuya-device-handlers==0.0.12",
"tuya-device-sharing-sdk==0.2.8"
]
}

View File

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

View File

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

View File

@@ -0,0 +1,54 @@
"""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]
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

@@ -0,0 +1,52 @@
"""Button platform for the UniFi Access integration."""
from __future__ import annotations
from unifi_access_api import ApiError, Door
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.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 ApiError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="unlock_failed",
) from err

View File

@@ -0,0 +1,68 @@
"""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

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

View File

@@ -0,0 +1,128 @@
"""Data update coordinator for the UniFi Access integration."""
from __future__ import annotations
import asyncio
import logging
from typing import cast
from unifi_access_api import (
ApiAuthError,
ApiConnectionError,
ApiError,
Door,
UnifiAccessApiClient,
WsMessageHandler,
)
from unifi_access_api.models.websocket import (
LocationUpdateState,
LocationUpdateV2,
V2LocationState,
V2LocationUpdate,
WebsocketMessage,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
type UnifiAccessConfigEntry = ConfigEntry[UnifiAccessCoordinator]
class UnifiAccessCoordinator(DataUpdateCoordinator[dict[str, Door]]):
"""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
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,
}
self.client.start_websocket(
handlers,
on_connect=self._on_ws_connect,
on_disconnect=self._on_ws_disconnect,
)
async def _async_update_data(self) -> dict[str, Door]:
"""Fetch all doors from the API."""
try:
async with asyncio.timeout(10):
doors = await self.client.get_doors()
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
return {door.id: door for door in doors}
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.debug("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:
return
if ws_state is None:
return
current_door = self.data[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"
updated_door = current_door.with_updates(**updates)
self.async_set_updated_data({**self.data, door_id: updated_door})

View File

@@ -0,0 +1,43 @@
"""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
@property
def _door(self) -> Door:
"""Return the current door state from coordinator data."""
return self.coordinator.data[self._door_id]

View File

@@ -0,0 +1,9 @@
{
"entity": {
"button": {
"unlock": {
"default": "mdi:lock-open"
}
}
}
}

View File

@@ -0,0 +1,12 @@
{
"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

@@ -0,0 +1,66 @@
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: todo
integration-owner: done
log-when-unavailable: todo
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: todo
reconfiguration-flow: todo
repair-issues: todo
stale-devices: todo
# Platinum
async-dependency: done
inject-websession: done
strict-typing: todo

View File

@@ -0,0 +1,38 @@
{
"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"
}
}
},
"exceptions": {
"unlock_failed": {
"message": "Failed to unlock the door."
}
}
}

View File

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

View File

@@ -248,7 +248,24 @@
"name": "[%key:component::victron_ble::common::starter_voltage%]"
},
"warning": {
"name": "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%]"
}
},
"yield_today": {
"name": "Yield today"

View File

@@ -156,6 +156,7 @@ 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,14 +20,8 @@ DEVICE_CLASSES_WINDOW: dict[str, str] = {
TRIGGERS: dict[str, type[Trigger]] = {
"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},
),
"opened": make_cover_opened_trigger(device_classes=DEVICE_CLASSES_WINDOW),
"closed": make_cover_closed_trigger(device_classes=DEVICE_CLASSES_WINDOW),
}

View File

@@ -103,6 +103,12 @@ 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."""
@@ -176,7 +182,12 @@ async def list_serial_ports(hass: HomeAssistant) -> list[USBDevice]:
ports.append(addon_port)
return ports
# Filter out ignored USB devices
return [
port
for port in ports
if (port.manufacturer, port.description) not in IGNORED_USB_DEVICES
]
class BaseZhaFlow(ConfigEntryBaseFlow):

View File

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

View File

@@ -82,6 +82,7 @@ FLOWS = {
"aurora_abb_powerone",
"aussie_broadband",
"autarco",
"autoskope",
"awair",
"aws_s3",
"axis",
@@ -750,6 +751,7 @@ FLOWS = {
"uhoo",
"ukraine_alarm",
"unifi",
"unifi_access",
"unifiprotect",
"upb",
"upcloud",

View File

@@ -647,6 +647,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"autoskope": {
"name": "Autoskope",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"avion": {
"name": "Avi-on",
"integration_type": "hub",
@@ -7373,6 +7379,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"unifi_access": {
"name": "UniFi Access",
"integration_type": "hub",
"config_flow": true,
"iot_class": "local_push"
},
"universal": {
"name": "Universal media player",
"integration_type": "hub",

View File

@@ -1,14 +1,68 @@
"""Helpers for automation."""
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from enum import Enum
from typing import Any
import voluptuous as vol
from homeassistant.const import CONF_OPTIONS
from homeassistant.core import HomeAssistant, split_entity_id
from .entity import get_device_class_or_undefined
from .typing import ConfigType
class AnyDeviceClassType(Enum):
"""Singleton type for matching any device class."""
_singleton = 0
ANY_DEVICE_CLASS = AnyDeviceClassType._singleton # noqa: SLF001
@dataclass(frozen=True, slots=True)
class DomainSpec:
"""Describes how to match and extract a value from an entity.
Used by triggers and conditions.
"""
device_class: str | None | AnyDeviceClassType = ANY_DEVICE_CLASS
value_source: str | None = None
"""Attribute name to extract the value from, or None for state.state."""
@dataclass(frozen=True, slots=True)
class NumericalDomainSpec(DomainSpec):
"""DomainSpec with an optional value converter for numerical triggers."""
value_converter: Callable[[Any], float] | None = None
"""Optional converter for numerical values (e.g. uint8 → percentage)."""
def filter_by_domain_specs(
hass: HomeAssistant,
domain_specs: Mapping[str, DomainSpec],
entities: set[str],
) -> set[str]:
"""Filter entities matching any of the domain specs."""
result: set[str] = set()
for entity_id in entities:
if not (domain_spec := domain_specs.get(split_entity_id(entity_id)[0])):
continue
if (
domain_spec.device_class is not ANY_DEVICE_CLASS
and get_device_class_or_undefined(hass, entity_id)
!= domain_spec.device_class
):
continue
result.add(entity_id)
return result
def get_absolute_description_key(domain: str, key: str) -> str:
"""Return the absolute description key."""
if not key.startswith("_"):

View File

@@ -4,7 +4,7 @@ from __future__ import annotations
import abc
from collections import deque
from collections.abc import Callable, Container, Coroutine, Generator, Iterable
from collections.abc import Callable, Container, Coroutine, Generator, Iterable, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, time as dt_time, timedelta
@@ -54,7 +54,7 @@ from homeassistant.const import (
STATE_UNKNOWN,
WEEKDAYS,
)
from homeassistant.core import HomeAssistant, State, callback, split_entity_id
from homeassistant.core import HomeAssistant, State, callback
from homeassistant.exceptions import (
ConditionError,
ConditionErrorContainer,
@@ -76,6 +76,8 @@ from homeassistant.util.yaml import load_yaml_dict
from . import config_validation as cv, entity_registry as er, selector
from .automation import (
DomainSpec,
filter_by_domain_specs,
get_absolute_description_key,
get_relative_description_key,
move_options_fields_to_top_level,
@@ -332,10 +334,10 @@ ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL = vol.Schema(
)
class EntityConditionBase(Condition):
class EntityConditionBase[DomainSpecT: DomainSpec = DomainSpec](Condition):
"""Base class for entity conditions."""
_domain: str
_domain_specs: Mapping[str, DomainSpecT]
_schema: vol.Schema = ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL
@override
@@ -356,12 +358,8 @@ class EntityConditionBase(Condition):
self._behavior = config.options[ATTR_BEHAVIOR]
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities of this domain."""
return {
entity_id
for entity_id in entities
if split_entity_id(entity_id)[0] == self._domain
}
"""Filter entities matching any of the domain specs."""
return filter_by_domain_specs(self._hass, self._domain_specs, entities)
@abc.abstractmethod
def is_valid_state(self, entity_state: State) -> bool:
@@ -428,7 +426,7 @@ def make_entity_state_condition(
class CustomCondition(EntityStateConditionBase):
"""Condition for entity state."""
_domain = domain
_domain_specs = {domain: DomainSpec()}
_states = states_set
return CustomCondition
@@ -458,7 +456,7 @@ def make_entity_state_attribute_condition(
class CustomCondition(EntityStateAttributeConditionBase):
"""Condition for entity attribute."""
_domain = domain
_domain_specs = {domain: DomainSpec()}
_attribute = attribute
_attribute_states = attribute_states_set

View File

@@ -169,6 +169,16 @@ def get_device_class(hass: HomeAssistant, entity_id: str) -> str | None:
return entry.device_class or entry.original_device_class
def get_device_class_or_undefined(
hass: HomeAssistant, entity_id: str
) -> str | None | UndefinedType:
"""Get the device class of an entity or UNDEFINED if not found."""
try:
return get_device_class(hass, entity_id)
except HomeAssistantError:
return UNDEFINED
def get_supported_features(hass: HomeAssistant, entity_id: str) -> int:
"""Get supported features for an entity.

View File

@@ -184,6 +184,52 @@ class IntentUnexpectedError(IntentError):
"""Unexpected error while handling intent."""
class MatchFailedError(IntentError):
"""Error when target matching fails."""
def __init__(
self,
result: MatchTargetsResult,
constraints: MatchTargetsConstraints,
preferences: MatchTargetsPreferences | None = None,
) -> None:
"""Initialize error."""
super().__init__()
self.result = result
self.constraints = constraints
self.preferences = preferences
def __str__(self) -> str:
"""Return string representation."""
return f"<MatchFailedError result={self.result}, constraints={self.constraints}, preferences={self.preferences}>"
class NoStatesMatchedError(MatchFailedError):
"""Error when no states match the intent's constraints."""
def __init__(
self,
reason: MatchFailedReason,
name: str | None = None,
area: str | None = None,
floor: str | None = None,
domains: set[str] | None = None,
device_classes: set[str] | None = None,
) -> None:
"""Initialize error."""
super().__init__(
result=MatchTargetsResult(False, reason),
constraints=MatchTargetsConstraints(
name=name,
area_name=area,
floor_name=floor,
domains=domains,
device_classes=device_classes,
),
)
class MatchFailedReason(Enum):
"""Possible reasons for match failure in async_match_targets."""
@@ -232,6 +278,29 @@ class MatchFailedReason(Enum):
)
@dataclass
class MatchTargetsResult:
"""Result from async_match_targets."""
is_match: bool
"""True if one or more entities matched."""
no_match_reason: MatchFailedReason | None = None
"""Reason for failed match when is_match = False."""
states: list[State] = field(default_factory=list)
"""List of matched entity states."""
no_match_name: str | None = None
"""Name of invalid area/floor or duplicate name when match fails for those reasons."""
areas: list[ar.AreaEntry] = field(default_factory=list)
"""Areas that were targeted."""
floors: list[fr.FloorEntry] = field(default_factory=list)
"""Floors that were targeted."""
@dataclass
class MatchTargetsConstraints:
"""Constraints for async_match_targets."""
@@ -292,75 +361,6 @@ class MatchTargetsPreferences:
"""Id of floor to use when deduplicating names."""
@dataclass
class MatchTargetsResult:
"""Result from async_match_targets."""
is_match: bool
"""True if one or more entities matched."""
no_match_reason: MatchFailedReason | None = None
"""Reason for failed match when is_match = False."""
states: list[State] = field(default_factory=list)
"""List of matched entity states."""
no_match_name: str | None = None
"""Name of invalid area/floor or duplicate name when match fails for those reasons."""
areas: list[ar.AreaEntry] = field(default_factory=list)
"""Areas that were targeted."""
floors: list[fr.FloorEntry] = field(default_factory=list)
"""Floors that were targeted."""
class MatchFailedError(IntentError):
"""Error when target matching fails."""
def __init__(
self,
result: MatchTargetsResult,
constraints: MatchTargetsConstraints,
preferences: MatchTargetsPreferences | None = None,
) -> None:
"""Initialize error."""
super().__init__()
self.result = result
self.constraints = constraints
self.preferences = preferences
def __str__(self) -> str:
"""Return string representation."""
return f"<MatchFailedError result={self.result}, constraints={self.constraints}, preferences={self.preferences}>"
class NoStatesMatchedError(MatchFailedError):
"""Error when no states match the intent's constraints."""
def __init__(
self,
reason: MatchFailedReason,
name: str | None = None,
area: str | None = None,
floor: str | None = None,
domains: set[str] | None = None,
device_classes: set[str] | None = None,
) -> None:
"""Initialize error."""
super().__init__(
result=MatchTargetsResult(False, reason),
constraints=MatchTargetsConstraints(
name=name,
area_name=area,
floor_name=floor,
domains=domains,
device_classes=device_classes,
),
)
@dataclass
class MatchTargetsCandidate:
"""Candidate for async_match_targets."""
@@ -915,7 +915,7 @@ class DynamicServiceIntentHandler(IntentHandler):
def __init__(
self,
intent_type: str,
speech: str | None = None,
*,
required_slots: _IntentSlotsType | None = None,
optional_slots: _IntentSlotsType | None = None,
required_domains: set[str] | None = None,
@@ -927,7 +927,6 @@ class DynamicServiceIntentHandler(IntentHandler):
) -> None:
"""Create Service Intent Handler."""
self.intent_type = intent_type
self.speech = speech
self.required_domains = required_domains
self.required_features = required_features
self.required_states = required_states
@@ -1114,7 +1113,6 @@ class DynamicServiceIntentHandler(IntentHandler):
)
for floor in match_result.floors
)
speech_name = match_result.floors[0].name
elif match_result.areas:
success_results.extend(
IntentResponseTarget(
@@ -1122,9 +1120,6 @@ class DynamicServiceIntentHandler(IntentHandler):
)
for area in match_result.areas
)
speech_name = match_result.areas[0].name
else:
speech_name = states[0].name
service_coros: list[Coroutine[Any, Any, None]] = []
for state in states:
@@ -1166,9 +1161,6 @@ class DynamicServiceIntentHandler(IntentHandler):
states = [hass.states.get(state.entity_id) or state for state in states]
response.async_set_states(states)
if self.speech is not None:
response.async_set_speech(self.speech.format(speech_name))
return response
async def async_call_service(
@@ -1231,7 +1223,7 @@ class ServiceIntentHandler(DynamicServiceIntentHandler):
intent_type: str,
domain: str,
service: str,
speech: str | None = None,
*,
required_slots: _IntentSlotsType | None = None,
optional_slots: _IntentSlotsType | None = None,
required_domains: set[str] | None = None,
@@ -1244,7 +1236,6 @@ class ServiceIntentHandler(DynamicServiceIntentHandler):
"""Create service handler."""
super().__init__(
intent_type,
speech=speech,
required_slots=required_slots,
optional_slots=optional_slots,
required_domains=required_domains,

View File

@@ -5,7 +5,7 @@ from __future__ import annotations
import abc
import asyncio
from collections import defaultdict
from collections.abc import Callable, Coroutine, Iterable
from collections.abc import Callable, Coroutine, Iterable, Mapping
from dataclasses import dataclass, field
from enum import StrEnum
import functools
@@ -69,11 +69,13 @@ from homeassistant.util.yaml import load_yaml_dict
from . import config_validation as cv, selector
from .automation import (
DomainSpec,
NumericalDomainSpec,
filter_by_domain_specs,
get_absolute_description_key,
get_relative_description_key,
move_options_fields_to_top_level,
)
from .entity import get_device_class
from .integration_platform import async_process_integration_platforms
from .selector import TargetSelector
from .target import (
@@ -81,7 +83,7 @@ from .target import (
async_track_target_selector_state_change_event,
)
from .template import Template
from .typing import UNDEFINED, ConfigType, TemplateVarsType, UndefinedType
from .typing import ConfigType, TemplateVarsType
_LOGGER = logging.getLogger(__name__)
@@ -334,20 +336,10 @@ ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST = ENTITY_STATE_TRIGGER_SCHEMA.extend(
)
def get_device_class_or_undefined(
hass: HomeAssistant, entity_id: str
) -> str | None | UndefinedType:
"""Get the device class of an entity or UNDEFINED if not found."""
try:
return get_device_class(hass, entity_id)
except HomeAssistantError:
return UNDEFINED
class EntityTriggerBase(Trigger):
class EntityTriggerBase[DomainSpecT: DomainSpec = DomainSpec](Trigger):
"""Trigger for entity state changes."""
_domains: set[str]
_domain_specs: Mapping[str, DomainSpecT]
_schema: vol.Schema = ENTITY_STATE_TRIGGER_SCHEMA_FIRST_LAST
@override
@@ -366,6 +358,10 @@ class EntityTriggerBase(Trigger):
self._options = config.options or {}
self._target = config.target
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities matching any of the domain specs."""
return filter_by_domain_specs(self._hass, self._domain_specs, entities)
def is_valid_transition(self, from_state: State, to_state: State) -> bool:
"""Check if the origin state is valid and the state has changed."""
if from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
@@ -396,14 +392,6 @@ class EntityTriggerBase(Trigger):
== 1
)
def entity_filter(self, entities: set[str]) -> set[str]:
"""Filter entities of these domains."""
return {
entity_id
for entity_id in entities
if split_entity_id(entity_id)[0] in self._domains
}
@override
async def async_attach_runner(
self, run_action: TriggerActionRunner
@@ -611,19 +599,22 @@ def _get_numerical_value(
return entity_or_float
class EntityNumericalStateBase(EntityTriggerBase):
class EntityNumericalStateBase(EntityTriggerBase[NumericalDomainSpec]):
"""Base class for numerical state and state attribute triggers."""
_attributes: dict[str, str | None]
_converter: Callable[[Any], float] = float
def _get_tracked_value(self, state: State) -> Any:
"""Get the tracked numerical value from a state."""
domain = split_entity_id(state.entity_id)[0]
source = self._attributes[domain]
if source is None:
domain_spec = self._domain_specs[split_entity_id(state.entity_id)[0]]
if domain_spec.value_source is None:
return state.state
return state.attributes.get(source)
return state.attributes.get(domain_spec.value_source)
def _get_converter(self, state: State) -> Callable[[Any], float]:
"""Get the value converter for an entity."""
domain_spec = self._domain_specs[split_entity_id(state.entity_id)[0]]
if domain_spec.value_converter is not None:
return domain_spec.value_converter
return float
class EntityNumericalStateAttributeChangedTriggerBase(EntityNumericalStateBase):
@@ -654,7 +645,7 @@ class EntityNumericalStateAttributeChangedTriggerBase(EntityNumericalStateBase):
return False
try:
current_value = self._converter(_attribute_value)
current_value = self._get_converter(state)(_attribute_value)
except TypeError, ValueError:
# Value is not a valid number, don't trigger
return False
@@ -780,7 +771,7 @@ class EntityNumericalStateAttributeCrossedThresholdTriggerBase(
return False
try:
current_value = self._converter(_attribute_value)
current_value = self._get_converter(state)(_attribute_value)
except TypeError, ValueError:
# Value is not a valid number, don't trigger
return False
@@ -812,7 +803,7 @@ def make_entity_target_state_trigger(
class CustomTrigger(EntityTargetStateTriggerBase):
"""Trigger for entity state changes."""
_domains = {domain}
_domain_specs = {domain: DomainSpec()}
_to_states = to_states_set
return CustomTrigger
@@ -826,7 +817,7 @@ def make_entity_transition_trigger(
class CustomTrigger(EntityTransitionTriggerBase):
"""Trigger for conditional entity state changes."""
_domains = {domain}
_domain_specs = {domain: DomainSpec()}
_from_states = from_states
_to_states = to_states
@@ -841,36 +832,34 @@ def make_entity_origin_state_trigger(
class CustomTrigger(EntityOriginStateTriggerBase):
"""Trigger for entity "from state" changes."""
_domains = {domain}
_domain_specs = {domain: DomainSpec()}
_from_state = from_state
return CustomTrigger
def make_entity_numerical_state_attribute_changed_trigger(
domains: set[str], attributes: dict[str, str | None]
def make_entity_numerical_state_changed_trigger(
domain_specs: Mapping[str, NumericalDomainSpec],
) -> type[EntityNumericalStateAttributeChangedTriggerBase]:
"""Create a trigger for numerical state attribute change."""
"""Create a trigger for numerical state value change."""
class CustomTrigger(EntityNumericalStateAttributeChangedTriggerBase):
"""Trigger for numerical state attribute changes."""
"""Trigger for numerical state value changes."""
_domains = domains
_attributes = attributes
_domain_specs = domain_specs
return CustomTrigger
def make_entity_numerical_state_attribute_crossed_threshold_trigger(
domains: set[str], attributes: dict[str, str | None]
def make_entity_numerical_state_crossed_threshold_trigger(
domain_specs: Mapping[str, NumericalDomainSpec],
) -> type[EntityNumericalStateAttributeCrossedThresholdTriggerBase]:
"""Create a trigger for numerical state attribute change."""
"""Create a trigger for numerical state value crossing a threshold."""
class CustomTrigger(EntityNumericalStateAttributeCrossedThresholdTriggerBase):
"""Trigger for numerical state attribute changes."""
"""Trigger for numerical state value crossing a threshold."""
_domains = domains
_attributes = attributes
_domain_specs = domain_specs
return CustomTrigger
@@ -883,7 +872,7 @@ def make_entity_target_state_attribute_trigger(
class CustomTrigger(EntityTargetStateAttributeTriggerBase):
"""Trigger for entity state changes."""
_domains = {domain}
_domain_specs = {domain: DomainSpec()}
_attribute = attribute
_attribute_to_state = to_state

14
requirements_all.txt generated
View File

@@ -579,6 +579,9 @@ autarco==3.2.0
# homeassistant.components.husqvarna_automower_ble
automower-ble==0.2.8
# homeassistant.components.autoskope
autoskope_client==1.4.1
# homeassistant.components.generic
# homeassistant.components.stream
av==16.0.1
@@ -1879,6 +1882,9 @@ py-sucks==0.9.11
# homeassistant.components.synology_dsm
py-synologydsm-api==2.7.3
# homeassistant.components.unifi_access
py-unifi-access==1.0.0
# homeassistant.components.atome
pyAtome==0.1.1
@@ -2155,7 +2161,7 @@ pyicloud==2.4.1
pyinsteon==1.6.4
# homeassistant.components.intelliclima
pyintelliclima==0.2.2
pyintelliclima==0.3.1
# homeassistant.components.intesishome
pyintesishome==1.8.0
@@ -2482,7 +2488,7 @@ pysmappee==0.2.29
pysmarlaapi==1.0.2
# homeassistant.components.smartthings
pysmartthings==3.6.0
pysmartthings==3.7.0
# homeassistant.components.smarty
pysmarty2==0.10.3
@@ -3130,7 +3136,7 @@ ttls==1.8.3
ttn_client==1.2.3
# homeassistant.components.tuya
tuya-device-handlers==0.0.11
tuya-device-handlers==0.0.12
# homeassistant.components.tuya
tuya-device-sharing-sdk==0.2.8
@@ -3356,7 +3362,7 @@ zeroconf==0.148.0
zeversolar==0.3.2
# homeassistant.components.zha
zha==1.0.1
zha==1.0.2
# homeassistant.components.zhong_hong
zhong-hong-hvac==1.0.13

View File

@@ -537,6 +537,9 @@ autarco==3.2.0
# homeassistant.components.husqvarna_automower_ble
automower-ble==0.2.8
# homeassistant.components.autoskope
autoskope_client==1.4.1
# homeassistant.components.generic
# homeassistant.components.stream
av==16.0.1
@@ -1628,6 +1631,9 @@ py-sucks==0.9.11
# homeassistant.components.synology_dsm
py-synologydsm-api==2.7.3
# homeassistant.components.unifi_access
py-unifi-access==1.0.0
# homeassistant.components.hdmi_cec
pyCEC==0.5.2
@@ -1844,7 +1850,7 @@ pyicloud==2.4.1
pyinsteon==1.6.4
# homeassistant.components.intelliclima
pyintelliclima==0.2.2
pyintelliclima==0.3.1
# homeassistant.components.ipma
pyipma==3.0.9
@@ -2114,7 +2120,7 @@ pysmappee==0.2.29
pysmarlaapi==1.0.2
# homeassistant.components.smartthings
pysmartthings==3.6.0
pysmartthings==3.7.0
# homeassistant.components.smarty
pysmarty2==0.10.3
@@ -2633,7 +2639,7 @@ ttls==1.8.3
ttn_client==1.2.3
# homeassistant.components.tuya
tuya-device-handlers==0.0.11
tuya-device-handlers==0.0.12
# homeassistant.components.tuya
tuya-device-sharing-sdk==0.2.8
@@ -2829,7 +2835,7 @@ zeroconf==0.148.0
zeversolar==0.3.2
# homeassistant.components.zha
zha==1.0.1
zha==1.0.2
# homeassistant.components.zinvolt
zinvolt==0.3.0

View File

@@ -0,0 +1,12 @@
"""Tests for Autoskope integration."""
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def setup_integration(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the Autoskope integration."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()

View File

@@ -0,0 +1,69 @@
"""Test fixtures for Autoskope integration."""
from collections.abc import Generator
from json import loads
from unittest.mock import AsyncMock, patch
from autoskope_client.models import Vehicle
import pytest
from homeassistant.components.autoskope.const import DEFAULT_HOST, DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from tests.common import MockConfigEntry, load_fixture
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Autoskope (test_user)",
data={
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
CONF_HOST: DEFAULT_HOST,
},
unique_id=f"test_user@{DEFAULT_HOST}",
entry_id="01AUTOSKOPE_TEST_ENTRY",
)
@pytest.fixture
def mock_vehicles() -> list[Vehicle]:
"""Return a list of mock vehicles from fixture data."""
data = loads(load_fixture("vehicles.json", DOMAIN))
return [
Vehicle.from_api(vehicle, data["positions"]) for vehicle in data["vehicles"]
]
@pytest.fixture
def mock_autoskope_client(mock_vehicles: list[Vehicle]) -> Generator[AsyncMock]:
"""Mock the Autoskope API client."""
with (
patch(
"homeassistant.components.autoskope.AutoskopeApi",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.autoskope.config_flow.AutoskopeApi",
new=mock_client,
),
):
client = mock_client.return_value
client.connect.return_value = None
client.get_vehicles.return_value = mock_vehicles
client.__aenter__.return_value = client
client.__aexit__.return_value = None
yield client
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.autoskope.async_setup_entry",
return_value=True,
) as mock_setup:
yield mock_setup

View File

@@ -0,0 +1,33 @@
{
"vehicles": [
{
"id": "12345",
"name": "Test Vehicle",
"ex_pow": 12.5,
"bat_pow": 3.7,
"hdop": 1.2,
"support_infos": {
"imei": "123456789012345"
},
"model": "Autoskope"
}
],
"positions": {
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [8.6821267, 50.1109221]
},
"properties": {
"carid": "12345",
"s": 0,
"dt": "2025-05-28T10:00:00Z",
"park": false
}
}
]
}
}

View File

@@ -0,0 +1,55 @@
# serializer version: 1
# name: test_all_entities[device_tracker.test_vehicle-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'device_tracker',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'device_tracker.test_vehicle',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': 'mdi:car',
'original_name': None,
'platform': 'autoskope',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '12345',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[device_tracker.test_vehicle-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Test Vehicle',
'gps_accuracy': 6.0,
'icon': 'mdi:car',
'latitude': 50.1109221,
'longitude': 8.6821267,
'source_type': <SourceType.GPS: 'gps'>,
}),
'context': <ANY>,
'entity_id': 'device_tracker.test_vehicle',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'not_home',
})
# ---

View File

@@ -0,0 +1,167 @@
"""Test Autoskope config flow."""
from unittest.mock import AsyncMock
from autoskope_client.models import CannotConnect, InvalidAuth
import pytest
from homeassistant.components.autoskope.const import (
DEFAULT_HOST,
DOMAIN,
SECTION_ADVANCED_SETTINGS,
)
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
USER_INPUT = {
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
SECTION_ADVANCED_SETTINGS: {
CONF_HOST: DEFAULT_HOST,
},
}
async def test_full_flow(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test full user config flow from form to entry creation."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "Autoskope (test_user)"
assert result["data"] == {
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
CONF_HOST: DEFAULT_HOST,
}
assert result["result"].unique_id == f"test_user@{DEFAULT_HOST}"
@pytest.mark.parametrize(
("exception", "error"),
[
(InvalidAuth("Invalid credentials"), "invalid_auth"),
(CannotConnect("Connection failed"), "cannot_connect"),
],
)
async def test_flow_errors(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
exception: Exception,
error: str,
) -> None:
"""Test config flow error handling with recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_autoskope_client.__aenter__.side_effect = exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": error}
# Recovery: clear the error and retry
mock_autoskope_client.__aenter__.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_flow_invalid_url(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test config flow rejects invalid URL with recovery."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
SECTION_ADVANCED_SETTINGS: {
CONF_HOST: "not-a-valid-url",
},
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "invalid_url"}
# Recovery: provide a valid URL
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_already_configured(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
) -> None:
"""Test aborting if already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT,
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_custom_host(
hass: HomeAssistant,
mock_autoskope_client: AsyncMock,
mock_setup_entry: AsyncMock,
) -> None:
"""Test config flow with a custom white-label host."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test_user",
CONF_PASSWORD: "test_password",
SECTION_ADVANCED_SETTINGS: {
CONF_HOST: "https://custom.autoskope.server",
},
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_HOST] == "https://custom.autoskope.server"
assert result["result"].unique_id == "test_user@https://custom.autoskope.server"

View File

@@ -0,0 +1,232 @@
"""Test Autoskope device tracker."""
from unittest.mock import AsyncMock
from autoskope_client.models import CannotConnect, InvalidAuth, Vehicle, VehiclePosition
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.autoskope.const import DOMAIN, UPDATE_INTERVAL
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from . import setup_integration
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
async def test_all_entities(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test all entities with snapshot."""
await setup_integration(hass, mock_config_entry)
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("speed", "park_mode", "has_position", "expected_icon"),
[
(50, False, True, "mdi:car-arrow-right"),
(0, True, True, "mdi:car-brake-parking"),
(2, False, True, "mdi:car"),
(0, False, False, "mdi:car-clock"),
],
ids=["moving", "parked", "idle", "no_position"],
)
async def test_vehicle_icons(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
speed: int,
park_mode: bool,
has_position: bool,
expected_icon: str,
) -> None:
"""Test device tracker icon for different vehicle states."""
position = (
VehiclePosition(
latitude=50.1109221,
longitude=8.6821267,
speed=speed,
timestamp="2025-05-28T10:00:00Z",
park_mode=park_mode,
)
if has_position
else None
)
mock_autoskope_client.get_vehicles.return_value = [
Vehicle(
id="12345",
name="Test Vehicle",
position=position,
external_voltage=12.5,
battery_voltage=3.7,
gps_quality=1.2,
imei="123456789012345",
model="Autoskope",
)
]
await setup_integration(hass, mock_config_entry)
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.attributes["icon"] == expected_icon
async def test_entity_unavailable_on_coordinator_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity becomes unavailable when coordinator update fails."""
await setup_integration(hass, mock_config_entry)
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state != STATE_UNAVAILABLE
# Simulate connection error on next update
mock_autoskope_client.get_vehicles.side_effect = CannotConnect("Connection lost")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_entity_recovers_after_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
mock_vehicles: list[Vehicle],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity recovers after a transient coordinator error."""
await setup_integration(hass, mock_config_entry)
# Simulate error
mock_autoskope_client.get_vehicles.side_effect = CannotConnect("Connection lost")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert hass.states.get("device_tracker.test_vehicle").state == STATE_UNAVAILABLE
# Recover
mock_autoskope_client.get_vehicles.side_effect = None
mock_autoskope_client.get_vehicles.return_value = mock_vehicles
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state.state != STATE_UNAVAILABLE
async def test_reauth_success(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
mock_vehicles: list[Vehicle],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity stays available after successful re-authentication."""
await setup_integration(hass, mock_config_entry)
# First get_vehicles raises InvalidAuth, retry after authenticate succeeds
mock_autoskope_client.get_vehicles.side_effect = [
InvalidAuth("Token expired"),
mock_vehicles,
]
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state != STATE_UNAVAILABLE
async def test_reauth_failure(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
mock_vehicles: list[Vehicle],
freezer: FrozenDateTimeFactory,
) -> None:
"""Test entity becomes unavailable on permanent auth failure."""
await setup_integration(hass, mock_config_entry)
# get_vehicles raises InvalidAuth, and re-authentication also fails
mock_autoskope_client.get_vehicles.side_effect = InvalidAuth("Token expired")
mock_autoskope_client.authenticate.side_effect = InvalidAuth("Invalid credentials")
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
state = hass.states.get("device_tracker.test_vehicle")
assert state is not None
assert state.state == STATE_UNAVAILABLE
# Clean up side effects to prevent teardown errors
mock_autoskope_client.get_vehicles.side_effect = None
mock_autoskope_client.authenticate.side_effect = None
mock_autoskope_client.get_vehicles.return_value = mock_vehicles
async def test_vehicle_name_update(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test device name updates in device registry when vehicle is renamed."""
await setup_integration(hass, mock_config_entry)
device_registry = dr.async_get(hass)
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry is not None
assert device_entry.name == "Test Vehicle"
# Simulate vehicle rename on Autoskope side
mock_autoskope_client.get_vehicles.return_value = [
Vehicle(
id="12345",
name="Renamed Vehicle",
position=VehiclePosition(
latitude=50.1109221,
longitude=8.6821267,
speed=0,
timestamp="2025-05-28T10:00:00Z",
park_mode=True,
),
external_voltage=12.5,
battery_voltage=3.7,
gps_quality=1.2,
imei="123456789012345",
model="Autoskope",
)
]
freezer.tick(UPDATE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
# Device registry should reflect the new name
device_entry = device_registry.async_get_device(identifiers={(DOMAIN, "12345")})
assert device_entry is not None
assert device_entry.name == "Renamed Vehicle"

View File

@@ -0,0 +1,48 @@
"""Test Autoskope integration setup."""
from unittest.mock import AsyncMock
from autoskope_client.models import CannotConnect, InvalidAuth
import pytest
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from . import setup_integration
from tests.common import MockConfigEntry
async def test_setup_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
) -> None:
"""Test successful setup and unload of entry."""
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("exception", "expected_state"),
[
(InvalidAuth("Invalid credentials"), ConfigEntryState.SETUP_ERROR),
(CannotConnect("Connection failed"), ConfigEntryState.SETUP_RETRY),
],
)
async def test_setup_entry_errors(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_autoskope_client: AsyncMock,
exception: Exception,
expected_state: ConfigEntryState,
) -> None:
"""Test setup with authentication and connection errors."""
mock_autoskope_client.connect.side_effect = exception
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is expected_state

View File

@@ -13,6 +13,7 @@ import string
from typing import TYPE_CHECKING, Any
from unittest.mock import AsyncMock, MagicMock, patch
from aiohasupervisor import SupervisorNotFoundError
from aiohasupervisor.models import (
Discovery,
GreenInfo,
@@ -313,6 +314,7 @@ def addon_not_installed_fixture(
"""Mock add-on not installed."""
from .hassio.common import mock_addon_not_installed # noqa: PLC0415
addon_info.side_effect = SupervisorNotFoundError
return mock_addon_not_installed(addon_store_info, addon_info)

View File

@@ -43,7 +43,7 @@ async def test_open_cover_intent(hass: HomeAssistant, slots: dict[str, Any]) ->
)
await hass.async_block_till_done()
assert response.speech["plain"]["speech"] == "Opening garage door"
assert response.response_type == intent.IntentResponseType.ACTION_DONE
assert len(calls) == 1
call = calls[0]
assert call.domain == DOMAIN
@@ -75,7 +75,7 @@ async def test_close_cover_intent(hass: HomeAssistant, slots: dict[str, Any]) ->
)
await hass.async_block_till_done()
assert response.speech["plain"]["speech"] == "Closing garage door"
assert response.response_type == intent.IntentResponseType.ACTION_DONE
assert len(calls) == 1
call = calls[0]
assert call.domain == DOMAIN

View File

@@ -3,11 +3,15 @@
from __future__ import annotations
import asyncio
from typing import Any
from unittest.mock import AsyncMock, call
from uuid import uuid4
from aiohasupervisor import SupervisorError
from aiohasupervisor import (
AddonNotSupportedArchitectureError,
AddonNotSupportedHomeAssistantVersionError,
AddonNotSupportedMachineTypeError,
SupervisorError,
)
from aiohasupervisor.models import AddonsOptions, Discovery, PartialBackupOptions
import pytest
@@ -20,10 +24,8 @@ from homeassistant.components.hassio.addon_manager import (
from homeassistant.core import HomeAssistant
async def test_not_installed_raises_exception(
addon_manager: AddonManager,
addon_not_installed: dict[str, Any],
) -> None:
@pytest.mark.usefixtures("addon_not_installed")
async def test_not_installed_raises_exception(addon_manager: AddonManager) -> None:
"""Test addon not installed raises exception."""
addon_config = {"test_key": "test"}
@@ -38,24 +40,40 @@ async def test_not_installed_raises_exception(
assert str(err.value) == "Test app is not installed"
@pytest.mark.parametrize(
"exception",
[
AddonNotSupportedArchitectureError(
"Add-on test not supported on this platform, supported architectures: test"
),
AddonNotSupportedHomeAssistantVersionError(
"Add-on test not supported on this system, requires Home Assistant version 2026.1.0 or greater"
),
AddonNotSupportedMachineTypeError(
"Add-on test not supported on this machine, supported machine types: test"
),
],
)
async def test_not_available_raises_exception(
addon_manager: AddonManager,
addon_store_info: AsyncMock,
supervisor_client: AsyncMock,
addon_info: AsyncMock,
exception: SupervisorError,
) -> None:
"""Test addon not available raises exception."""
addon_store_info.return_value.available = False
addon_info.return_value.available = False
supervisor_client.store.addon_availability.side_effect = exception
supervisor_client.store.install_addon.side_effect = exception
addon_info.return_value.update_available = True
with pytest.raises(AddonError) as err:
await addon_manager.async_install_addon()
assert str(err.value) == "Test app is not available"
assert str(err.value) == f"Test app is not available: {exception!s}"
with pytest.raises(AddonError) as err:
await addon_manager.async_update_addon()
assert str(err.value) == "Test app is not available"
assert str(err.value) == f"Test app is not available: {exception!s}"
async def test_get_addon_discovery_info(
@@ -496,11 +514,10 @@ async def test_stop_addon_error(
assert stop_addon.call_count == 1
@pytest.mark.usefixtures("hass", "addon_installed")
async def test_update_addon(
hass: HomeAssistant,
addon_manager: AddonManager,
addon_info: AsyncMock,
addon_installed: AsyncMock,
create_backup: AsyncMock,
update_addon: AsyncMock,
) -> None:
@@ -509,7 +526,7 @@ async def test_update_addon(
await addon_manager.async_update_addon()
assert addon_info.call_count == 2
assert addon_info.call_count == 1
assert create_backup.call_count == 1
assert create_backup.call_args == call(
PartialBackupOptions(name="addon_test_addon_1.0.0", addons={"test_addon"})
@@ -517,10 +534,10 @@ async def test_update_addon(
assert update_addon.call_count == 1
@pytest.mark.usefixtures("addon_installed")
async def test_update_addon_no_update(
addon_manager: AddonManager,
addon_info: AsyncMock,
addon_installed: AsyncMock,
create_backup: AsyncMock,
update_addon: AsyncMock,
) -> None:
@@ -534,11 +551,10 @@ async def test_update_addon_no_update(
assert update_addon.call_count == 0
@pytest.mark.usefixtures("hass", "addon_installed")
async def test_update_addon_error(
hass: HomeAssistant,
addon_manager: AddonManager,
addon_info: AsyncMock,
addon_installed: AsyncMock,
create_backup: AsyncMock,
update_addon: AsyncMock,
) -> None:
@@ -551,7 +567,7 @@ async def test_update_addon_error(
assert str(err.value) == "Failed to update the Test app: Boom"
assert addon_info.call_count == 2
assert addon_info.call_count == 1
assert create_backup.call_count == 1
assert create_backup.call_args == call(
PartialBackupOptions(name="addon_test_addon_1.0.0", addons={"test_addon"})
@@ -559,11 +575,10 @@ async def test_update_addon_error(
assert update_addon.call_count == 1
@pytest.mark.usefixtures("hass", "addon_installed")
async def test_schedule_update_addon(
hass: HomeAssistant,
addon_manager: AddonManager,
addon_info: AsyncMock,
addon_installed: AsyncMock,
create_backup: AsyncMock,
update_addon: AsyncMock,
) -> None:
@@ -589,7 +604,7 @@ async def test_schedule_update_addon(
await asyncio.gather(update_task, update_task_two)
assert addon_manager.task_in_progress() is False
assert addon_info.call_count == 3
assert addon_info.call_count == 2
assert create_backup.call_count == 1
assert create_backup.call_args == call(
PartialBackupOptions(name="addon_test_addon_1.0.0", addons={"test_addon"})

View File

@@ -906,7 +906,6 @@ async def test_config_flow_thread_addon_already_installed(
}
@pytest.mark.usefixtures("addon_not_installed")
async def test_options_flow_zigbee_to_thread(
hass: HomeAssistant,
install_addon: AsyncMock,

View File

@@ -4,6 +4,7 @@ from collections.abc import Generator
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from pyintelliclima.const import FanMode, FanSpeed
from pyintelliclima.intelliclima_types import (
IntelliClimaDevices,
IntelliClimaECO,
@@ -50,9 +51,9 @@ def single_eco_device() -> IntelliClimaDevices:
model=IntelliClimaModelType(modello="ECO", tipo="wifi"),
name="Test VMC",
houses_id="12345",
mode_set="1",
mode_set=FanMode.inward,
mode_state="1",
speed_set="3",
speed_set=FanSpeed.medium,
speed_state="3",
last_online="2025-11-18 10:22:51",
creation_date="2025-11-18 10:22:51",

View File

@@ -3,6 +3,7 @@
from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock, patch
from pyintelliclima.const import FanMode, FanSpeed
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -103,7 +104,7 @@ async def test_fan_turn_on_service_calls_api(
# Device serial from single_eco_device.crono_sn
mock_cloud_interface.ecocomfort.set_mode_speed.assert_awaited_once_with(
"11223344", "1", "2"
"11223344", FanMode.inward, FanSpeed.low
)
@@ -119,10 +120,10 @@ async def test_fan_set_percentage_maps_to_speed(
{ATTR_ENTITY_ID: FAN_ENTITY_ID, ATTR_PERCENTAGE: 15},
blocking=True,
)
# Initial mode_set="1" (forward) from single_eco_device.
# Sleep speed is "1" (25%).
# Initial mode_set=FanMode.inward from single_eco_device.
# Sleep speed is FanSpeed.sleep (25%).
mock_cloud_interface.ecocomfort.set_mode_speed.assert_awaited_once_with(
"11223344", "1", "1"
"11223344", FanMode.inward, FanSpeed.sleep
)
@@ -165,18 +166,18 @@ async def test_fan_set_percentage_zero_turns_off(
("service_data", "expected_mode", "expected_speed"),
[
# percentage=None, preset_mode=None -> defaults to previous speed > 75% (medium),
# previous mode > "inward"
({}, "1", "3"),
# percentage=0, preset_mode=None -> default 25% (sleep), previous mode (inward)
({ATTR_PERCENTAGE: 0}, "1", "1"),
# previous mode > FanMode.inward
({}, FanMode.inward, FanSpeed.medium),
# percentage=0, preset_mode=None -> default 25% (FanSpeed.sleep), previous mode (inward)
({ATTR_PERCENTAGE: 0}, FanMode.inward, FanSpeed.sleep),
],
)
async def test_fan_turn_on_defaulting_behavior(
hass: HomeAssistant,
mock_cloud_interface: AsyncMock,
service_data: dict,
expected_mode: str,
expected_speed: str,
expected_mode: FanMode,
expected_speed: FanSpeed,
) -> None:
"""turn_on defaults percentage/preset as expected."""
data = {ATTR_ENTITY_ID: FAN_ENTITY_ID} | service_data

View File

@@ -86,10 +86,10 @@ async def test_select_option_keeps_current_speed(
{ATTR_ENTITY_ID: SELECT_ENTITY_ID, ATTR_OPTION: option},
blocking=True,
)
# Device starts with speed_set="3" (from single_eco_device in conftest),
# Device starts with speed_set=FanSpeed.medium (from single_eco_device in conftest),
# mode is not off and not auto, so current speed is preserved.
mock_cloud_interface.ecocomfort.set_mode_speed.assert_awaited_once_with(
"11223344", expected_mode, "3"
"11223344", expected_mode, FanSpeed.medium
)
@@ -119,9 +119,9 @@ async def test_select_option_in_auto_mode_defaults_speed_to_sleep(
mock_cloud_interface: AsyncMock,
single_eco_device,
) -> None:
"""When speed_set is FanSpeed.auto (auto preset), selecting an option defaults to sleep speed."""
"""When speed_set is FanSpeed.auto_get (auto preset), selecting an option defaults to sleep speed."""
eco = list(single_eco_device.ecocomfort2_devices.values())[0]
eco.speed_set = FanSpeed.auto
eco.speed_set = FanSpeed.auto_get
eco.mode_set = FanMode.sensor
await hass.services.async_call(

View File

@@ -248,12 +248,11 @@ async def test_cover_intents_loading(hass: HomeAssistant) -> None:
hass.states.async_set("cover.garage_door", "closed")
calls = async_mock_service(hass, "cover", SERVICE_OPEN_COVER)
response = await intent.async_handle(
await intent.async_handle(
hass, "test", "HassOpenCover", {"name": {"value": "garage door"}}
)
await hass.async_block_till_done()
assert response.speech["plain"]["speech"] == "Opening garage door"
assert len(calls) == 1
call = calls[0]
assert call.domain == "cover"

View File

@@ -0,0 +1,24 @@
# serializer version: 1
# name: test_diagnostics
dict({
'pets': list([
]),
'robots': list([
dict({
'cleanCycleWaitTimeMinutes': '7',
'cycleCapacity': '30',
'cycleCount': '15',
'cyclesAfterDrawerFull': '0',
'lastSeen': '2022-09-17T13:06:37.884Z',
'litterRobotId': '**REDACTED**',
'litterRobotNickname': 'Test',
'litterRobotSerial': '**REDACTED**',
'nightLightActive': '1',
'panelLockActive': '0',
'powerStatus': 'AC',
'sleepModeActive': '112:50:19',
'unitStatus': 'RDY',
}),
]),
})
# ---

View File

@@ -3,10 +3,12 @@
from unittest.mock import MagicMock
from freezegun import freeze_time
import pytest
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .conftest import setup_integration
@@ -42,3 +44,18 @@ async def test_button(
state = hass.states.get(BUTTON_ENTITY)
assert state
assert state.state == "2021-11-15T10:37:00+00:00"
async def test_button_command_exception(
hass: HomeAssistant, mock_account_with_side_effects: MagicMock
) -> None:
"""Test that LitterRobotException is wrapped in HomeAssistantError."""
await setup_integration(hass, mock_account_with_side_effects, BUTTON_DOMAIN)
with pytest.raises(HomeAssistantError, match="Invalid command: oops"):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: BUTTON_ENTITY},
blocking=True,
)

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