Move zone conditions to the zone integration (#148157)

This commit is contained in:
Erik Montnemery 2025-07-07 18:27:44 +02:00 committed by GitHub
parent 090b8f0659
commit 6396f54e0d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 368 additions and 312 deletions

View File

@ -7,6 +7,7 @@ from typing import Final
import voluptuous as vol
from homeassistant.components.zone import condition as zone_condition
from homeassistant.const import CONF_EVENT, CONF_PLATFORM, CONF_SOURCE, CONF_ZONE
from homeassistant.core import (
CALLBACK_TYPE,
@ -17,7 +18,7 @@ from homeassistant.core import (
State,
callback,
)
from homeassistant.helpers import condition, config_validation as cv
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.config_validation import entity_domain
from homeassistant.helpers.event import TrackStates, async_track_state_change_filtered
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
@ -79,9 +80,11 @@ async def async_attach_trigger(
return
from_match = (
condition.zone(hass, zone_state, from_state) if from_state else False
zone_condition.zone(hass, zone_state, from_state) if from_state else False
)
to_match = (
zone_condition.zone(hass, zone_state, to_state) if to_state else False
)
to_match = condition.zone(hass, zone_state, to_state) if to_state else False
if (trigger_event == EVENT_ENTER and not from_match and to_match) or (
trigger_event == EVENT_LEAVE and from_match and not to_match

View File

@ -0,0 +1,156 @@
"""Offer zone automation rules."""
from __future__ import annotations
import voluptuous as vol
from homeassistant.const import (
ATTR_GPS_ACCURACY,
ATTR_LATITUDE,
ATTR_LONGITUDE,
CONF_CONDITION,
CONF_ENTITY_ID,
CONF_ZONE,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import ConditionErrorContainer, ConditionErrorMessage
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.condition import (
Condition,
ConditionCheckerType,
trace_condition_function,
)
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from . import in_zone
_CONDITION_SCHEMA = vol.Schema(
{
**cv.CONDITION_BASE_SCHEMA,
vol.Required(CONF_CONDITION): "zone",
vol.Required(CONF_ENTITY_ID): cv.entity_ids,
vol.Required("zone"): cv.entity_ids,
# To support use_trigger_value in automation
# Deprecated 2016/04/25
vol.Optional("event"): vol.Any("enter", "leave"),
}
)
def zone(
hass: HomeAssistant,
zone_ent: str | State | None,
entity: str | State | None,
) -> bool:
"""Test if zone-condition matches.
Async friendly.
"""
if zone_ent is None:
raise ConditionErrorMessage("zone", "no zone specified")
if isinstance(zone_ent, str):
zone_ent_id = zone_ent
if (zone_ent := hass.states.get(zone_ent)) is None:
raise ConditionErrorMessage("zone", f"unknown zone {zone_ent_id}")
if entity is None:
raise ConditionErrorMessage("zone", "no entity specified")
if isinstance(entity, str):
entity_id = entity
if (entity := hass.states.get(entity)) is None:
raise ConditionErrorMessage("zone", f"unknown entity {entity_id}")
else:
entity_id = entity.entity_id
if entity.state in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
return False
latitude = entity.attributes.get(ATTR_LATITUDE)
longitude = entity.attributes.get(ATTR_LONGITUDE)
if latitude is None:
raise ConditionErrorMessage(
"zone", f"entity {entity_id} has no 'latitude' attribute"
)
if longitude is None:
raise ConditionErrorMessage(
"zone", f"entity {entity_id} has no 'longitude' attribute"
)
return in_zone(
zone_ent, latitude, longitude, entity.attributes.get(ATTR_GPS_ACCURACY, 0)
)
class ZoneCondition(Condition):
"""Zone condition."""
def __init__(self, hass: HomeAssistant, config: ConfigType) -> None:
"""Initialize condition."""
self._config = config
@classmethod
async def async_validate_condition_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return _CONDITION_SCHEMA(config) # type: ignore[no-any-return]
async def async_condition_from_config(self) -> ConditionCheckerType:
"""Wrap action method with zone based condition."""
entity_ids = self._config.get(CONF_ENTITY_ID, [])
zone_entity_ids = self._config.get(CONF_ZONE, [])
@trace_condition_function
def if_in_zone(hass: HomeAssistant, variables: TemplateVarsType = None) -> bool:
"""Test if condition."""
errors = []
all_ok = True
for entity_id in entity_ids:
entity_ok = False
for zone_entity_id in zone_entity_ids:
try:
if zone(hass, zone_entity_id, entity_id):
entity_ok = True
except ConditionErrorMessage as ex:
errors.append(
ConditionErrorMessage(
"zone",
(
f"error matching {entity_id} with {zone_entity_id}:"
f" {ex.message}"
),
)
)
if not entity_ok:
all_ok = False
# Raise the errors only if no definitive result was found
if errors and not all_ok:
raise ConditionErrorContainer("zone", errors=errors)
return all_ok
return if_in_zone
CONDITIONS: dict[str, type[Condition]] = {
"zone": ZoneCondition,
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the sun conditions."""
return CONDITIONS

View File

@ -22,7 +22,6 @@ from homeassistant.core import (
callback,
)
from homeassistant.helpers import (
condition,
config_validation as cv,
entity_registry as er,
location,
@ -31,6 +30,8 @@ from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
from . import condition
EVENT_ENTER = "enter"
EVENT_LEAVE = "leave"
DEFAULT_EVENT = EVENT_ENTER

View File

@ -18,9 +18,6 @@ import voluptuous as vol
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_GPS_ACCURACY,
ATTR_LATITUDE,
ATTR_LONGITUDE,
CONF_ABOVE,
CONF_AFTER,
CONF_ATTRIBUTE,
@ -36,7 +33,6 @@ from homeassistant.const import (
CONF_STATE,
CONF_VALUE_TEMPLATE,
CONF_WEEKDAY,
CONF_ZONE,
ENTITY_MATCH_ALL,
ENTITY_MATCH_ANY,
STATE_UNAVAILABLE,
@ -95,7 +91,6 @@ _PLATFORM_ALIASES: dict[str | None, str | None] = {
"template": None,
"time": None,
"trigger": None,
"zone": None,
}
INPUT_ENTITY_ID = re.compile(
@ -919,101 +914,6 @@ def time_from_config(config: ConfigType) -> ConditionCheckerType:
return time_if
def zone(
hass: HomeAssistant,
zone_ent: str | State | None,
entity: str | State | None,
) -> bool:
"""Test if zone-condition matches.
Async friendly.
"""
from homeassistant.components import zone as zone_cmp # noqa: PLC0415
if zone_ent is None:
raise ConditionErrorMessage("zone", "no zone specified")
if isinstance(zone_ent, str):
zone_ent_id = zone_ent
if (zone_ent := hass.states.get(zone_ent)) is None:
raise ConditionErrorMessage("zone", f"unknown zone {zone_ent_id}")
if entity is None:
raise ConditionErrorMessage("zone", "no entity specified")
if isinstance(entity, str):
entity_id = entity
if (entity := hass.states.get(entity)) is None:
raise ConditionErrorMessage("zone", f"unknown entity {entity_id}")
else:
entity_id = entity.entity_id
if entity.state in (
STATE_UNAVAILABLE,
STATE_UNKNOWN,
):
return False
latitude = entity.attributes.get(ATTR_LATITUDE)
longitude = entity.attributes.get(ATTR_LONGITUDE)
if latitude is None:
raise ConditionErrorMessage(
"zone", f"entity {entity_id} has no 'latitude' attribute"
)
if longitude is None:
raise ConditionErrorMessage(
"zone", f"entity {entity_id} has no 'longitude' attribute"
)
return zone_cmp.in_zone(
zone_ent, latitude, longitude, entity.attributes.get(ATTR_GPS_ACCURACY, 0)
)
def zone_from_config(config: ConfigType) -> ConditionCheckerType:
"""Wrap action method with zone based condition."""
entity_ids = config.get(CONF_ENTITY_ID, [])
zone_entity_ids = config.get(CONF_ZONE, [])
@trace_condition_function
def if_in_zone(hass: HomeAssistant, variables: TemplateVarsType = None) -> bool:
"""Test if condition."""
errors = []
all_ok = True
for entity_id in entity_ids:
entity_ok = False
for zone_entity_id in zone_entity_ids:
try:
if zone(hass, zone_entity_id, entity_id):
entity_ok = True
except ConditionErrorMessage as ex:
errors.append(
ConditionErrorMessage(
"zone",
(
f"error matching {entity_id} with {zone_entity_id}:"
f" {ex.message}"
),
)
)
if not entity_ok:
all_ok = False
# Raise the errors only if no definitive result was found
if errors and not all_ok:
raise ConditionErrorContainer("zone", errors=errors)
return all_ok
return if_in_zone
async def async_trigger_from_config(
hass: HomeAssistant, config: ConfigType
) -> ConditionCheckerType:

View File

@ -1570,18 +1570,6 @@ TRIGGER_CONDITION_SCHEMA = vol.Schema(
}
)
ZONE_CONDITION_SCHEMA = vol.Schema(
{
**CONDITION_BASE_SCHEMA,
vol.Required(CONF_CONDITION): "zone",
vol.Required(CONF_ENTITY_ID): entity_ids,
vol.Required("zone"): entity_ids,
# To support use_trigger_value in automation
# Deprecated 2016/04/25
vol.Optional("event"): vol.Any("enter", "leave"),
}
)
AND_CONDITION_SCHEMA = vol.Schema(
{
**CONDITION_BASE_SCHEMA,
@ -1729,7 +1717,6 @@ BUILT_IN_CONDITIONS: ValueSchemas = {
"template": TEMPLATE_CONDITION_SCHEMA,
"time": TIME_CONDITION_SCHEMA,
"trigger": TRIGGER_CONDITION_SCHEMA,
"zone": ZONE_CONDITION_SCHEMA,
}

View File

@ -54,6 +54,7 @@ CONDITIONS_SCHEMA = vol.Schema(
NON_MIGRATED_INTEGRATIONS = {
"device_automation",
"sun",
"zone",
}

View File

@ -0,0 +1,203 @@
"""The tests for the location condition."""
import pytest
from homeassistant.components.zone import condition as zone_condition
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConditionError
from homeassistant.helpers import condition, config_validation as cv
async def test_zone_raises(hass: HomeAssistant) -> None:
"""Test that zone raises ConditionError on errors."""
config = {
"condition": "zone",
"entity_id": "device_tracker.cat",
"zone": "zone.home",
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
with pytest.raises(ConditionError, match="no zone"):
zone_condition.zone(hass, zone_ent=None, entity="sensor.any")
with pytest.raises(ConditionError, match="unknown zone"):
test(hass)
hass.states.async_set(
"zone.home",
"zoning",
{"name": "home", "latitude": 2.1, "longitude": 1.1, "radius": 10},
)
with pytest.raises(ConditionError, match="no entity"):
zone_condition.zone(hass, zone_ent="zone.home", entity=None)
with pytest.raises(ConditionError, match="unknown entity"):
test(hass)
hass.states.async_set(
"device_tracker.cat",
"home",
{"friendly_name": "cat"},
)
with pytest.raises(ConditionError, match="latitude"):
test(hass)
hass.states.async_set(
"device_tracker.cat",
"home",
{"friendly_name": "cat", "latitude": 2.1},
)
with pytest.raises(ConditionError, match="longitude"):
test(hass)
hass.states.async_set(
"device_tracker.cat",
"home",
{"friendly_name": "cat", "latitude": 2.1, "longitude": 1.1},
)
# All okay, now test multiple failed conditions
assert test(hass)
config = {
"condition": "zone",
"entity_id": ["device_tracker.cat", "device_tracker.dog"],
"zone": ["zone.home", "zone.work"],
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
with pytest.raises(ConditionError, match="dog"):
test(hass)
with pytest.raises(ConditionError, match="work"):
test(hass)
hass.states.async_set(
"zone.work",
"zoning",
{"name": "work", "latitude": 20, "longitude": 10, "radius": 25000},
)
hass.states.async_set(
"device_tracker.dog",
"work",
{"friendly_name": "dog", "latitude": 20.1, "longitude": 10.1},
)
assert test(hass)
async def test_zone_multiple_entities(hass: HomeAssistant) -> None:
"""Test with multiple entities in condition."""
config = {
"condition": "and",
"conditions": [
{
"alias": "Zone Condition",
"condition": "zone",
"entity_id": ["device_tracker.person_1", "device_tracker.person_2"],
"zone": "zone.home",
},
],
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
hass.states.async_set(
"zone.home",
"zoning",
{"name": "home", "latitude": 2.1, "longitude": 1.1, "radius": 10},
)
hass.states.async_set(
"device_tracker.person_1",
"home",
{"friendly_name": "person_1", "latitude": 2.1, "longitude": 1.1},
)
hass.states.async_set(
"device_tracker.person_2",
"home",
{"friendly_name": "person_2", "latitude": 2.1, "longitude": 1.1},
)
assert test(hass)
hass.states.async_set(
"device_tracker.person_1",
"home",
{"friendly_name": "person_1", "latitude": 20.1, "longitude": 10.1},
)
hass.states.async_set(
"device_tracker.person_2",
"home",
{"friendly_name": "person_2", "latitude": 2.1, "longitude": 1.1},
)
assert not test(hass)
hass.states.async_set(
"device_tracker.person_1",
"home",
{"friendly_name": "person_1", "latitude": 2.1, "longitude": 1.1},
)
hass.states.async_set(
"device_tracker.person_2",
"home",
{"friendly_name": "person_2", "latitude": 20.1, "longitude": 10.1},
)
assert not test(hass)
async def test_multiple_zones(hass: HomeAssistant) -> None:
"""Test with multiple entities in condition."""
config = {
"condition": "and",
"conditions": [
{
"condition": "zone",
"entity_id": "device_tracker.person",
"zone": ["zone.home", "zone.work"],
},
],
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
hass.states.async_set(
"zone.home",
"zoning",
{"name": "home", "latitude": 2.1, "longitude": 1.1, "radius": 10},
)
hass.states.async_set(
"zone.work",
"zoning",
{"name": "work", "latitude": 20.1, "longitude": 10.1, "radius": 10},
)
hass.states.async_set(
"device_tracker.person",
"home",
{"friendly_name": "person", "latitude": 2.1, "longitude": 1.1},
)
assert test(hass)
hass.states.async_set(
"device_tracker.person",
"home",
{"friendly_name": "person", "latitude": 20.1, "longitude": 10.1},
)
assert test(hass)
hass.states.async_set(
"device_tracker.person",
"home",
{"friendly_name": "person", "latitude": 50.1, "longitude": 20.1},
)
assert not test(hass)

View File

@ -1892,201 +1892,6 @@ async def test_numeric_state_using_input_number(hass: HomeAssistant) -> None:
)
async def test_zone_raises(hass: HomeAssistant) -> None:
"""Test that zone raises ConditionError on errors."""
config = {
"condition": "zone",
"entity_id": "device_tracker.cat",
"zone": "zone.home",
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
with pytest.raises(ConditionError, match="no zone"):
condition.zone(hass, zone_ent=None, entity="sensor.any")
with pytest.raises(ConditionError, match="unknown zone"):
test(hass)
hass.states.async_set(
"zone.home",
"zoning",
{"name": "home", "latitude": 2.1, "longitude": 1.1, "radius": 10},
)
with pytest.raises(ConditionError, match="no entity"):
condition.zone(hass, zone_ent="zone.home", entity=None)
with pytest.raises(ConditionError, match="unknown entity"):
test(hass)
hass.states.async_set(
"device_tracker.cat",
"home",
{"friendly_name": "cat"},
)
with pytest.raises(ConditionError, match="latitude"):
test(hass)
hass.states.async_set(
"device_tracker.cat",
"home",
{"friendly_name": "cat", "latitude": 2.1},
)
with pytest.raises(ConditionError, match="longitude"):
test(hass)
hass.states.async_set(
"device_tracker.cat",
"home",
{"friendly_name": "cat", "latitude": 2.1, "longitude": 1.1},
)
# All okay, now test multiple failed conditions
assert test(hass)
config = {
"condition": "zone",
"entity_id": ["device_tracker.cat", "device_tracker.dog"],
"zone": ["zone.home", "zone.work"],
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
with pytest.raises(ConditionError, match="dog"):
test(hass)
with pytest.raises(ConditionError, match="work"):
test(hass)
hass.states.async_set(
"zone.work",
"zoning",
{"name": "work", "latitude": 20, "longitude": 10, "radius": 25000},
)
hass.states.async_set(
"device_tracker.dog",
"work",
{"friendly_name": "dog", "latitude": 20.1, "longitude": 10.1},
)
assert test(hass)
async def test_zone_multiple_entities(hass: HomeAssistant) -> None:
"""Test with multiple entities in condition."""
config = {
"condition": "and",
"conditions": [
{
"alias": "Zone Condition",
"condition": "zone",
"entity_id": ["device_tracker.person_1", "device_tracker.person_2"],
"zone": "zone.home",
},
],
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
hass.states.async_set(
"zone.home",
"zoning",
{"name": "home", "latitude": 2.1, "longitude": 1.1, "radius": 10},
)
hass.states.async_set(
"device_tracker.person_1",
"home",
{"friendly_name": "person_1", "latitude": 2.1, "longitude": 1.1},
)
hass.states.async_set(
"device_tracker.person_2",
"home",
{"friendly_name": "person_2", "latitude": 2.1, "longitude": 1.1},
)
assert test(hass)
hass.states.async_set(
"device_tracker.person_1",
"home",
{"friendly_name": "person_1", "latitude": 20.1, "longitude": 10.1},
)
hass.states.async_set(
"device_tracker.person_2",
"home",
{"friendly_name": "person_2", "latitude": 2.1, "longitude": 1.1},
)
assert not test(hass)
hass.states.async_set(
"device_tracker.person_1",
"home",
{"friendly_name": "person_1", "latitude": 2.1, "longitude": 1.1},
)
hass.states.async_set(
"device_tracker.person_2",
"home",
{"friendly_name": "person_2", "latitude": 20.1, "longitude": 10.1},
)
assert not test(hass)
async def test_multiple_zones(hass: HomeAssistant) -> None:
"""Test with multiple entities in condition."""
config = {
"condition": "and",
"conditions": [
{
"condition": "zone",
"entity_id": "device_tracker.person",
"zone": ["zone.home", "zone.work"],
},
],
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
hass.states.async_set(
"zone.home",
"zoning",
{"name": "home", "latitude": 2.1, "longitude": 1.1, "radius": 10},
)
hass.states.async_set(
"zone.work",
"zoning",
{"name": "work", "latitude": 20.1, "longitude": 10.1, "radius": 10},
)
hass.states.async_set(
"device_tracker.person",
"home",
{"friendly_name": "person", "latitude": 2.1, "longitude": 1.1},
)
assert test(hass)
hass.states.async_set(
"device_tracker.person",
"home",
{"friendly_name": "person", "latitude": 20.1, "longitude": 10.1},
)
assert test(hass)
hass.states.async_set(
"device_tracker.person",
"home",
{"friendly_name": "person", "latitude": 50.1, "longitude": 20.1},
)
assert not test(hass)
@pytest.mark.usefixtures("hass")
async def test_extract_entities() -> None:
"""Test extracting entities."""