Compare commits

...

1 Commits

Author SHA1 Message Date
Paulus Schoutsen
84e4a0f22e Allow finding relevant blueprints 2025-11-15 14:01:37 -05:00
2 changed files with 122 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,61 @@ 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 = [
entry
for entry in er.async_entries_for_device(er.async_get(hass), device_id)
if not entry.entity_category
]
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",
]
},
}
]
}