Compare commits

..

16 Commits

Author SHA1 Message Date
Erik
7917f0a2b9 Adjust entity condition strings 2026-01-16 11:04:55 +01:00
epenet
7c6a31861e Improve type hints in egardia (#161048) 2026-01-16 10:08:24 +01:00
Robert Resch
b2b25ca28c Revert "Add SmartThings media-player audio notifications" (#161049) 2026-01-16 10:06:30 +01:00
epenet
ad9efab16a Improve type hints in concord232 (#161045) 2026-01-16 09:46:53 +01:00
Matthias Alphart
e967d33911 Update knx-frontend to 2026.1.15.112308 (#161004) 2026-01-16 09:37:09 +01:00
epenet
86bacdbdde Use shorthand attributes in oasa_telematics (#160990) 2026-01-16 09:34:51 +01:00
Robert Resch
644a40674d Make shebang matcher stricter (#160986) 2026-01-16 09:21:19 +01:00
Raphael Hehl
2cf813758e Add per-camera ring volume control for UniFi Protect chimes (#161031)
Co-authored-by: RaHehl <rahehl@users.noreply.github.com>
2026-01-16 08:29:35 +01:00
DeerMaximum
ad47eccf5f Bump pynina to 1.0.2 (#161013) 2026-01-16 08:24:58 +01:00
epenet
581b554a66 Improve type hints in digital_ocean (#161006) 2026-01-16 08:23:13 +01:00
epenet
e4def9eb03 Improve type hints in envisalink (#161005) 2026-01-16 08:22:15 +01:00
epenet
5f2d17faf6 Improve type hints in homematic (#161002) 2026-01-16 08:21:30 +01:00
TheJulianJES
e17565c069 Add Resideo X2S Smart Thermostat diagnostics to Matter fixture (#161037) 2026-01-16 08:20:42 +01:00
Erik Montnemery
b856e04825 Add assist_satellite conditions (#161019) 2026-01-16 07:39:59 +01:00
epenet
67e676df4f Fix duplicate HVACMode in Tuya climate (#160918) 2026-01-15 22:12:24 +01:00
Erik Montnemery
e2e7485e30 Remove unused test fixture from light condition tests (#160925) 2026-01-15 22:03:18 +01:00
69 changed files with 1251 additions and 1839 deletions

View File

@@ -247,17 +247,11 @@ jobs:
&& github.event.inputs.audit-licenses-only != 'true'
steps:
- *checkout
- name: Register yamllint problem matcher
- name: Register problem matchers
run: |
echo "::add-matcher::.github/workflows/matchers/yamllint.json"
- name: Register check-json problem matcher
run: |
echo "::add-matcher::.github/workflows/matchers/check-json.json"
- name: Register check executables problem matcher
run: |
echo "::add-matcher::.github/workflows/matchers/check-executables-have-shebangs.json"
- name: Register codespell problem matcher
run: |
echo "::add-matcher::.github/workflows/matchers/codespell.json"
- name: Run prek
uses: j178/prek-action@9d6a3097e0c1865ecce00cfb89fe80f2ee91b547 # v1.0.12

View File

@@ -4,7 +4,7 @@
"owner": "check-executables-have-shebangs",
"pattern": [
{
"regexp": "^(.+):\\s(.+)$",
"regexp": "^(.+):\\s(marked executable but has no \\(or invalid\\) shebang!.*)$",
"file": 1,
"message": 2
}

View File

@@ -14,7 +14,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is armed"
"name": "Alarm is armed"
},
"is_armed_away": {
"description": "Tests if one or more alarms are armed in away mode.",
@@ -24,7 +24,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is armed away"
"name": "Alarm is armed away"
},
"is_armed_home": {
"description": "Tests if one or more alarms are armed in home mode.",
@@ -34,7 +34,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is armed home"
"name": "Alarm is armed home"
},
"is_armed_night": {
"description": "Tests if one or more alarms are armed in night mode.",
@@ -44,7 +44,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is armed night"
"name": "Alarm is armed night"
},
"is_armed_vacation": {
"description": "Tests if one or more alarms are armed in vacation mode.",
@@ -54,7 +54,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is armed vacation"
"name": "Alarm is armed vacation"
},
"is_disarmed": {
"description": "Tests if one or more alarms are disarmed.",
@@ -64,7 +64,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is disarmed"
"name": "Alarm is disarmed"
},
"is_triggered": {
"description": "Tests if one or more alarms are triggered.",
@@ -74,7 +74,7 @@
"name": "[%key:component::alarm_control_panel::common::condition_behavior_name%]"
}
},
"name": "If an alarm is triggered"
"name": "Alarm is triggered"
}
},
"device_automation": {

View File

@@ -0,0 +1,23 @@
"""Provides conditions for assist satellites."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.condition import Condition, make_entity_state_condition
from .const import DOMAIN
from .entity import AssistSatelliteState
CONDITIONS: dict[str, type[Condition]] = {
"is_idle": make_entity_state_condition(DOMAIN, AssistSatelliteState.IDLE),
"is_listening": make_entity_state_condition(DOMAIN, AssistSatelliteState.LISTENING),
"is_processing": make_entity_state_condition(
DOMAIN, AssistSatelliteState.PROCESSING
),
"is_responding": make_entity_state_condition(
DOMAIN, AssistSatelliteState.RESPONDING
),
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the assist satellite conditions."""
return CONDITIONS

View File

@@ -1,7 +1,7 @@
.condition_common: &condition_common
target:
entity:
domain: climate
domain: assist_satellite
fields:
behavior:
required: true
@@ -13,8 +13,7 @@
- all
- any
is_off: *condition_common
is_on: *condition_common
is_cooling: *condition_common
is_drying: *condition_common
is_heating: *condition_common
is_idle: *condition_common
is_listening: *condition_common
is_processing: *condition_common
is_responding: *condition_common

View File

@@ -1,4 +1,18 @@
{
"conditions": {
"is_idle": {
"condition": "mdi:chat-sleep"
},
"is_listening": {
"condition": "mdi:chat-question"
},
"is_processing": {
"condition": "mdi:chat-processing"
},
"is_responding": {
"condition": "mdi:chat-alert"
}
},
"entity_component": {
"_": {
"default": "mdi:account-voice"

View File

@@ -1,8 +1,52 @@
{
"common": {
"condition_behavior_description": "How the state should match on the targeted Assist satellites.",
"condition_behavior_name": "Behavior",
"trigger_behavior_description": "The behavior of the targeted Assist satellites to trigger on.",
"trigger_behavior_name": "Behavior"
},
"conditions": {
"is_idle": {
"description": "Tests if one or more Assist satellites are idle.",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::condition_behavior_description%]",
"name": "[%key:component::assist_satellite::common::condition_behavior_name%]"
}
},
"name": "Satellite is idle"
},
"is_listening": {
"description": "Tests if one or more Assist satellites are listening.",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::condition_behavior_description%]",
"name": "[%key:component::assist_satellite::common::condition_behavior_name%]"
}
},
"name": "Satellite is listening"
},
"is_processing": {
"description": "Tests if one or more Assist satellites are processing.",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::condition_behavior_description%]",
"name": "[%key:component::assist_satellite::common::condition_behavior_name%]"
}
},
"name": "Satellite is processing"
},
"is_responding": {
"description": "Tests if one or more Assist satellites are responding.",
"fields": {
"behavior": {
"description": "[%key:component::assist_satellite::common::condition_behavior_description%]",
"name": "[%key:component::assist_satellite::common::condition_behavior_name%]"
}
},
"name": "Satellite is responding"
}
},
"entity_component": {
"_": {
"name": "Assist satellite",
@@ -21,6 +65,12 @@
"sentences": "Sentences"
}
},
"condition_behavior": {
"options": {
"all": "All",
"any": "Any"
}
},
"trigger_behavior": {
"options": {
"any": "Any",

View File

@@ -124,7 +124,7 @@ NEW_TRIGGERS_CONDITIONS_FEATURE_FLAG = "new_triggers_conditions"
_EXPERIMENTAL_CONDITION_PLATFORMS = {
"alarm_control_panel",
"climate",
"assist_satellite",
"fan",
"light",
}

View File

@@ -1,39 +0,0 @@
"""Provides conditions for climates."""
from homeassistant.core import HomeAssistant
from homeassistant.helpers.condition import (
Condition,
make_entity_state_attribute_condition,
make_entity_state_condition,
)
from .const import ATTR_HVAC_ACTION, DOMAIN, HVACAction, HVACMode
CONDITIONS: dict[str, type[Condition]] = {
"is_off": make_entity_state_condition(DOMAIN, HVACMode.OFF),
"is_on": make_entity_state_condition(
DOMAIN,
{
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
},
),
"is_cooling": make_entity_state_attribute_condition(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.COOLING
),
"is_drying": make_entity_state_attribute_condition(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.DRYING
),
"is_heating": make_entity_state_attribute_condition(
DOMAIN, ATTR_HVAC_ACTION, HVACAction.HEATING
),
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the climate conditions."""
return CONDITIONS

View File

@@ -1,21 +1,4 @@
{
"conditions": {
"is_cooling": {
"condition": "mdi:snowflake"
},
"is_drying": {
"condition": "mdi:water-percent"
},
"is_heating": {
"condition": "mdi:fire"
},
"is_off": {
"condition": "mdi:power-off"
},
"is_on": {
"condition": "mdi:power-on"
}
},
"entity_component": {
"_": {
"default": "mdi:thermostat",

View File

@@ -1,62 +1,8 @@
{
"common": {
"condition_behavior_description": "How the state should match on the targeted climate-control devices.",
"condition_behavior_name": "Behavior",
"trigger_behavior_description": "The behavior of the targeted climates to trigger on.",
"trigger_behavior_name": "Behavior"
},
"conditions": {
"is_cooling": {
"description": "Tests if one or more climate-control devices are cooling.",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::condition_behavior_description%]",
"name": "[%key:component::climate::common::condition_behavior_name%]"
}
},
"name": "If a climate-control device is cooling"
},
"is_drying": {
"description": "Tests if one or more climate-control devices are drying.",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::condition_behavior_description%]",
"name": "[%key:component::climate::common::condition_behavior_name%]"
}
},
"name": "If a climate-control device is drying"
},
"is_heating": {
"description": "Tests if one or more climate-control devices are heating.",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::condition_behavior_description%]",
"name": "[%key:component::climate::common::condition_behavior_name%]"
}
},
"name": "If a climate-control device is heating"
},
"is_off": {
"description": "Tests if one or more climate-control devices are off.",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::condition_behavior_description%]",
"name": "[%key:component::climate::common::condition_behavior_name%]"
}
},
"name": "If a climate-control device is off"
},
"is_on": {
"description": "Tests if one or more climate-control devices are on.",
"fields": {
"behavior": {
"description": "[%key:component::climate::common::condition_behavior_description%]",
"name": "[%key:component::climate::common::condition_behavior_name%]"
}
},
"name": "If a climate-control device is on"
}
},
"device_automation": {
"action_type": {
"set_hvac_mode": "Change HVAC mode on {entity_name}",
@@ -235,12 +181,6 @@
}
},
"selector": {
"condition_behavior": {
"options": {
"all": "All",
"any": "Any"
}
},
"hvac_mode": {
"options": {
"auto": "[%key:common::state::auto%]",

View File

@@ -49,11 +49,11 @@ def setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Concord232 alarm control panel platform."""
name = config[CONF_NAME]
code = config.get(CONF_CODE)
mode = config[CONF_MODE]
host = config[CONF_HOST]
port = config[CONF_PORT]
name: str = config[CONF_NAME]
code: str | None = config.get(CONF_CODE)
mode: str = config[CONF_MODE]
host: str = config[CONF_HOST]
port: int = config[CONF_PORT]
url = f"http://{host}:{port}"
@@ -72,7 +72,7 @@ class Concord232Alarm(AlarmControlPanelEntity):
| AlarmControlPanelEntityFeature.ARM_AWAY
)
def __init__(self, url, name, code, mode):
def __init__(self, url: str, name: str, code: str | None, mode: str) -> None:
"""Initialize the Concord232 alarm panel."""
self._attr_name = name
@@ -125,7 +125,7 @@ class Concord232Alarm(AlarmControlPanelEntity):
return
self._alarm.arm("away")
def _validate_code(self, code, state):
def _validate_code(self, code: str | None, state: AlarmControlPanelState) -> bool:
"""Validate given code."""
if self._code is None:
return True

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import datetime
import logging
from typing import Any
from concord232 import client as concord232_client
import requests
@@ -29,8 +30,7 @@ CONF_ZONE_TYPES = "zone_types"
DEFAULT_HOST = "localhost"
DEFAULT_NAME = "Alarm"
DEFAULT_PORT = "5007"
DEFAULT_SSL = False
DEFAULT_PORT = 5007
SCAN_INTERVAL = datetime.timedelta(seconds=10)
@@ -56,10 +56,10 @@ def setup_platform(
) -> None:
"""Set up the Concord232 binary sensor platform."""
host = config[CONF_HOST]
port = config[CONF_PORT]
exclude = config[CONF_EXCLUDE_ZONES]
zone_types = config[CONF_ZONE_TYPES]
host: str = config[CONF_HOST]
port: int = config[CONF_PORT]
exclude: list[int] = config[CONF_EXCLUDE_ZONES]
zone_types: dict[int, BinarySensorDeviceClass] = config[CONF_ZONE_TYPES]
sensors = []
try:
@@ -84,7 +84,6 @@ def setup_platform(
if zone["number"] not in exclude:
sensors.append(
Concord232ZoneSensor(
hass,
client,
zone,
zone_types.get(zone["number"], get_opening_type(zone)),
@@ -110,26 +109,25 @@ def get_opening_type(zone):
class Concord232ZoneSensor(BinarySensorEntity):
"""Representation of a Concord232 zone as a sensor."""
def __init__(self, hass, client, zone, zone_type):
def __init__(
self,
client: concord232_client.Client,
zone: dict[str, Any],
zone_type: BinarySensorDeviceClass,
) -> None:
"""Initialize the Concord232 binary sensor."""
self._hass = hass
self._client = client
self._zone = zone
self._number = zone["number"]
self._zone_type = zone_type
self._attr_device_class = zone_type
@property
def device_class(self) -> BinarySensorDeviceClass:
"""Return the class of this sensor, from DEVICE_CLASSES."""
return self._zone_type
@property
def name(self):
def name(self) -> str:
"""Return the name of the binary sensor."""
return self._zone["name"]
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if the binary sensor is on."""
# True means "faulted" or "open" or "abnormal state"
return bool(self._zone["state"] != "Normal")
@@ -145,5 +143,5 @@ class Concord232ZoneSensor(BinarySensorEntity):
if hasattr(self._client, "zones"):
self._zone = next(
(x for x in self._client.zones if x["number"] == self._number), None
x for x in self._client.zones if x["number"] == self._number
)

View File

@@ -1,6 +1,7 @@
"""Support for Digital Ocean."""
from datetime import timedelta
from __future__ import annotations
import logging
import digitalocean
@@ -12,27 +13,12 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import Throttle
from .const import DATA_DIGITAL_OCEAN, DOMAIN, MIN_TIME_BETWEEN_UPDATES
_LOGGER = logging.getLogger(__name__)
ATTR_CREATED_AT = "created_at"
ATTR_DROPLET_ID = "droplet_id"
ATTR_DROPLET_NAME = "droplet_name"
ATTR_FEATURES = "features"
ATTR_IPV4_ADDRESS = "ipv4_address"
ATTR_IPV6_ADDRESS = "ipv6_address"
ATTR_MEMORY = "memory"
ATTR_REGION = "region"
ATTR_VCPUS = "vcpus"
ATTRIBUTION = "Data provided by Digital Ocean"
CONF_DROPLETS = "droplets"
DATA_DIGITAL_OCEAN = "data_do"
DIGITAL_OCEAN_PLATFORMS = [Platform.SWITCH, Platform.BINARY_SENSOR]
DOMAIN = "digital_ocean"
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)
CONFIG_SCHEMA = vol.Schema(
{DOMAIN: vol.Schema({vol.Required(CONF_ACCESS_TOKEN): cv.string})},

View File

@@ -3,6 +3,7 @@
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
@@ -16,7 +17,7 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import (
from .const import (
ATTR_CREATED_AT,
ATTR_DROPLET_ID,
ATTR_DROPLET_NAME,
@@ -65,6 +66,7 @@ class DigitalOceanBinarySensor(BinarySensorEntity):
"""Representation of a Digital Ocean droplet sensor."""
_attr_attribution = ATTRIBUTION
_attr_device_class = BinarySensorDeviceClass.MOVING
def __init__(self, do, droplet_id):
"""Initialize a new Digital Ocean sensor."""
@@ -79,17 +81,12 @@ class DigitalOceanBinarySensor(BinarySensorEntity):
return self.data.name
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if the binary sensor is on."""
return self.data.status == "active"
@property
def device_class(self) -> BinarySensorDeviceClass:
"""Return the class of this sensor."""
return BinarySensorDeviceClass.MOVING
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the Digital Ocean droplet."""
return {
ATTR_CREATED_AT: self.data.created_at,

View File

@@ -0,0 +1,30 @@
"""Support for Digital Ocean."""
from __future__ import annotations
from datetime import timedelta
from typing import TYPE_CHECKING
from homeassistant.util.hass_dict import HassKey
if TYPE_CHECKING:
from . import DigitalOcean
ATTR_CREATED_AT = "created_at"
ATTR_DROPLET_ID = "droplet_id"
ATTR_DROPLET_NAME = "droplet_name"
ATTR_FEATURES = "features"
ATTR_IPV4_ADDRESS = "ipv4_address"
ATTR_IPV6_ADDRESS = "ipv6_address"
ATTR_MEMORY = "memory"
ATTR_REGION = "region"
ATTR_VCPUS = "vcpus"
ATTRIBUTION = "Data provided by Digital Ocean"
CONF_DROPLETS = "droplets"
DOMAIN = "digital_ocean"
DATA_DIGITAL_OCEAN: HassKey[DigitalOcean] = HassKey(DOMAIN)
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=60)

View File

@@ -16,7 +16,7 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import (
from .const import (
ATTR_CREATED_AT,
ATTR_DROPLET_ID,
ATTR_DROPLET_NAME,
@@ -80,12 +80,12 @@ class DigitalOceanSwitch(SwitchEntity):
return self.data.name
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if switch is on."""
return self.data.status == "active"
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes of the Digital Ocean droplet."""
return {
ATTR_CREATED_AT: self.data.created_at,

View File

@@ -18,6 +18,7 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, discovery
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
_LOGGER = logging.getLogger(__name__)
@@ -35,7 +36,7 @@ DEFAULT_REPORT_SERVER_PORT = 52010
DEFAULT_VERSION = "GATE-01"
DOMAIN = "egardia"
EGARDIA_DEVICE = "egardiadevice"
EGARDIA_DEVICE: HassKey[egardiadevice.EgardiaDevice] = HassKey(DOMAIN)
EGARDIA_NAME = "egardianame"
EGARDIA_REPORT_SERVER_CODES = "egardia_rs_codes"
EGARDIA_REPORT_SERVER_ENABLED = "egardia_rs_enabled"

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import logging
from pythonegardia.egardiadevice import EgardiaDevice
import requests
from homeassistant.components.alarm_control_panel import (
@@ -11,6 +12,7 @@ from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntityFeature,
AlarmControlPanelState,
)
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@@ -47,10 +49,10 @@ def setup_platform(
if discovery_info is None:
return
device = EgardiaAlarm(
discovery_info["name"],
discovery_info[CONF_NAME],
hass.data[EGARDIA_DEVICE],
discovery_info[CONF_REPORT_SERVER_ENABLED],
discovery_info.get(CONF_REPORT_SERVER_CODES),
discovery_info[CONF_REPORT_SERVER_CODES],
discovery_info[CONF_REPORT_SERVER_PORT],
)
@@ -67,8 +69,13 @@ class EgardiaAlarm(AlarmControlPanelEntity):
)
def __init__(
self, name, egardiasystem, rs_enabled=False, rs_codes=None, rs_port=52010
):
self,
name: str,
egardiasystem: EgardiaDevice,
rs_enabled: bool,
rs_codes: dict[str, list[str]],
rs_port: int,
) -> None:
"""Initialize the Egardia alarm."""
self._attr_name = name
self._egardiasystem = egardiasystem
@@ -85,9 +92,7 @@ class EgardiaAlarm(AlarmControlPanelEntity):
@property
def should_poll(self) -> bool:
"""Poll if no report server is enabled."""
if not self._rs_enabled:
return True
return False
return not self._rs_enabled
def handle_status_event(self, event):
"""Handle the Egardia system status event."""

View File

@@ -2,11 +2,12 @@
from __future__ import annotations
from pythonegardia.egardiadevice import EgardiaDevice
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
BinarySensorEntity,
)
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
@@ -51,30 +52,20 @@ async def async_setup_platform(
class EgardiaBinarySensor(BinarySensorEntity):
"""Represents a sensor based on an Egardia sensor (IR, Door Contact)."""
def __init__(self, sensor_id, name, egardia_system, device_class):
def __init__(
self,
sensor_id: str,
name: str,
egardia_system: EgardiaDevice,
device_class: BinarySensorDeviceClass | None,
) -> None:
"""Initialize the sensor device."""
self._id = sensor_id
self._name = name
self._state = None
self._device_class = device_class
self._attr_name = name
self._attr_device_class = device_class
self._egardia_system = egardia_system
def update(self) -> None:
"""Update the status."""
egardia_input = self._egardia_system.getsensorstate(self._id)
self._state = STATE_ON if egardia_input else STATE_OFF
@property
def name(self):
"""Return the name of the device."""
return self._name
@property
def is_on(self):
"""Whether the device is switched on."""
return self._state == STATE_ON
@property
def device_class(self) -> BinarySensorDeviceClass | None:
"""Return the device class."""
return self._device_class
self._attr_is_on = bool(egardia_input)

View File

@@ -18,12 +18,13 @@ from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.helpers.dispatcher import async_dispatcher_send
from homeassistant.helpers.typing import ConfigType
from homeassistant.util.hass_dict import HassKey
_LOGGER = logging.getLogger(__name__)
DOMAIN = "envisalink"
DATA_EVL = "envisalink"
DATA_EVL: HassKey[EnvisalinkAlarmPanel] = HassKey(DOMAIN)
CONF_EVL_KEEPALIVE = "keepalive_interval"
CONF_EVL_PORT = "port"

View File

@@ -3,7 +3,9 @@
from __future__ import annotations
import logging
from typing import Any
from pyenvisalink import EnvisalinkAlarmPanel
import voluptuous as vol
from homeassistant.components.alarm_control_panel import (
@@ -22,6 +24,7 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import (
CONF_PANIC,
CONF_PARTITIONNAME,
CONF_PARTITIONS,
DATA_EVL,
DOMAIN,
PARTITION_SCHEMA,
@@ -51,15 +54,14 @@ async def async_setup_platform(
"""Perform the setup for Envisalink alarm panels."""
if not discovery_info:
return
configured_partitions = discovery_info["partitions"]
code = discovery_info[CONF_CODE]
panic_type = discovery_info[CONF_PANIC]
configured_partitions: dict[int, dict[str, Any]] = discovery_info[CONF_PARTITIONS]
code: str | None = discovery_info[CONF_CODE]
panic_type: str = discovery_info[CONF_PANIC]
entities = []
for part_num in configured_partitions:
entity_config_data = PARTITION_SCHEMA(configured_partitions[part_num])
for part_num, part_config in configured_partitions.items():
entity_config_data = PARTITION_SCHEMA(part_config)
entity = EnvisalinkAlarm(
hass,
part_num,
entity_config_data[CONF_PARTITIONNAME],
code,
@@ -103,8 +105,14 @@ class EnvisalinkAlarm(EnvisalinkEntity, AlarmControlPanelEntity):
)
def __init__(
self, hass, partition_number, alarm_name, code, panic_type, info, controller
):
self,
partition_number: int,
alarm_name: str,
code: str | None,
panic_type: str,
info: dict[str, Any],
controller: EnvisalinkAlarmPanel,
) -> None:
"""Initialize the alarm panel."""
self._partition_number = partition_number
self._panic_type = panic_type

View File

@@ -4,6 +4,9 @@ from __future__ import annotations
import datetime
import logging
from typing import Any
from pyenvisalink import EnvisalinkAlarmPanel
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@@ -16,7 +19,14 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
from . import CONF_ZONENAME, CONF_ZONETYPE, DATA_EVL, SIGNAL_ZONE_UPDATE, ZONE_SCHEMA
from . import (
CONF_ZONENAME,
CONF_ZONES,
CONF_ZONETYPE,
DATA_EVL,
SIGNAL_ZONE_UPDATE,
ZONE_SCHEMA,
)
from .entity import EnvisalinkEntity
_LOGGER = logging.getLogger(__name__)
@@ -31,13 +41,12 @@ async def async_setup_platform(
"""Set up the Envisalink binary sensor entities."""
if not discovery_info:
return
configured_zones = discovery_info["zones"]
configured_zones: dict[int, dict[str, Any]] = discovery_info[CONF_ZONES]
entities = []
for zone_num in configured_zones:
entity_config_data = ZONE_SCHEMA(configured_zones[zone_num])
for zone_num, zone_data in configured_zones.items():
entity_config_data = ZONE_SCHEMA(zone_data)
entity = EnvisalinkBinarySensor(
hass,
zone_num,
entity_config_data[CONF_ZONENAME],
entity_config_data[CONF_ZONETYPE],
@@ -52,9 +61,16 @@ async def async_setup_platform(
class EnvisalinkBinarySensor(EnvisalinkEntity, BinarySensorEntity):
"""Representation of an Envisalink binary sensor."""
def __init__(self, hass, zone_number, zone_name, zone_type, info, controller):
def __init__(
self,
zone_number: int,
zone_name: str,
zone_type: BinarySensorDeviceClass,
info: dict[str, Any],
controller: EnvisalinkAlarmPanel,
) -> None:
"""Initialize the binary_sensor."""
self._zone_type = zone_type
self._attr_device_class = zone_type
self._zone_number = zone_number
_LOGGER.debug("Setting up zone: %s", zone_name)
@@ -69,9 +85,9 @@ class EnvisalinkBinarySensor(EnvisalinkEntity, BinarySensorEntity):
)
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
attr = {}
attr: dict[str, Any] = {}
# The Envisalink library returns a "last_fault" value that's the
# number of seconds since the last fault, up to a maximum of 327680
@@ -104,11 +120,6 @@ class EnvisalinkBinarySensor(EnvisalinkEntity, BinarySensorEntity):
"""Return true if sensor is on."""
return self._info["status"]["open"]
@property
def device_class(self) -> BinarySensorDeviceClass:
"""Return the class of this sensor, from DEVICE_CLASSES."""
return self._zone_type
@callback
def async_update_callback(self, zone):
"""Update the zone's state, if needed."""

View File

@@ -1,5 +1,9 @@
"""Support for Envisalink devices."""
from typing import Any
from pyenvisalink import EnvisalinkAlarmPanel
from homeassistant.helpers.entity import Entity
@@ -8,13 +12,10 @@ class EnvisalinkEntity(Entity):
_attr_should_poll = False
def __init__(self, name, info, controller):
def __init__(
self, name: str, info: dict[str, Any], controller: EnvisalinkAlarmPanel
) -> None:
"""Initialize the device."""
self._controller = controller
self._info = info
self._name = name
@property
def name(self):
"""Return the name of the device."""
return self._name
self._attr_name = name

View File

@@ -3,6 +3,9 @@
from __future__ import annotations
import logging
from typing import Any
from pyenvisalink import EnvisalinkAlarmPanel
from homeassistant.components.sensor import SensorEntity
from homeassistant.core import HomeAssistant, callback
@@ -12,6 +15,7 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import (
CONF_PARTITIONNAME,
CONF_PARTITIONS,
DATA_EVL,
PARTITION_SCHEMA,
SIGNAL_KEYPAD_UPDATE,
@@ -31,13 +35,12 @@ async def async_setup_platform(
"""Perform the setup for Envisalink sensor entities."""
if not discovery_info:
return
configured_partitions = discovery_info["partitions"]
configured_partitions: dict[int, dict[str, Any]] = discovery_info[CONF_PARTITIONS]
entities = []
for part_num in configured_partitions:
entity_config_data = PARTITION_SCHEMA(configured_partitions[part_num])
for part_num, part_config in configured_partitions.items():
entity_config_data = PARTITION_SCHEMA(part_config)
entity = EnvisalinkSensor(
hass,
entity_config_data[CONF_PARTITIONNAME],
part_num,
hass.data[DATA_EVL].alarm_state["partition"][part_num],
@@ -52,9 +55,16 @@ async def async_setup_platform(
class EnvisalinkSensor(EnvisalinkEntity, SensorEntity):
"""Representation of an Envisalink keypad."""
def __init__(self, hass, partition_name, partition_number, info, controller):
_attr_icon = "mdi:alarm"
def __init__(
self,
partition_name: str,
partition_number: int,
info: dict[str, Any],
controller: EnvisalinkAlarmPanel,
) -> None:
"""Initialize the sensor."""
self._icon = "mdi:alarm"
self._partition_number = partition_number
_LOGGER.debug("Setting up sensor for partition: %s", partition_name)
@@ -73,11 +83,6 @@ class EnvisalinkSensor(EnvisalinkEntity, SensorEntity):
)
)
@property
def icon(self):
"""Return the icon if any."""
return self._icon
@property
def native_value(self):
"""Return the overall state."""

View File

@@ -5,13 +5,21 @@ from __future__ import annotations
import logging
from typing import Any
from pyenvisalink import EnvisalinkAlarmPanel
from homeassistant.components.switch import SwitchEntity
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import CONF_ZONENAME, DATA_EVL, SIGNAL_ZONE_BYPASS_UPDATE, ZONE_SCHEMA
from . import (
CONF_ZONENAME,
CONF_ZONES,
DATA_EVL,
SIGNAL_ZONE_BYPASS_UPDATE,
ZONE_SCHEMA,
)
from .entity import EnvisalinkEntity
_LOGGER = logging.getLogger(__name__)
@@ -26,16 +34,15 @@ async def async_setup_platform(
"""Set up the Envisalink switch entities."""
if not discovery_info:
return
configured_zones = discovery_info["zones"]
configured_zones: dict[int, dict[str, Any]] = discovery_info[CONF_ZONES]
entities = []
for zone_num in configured_zones:
entity_config_data = ZONE_SCHEMA(configured_zones[zone_num])
for zone_num, zone_data in configured_zones.items():
entity_config_data = ZONE_SCHEMA(zone_data)
zone_name = f"{entity_config_data[CONF_ZONENAME]}_bypass"
_LOGGER.debug("Setting up zone_bypass switch: %s", zone_name)
entity = EnvisalinkSwitch(
hass,
zone_num,
zone_name,
hass.data[DATA_EVL].alarm_state["zone"][zone_num],
@@ -49,7 +56,13 @@ async def async_setup_platform(
class EnvisalinkSwitch(EnvisalinkEntity, SwitchEntity):
"""Representation of an Envisalink switch."""
def __init__(self, hass, zone_number, zone_name, info, controller):
def __init__(
self,
zone_number: int,
zone_name: str,
info: dict[str, Any],
controller: EnvisalinkAlarmPanel,
) -> None:
"""Initialize the switch."""
self._zone_number = zone_number

View File

@@ -14,7 +14,7 @@
"name": "[%key:component::fan::common::condition_behavior_name%]"
}
},
"name": "If a fan is off"
"name": "Fan is off"
},
"is_on": {
"description": "Tests if one or more fans are on.",
@@ -24,7 +24,7 @@
"name": "[%key:component::fan::common::condition_behavior_name%]"
}
},
"name": "If a fan is on"
"name": "Fan is on"
}
},
"device_automation": {

View File

@@ -59,7 +59,7 @@ class HMBinarySensor(HMDevice, BinarySensorEntity):
"""Representation of a binary HomeMatic device."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if switch is on."""
if not self.available:
return False
@@ -73,7 +73,7 @@ class HMBinarySensor(HMDevice, BinarySensorEntity):
return BinarySensorDeviceClass.MOTION
return SENSOR_TYPES_CLASS.get(self._hmdevice.__class__.__name__)
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate the data dictionary (self._data) from metadata."""
# Add state to data struct
if self._state:
@@ -86,11 +86,11 @@ class HMBatterySensor(HMDevice, BinarySensorEntity):
_attr_device_class = BinarySensorDeviceClass.BATTERY
@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if battery is low."""
return bool(self._hm_get_state())
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate the data dictionary (self._data) from metadata."""
# Add state to data struct
if self._state:

View File

@@ -178,7 +178,7 @@ class HMThermostat(HMDevice, ClimateEntity):
# Homematic
return self._data.get("CONTROL_MODE")
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate a data dict (self._data) from the Homematic metadata."""
self._state = next(iter(self._hmdevice.WRITENODE.keys()))
self._data[self._state] = None

View File

@@ -78,7 +78,7 @@ class HMCover(HMDevice, CoverEntity):
"""Stop the device if in motion."""
self._hmdevice.stop(self._channel)
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate a data dictionary (self._data) from metadata."""
self._state = "LEVEL"
self._data.update({self._state: None})
@@ -138,7 +138,7 @@ class HMGarage(HMCover):
"""Return whether the cover is closed."""
return self._hmdevice.is_closed(self._hm_get_state())
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate a data dictionary (self._data) from metadata."""
self._state = "DOOR_STATE"
self._data.update({self._state: None})

View File

@@ -204,7 +204,7 @@ class HMDevice(Entity):
self._init_data_struct()
@abstractmethod
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate a data dictionary from the HomeMatic device metadata."""

View File

@@ -51,7 +51,7 @@ class HMLight(HMDevice, LightEntity):
_attr_max_color_temp_kelvin = 6500 # 153 Mireds
@property
def brightness(self):
def brightness(self) -> int | None:
"""Return the brightness of this light between 0..255."""
# Is dimmer?
if self._state == "LEVEL":
@@ -59,7 +59,7 @@ class HMLight(HMDevice, LightEntity):
return None
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if light is on."""
try:
return self._hm_get_state() > 0
@@ -98,7 +98,7 @@ class HMLight(HMDevice, LightEntity):
return features
@property
def hs_color(self):
def hs_color(self) -> tuple[float, float] | None:
"""Return the hue and saturation color value [float, float]."""
if ColorMode.HS not in self.supported_color_modes:
return None
@@ -116,14 +116,14 @@ class HMLight(HMDevice, LightEntity):
)
@property
def effect_list(self):
def effect_list(self) -> list[str] | None:
"""Return the list of supported effects."""
if not self.supported_features & LightEntityFeature.EFFECT:
return None
return self._hmdevice.get_effect_list()
@property
def effect(self):
def effect(self) -> str | None:
"""Return the current color change program of the light."""
if not self.supported_features & LightEntityFeature.EFFECT:
return None
@@ -166,7 +166,7 @@ class HMLight(HMDevice, LightEntity):
self._hmdevice.off(self._channel)
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate a data dict (self._data) from the Homematic metadata."""
# Use LEVEL
self._state = "LEVEL"

View File

@@ -48,7 +48,7 @@ class HMLock(HMDevice, LockEntity):
"""Open the door latch."""
self._hmdevice.open()
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate the data dictionary (self._data) from metadata."""
self._state = "STATE"
self._data.update({self._state: None})

View File

@@ -339,7 +339,7 @@ class HMSensor(HMDevice, SensorEntity):
# No cast, return original value
return self._hm_get_state()
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate a data dictionary (self._data) from metadata."""
if self._state:
self._data.update({self._state: None})

View File

@@ -35,7 +35,7 @@ class HMSwitch(HMDevice, SwitchEntity):
"""Representation of a HomeMatic switch."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if switch is on."""
try:
return self._hm_get_state() > 0
@@ -43,7 +43,7 @@ class HMSwitch(HMDevice, SwitchEntity):
return False
@property
def today_energy_kwh(self):
def today_energy_kwh(self) -> float | None:
"""Return the current power usage in kWh."""
if "ENERGY_COUNTER" in self._data:
try:
@@ -61,7 +61,7 @@ class HMSwitch(HMDevice, SwitchEntity):
"""Turn the switch off."""
self._hmdevice.off(self._channel)
def _init_data_struct(self):
def _init_data_struct(self) -> None:
"""Generate the data dictionary (self._data) from metadata."""
self._state = "STATE"
self._data.update({self._state: None})

View File

@@ -13,7 +13,7 @@
"requirements": [
"xknx==3.14.0",
"xknxproject==3.8.2",
"knx-frontend==2025.12.30.151231"
"knx-frontend==2026.1.15.112308"
],
"single_config_entry": true
}

View File

@@ -49,7 +49,7 @@
"name": "[%key:component::light::common::condition_behavior_name%]"
}
},
"name": "If a light is off"
"name": "Light is off"
},
"is_on": {
"description": "Tests if one or more lights are on.",
@@ -59,7 +59,7 @@
"name": "[%key:component::light::common::condition_behavior_name%]"
}
},
"name": "If a light is on"
"name": "Light is on"
}
},
"device_automation": {

View File

@@ -141,7 +141,7 @@ class NinaConfigFlow(ConfigFlow, domain=DOMAIN):
try:
self._all_region_codes_sorted = swap_key_value(
await nina.getAllRegionalCodes()
await nina.get_all_regional_codes()
)
except ApiError:
return self.async_abort(reason="no_fetch")
@@ -221,7 +221,7 @@ class OptionsFlowHandler(OptionsFlowWithReload):
try:
self._all_region_codes_sorted = swap_key_value(
await nina.getAllRegionalCodes()
await nina.get_all_regional_codes()
)
except ApiError:
return self.async_abort(reason="no_fetch")

View File

@@ -66,7 +66,7 @@ class NINADataUpdateCoordinator(
regions: dict[str, str] = config_entry.data[CONF_REGIONS]
for region in regions:
self._nina.addRegion(region)
self._nina.add_region(region)
super().__init__(
hass,
@@ -151,7 +151,7 @@ class NINADataUpdateCoordinator(
raw_warn.sent or "",
raw_warn.start or "",
raw_warn.expires or "",
raw_warn.isValid(),
raw_warn.is_valid,
)
warnings_for_regions.append(warning_data)

View File

@@ -8,6 +8,6 @@
"iot_class": "cloud_polling",
"loggers": ["pynina"],
"quality_scale": "bronze",
"requirements": ["pynina==0.3.6"],
"requirements": ["pynina==1.0.2"],
"single_config_entry": true
}

View File

@@ -2,9 +2,10 @@
from __future__ import annotations
from datetime import timedelta
from datetime import datetime, timedelta
import logging
from operator import itemgetter
from typing import Any
import oasatelematics
import voluptuous as vol
@@ -55,9 +56,9 @@ def setup_platform(
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the OASA Telematics sensor."""
name = config[CONF_NAME]
stop_id = config[CONF_STOP_ID]
route_id = config.get(CONF_ROUTE_ID)
name: str = config[CONF_NAME]
stop_id: str = config[CONF_STOP_ID]
route_id: str = config[CONF_ROUTE_ID]
data = OASATelematicsData(stop_id, route_id)
@@ -68,42 +69,31 @@ class OASATelematicsSensor(SensorEntity):
"""Implementation of the OASA Telematics sensor."""
_attr_attribution = "Data retrieved from telematics.oasa.gr"
_attr_device_class = SensorDeviceClass.TIMESTAMP
_attr_icon = "mdi:bus"
def __init__(self, data, stop_id, route_id, name):
def __init__(
self, data: OASATelematicsData, stop_id: str, route_id: str, name: str
) -> None:
"""Initialize the sensor."""
self.data = data
self._name = name
self._attr_name = name
self._stop_id = stop_id
self._route_id = route_id
self._name_data = self._times = self._state = None
self._name_data: dict[str, Any] | None = None
self._times: list[dict[str, Any]] | None = None
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def device_class(self) -> SensorDeviceClass:
"""Return the class of this sensor."""
return SensorDeviceClass.TIMESTAMP
@property
def native_value(self):
"""Return the state of the sensor."""
return self._state
@property
def extra_state_attributes(self):
def extra_state_attributes(self) -> dict[str, Any]:
"""Return the state attributes."""
params = {}
if self._times is not None:
next_arrival_data = self._times[0]
if ATTR_NEXT_ARRIVAL in next_arrival_data:
next_arrival = next_arrival_data[ATTR_NEXT_ARRIVAL]
next_arrival: datetime = next_arrival_data[ATTR_NEXT_ARRIVAL]
params.update({ATTR_NEXT_ARRIVAL: next_arrival.isoformat()})
if len(self._times) > 1:
second_next_arrival_time = self._times[1][ATTR_NEXT_ARRIVAL]
second_next_arrival_time: datetime = self._times[1][ATTR_NEXT_ARRIVAL]
if second_next_arrival_time is not None:
second_arrival = second_next_arrival_time
params.update(
@@ -115,12 +105,13 @@ class OASATelematicsSensor(SensorEntity):
ATTR_STOP_ID: self._stop_id,
}
)
params.update(
{
ATTR_ROUTE_NAME: self._name_data[ATTR_ROUTE_NAME],
ATTR_STOP_NAME: self._name_data[ATTR_STOP_NAME],
}
)
if self._name_data is not None:
params.update(
{
ATTR_ROUTE_NAME: self._name_data[ATTR_ROUTE_NAME],
ATTR_STOP_NAME: self._name_data[ATTR_STOP_NAME],
}
)
return {k: v for k, v in params.items() if v}
def update(self) -> None:
@@ -130,7 +121,7 @@ class OASATelematicsSensor(SensorEntity):
self._name_data = self.data.name_data
next_arrival_data = self._times[0]
if ATTR_NEXT_ARRIVAL in next_arrival_data:
self._state = next_arrival_data[ATTR_NEXT_ARRIVAL]
self._attr_native_value = next_arrival_data[ATTR_NEXT_ARRIVAL]
class OASATelematicsData:

View File

@@ -1,265 +0,0 @@
"""Audio helper for SmartThings audio notifications."""
from __future__ import annotations
import asyncio
import contextlib
from dataclasses import dataclass
from datetime import timedelta
import logging
import secrets
from aiohttp import hdrs, web
from homeassistant.components import ffmpeg
from homeassistant.components.http import HomeAssistantView
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.network import NoURLAvailableError, get_url
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
PCM_SAMPLE_RATE = 24000
PCM_SAMPLE_WIDTH = 2
PCM_CHANNELS = 1
PCM_MIME = "audio/L16"
PCM_EXTENSION = ".pcm"
WARNING_DURATION_SECONDS = 40
FFMPEG_MAX_DURATION_SECONDS = 10 * 60
TRANSCODE_TIMEOUT_SECONDS = WARNING_DURATION_SECONDS + 10
_TRUNCATION_EPSILON = 1 / PCM_SAMPLE_RATE
ENTRY_TTL = timedelta(minutes=5)
MAX_STORED_ENTRIES = 4 # Limit the number of cached notifications.
PCM_FRAME_BYTES = PCM_SAMPLE_WIDTH * PCM_CHANNELS
DATA_AUDIO_MANAGER = "audio_manager"
class SmartThingsAudioError(HomeAssistantError):
"""Error raised when SmartThings audio preparation fails."""
@dataclass
class _AudioEntry:
"""Stored PCM audio entry."""
pcm: bytes
created: float
expires: float
class SmartThingsAudioManager(HomeAssistantView):
"""Manage PCM proxy URLs for SmartThings audio notifications."""
url = "/api/smartthings/audio/{token}"
name = "api:smartthings:audio"
requires_auth = False
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize the manager."""
self.hass = hass
self._entries: dict[str, _AudioEntry] = {}
self._cleanup_handle: asyncio.TimerHandle | None = None
async def async_prepare_notification(self, source_url: str) -> str:
"""Generate an externally accessible PCM URL for SmartThings."""
pcm, duration, truncated = await self._transcode_to_pcm(source_url)
if not pcm:
raise SmartThingsAudioError("Converted audio is empty")
if truncated:
_LOGGER.warning(
"SmartThings audio notification truncated to %s seconds (output length %.1fs); longer sources may be cut off",
FFMPEG_MAX_DURATION_SECONDS,
duration,
)
elif duration > WARNING_DURATION_SECONDS:
_LOGGER.warning(
"SmartThings audio notification is %.1fs; playback over %s seconds may be cut off",
duration,
WARNING_DURATION_SECONDS,
)
token = secrets.token_urlsafe(
16
) # Shorter tokens avoid playback issues in some devices.
now = self.hass.loop.time()
entry = _AudioEntry(
pcm=pcm,
created=now,
expires=now + ENTRY_TTL.total_seconds(),
)
self._cleanup(now)
while token in self._entries:
token = secrets.token_urlsafe(16)
self._entries[token] = entry
while len(self._entries) > MAX_STORED_ENTRIES:
dropped_token = next(iter(self._entries))
self._entries.pop(dropped_token, None)
_LOGGER.debug(
"Dropped oldest SmartThings audio token %s to cap cache",
dropped_token,
)
self._schedule_cleanup()
path = f"/api/smartthings/audio/{token}{PCM_EXTENSION}"
try:
base_url = get_url(
self.hass,
allow_internal=True,
allow_external=True,
allow_cloud=True,
prefer_external=False, # Prevent NAT loopback failures; may break non-local access for devices outside the LAN.
prefer_cloud=True,
)
except NoURLAvailableError as err:
self._entries.pop(token, None)
self._schedule_cleanup()
raise SmartThingsAudioError(
"SmartThings audio notifications require an accessible Home Assistant URL"
) from err
return f"{base_url}{path}"
async def get(self, request: web.Request, token: str) -> web.StreamResponse:
"""Serve a PCM audio response."""
token = token.removesuffix(PCM_EXTENSION)
now = self.hass.loop.time()
self._cleanup(now)
self._schedule_cleanup()
entry = self._entries.get(token)
if entry is None:
raise web.HTTPNotFound
_LOGGER.debug("Serving SmartThings audio token=%s to %s", token, request.remote)
response = web.Response(body=entry.pcm, content_type=PCM_MIME)
response.headers[hdrs.CACHE_CONTROL] = "no-store"
response.headers[hdrs.ACCEPT_RANGES] = "none"
response.headers[hdrs.CONTENT_DISPOSITION] = (
f'inline; filename="{token}{PCM_EXTENSION}"'
)
return response
async def _transcode_to_pcm(self, source_url: str) -> tuple[bytes, float, bool]:
"""Use ffmpeg to convert the source media to 24kHz mono PCM."""
manager = ffmpeg.get_ffmpeg_manager(self.hass)
command = [
manager.binary,
"-hide_banner",
"-loglevel",
"error",
"-nostdin",
"-i",
source_url,
"-ac",
str(PCM_CHANNELS),
"-ar",
str(PCM_SAMPLE_RATE),
"-c:a",
"pcm_s16le",
"-t",
str(FFMPEG_MAX_DURATION_SECONDS),
"-f",
"s16le",
"pipe:1",
]
try:
process = await asyncio.create_subprocess_exec(
*command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
except FileNotFoundError as err:
raise SmartThingsAudioError(
"FFmpeg is required for SmartThings audio notifications"
) from err
try:
stdout, stderr = await asyncio.wait_for(
process.communicate(), timeout=TRANSCODE_TIMEOUT_SECONDS
)
except TimeoutError:
_LOGGER.warning(
"FFmpeg timed out after %s seconds while converting SmartThings audio from %s",
TRANSCODE_TIMEOUT_SECONDS,
source_url,
)
with contextlib.suppress(ProcessLookupError):
process.kill()
stdout, stderr = await process.communicate()
if process.returncode != 0:
message = stderr.decode().strip() or "unknown error"
_LOGGER.error(
"FFmpeg failed to convert SmartThings audio from %s: %s",
source_url,
message,
)
raise SmartThingsAudioError(
"Unable to convert audio to PCM for SmartThings"
)
if not stdout:
return b"", 0.0, False
frame_count, remainder = divmod(len(stdout), PCM_FRAME_BYTES)
if remainder:
_LOGGER.debug(
"SmartThings audio conversion produced misaligned PCM: dropping %s extra byte(s)",
remainder,
)
stdout = stdout[: len(stdout) - remainder]
frame_count = len(stdout) // PCM_FRAME_BYTES
if frame_count == 0:
return b"", 0.0, False
duration = frame_count / PCM_SAMPLE_RATE
truncated = duration >= (FFMPEG_MAX_DURATION_SECONDS - _TRUNCATION_EPSILON)
return stdout, duration, truncated
@callback
def _schedule_cleanup(self) -> None:
"""Schedule the next cleanup based on entry expiry."""
if self._cleanup_handle is not None:
self._cleanup_handle.cancel()
self._cleanup_handle = None
if not self._entries:
return
next_expiry = min(entry.expires for entry in self._entries.values())
delay = max(0.0, next_expiry - self.hass.loop.time())
self._cleanup_handle = self.hass.loop.call_later(delay, self._cleanup_callback)
@callback
def _cleanup_callback(self) -> None:
"""Run a cleanup pass."""
self._cleanup_handle = None
now = self.hass.loop.time()
self._cleanup(now)
self._schedule_cleanup()
def _cleanup(self, now: float) -> None:
"""Remove expired entries."""
expired = [
token for token, entry in self._entries.items() if entry.expires <= now
]
for token in expired:
self._entries.pop(token, None)
async def async_get_audio_manager(hass: HomeAssistant) -> SmartThingsAudioManager:
"""Return the shared SmartThings audio manager."""
domain_data = hass.data.setdefault(DOMAIN, {})
if (manager := domain_data.get(DATA_AUDIO_MANAGER)) is None:
manager = SmartThingsAudioManager(hass)
hass.http.register_view(manager)
domain_data[DATA_AUDIO_MANAGER] = manager
return manager

View File

@@ -3,7 +3,7 @@
"name": "SmartThings",
"codeowners": ["@joostlek"],
"config_flow": true,
"dependencies": ["application_credentials", "http", "ffmpeg"],
"dependencies": ["application_credentials"],
"dhcp": [
{
"hostname": "st*",

View File

@@ -6,22 +6,17 @@ from typing import Any
from pysmartthings import Attribute, Capability, Category, Command, SmartThings
from homeassistant.components import media_source
from homeassistant.components.media_player import (
MediaPlayerDeviceClass,
MediaPlayerEntity,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
RepeatMode,
async_process_play_media_url,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import FullDevice, SmartThingsConfigEntry
from .audio import SmartThingsAudioError, async_get_audio_manager
from .const import MAIN
from .entity import SmartThingsEntity
@@ -89,7 +84,6 @@ class SmartThingsMediaPlayer(SmartThingsEntity, MediaPlayerEntity):
Capability.AUDIO_MUTE,
Capability.AUDIO_TRACK_DATA,
Capability.AUDIO_VOLUME,
Capability.AUDIO_NOTIFICATION,
Capability.MEDIA_INPUT_SOURCE,
Capability.MEDIA_PLAYBACK,
Capability.MEDIA_PLAYBACK_REPEAT,
@@ -134,8 +128,6 @@ class SmartThingsMediaPlayer(SmartThingsEntity, MediaPlayerEntity):
flags |= MediaPlayerEntityFeature.SHUFFLE_SET
if self.supports_capability(Capability.MEDIA_PLAYBACK_REPEAT):
flags |= MediaPlayerEntityFeature.REPEAT_SET
if self.supports_capability(Capability.AUDIO_NOTIFICATION):
flags |= MediaPlayerEntityFeature.PLAY_MEDIA
return flags
async def async_turn_off(self, **kwargs: Any) -> None:
@@ -241,40 +233,6 @@ class SmartThingsMediaPlayer(SmartThingsEntity, MediaPlayerEntity):
argument=HA_REPEAT_MODE_TO_SMARTTHINGS[repeat],
)
async def async_play_media(
self, media_type: MediaType | str, media_id: str, **kwargs: Any
) -> None:
"""Play media using SmartThings audio notifications."""
if not self.supports_capability(Capability.AUDIO_NOTIFICATION):
raise HomeAssistantError("Device does not support audio notifications")
if media_type not in (MediaType.MUSIC,):
raise HomeAssistantError(
"Unsupported media type for SmartThings audio notification"
)
if media_source.is_media_source_id(media_id):
play_item = await media_source.async_resolve_media(
self.hass, media_id, self.entity_id
)
media_id = async_process_play_media_url(self.hass, play_item.url)
else:
media_id = async_process_play_media_url(self.hass, media_id)
audio_manager = await async_get_audio_manager(self.hass)
try:
proxy_url = await audio_manager.async_prepare_notification(media_id)
except SmartThingsAudioError as err:
raise HomeAssistantError(str(err)) from err
command = Command("playTrackAndResume")
await self.execute_device_command(
Capability.AUDIO_NOTIFICATION,
command,
argument=[proxy_url],
)
@property
def media_title(self) -> str | None:
"""Title of current playing media."""

View File

@@ -2,6 +2,7 @@
from __future__ import annotations
import collections
from dataclasses import dataclass
from typing import Any, Self
@@ -140,6 +141,22 @@ class _SwingModeWrapper(DeviceWrapper):
return commands
def _filter_hvac_mode_mappings(tuya_range: list[str]) -> dict[str, HVACMode | None]:
"""Filter TUYA_HVAC_TO_HA modes that are not in the range.
If multiple Tuya modes map to the same HA mode, set the mapping to None to avoid
ambiguity when converting back from HA to Tuya modes.
"""
modes_in_range = {
tuya_mode: TUYA_HVAC_TO_HA.get(tuya_mode) for tuya_mode in tuya_range
}
modes_occurrences = collections.Counter(modes_in_range.values())
for key, value in modes_in_range.items():
if value is not None and modes_occurrences[value] > 1:
modes_in_range[key] = None
return modes_in_range
class _HvacModeWrapper(DPCodeEnumWrapper):
"""Wrapper for managing climate HVACMode."""
@@ -148,10 +165,9 @@ class _HvacModeWrapper(DPCodeEnumWrapper):
def __init__(self, dpcode: str, type_information: EnumTypeInformation) -> None:
"""Init _HvacModeWrapper."""
super().__init__(dpcode, type_information)
self._mappings = _filter_hvac_mode_mappings(type_information.range)
self.options = [
TUYA_HVAC_TO_HA[tuya_mode]
for tuya_mode in type_information.range
if tuya_mode in TUYA_HVAC_TO_HA
ha_mode for ha_mode in self._mappings.values() if ha_mode is not None
]
def read_device_status(self, device: CustomerDevice) -> HVACMode | None:
@@ -166,7 +182,7 @@ class _HvacModeWrapper(DPCodeEnumWrapper):
"""Convert value to raw value."""
return next(
tuya_mode
for tuya_mode, ha_mode in TUYA_HVAC_TO_HA.items()
for tuya_mode, ha_mode in self._mappings.items()
if ha_mode == value
)
@@ -179,10 +195,9 @@ class _PresetWrapper(DPCodeEnumWrapper):
def __init__(self, dpcode: str, type_information: EnumTypeInformation) -> None:
"""Init _PresetWrapper."""
super().__init__(dpcode, type_information)
mappings = _filter_hvac_mode_mappings(type_information.range)
self.options = [
tuya_mode
for tuya_mode in type_information.range
if tuya_mode not in TUYA_HVAC_TO_HA
tuya_mode for tuya_mode, ha_mode in mappings.items() if ha_mode is None
]
def read_device_status(self, device: CustomerDevice) -> str | None:

View File

@@ -5,9 +5,11 @@ from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import timedelta
import logging
from uiprotect.data import (
Camera,
Chime,
Doorlock,
Light,
ModelType,
@@ -30,6 +32,8 @@ from .entity import (
)
from .utils import async_ufp_instance_command
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
@@ -245,6 +249,51 @@ _MODEL_DESCRIPTIONS: dict[ModelType, Sequence[ProtectEntityDescription]] = {
}
def _async_chime_ring_volume_entities(
data: ProtectData,
chime: Chime,
) -> list[ChimeRingVolumeNumber]:
"""Generate ring volume entities for each paired camera on a chime."""
entities: list[ChimeRingVolumeNumber] = []
if not chime.is_adopted_by_us:
return entities
auth_user = data.api.bootstrap.auth_user
if not chime.can_write(auth_user):
return entities
for ring_setting in chime.ring_settings:
camera = data.api.bootstrap.cameras.get(ring_setting.camera_id)
if camera is None:
_LOGGER.debug(
"Camera %s not found for chime %s ring volume",
ring_setting.camera_id,
chime.display_name,
)
continue
entities.append(ChimeRingVolumeNumber(data, chime, camera))
return entities
def _async_all_chime_ring_volume_entities(
data: ProtectData,
chime: Chime | None = None,
) -> list[ChimeRingVolumeNumber]:
"""Generate all ring volume entities for chimes."""
entities: list[ChimeRingVolumeNumber] = []
if chime is not None:
return _async_chime_ring_volume_entities(data, chime)
for device in data.get_by_types({ModelType.CHIME}):
if isinstance(device, Chime):
entities.extend(_async_chime_ring_volume_entities(data, device))
return entities
async def async_setup_entry(
hass: HomeAssistant,
entry: UFPConfigEntry,
@@ -255,23 +304,26 @@ async def async_setup_entry(
@callback
def _add_new_device(device: ProtectAdoptableDeviceModel) -> None:
async_add_entities(
async_all_device_entities(
data,
ProtectNumbers,
model_descriptions=_MODEL_DESCRIPTIONS,
ufp_device=device,
)
)
data.async_subscribe_adopt(_add_new_device)
async_add_entities(
async_all_device_entities(
entities = async_all_device_entities(
data,
ProtectNumbers,
model_descriptions=_MODEL_DESCRIPTIONS,
ufp_device=device,
)
# Add ring volume entities for chimes
if isinstance(device, Chime):
entities += _async_all_chime_ring_volume_entities(data, device)
async_add_entities(entities)
data.async_subscribe_adopt(_add_new_device)
entities = async_all_device_entities(
data,
ProtectNumbers,
model_descriptions=_MODEL_DESCRIPTIONS,
)
# Add ring volume entities for all chimes
entities += _async_all_chime_ring_volume_entities(data)
async_add_entities(entities)
class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
@@ -302,3 +354,62 @@ class ProtectNumbers(ProtectDeviceEntity, NumberEntity):
async def async_set_native_value(self, value: float) -> None:
"""Set new value."""
await self.entity_description.ufp_set(self.device, value)
class ChimeRingVolumeNumber(ProtectDeviceEntity, NumberEntity):
"""A UniFi Protect Number Entity for ring volume per camera on a chime."""
device: Chime
_state_attrs = ("_attr_available", "_attr_native_value")
_attr_native_max_value: float = 100
_attr_native_min_value: float = 0
_attr_native_step: float = 1
_attr_native_unit_of_measurement = PERCENTAGE
_attr_entity_category = EntityCategory.CONFIG
def __init__(
self,
data: ProtectData,
chime: Chime,
camera: Camera,
) -> None:
"""Initialize the ring volume number entity."""
self._camera_id = camera.id
# Use chime MAC and camera ID for unique ID
super().__init__(data, chime)
self._attr_unique_id = f"{chime.mac}_ring_volume_{camera.id}"
self._attr_translation_key = "chime_ring_volume"
self._attr_translation_placeholders = {"camera_name": camera.display_name}
# BaseProtectEntity sets _attr_name = None when no description is passed,
# which prevents translation_key from being used. Delete to enable translations.
del self._attr_name
@callback
def _async_update_device_from_protect(self, device: ProtectDeviceType) -> None:
"""Update entity from protect device."""
super()._async_update_device_from_protect(device)
self._attr_native_value = self._get_ring_volume()
def _get_ring_volume(self) -> int | None:
"""Get the ring volume for this camera from the chime's ring settings."""
for ring_setting in self.device.ring_settings:
if ring_setting.camera_id == self._camera_id:
return ring_setting.volume
return None
@property
def available(self) -> bool:
"""Return if entity is available."""
# Entity is unavailable if the camera is no longer paired with the chime
return super().available and self._get_ring_volume() is not None
@async_ufp_instance_command
async def async_set_native_value(self, value: float) -> None:
"""Set new ring volume value."""
camera = self.data.api.bootstrap.cameras.get(self._camera_id)
if camera is None:
_LOGGER.warning(
"Cannot set ring volume: camera %s not found", self._camera_id
)
return
await self.device.set_volume_for_camera_public(camera, int(value))

View File

@@ -323,6 +323,9 @@
"chime_duration": {
"name": "Chime duration"
},
"chime_ring_volume": {
"name": "Ring volume ({camera_name})"
},
"doorbell_ring_volume": {
"name": "Doorbell ring volume"
},

View File

@@ -330,11 +330,12 @@ ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL = vol.Schema(
)
class EntityConditionBase(Condition):
"""Base class for entity conditions."""
class EntityStateConditionBase(Condition):
"""State condition."""
_domain: str
_schema: vol.Schema = ENTITY_STATE_CONDITION_SCHEMA_ANY_ALL
_states: set[str]
@override
@classmethod
@@ -361,23 +362,19 @@ class EntityConditionBase(Condition):
if split_entity_id(entity_id)[0] == self._domain
}
@abc.abstractmethod
def is_valid_state(self, entity_state: State) -> bool:
"""Check if the state matches the expected state(s)."""
@override
async def async_get_checker(self) -> ConditionChecker:
"""Get the condition checker."""
def check_any_match_state(states: list[State]) -> bool:
"""Test if any entity matches the state."""
return any(self.is_valid_state(state) for state in states)
def check_any_match_state(states: list[str]) -> bool:
"""Test if any entity match the state."""
return any(state in self._states for state in states)
def check_all_match_state(states: list[State]) -> bool:
def check_all_match_state(states: list[str]) -> bool:
"""Test if all entities match the state."""
return all(self.is_valid_state(state) for state in states)
return all(state in self._states for state in states)
matcher: Callable[[list[State]], bool]
matcher: Callable[[list[str]], bool]
if self._behavior == BEHAVIOR_ANY:
matcher = check_any_match_state
elif self._behavior == BEHAVIOR_ALL:
@@ -393,7 +390,7 @@ class EntityConditionBase(Condition):
)
filtered_entity_ids = self.entity_filter(referenced_entity_ids)
entity_states = [
_state
_state.state
for entity_id in filtered_entity_ids
if (_state := self._hass.states.get(entity_id))
and _state.state not in (STATE_UNAVAILABLE, STATE_UNKNOWN)
@@ -403,16 +400,6 @@ class EntityConditionBase(Condition):
return test_state
class EntityStateConditionBase(EntityConditionBase):
"""State condition."""
_states: set[str]
def is_valid_state(self, entity_state: State) -> bool:
"""Check if the state matches the expected state(s)."""
return entity_state.state in self._states
def make_entity_state_condition(
domain: str, states: str | set[str]
) -> type[EntityStateConditionBase]:
@@ -432,37 +419,6 @@ def make_entity_state_condition(
return CustomCondition
class EntityStateAttributeConditionBase(EntityStateConditionBase):
"""State attribute condition."""
_attribute: str
_attribute_states: set[str]
def is_valid_state(self, entity_state: State) -> bool:
"""Check if the state matches the expected state(s)."""
return entity_state.attributes.get(self._attribute) in self._attribute_states
def make_entity_state_attribute_condition(
domain: str, attribute: str, attribute_states: str | set[str]
) -> type[EntityStateAttributeConditionBase]:
"""Create a condition for entity attribute matching specific state(s)."""
if isinstance(attribute_states, str):
attribute_states_set = {attribute_states}
else:
attribute_states_set = attribute_states
class CustomCondition(EntityStateAttributeConditionBase):
"""Condition for entity attribute."""
_domain = domain
_attribute = attribute
_attribute_states = attribute_states_set
return CustomCondition
class ConditionProtocol(Protocol):
"""Define the format of condition modules."""

4
requirements_all.txt generated
View File

@@ -1351,7 +1351,7 @@ kiwiki-client==0.1.1
knocki==0.4.2
# homeassistant.components.knx
knx-frontend==2025.12.30.151231
knx-frontend==2026.1.15.112308
# homeassistant.components.konnected
konnected==1.2.0
@@ -2234,7 +2234,7 @@ pynetgear==0.10.10
pynetio==0.1.9.1
# homeassistant.components.nina
pynina==0.3.6
pynina==1.0.2
# homeassistant.components.nintendo_parental_controls
pynintendoauth==1.0.2

View File

@@ -1185,7 +1185,7 @@ kegtron-ble==1.0.2
knocki==0.4.2
# homeassistant.components.knx
knx-frontend==2025.12.30.151231
knx-frontend==2026.1.15.112308
# homeassistant.components.konnected
konnected==1.2.0
@@ -1887,7 +1887,7 @@ pynecil==4.2.1
pynetgear==0.10.10
# homeassistant.components.nina
pynina==0.3.6
pynina==1.0.2
# homeassistant.components.nintendo_parental_controls
pynintendoauth==1.0.2

View File

@@ -0,0 +1,190 @@
"""Test assist satellite conditions."""
from typing import Any
import pytest
from homeassistant.components.assist_satellite.entity import AssistSatelliteState
from homeassistant.core import HomeAssistant
from tests.components import (
ConditionStateDescription,
assert_condition_gated_by_labs_flag,
create_target_condition,
other_states,
parametrize_condition_states,
parametrize_target_entities,
set_or_remove_state,
target_entities,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
async def target_assist_satellites(hass: HomeAssistant) -> list[str]:
"""Create multiple assist satellite entities associated with different targets."""
return (await target_entities(hass, "assist_satellite"))["included"]
@pytest.mark.parametrize(
"condition",
[
"assist_satellite.is_idle",
"assist_satellite.is_listening",
"assist_satellite.is_processing",
"assist_satellite.is_responding",
],
)
async def test_assist_satellite_conditions_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, condition: str
) -> None:
"""Test the assist satellite conditions are gated by the labs flag."""
await assert_condition_gated_by_labs_flag(hass, caplog, condition)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("assist_satellite"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states(
condition="assist_satellite.is_idle",
target_states=[AssistSatelliteState.IDLE],
other_states=other_states(AssistSatelliteState.IDLE),
),
*parametrize_condition_states(
condition="assist_satellite.is_listening",
target_states=[AssistSatelliteState.LISTENING],
other_states=other_states(AssistSatelliteState.LISTENING),
),
*parametrize_condition_states(
condition="assist_satellite.is_processing",
target_states=[AssistSatelliteState.PROCESSING],
other_states=other_states(AssistSatelliteState.PROCESSING),
),
*parametrize_condition_states(
condition="assist_satellite.is_responding",
target_states=[AssistSatelliteState.RESPONDING],
other_states=other_states(AssistSatelliteState.RESPONDING),
),
],
)
async def test_assist_satellite_state_condition_behavior_any(
hass: HomeAssistant,
target_assist_satellites: list[str],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the assist satellite state condition with the 'any' behavior."""
other_entity_ids = set(target_assist_satellites) - {entity_id}
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
condition = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
behavior="any",
)
for state in states:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
# Check if changing other assist satellites also passes the condition
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("assist_satellite"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states(
condition="assist_satellite.is_idle",
target_states=[AssistSatelliteState.IDLE],
other_states=other_states(AssistSatelliteState.IDLE),
),
*parametrize_condition_states(
condition="assist_satellite.is_listening",
target_states=[AssistSatelliteState.LISTENING],
other_states=other_states(AssistSatelliteState.LISTENING),
),
*parametrize_condition_states(
condition="assist_satellite.is_processing",
target_states=[AssistSatelliteState.PROCESSING],
other_states=other_states(AssistSatelliteState.PROCESSING),
),
*parametrize_condition_states(
condition="assist_satellite.is_responding",
target_states=[AssistSatelliteState.RESPONDING],
other_states=other_states(AssistSatelliteState.RESPONDING),
),
],
)
async def test_assist_satellite_state_condition_behavior_all(
hass: HomeAssistant,
target_assist_satellites: list[str],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the assist satellite state condition with the 'all' behavior."""
other_entity_ids = set(target_assist_satellites) - {entity_id}
# Set all assist satellites, including the tested one, to the initial state
for eid in target_assist_satellites:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
condition = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
behavior="all",
)
for state in states:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
# The condition passes if all entities are either in a target state or invalid
assert condition(hass) == (
(not state["state_valid"])
or (state["condition_true"] and entities_in_target == 1)
)
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
# The condition passes if all entities are either in a target state or invalid
assert condition(hass) == (
(not state["state_valid"]) or state["condition_true"]
)

View File

@@ -1,323 +0,0 @@
"""Test climate conditions."""
from typing import Any
import pytest
from homeassistant.components.climate.const import (
ATTR_HVAC_ACTION,
HVACAction,
HVACMode,
)
from homeassistant.core import HomeAssistant
from tests.components import (
ConditionStateDescription,
assert_condition_gated_by_labs_flag,
create_target_condition,
other_states,
parametrize_condition_states,
parametrize_target_entities,
set_or_remove_state,
target_entities,
)
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
"""Stub copying the blueprints to the config folder."""
@pytest.fixture
async def target_climates(hass: HomeAssistant) -> list[str]:
"""Create multiple climate entities associated with different targets."""
return (await target_entities(hass, "climate"))["included"]
@pytest.mark.parametrize(
"condition",
[
"climate.is_off",
"climate.is_on",
"climate.is_cooling",
"climate.is_drying",
"climate.is_heating",
],
)
async def test_climate_conditions_gated_by_labs_flag(
hass: HomeAssistant, caplog: pytest.LogCaptureFixture, condition: str
) -> None:
"""Test the climate conditions are gated by the labs flag."""
await assert_condition_gated_by_labs_flag(hass, caplog, condition)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states(
condition="climate.is_off",
target_states=[HVACMode.OFF],
other_states=other_states(HVACMode.OFF),
),
*parametrize_condition_states(
condition="climate.is_on",
target_states=[
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
],
other_states=[HVACMode.OFF],
),
],
)
async def test_climate_state_condition_behavior_any(
hass: HomeAssistant,
target_climates: list[str],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the climate state condition with the 'any' behavior."""
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
condition = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
behavior="any",
)
for state in states:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
# Check if changing other climates also passes the condition
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states(
condition="climate.is_off",
target_states=[HVACMode.OFF],
other_states=other_states(HVACMode.OFF),
),
*parametrize_condition_states(
condition="climate.is_on",
target_states=[
HVACMode.AUTO,
HVACMode.COOL,
HVACMode.DRY,
HVACMode.FAN_ONLY,
HVACMode.HEAT,
HVACMode.HEAT_COOL,
],
other_states=[HVACMode.OFF],
),
],
)
async def test_climate_state_condition_behavior_all(
hass: HomeAssistant,
target_climates: list[str],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the climate state condition with the 'all' behavior."""
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
condition = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
behavior="all",
)
for state in states:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
# The condition passes if all entities are either in a target state or invalid
assert condition(hass) == (
(not state["state_valid"])
or (state["condition_true"] and entities_in_target == 1)
)
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
# The condition passes if all entities are either in a target state or invalid
assert condition(hass) == (
(not state["state_valid"]) or state["condition_true"]
)
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states(
condition="climate.is_cooling",
target_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.COOLING})],
other_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.IDLE})],
),
*parametrize_condition_states(
condition="climate.is_drying",
target_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.DRYING})],
other_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.IDLE})],
),
*parametrize_condition_states(
condition="climate.is_heating",
target_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.HEATING})],
other_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.IDLE})],
),
],
)
async def test_climate_attribute_condition_behavior_any(
hass: HomeAssistant,
target_climates: list[str],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the climate attribute condition with the 'any' behavior."""
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
condition = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
behavior="any",
)
for state in states:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
# Check if changing other climates also passes the condition
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
assert condition(hass) == state["condition_true"]
@pytest.mark.usefixtures("enable_labs_preview_features")
@pytest.mark.parametrize(
("condition_target_config", "entity_id", "entities_in_target"),
parametrize_target_entities("climate"),
)
@pytest.mark.parametrize(
("condition", "condition_options", "states"),
[
*parametrize_condition_states(
condition="climate.is_cooling",
target_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.COOLING})],
other_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.IDLE})],
),
*parametrize_condition_states(
condition="climate.is_drying",
target_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.DRYING})],
other_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.IDLE})],
),
*parametrize_condition_states(
condition="climate.is_heating",
target_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.HEATING})],
other_states=[(HVACMode.AUTO, {ATTR_HVAC_ACTION: HVACAction.IDLE})],
),
],
)
async def test_climate_attribute_condition_behavior_all(
hass: HomeAssistant,
target_climates: list[str],
condition_target_config: dict,
entity_id: str,
entities_in_target: int,
condition: str,
condition_options: dict[str, Any],
states: list[ConditionStateDescription],
) -> None:
"""Test the climate attribute condition with the 'all' behavior."""
other_entity_ids = set(target_climates) - {entity_id}
# Set all climates, including the tested climate, to the initial state
for eid in target_climates:
set_or_remove_state(hass, eid, states[0]["included"])
await hass.async_block_till_done()
condition = await create_target_condition(
hass,
condition=condition,
target=condition_target_config,
behavior="all",
)
for state in states:
included_state = state["included"]
set_or_remove_state(hass, entity_id, included_state)
await hass.async_block_till_done()
# The condition passes if all entities are either in a target state or invalid
assert condition(hass) == (
(not state["state_valid"])
or (state["condition_true"] and entities_in_target == 1)
)
for other_entity_id in other_entity_ids:
set_or_remove_state(hass, other_entity_id, included_state)
await hass.async_block_till_done()
# The condition passes if all entities are either in a target state or invalid
assert condition(hass) == (
(not state["state_valid"]) or state["condition_true"]
)

View File

@@ -5,7 +5,7 @@ from typing import Any
import pytest
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.core import HomeAssistant
from tests.components import (
ConditionStateDescription,
@@ -137,7 +137,6 @@ async def test_light_state_condition_behavior_any(
)
async def test_light_state_condition_behavior_all(
hass: HomeAssistant,
service_calls: list[ServiceCall],
target_lights: list[str],
condition_target_config: dict,
entity_id: str,

View File

@@ -129,6 +129,7 @@ async def integration_fixture(
"oven",
"pressure_sensor",
"pump",
"resideo_x2s_thermostat",
"room_airconditioner",
"secuyou_smart_lock",
"silabs_dishwasher",

View File

@@ -0,0 +1,190 @@
{
"node_id": 4,
"date_commissioned": "2026-01-04T01:49:35.244151",
"last_interview": "2026-01-04T03:11:54.520702",
"interview_version": 6,
"available": true,
"is_bridge": false,
"attributes": {
"0/29/0": [
{
"0": 22,
"1": 1
}
],
"0/29/1": [29, 31, 40, 48, 51, 60, 62, 63],
"0/29/2": [],
"0/29/3": [1],
"0/29/65532": 0,
"0/29/65533": 2,
"0/29/65528": [],
"0/29/65529": [],
"0/29/65531": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"0/31/0": [
{
"1": 5,
"2": 2,
"3": [112233],
"4": null,
"254": 1
}
],
"0/31/2": 4,
"0/31/3": 3,
"0/31/4": 4,
"0/31/65532": 0,
"0/31/65533": 1,
"0/31/65528": [],
"0/31/65529": [],
"0/31/65531": [0, 2, 3, 4, 65528, 65529, 65531, 65532, 65533],
"0/40/0": 17,
"0/40/1": "Resideo",
"0/40/2": 4890,
"0/40/3": "X2S Smart Thermostat",
"0/40/4": 4096,
"0/40/5": "",
"0/40/6": "**REDACTED**",
"0/40/7": 2,
"0/40/8": "X2S_STAT_NPS_002",
"0/40/9": 1,
"0/40/10": "2.0.0.0",
"0/40/15": "**REDACTED**",
"0/40/19": {
"0": 3,
"1": 3
},
"0/40/21": 16973824,
"0/40/22": 1,
"0/40/65532": 0,
"0/40/65533": 3,
"0/40/65528": [],
"0/40/65529": [],
"0/40/65531": [
0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 15, 19, 21, 22, 65528, 65529, 65531,
65532, 65533
],
"0/48/0": 0,
"0/48/1": {
"0": 60,
"1": 900
},
"0/48/2": 0,
"0/48/3": 2,
"0/48/4": true,
"0/48/65532": 0,
"0/48/65533": 1,
"0/48/65528": [1, 3, 5],
"0/48/65529": [0, 2, 4],
"0/48/65531": [0, 1, 2, 3, 4, 65528, 65529, 65531, 65532, 65533],
"0/51/0": [
{
"0": "r0",
"1": true,
"2": null,
"3": null,
"4": "XPzhnNpQ",
"5": ["wKgJoQ=="],
"6": ["/oAAAAAAAABe/OH//pzaUA=="],
"7": 1
}
],
"0/51/1": 2,
"0/51/2": 5105,
"0/51/8": false,
"0/51/65532": 0,
"0/51/65533": 2,
"0/51/65528": [2],
"0/51/65529": [0, 1],
"0/51/65531": [0, 1, 2, 8, 65528, 65529, 65531, 65532, 65533],
"0/60/0": 0,
"0/60/1": null,
"0/60/2": null,
"0/60/65532": 0,
"0/60/65533": 1,
"0/60/65528": [],
"0/60/65529": [0, 2],
"0/60/65531": [0, 1, 2, 65528, 65529, 65531, 65532, 65533],
"0/62/0": [
{
"1": "FTABAQEkAgE3AyQTAhgmBIAigScmBYAlTTo3BiQVAiQRBBgkBwEkCAEwCUEEKl96Nj13XjcDn1kF4aoFwMWb9leBXP2Urts/tvLi1DF1UZkPEBrfZ5YYqd5tps3ELof6pBX91oACxfbnYF7UyzcKNQEoARgkAgE2AwQCBAEYMAQUY8nv41nGGNtJapsJ0+8/6EAnt9owBRRnrnI3xp/0zhgwIJN0RMbKS99orRgwC0BsruLcJINuIyVVZHD5AlYCuha4XhnLxtIjyYCXIIHGNuu39D6u/j94efSHPrOvVjAHXY56+z5KJguTzlTBOC5tGA==",
"2": "FTABAQEkAgE3AyQUARgmBIAigScmBYAlTTo3BiQTAhgkBwEkCAEwCUEEOoay7Kv2Hog5xV7knNJLl+Ywx5Sr/jrp6/PV5XF57NXm4UJfgdb6Ja7rZ+965UjigpYh+JVAVvCRK1xNgkikiDcKNQEpARgkAmAwBBRnrnI3xp/0zhgwIJN0RMbKS99orTAFFEHaAQy9nUPjHiRv7FIwcIp50v+EGDALQF5JHY0EJKgFC63BM4uO0mrkHpeTCSDpUEEz7IsvkdxAgUToWftgJSC3B7gqDelohC4uqReJpmeQ64F5XqYtB3AY",
"254": 1
}
],
"0/62/1": [
{
"1": "BGkTBQSFwwkc5WoOUncXmIahsjWs9bKfHyZRWpArIFMjhyjNKqURWvFS8xbVXTFf+UlFmJF2JnlMX4WgKjXkOLo=",
"2": 4939,
"3": 2,
"4": 4,
"5": "Home",
"254": 1
}
],
"0/62/2": 5,
"0/62/3": 1,
"0/62/4": [
"FTABAQEkAgE3AyQUARgmBIAigScmBYAlTTo3BiQUARgkBwEkCAEwCUEEaRMFBIXDCRzlag5SdxeYhqGyNaz1sp8fJlFakCsgUyOHKM0qpRFa8VLzFtVdMV/5SUWYkXYmeUxfhaAqNeQ4ujcKNQEpARgkAmAwBBRB2gEMvZ1D4x4kb+xSMHCKedL/hDAFFEHaAQy9nUPjHiRv7FIwcIp50v+EGDALQE/fBBea6WzXom6INogGzGdop0w7g8j4dcIo6v8Id2k+sttWqeL5we7dDJonx/m2MgVsQTKCeVhtN/nzT4stvmEY"
],
"0/62/5": 1,
"0/62/65532": 0,
"0/62/65533": 1,
"0/62/65528": [1, 3, 5, 8],
"0/62/65529": [0, 2, 4, 6, 7, 9, 10, 11],
"0/62/65531": [0, 1, 2, 3, 4, 5, 65528, 65529, 65531, 65532, 65533],
"0/63/0": [],
"0/63/1": [],
"0/63/2": 4,
"0/63/3": 3,
"0/63/65532": 0,
"0/63/65533": 2,
"0/63/65528": [2, 5],
"0/63/65529": [0, 1, 3, 4],
"0/63/65531": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"1/3/0": 0,
"1/3/1": 4,
"1/3/65532": 0,
"1/3/65533": 4,
"1/3/65528": [],
"1/3/65529": [0],
"1/3/65531": [0, 1, 65528, 65529, 65531, 65532, 65533],
"1/4/0": 128,
"1/4/65532": 1,
"1/4/65533": 4,
"1/4/65528": [0, 1, 2, 3],
"1/4/65529": [0, 1, 2, 3, 4, 5],
"1/4/65531": [0, 65528, 65529, 65531, 65532, 65533],
"1/29/0": [
{
"0": 769,
"1": 1
}
],
"1/29/1": [3, 4, 29, 513],
"1/29/2": [],
"1/29/3": [],
"1/29/65532": 0,
"1/29/65533": 2,
"1/29/65528": [],
"1/29/65529": [],
"1/29/65531": [0, 1, 2, 3, 65528, 65529, 65531, 65532, 65533],
"1/513/0": 2055,
"1/513/3": 443,
"1/513/4": 3221,
"1/513/5": 1000,
"1/513/6": 3721,
"1/513/17": 2666,
"1/513/18": 2166,
"1/513/25": 0,
"1/513/27": 4,
"1/513/28": 0,
"1/513/65532": 3,
"1/513/65533": 6,
"1/513/65528": [],
"1/513/65529": [0],
"1/513/65531": [
0, 3, 4, 5, 6, 17, 18, 25, 27, 28, 65528, 65529, 65531, 65532, 65533
]
},
"attribute_subscriptions": []
}

View File

@@ -3087,6 +3087,56 @@
'state': 'unknown',
})
# ---
# name: test_buttons[resideo_x2s_thermostat][button.x2s_smart_thermostat_identify-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': 'button',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'button.x2s_smart_thermostat_identify',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Identify',
'options': dict({
}),
'original_device_class': <ButtonDeviceClass.IDENTIFY: 'identify'>,
'original_icon': None,
'original_name': 'Identify',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '00000000000004D2-0000000000000004-MatterNodeDevice-1-IdentifyButton-3-1',
'unit_of_measurement': None,
})
# ---
# name: test_buttons[resideo_x2s_thermostat][button.x2s_smart_thermostat_identify-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'identify',
'friendly_name': 'X2S Smart Thermostat Identify',
}),
'context': <ANY>,
'entity_id': 'button.x2s_smart_thermostat_identify',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_buttons[secuyou_smart_lock][button.secuyou_smart_lock_identify-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@@ -467,6 +467,75 @@
'state': 'heat_cool',
})
# ---
# name: test_climates[resideo_x2s_thermostat][climate.x2s_smart_thermostat-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT: 'heat'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 32.2,
'min_temp': 4.4,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.x2s_smart_thermostat',
'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': None,
'original_name': None,
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'translation_key': None,
'unique_id': '00000000000004D2-0000000000000004-MatterNodeDevice-1-MatterThermostat-513-0',
'unit_of_measurement': None,
})
# ---
# name: test_climates[resideo_x2s_thermostat][climate.x2s_smart_thermostat-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 20.6,
'friendly_name': 'X2S Smart Thermostat',
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT: 'heat'>,
<HVACMode.COOL: 'cool'>,
<HVACMode.FAN_ONLY: 'fan_only'>,
]),
'max_temp': 32.2,
'min_temp': 4.4,
'supported_features': <ClimateEntityFeature: 385>,
'temperature': 21.7,
}),
'context': <ANY>,
'entity_id': 'climate.x2s_smart_thermostat',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_climates[room_airconditioner][climate.room_airconditioner-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@@ -9748,6 +9748,63 @@
'state': '60.0',
})
# ---
# name: test_sensors[resideo_x2s_thermostat][sensor.x2s_smart_thermostat_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': None,
'entity_id': 'sensor.x2s_smart_thermostat_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Temperature',
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.TEMPERATURE: 'temperature'>,
'original_icon': None,
'original_name': 'Temperature',
'platform': 'matter',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': None,
'unique_id': '00000000000004D2-0000000000000004-MatterNodeDevice-1-ThermostatLocalTemperature-513-0',
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
})
# ---
# name: test_sensors[resideo_x2s_thermostat][sensor.x2s_smart_thermostat_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'temperature',
'friendly_name': 'X2S Smart Thermostat Temperature',
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
'unit_of_measurement': <UnitOfTemperature.CELSIUS: '°C'>,
}),
'context': <ANY>,
'entity_id': 'sensor.x2s_smart_thermostat_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '20.55',
})
# ---
# name: test_sensors[room_airconditioner][sensor.room_airconditioner_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': set({

View File

@@ -17,7 +17,7 @@ from tests.common import (
async def setup_platform(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
"""Set up the NINA platform."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
await hass.config_entries.async_setup(config_entry.entry_id)

View File

@@ -25,7 +25,7 @@
'id': 'biw.BIWAPP-69634',
'is_valid': False,
'recommended_actions': '',
'sender': '',
'sender': None,
'sent': '1999-08-07T10:59:00+02:00',
'severity': 'Minor',
'start': '',

View File

@@ -49,7 +49,7 @@ def assert_dummy_entry_created(result: dict[str, Any]) -> None:
async def test_step_user_connection_error(hass: HomeAssistant) -> None:
"""Test starting a flow by user but no connection."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
side_effect=ApiError("Could not connect to Api"),
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@@ -63,7 +63,7 @@ async def test_step_user_connection_error(hass: HomeAssistant) -> None:
async def test_step_user_unexpected_exception(hass: HomeAssistant) -> None:
"""Test starting a flow by user but with an unexpected exception."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
side_effect=Exception("DUMMY"),
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@@ -77,7 +77,7 @@ async def test_step_user_unexpected_exception(hass: HomeAssistant) -> None:
async def test_step_user(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test starting a flow by user with valid values."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@@ -95,7 +95,7 @@ async def test_step_user(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> No
async def test_step_user_no_selection(hass: HomeAssistant) -> None:
"""Test starting a flow by user with no selection."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
result: dict[str, Any] = await hass.config_entries.flow.async_init(
@@ -121,7 +121,7 @@ async def test_step_user_already_configured(
) -> None:
"""Test starting a flow by user, but it was already configured."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
result = await hass.config_entries.flow.async_init(
@@ -141,7 +141,7 @@ async def test_options_flow_init(
with (
patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
),
):
@@ -195,7 +195,7 @@ async def test_options_flow_with_no_selection(
with (
patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
),
):
@@ -263,7 +263,7 @@ async def test_options_flow_connection_error(
await setup_platform(hass, mock_config_entry)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
side_effect=ApiError("Could not connect to Api"),
):
result = await hass.config_entries.options.async_init(
@@ -283,7 +283,7 @@ async def test_options_flow_unexpected_exception(
with (
patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
side_effect=Exception("DUMMY"),
),
):
@@ -312,7 +312,7 @@ async def test_options_flow_entity_removal(
with (
patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
),
):

View File

@@ -32,7 +32,7 @@ async def test_diagnostics(
"""Test diagnostics."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
config_entry: MockConfigEntry = MockConfigEntry(

View File

@@ -28,7 +28,7 @@ async def init_integration(hass: HomeAssistant) -> MockConfigEntry:
"""Set up the NINA integration in Home Assistant."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
entry: MockConfigEntry = MockConfigEntry(
@@ -54,7 +54,7 @@ async def test_config_migration_from1_1(hass: HomeAssistant) -> None:
)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
old_conf_entry.add_to_hass(hass)
@@ -82,7 +82,7 @@ async def test_config_migration_from1_2(hass: HomeAssistant) -> None:
)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
old_conf_entry.add_to_hass(hass)
@@ -104,7 +104,7 @@ async def test_config_migration_downgrade(hass: HomeAssistant) -> None:
)
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
wraps=mocked_request_function,
):
conf_entry.add_to_hass(hass)
@@ -126,7 +126,7 @@ async def test_config_entry_not_ready(hass: HomeAssistant) -> None:
async def test_sensors_connection_error(hass: HomeAssistant) -> None:
"""Test the creation and values of the NINA sensors with no connected."""
with patch(
"pynina.baseApi.BaseAPI._makeRequest",
"pynina.api_client.APIClient.make_request",
side_effect=ApiError("Could not connect to Api"),
):
conf_entry: MockConfigEntry = MockConfigEntry(

View File

@@ -1,7 +1,5 @@
"""Tests for the SmartThings integration."""
import sys
import types
from typing import Any
from unittest.mock import AsyncMock
@@ -92,38 +90,3 @@ async def trigger_health_update(
if call[0][0] == device_id:
call[0][1](event)
await hass.async_block_till_done()
def ensure_haffmpeg_stubs() -> None:
"""Ensure haffmpeg stubs are available for SmartThings tests."""
if "haffmpeg" in sys.modules:
return
haffmpeg_module = types.ModuleType("haffmpeg")
haffmpeg_core_module = types.ModuleType("haffmpeg.core")
haffmpeg_tools_module = types.ModuleType("haffmpeg.tools")
class _StubHAFFmpeg: ...
class _StubFFVersion:
def __init__(self, bin_path: str | None = None) -> None:
self.bin_path = bin_path
async def get_version(self) -> str:
return "4.0.0"
class _StubImageFrame: ...
haffmpeg_core_module.HAFFmpeg = _StubHAFFmpeg
haffmpeg_tools_module.IMAGE_JPEG = b""
haffmpeg_tools_module.FFVersion = _StubFFVersion
haffmpeg_tools_module.ImageFrame = _StubImageFrame
haffmpeg_module.core = haffmpeg_core_module
haffmpeg_module.tools = haffmpeg_tools_module
sys.modules["haffmpeg"] = haffmpeg_module
sys.modules["haffmpeg.core"] = haffmpeg_core_module
sys.modules["haffmpeg.tools"] = haffmpeg_tools_module
ensure_haffmpeg_stubs()

View File

@@ -37,7 +37,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 24461>,
'supported_features': <MediaPlayerEntityFeature: 23949>,
'translation_key': None,
'unique_id': 'afcf3b91-0000-1111-2222-ddff2a0a6577_main',
'unit_of_measurement': None,
@@ -59,7 +59,7 @@
'HDMI2',
'digital',
]),
'supported_features': <MediaPlayerEntityFeature: 24461>,
'supported_features': <MediaPlayerEntityFeature: 23949>,
'volume_level': 0.01,
}),
'context': <ANY>,
@@ -101,7 +101,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 318989>,
'supported_features': <MediaPlayerEntityFeature: 318477>,
'translation_key': None,
'unique_id': 'c9276e43-fe3c-88c3-1dcc-2eb79e292b8c_main',
'unit_of_measurement': None,
@@ -115,7 +115,7 @@
'is_volume_muted': False,
'repeat': <RepeatMode.OFF: 'off'>,
'shuffle': False,
'supported_features': <MediaPlayerEntityFeature: 318989>,
'supported_features': <MediaPlayerEntityFeature: 318477>,
'volume_level': 0.52,
}),
'context': <ANY>,
@@ -157,7 +157,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 22029>,
'supported_features': <MediaPlayerEntityFeature: 21517>,
'translation_key': None,
'unique_id': 'c85fced9-c474-4a47-93c2-037cc7829536_main',
'unit_of_measurement': None,
@@ -171,7 +171,7 @@
'is_volume_muted': False,
'media_artist': 'David Guetta',
'media_title': 'Forever Young',
'supported_features': <MediaPlayerEntityFeature: 22029>,
'supported_features': <MediaPlayerEntityFeature: 21517>,
'volume_level': 0.15,
}),
'context': <ANY>,
@@ -213,7 +213,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 22413>,
'supported_features': <MediaPlayerEntityFeature: 21901>,
'translation_key': None,
'unique_id': '0d94e5db-8501-2355-eb4f-214163702cac_main',
'unit_of_measurement': None,
@@ -228,7 +228,7 @@
'media_artist': '',
'media_title': '',
'source': 'HDMI1',
'supported_features': <MediaPlayerEntityFeature: 22413>,
'supported_features': <MediaPlayerEntityFeature: 21901>,
'volume_level': 0.17,
}),
'context': <ANY>,
@@ -270,7 +270,7 @@
'platform': 'smartthings',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <MediaPlayerEntityFeature: 1932>,
'supported_features': <MediaPlayerEntityFeature: 1420>,
'translation_key': None,
'unique_id': 'a75cb1e1-03fd-3c77-ca9f-d4e56c4096c6_main',
'unit_of_measurement': None,
@@ -281,7 +281,7 @@
'attributes': ReadOnlyDict({
'device_class': 'speaker',
'friendly_name': 'Soundbar',
'supported_features': <MediaPlayerEntityFeature: 1932>,
'supported_features': <MediaPlayerEntityFeature: 1420>,
}),
'context': <ANY>,
'entity_id': 'media_player.soundbar',

