Allow finding relevant blueprints

This commit is contained in:
Paulus Schoutsen 2025-06-19 03:23:38 +00:00
parent 3dba7e5bd2
commit 410cbdd99b
2 changed files with 118 additions and 1 deletions

View File

@ -1,7 +1,13 @@
"""The blueprint integration.""" """The blueprint integration."""
from typing import Any
import voluptuous as vol
from homeassistant.const import CONF_NAME, CONF_SELECTOR
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.selector import selector as create_selector
from homeassistant.helpers.typing import ConfigType from homeassistant.helpers.typing import ConfigType
from . import websocket_api from . import websocket_api
@ -29,3 +35,57 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the blueprint integration.""" """Set up the blueprint integration."""
websocket_api.async_setup(hass) websocket_api.async_setup(hass)
return True return True
async def async_find_relevant_blueprints(
hass: HomeAssistant, device_id: str
) -> dict[str, list[dict[str, Any]]]:
"""Find all blueprints relevant to a specific device."""
results = {}
entities = er.async_entries_for_device(er.async_get(hass), device_id)
async def all_blueprints_generator(hass: HomeAssistant):
"""Yield all blueprints from all domains."""
blueprint_domains: dict[str, DomainBlueprints] = hass.data[DOMAIN]
for blueprint_domain in blueprint_domains.values():
blueprints = await blueprint_domain.async_get_blueprints()
for blueprint in blueprints.values():
yield blueprint
async for blueprint in all_blueprints_generator(hass):
blueprint_input_matches: dict[str, list[str]] = {}
for info in blueprint.inputs.values():
if (
not info
or not (selector_conf := info.get(CONF_SELECTOR))
or "entity" not in selector_conf
):
continue
selector = create_selector(selector_conf)
matched = []
for entity in entities:
try:
entity.entity_id, selector(entity.entity_id)
except vol.Invalid:
continue
matched.append(entity.entity_id)
if matched:
blueprint_input_matches[info[CONF_NAME]] = matched
if not blueprint_input_matches:
continue
results.setdefault(blueprint.domain, []).append(
{
"blueprint": blueprint,
"matched_input": blueprint_input_matches,
}
)
return results

View File

@ -1 +1,58 @@
"""Tests for the blueprint init.""" """Tests for the blueprint init."""
from pathlib import Path
from unittest.mock import patch
from homeassistant.components import automation, blueprint
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry
async def test_find_relevant_blueprints(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test finding relevant blueprints."""
config_entry = MockConfigEntry()
config_entry.add_to_hass(hass)
device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={("test_domain", "test_device")},
name="Test Device",
)
entity_registry.async_get_or_create(
"person",
"test_domain",
"test_entity",
device_id=device.id,
original_name="Test Person",
)
with patch.object(
hass.config,
"path",
return_value=Path(automation.__file__).parent / "blueprints",
):
automation.async_get_blueprints(hass)
results = await blueprint.async_find_relevant_blueprints(hass, device.id)
for matches in results.values():
for match in matches:
match["blueprint"] = match["blueprint"].name
assert results == {
"automation": [
{
"blueprint": "Motion-activated Light",
"matched_input": {
"Person": [
"person.test_domain_test_entity",
]
},
}
]
}