View File

@@ -1,531 +0,0 @@
"""Tests for SmartThings audio helper."""
from __future__ import annotations
import asyncio
import logging
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from urllib.parse import urlsplit
import pytest
from homeassistant.components.smartthings.audio import (
FFMPEG_MAX_DURATION_SECONDS,
MAX_STORED_ENTRIES,
PCM_CHANNELS,
PCM_MIME,
PCM_SAMPLE_RATE,
PCM_SAMPLE_WIDTH,
TRANSCODE_TIMEOUT_SECONDS,
WARNING_DURATION_SECONDS,
SmartThingsAudioError,
async_get_audio_manager,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.network import NoURLAvailableError
from tests.typing import ClientSessionGenerator
class _FakeProcess:
"""Async subprocess stand-in that provides communicate."""
def __init__(self, stdout: bytes, stderr: bytes, returncode: int) -> None:
self._stdout = stdout
self._stderr = stderr
self.returncode = returncode
self.killed = False
async def communicate(self) -> tuple[bytes, bytes]:
return self._stdout, self._stderr
def kill(self) -> None:
self.killed = True
def _build_pcm(
duration_seconds: float = 1.0,
*,
sample_rate: int = PCM_SAMPLE_RATE,
sample_width: int = PCM_SAMPLE_WIDTH,
channels: int = PCM_CHANNELS,
) -> bytes:
"""Generate silent raw PCM bytes for testing."""
frame_count = int(sample_rate * duration_seconds)
return b"\x00" * frame_count * sample_width * channels
async def test_prepare_notification_creates_url(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Ensure PCM proxy URLs are generated and served."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with patch.object(
manager, "_transcode_to_pcm", AsyncMock(return_value=(pcm_bytes, 1.0, False))
):
url = await manager.async_prepare_notification("https://example.com/source.mp3")
parsed = urlsplit(url)
assert parsed.path.endswith(".pcm")
assert not parsed.query
client = await hass_client_no_auth()
response = await client.get(parsed.path)
assert response.status == 200
assert response.headers["Content-Type"] == PCM_MIME
assert response.headers["Cache-Control"] == "no-store"
body = await response.read()
assert body == pcm_bytes
@pytest.mark.asyncio
async def test_prepare_notification_uses_internal_url_when_external_missing(
hass: HomeAssistant,
) -> None:
"""Fallback to the internal URL if no external URL is available."""
hass.config.external_url = None
hass.config.internal_url = "http://homeassistant.local:8123"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with patch.object(
manager, "_transcode_to_pcm", AsyncMock(return_value=(pcm_bytes, 1.0, False))
):
url = await manager.async_prepare_notification("https://example.com/source.mp3")
parsed = urlsplit(url)
assert parsed.scheme == "http"
assert parsed.netloc == "homeassistant.local:8123"
assert parsed.path.endswith(".pcm")
@pytest.mark.asyncio
async def test_prepare_notification_requires_accessible_url(
hass: HomeAssistant,
) -> None:
"""Fail if neither external nor internal URLs are available."""
hass.config.external_url = None
hass.config.internal_url = None
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with (
patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
),
patch(
"homeassistant.components.smartthings.audio.get_url",
side_effect=NoURLAvailableError,
) as mock_get_url,
pytest.raises(SmartThingsAudioError),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert mock_get_url.called
# Stored entry should be cleaned up after failure so subsequent requests
# don't leak memory or serve stale audio.
assert not manager._entries
async def test_audio_view_returns_404_for_unknown_token(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
) -> None:
"""Unknown tokens should return 404."""
await async_get_audio_manager(hass)
client = await hass_client_no_auth()
response = await client.get("/api/smartthings/audio/invalid-token.pcm")
assert response.status == 404
@pytest.mark.asyncio
async def test_prepare_notification_raises_when_transcode_empty(
hass: HomeAssistant,
) -> None:
"""Transcoding empty audio results in an error."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
with (
patch.object(
manager, "_transcode_to_pcm", AsyncMock(return_value=(b"", 0.0, False))
),
pytest.raises(SmartThingsAudioError, match="Converted audio is empty"),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
@pytest.mark.asyncio
async def test_prepare_notification_warns_when_duration_exceeds_max(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Warn when transcoded audio exceeds the SmartThings duration limit."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = b"pcm"
caplog.set_level(logging.WARNING)
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, FFMPEG_MAX_DURATION_SECONDS + 1.0, True)),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert any("truncated" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_prepare_notification_warns_when_duration_exceeds_warning(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Warn when transcoded audio exceeds the SmartThings warning threshold."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm(duration_seconds=WARNING_DURATION_SECONDS + 1)
caplog.set_level(logging.WARNING)
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, WARNING_DURATION_SECONDS + 1.0, False)),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert any(
"playback over" in record.message and "truncated" not in record.message
for record in caplog.records
)
@pytest.mark.asyncio
async def test_prepare_notification_regenerates_token_on_collision(
hass: HomeAssistant,
) -> None:
"""Regenerate tokens when a collision is detected."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with (
patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
),
patch(
"homeassistant.components.smartthings.audio.secrets.token_urlsafe",
side_effect=["dup", "dup", "unique"],
),
):
url1 = await manager.async_prepare_notification(
"https://example.com/source.mp3"
)
url2 = await manager.async_prepare_notification(
"https://example.com/source.mp3"
)
assert urlsplit(url1).path.endswith("/dup.pcm")
assert urlsplit(url2).path.endswith("/unique.pcm")
@pytest.mark.asyncio
async def test_prepare_notification_schedules_cleanup(
hass: HomeAssistant,
) -> None:
"""Ensure cached entries are scheduled for cleanup."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert manager._cleanup_handle is not None
for entry in manager._entries.values():
entry.expires = 0
manager._cleanup_callback()
assert not manager._entries
@pytest.mark.asyncio
async def test_prepare_notification_caps_entry_count(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Ensure cached entries are capped."""
hass.config.external_url = "https://example.com"
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm()
caplog.set_level(logging.DEBUG)
with patch.object(
manager,
"_transcode_to_pcm",
AsyncMock(return_value=(pcm_bytes, 1.0, False)),
):
for _ in range(MAX_STORED_ENTRIES + 2):
await manager.async_prepare_notification("https://example.com/source.mp3")
assert len(manager._entries) == MAX_STORED_ENTRIES
assert any(
"Dropped oldest SmartThings audio token" in record.message
for record in caplog.records
)
@pytest.mark.asyncio
async def test_transcode_to_pcm_handles_missing_ffmpeg(
hass: HomeAssistant,
) -> None:
"""Raise friendly error when ffmpeg is unavailable."""
manager = await async_get_audio_manager(hass)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
side_effect=FileNotFoundError,
),
pytest.raises(SmartThingsAudioError, match="FFmpeg is required"),
):
await manager._transcode_to_pcm("https://example.com/source.mp3")
@pytest.mark.asyncio
async def test_transcode_to_pcm_handles_process_failure(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Raise when ffmpeg reports an error."""
manager = await async_get_audio_manager(hass)
caplog.set_level(logging.ERROR)
fake_process = _FakeProcess(stdout=b"", stderr=b"boom", returncode=1)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
pytest.raises(SmartThingsAudioError, match="Unable to convert"),
):
await manager._transcode_to_pcm("https://example.com/source.mp3")
assert any("FFmpeg failed" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_transcode_to_pcm_times_out_and_kills_process(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Kill ffmpeg when the transcode times out."""
manager = await async_get_audio_manager(hass)
fake_process = _FakeProcess(stdout=b"\x00\x00", stderr=b"", returncode=0)
caplog.set_level(logging.WARNING)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.wait_for",
side_effect=TimeoutError,
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert fake_process.killed is True
assert pcm == b"\x00\x00"
assert duration == pytest.approx(1 / PCM_SAMPLE_RATE)
assert truncated is False
assert any("FFmpeg timed out" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_transcode_to_pcm_returns_empty_audio(
hass: HomeAssistant,
) -> None:
"""Return empty payload when ffmpeg produced nothing."""
manager = await async_get_audio_manager(hass)
fake_process = _FakeProcess(stdout=b"", stderr=b"", returncode=0)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
) as mock_exec,
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert pcm == b""
assert duration == 0.0
assert truncated is False
mock_exec.assert_awaited_once()
@pytest.mark.asyncio
async def test_transcode_to_pcm_enforces_duration_cap(
hass: HomeAssistant,
) -> None:
"""Ensure ffmpeg is instructed to limit duration and timeout is enforced."""
manager = await async_get_audio_manager(hass)
pcm_bytes = _build_pcm(duration_seconds=FFMPEG_MAX_DURATION_SECONDS)
fake_process = _FakeProcess(stdout=pcm_bytes, stderr=b"", returncode=0)
timeouts: list[float] = []
original_wait_for = asyncio.wait_for
async def _wait_for(awaitable, timeout):
timeouts.append(timeout)
return await original_wait_for(awaitable, timeout)
mock_exec = AsyncMock(return_value=fake_process)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
mock_exec,
),
patch(
"homeassistant.components.smartthings.audio.asyncio.wait_for",
new=_wait_for,
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
command = list(mock_exec.await_args.args)
assert "-t" in command
assert command[command.index("-t") + 1] == str(FFMPEG_MAX_DURATION_SECONDS)
assert timeouts == [TRANSCODE_TIMEOUT_SECONDS]
assert pcm == pcm_bytes
assert duration == pytest.approx(FFMPEG_MAX_DURATION_SECONDS)
assert truncated is True
async def test_transcode_to_pcm_logs_misaligned_pcm(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Log debug output when ffmpeg output contains a partial frame."""
manager = await async_get_audio_manager(hass)
caplog.set_level(logging.DEBUG)
pcm_bytes = _build_pcm() + b"\xaa"
fake_process = _FakeProcess(stdout=pcm_bytes, stderr=b"", returncode=0)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert pcm == _build_pcm()
assert duration > 0
assert truncated is False
assert any("misaligned PCM" in record.message for record in caplog.records)
@pytest.mark.asyncio
async def test_transcode_to_pcm_drops_partial_frame_payload(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Drop audio entirely when ffmpeg returns fewer bytes than a full frame."""
manager = await async_get_audio_manager(hass)
caplog.set_level(logging.DEBUG)
fake_process = _FakeProcess(stdout=b"\x00", stderr=b"", returncode=0)
with (
patch(
"homeassistant.components.smartthings.audio.ffmpeg.get_ffmpeg_manager",
return_value=SimpleNamespace(binary="ffmpeg"),
),
patch(
"homeassistant.components.smartthings.audio.asyncio.create_subprocess_exec",
AsyncMock(return_value=fake_process),
),
):
pcm, duration, truncated = await manager._transcode_to_pcm(
"https://example.com/source.mp3"
)
assert pcm == b""
assert duration == 0.0
assert truncated is False
assert any("misaligned PCM" in record.message for record in caplog.records)

View File

@@ -1,7 +1,6 @@
"""Test for the SmartThings media player platform."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from unittest.mock import AsyncMock
from pysmartthings import Attribute, Capability, Command, Status
from pysmartthings.models import HealthStatus
@@ -10,19 +9,14 @@ from syrupy.assertion import SnapshotAssertion
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
ATTR_MEDIA_REPEAT,
ATTR_MEDIA_SHUFFLE,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
SERVICE_SELECT_SOURCE,
MediaType,
RepeatMode,
)
from homeassistant.components.smartthings.audio import SmartThingsAudioError
from homeassistant.components.smartthings.const import MAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
@@ -45,7 +39,6 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from . import (
@@ -205,176 +198,6 @@ async def test_volume_down(
)
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_notification(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test playing media via SmartThings audio notification."""
await setup_integration(hass, mock_config_entry)
manager = AsyncMock()
manager.async_prepare_notification.return_value = "https://example.com/audio.pcm"
with (
patch(
"homeassistant.components.smartthings.media_player.async_get_audio_manager",
AsyncMock(return_value=manager),
),
patch(
"homeassistant.components.smartthings.media_player.async_process_play_media_url",
return_value="https://example.com/source.mp3",
),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
{
ATTR_ENTITY_ID: "media_player.soundbar",
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_ID: "https://example.com/source.mp3",
},
blocking=True,
)
expected_command = Command("playTrackAndResume")
devices.execute_device_command.assert_called_once_with(
"afcf3b91-0000-1111-2222-ddff2a0a6577",
Capability.AUDIO_NOTIFICATION,
expected_command,
MAIN,
argument=["https://example.com/audio.pcm"],
)
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_requires_audio_notification_capability(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Expect an error if the device lacks audio notification support."""
devices.get_device_status.return_value[MAIN].pop(
Capability.AUDIO_NOTIFICATION, None
)
await setup_integration(hass, mock_config_entry)
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
with pytest.raises(
HomeAssistantError, match="Device does not support audio notifications"
):
await entity.async_play_media(MediaType.MUSIC, "https://example.com/source.mp3")
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_rejects_unsupported_media_type(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Unsupported media types should raise an error."""
await setup_integration(hass, mock_config_entry)
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
with pytest.raises(
HomeAssistantError, match="Unsupported media type for SmartThings audio"
):
await entity.async_play_media(
MediaType.TVSHOW, "https://example.com/source.mp3"
)
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_uses_media_source_resolution(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Media source IDs are resolved and processed before playback."""
await setup_integration(hass, mock_config_entry)
manager = AsyncMock()
manager.async_prepare_notification.return_value = "https://example.com/audio.pcm"
with (
patch(
"homeassistant.components.smartthings.media_player.async_get_audio_manager",
AsyncMock(return_value=manager),
),
patch(
"homeassistant.components.smartthings.media_player.async_process_play_media_url",
return_value="https://example.com/processed.mp3",
) as mock_process,
patch(
"homeassistant.components.smartthings.media_player.media_source.is_media_source_id",
return_value=True,
) as mock_is_media,
patch(
"homeassistant.components.smartthings.media_player.media_source.async_resolve_media",
AsyncMock(
return_value=SimpleNamespace(url="https://example.com/from_source")
),
) as mock_resolve,
):
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
await entity.async_play_media(MediaType.MUSIC, "media-source://foo")
mock_is_media.assert_called_once()
mock_resolve.assert_called_once()
mock_process.assert_called_with(hass, "https://example.com/from_source")
devices.execute_device_command.assert_called_once()
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_play_media_wraps_audio_errors(
hass: HomeAssistant,
devices: AsyncMock,
mock_config_entry: MockConfigEntry,
) -> None:
"""SmartThings audio errors propagate as HomeAssistantError."""
await setup_integration(hass, mock_config_entry)
manager = AsyncMock()
manager.async_prepare_notification.side_effect = SmartThingsAudioError("boom")
entity = hass.data["entity_components"][MEDIA_PLAYER_DOMAIN].get_entity(
"media_player.soundbar"
)
assert entity is not None
with (
patch(
"homeassistant.components.smartthings.media_player.async_get_audio_manager",
AsyncMock(return_value=manager),
),
patch(
"homeassistant.components.smartthings.media_player.async_process_play_media_url",
return_value="https://example.com/source.mp3",
),
pytest.raises(HomeAssistantError, match="boom"),
):
await entity.async_play_media(MediaType.MUSIC, "https://example.com/source.mp3")
@pytest.mark.parametrize("device_fixture", ["hw_q80r_soundbar"])
async def test_media_play(
hass: HomeAssistant,

View File

@@ -83,13 +83,13 @@
'capabilities': dict({
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT: 'heat'>,
]),
'max_temp': 35.0,
'min_temp': 5.0,
'preset_modes': list([
'auto',
'manual',
'off',
]),
'target_temp_step': 1.0,
@@ -130,13 +130,13 @@
'friendly_name': 'Anbau',
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT: 'heat'>,
]),
'max_temp': 35.0,
'min_temp': 5.0,
'preset_modes': list([
'auto',
'manual',
'off',
]),
'supported_features': <ClimateEntityFeature: 17>,
@@ -159,13 +159,13 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 70.0,
'min_temp': 1.0,
'preset_modes': list([
'holiday',
'auto',
'manual',
'eco',
]),
'target_temp_step': 0.5,
@@ -208,14 +208,14 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 70.0,
'min_temp': 1.0,
'preset_mode': None,
'preset_modes': list([
'holiday',
'auto',
'manual',
'eco',
]),
'supported_features': <ClimateEntityFeature: 17>,
@@ -453,13 +453,13 @@
'capabilities': dict({
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT: 'heat'>,
]),
'max_temp': 35.0,
'min_temp': 5.0,
'preset_modes': list([
'auto',
'manual',
'off',
]),
'target_temp_step': 1.0,
@@ -501,14 +501,14 @@
'friendly_name': 'Empore',
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT: 'heat'>,
]),
'max_temp': 35.0,
'min_temp': 5.0,
'preset_mode': None,
'preset_modes': list([
'auto',
'manual',
'off',
]),
'supported_features': <ClimateEntityFeature: 17>,
@@ -532,12 +532,12 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 35.0,
'min_temp': 5.0,
'preset_modes': list([
'auto',
'manual',
'off',
]),
'target_temp_step': 1.0,
@@ -580,13 +580,13 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 35.0,
'min_temp': 5.0,
'preset_mode': None,
'preset_modes': list([
'auto',
'manual',
'off',
]),
'supported_features': <ClimateEntityFeature: 401>,
@@ -1107,12 +1107,12 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 5.9,
'min_temp': 0.1,
'preset_modes': list([
'auto',
'manual',
'holiday',
]),
'target_temp_step': 0.5,
@@ -1155,13 +1155,13 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 5.9,
'min_temp': 0.1,
'preset_mode': None,
'preset_modes': list([
'auto',
'manual',
'holiday',
]),
'supported_features': <ClimateEntityFeature: 17>,
@@ -1185,10 +1185,13 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 90.0,
'min_temp': 5.0,
'preset_modes': list([
'auto',
'manual',
]),
'target_temp_step': 1.0,
}),
'config_entry_id': <ANY>,
@@ -1215,7 +1218,7 @@
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <ClimateEntityFeature: 385>,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': None,
'unique_id': 'tuya.sb3zdertrw50bgogkw',
'unit_of_measurement': None,
@@ -1229,11 +1232,15 @@
'hvac_modes': list([
<HVACMode.OFF: 'off'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.HEAT_COOL: 'heat_cool'>,
]),
'max_temp': 90.0,
'min_temp': 5.0,
'supported_features': <ClimateEntityFeature: 385>,
'preset_mode': None,
'preset_modes': list([
'auto',
'manual',
]),
'supported_features': <ClimateEntityFeature: 401>,
'target_temp_step': 1.0,
'temperature': 12.0,
}),

View File

@@ -3,10 +3,10 @@
from __future__ import annotations
from datetime import timedelta
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, Mock
import pytest
from uiprotect.data import Camera, Doorlock, IRLEDMode, Light
from uiprotect.data import Camera, Chime, Doorlock, IRLEDMode, Light, RingSetting
from homeassistant.components.unifiprotect.const import DEFAULT_ATTRIBUTION
from homeassistant.components.unifiprotect.number import (
@@ -264,3 +264,140 @@ async def test_number_lock_auto_close(
)
mock_method.assert_called_once_with(timedelta(seconds=15.0))
def _setup_chime_with_doorbell(
chime: Chime, doorbell: Camera, volume: int = 50
) -> None:
"""Set up chime with paired doorbell for testing."""
chime.camera_ids = [doorbell.id]
chime.ring_settings = [
RingSetting(
camera_id=doorbell.id,
repeat_times=1,
ringtone_id="test-ringtone-id",
volume=volume,
)
]
async def test_chime_ring_volume_setup(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
ufp: MockUFPFixture,
chime: Chime,
doorbell: Camera,
) -> None:
"""Test chime ring volume number entity setup."""
_setup_chime_with_doorbell(chime, doorbell, volume=75)
await init_entry(hass, ufp, [chime, doorbell], regenerate_ids=False)
entity_id = "number.test_chime_ring_volume_test_camera"
entity = entity_registry.async_get(entity_id)
assert entity is not None
assert entity.unique_id == f"{chime.mac}_ring_volume_{doorbell.id}"
state = hass.states.get(entity_id)
assert state is not None
assert state.state == "75"
assert state.attributes[ATTR_ATTRIBUTION] == DEFAULT_ATTRIBUTION
async def test_chime_ring_volume_set_value(
hass: HomeAssistant,
ufp: MockUFPFixture,
chime: Chime,
doorbell: Camera,
) -> None:
"""Test setting chime ring volume."""
_setup_chime_with_doorbell(chime, doorbell)
await init_entry(hass, ufp, [chime, doorbell], regenerate_ids=False)
entity_id = "number.test_chime_ring_volume_test_camera"
with patch_ufp_method(
chime, "set_volume_for_camera_public", new_callable=AsyncMock
) as mock_method:
await hass.services.async_call(
"number",
"set_value",
{ATTR_ENTITY_ID: entity_id, "value": 80.0},
blocking=True,
)
mock_method.assert_called_once_with(doorbell, 80)
async def test_chime_ring_volume_multiple_cameras(
hass: HomeAssistant,
ufp: MockUFPFixture,
chime: Chime,
doorbell: Camera,
) -> None:
"""Test chime ring volume with multiple paired cameras."""
doorbell2 = doorbell.model_copy()
doorbell2.id = "test-doorbell-2"
doorbell2.name = "Test Doorbell 2"
doorbell2.mac = "aa:bb:cc:dd:ee:02"
chime.camera_ids = [doorbell.id, doorbell2.id]
chime.ring_settings = [
RingSetting(
camera_id=doorbell.id,
repeat_times=1,
ringtone_id="test-ringtone-id",
volume=60,
),
RingSetting(
camera_id=doorbell2.id,
repeat_times=2,
ringtone_id="test-ringtone-id-2",
volume=80,
),
]
await init_entry(hass, ufp, [chime, doorbell, doorbell2], regenerate_ids=False)
state1 = hass.states.get("number.test_chime_ring_volume_test_camera")
assert state1 is not None
assert state1.state == "60"
state2 = hass.states.get("number.test_chime_ring_volume_test_doorbell_2")
assert state2 is not None
assert state2.state == "80"
async def test_chime_ring_volume_unavailable_when_unpaired(
hass: HomeAssistant,
ufp: MockUFPFixture,
chime: Chime,
doorbell: Camera,
) -> None:
"""Test chime ring volume becomes unavailable when camera is unpaired."""
_setup_chime_with_doorbell(chime, doorbell)
await init_entry(hass, ufp, [chime, doorbell], regenerate_ids=False)
entity_id = "number.test_chime_ring_volume_test_camera"
state = hass.states.get(entity_id)
assert state
assert state.state == "50"
# Simulate removing the camera pairing
new_chime = chime.model_copy()
new_chime.ring_settings = []
ufp.api.bootstrap.chimes = {new_chime.id: new_chime}
ufp.api.bootstrap.nvr.system_info.ustorage = None
mock_msg = Mock()
mock_msg.changed_data = {}
mock_msg.new_obj = new_chime
ufp.ws_msg(mock_msg)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "unavailable"