980 lines
41 KiB
Python
980 lines
41 KiB
Python
import asyncio
|
|
from collections import defaultdict
|
|
from collections.abc import Iterable, Iterator
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timedelta
|
|
from enum import StrEnum
|
|
import logging
|
|
import re
|
|
from typing import Any
|
|
|
|
from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
|
|
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
|
from homeassistant.config_entries import SOURCE_INTEGRATION_DISCOVERY, SOURCE_USER, ConfigEntry
|
|
from homeassistant.const import CONF_DEVICE, CONF_ENTITY_ID, CONF_PLATFORM, CONF_UNIQUE_ID
|
|
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant
|
|
from homeassistant.helpers import discovery_flow
|
|
import homeassistant.helpers.device_registry as dr
|
|
from homeassistant.helpers.entity import EntityCategory
|
|
import homeassistant.helpers.entity_registry as er
|
|
from homeassistant.helpers.event import async_call_later, async_track_time_interval
|
|
from homeassistant.helpers.typing import ConfigType
|
|
from homeassistant.loader import IntegrationNotFound, async_get_integration
|
|
|
|
from .common import SourceEntity, create_source_entity
|
|
from .const import (
|
|
CONF_MANUFACTURER,
|
|
CONF_MODE,
|
|
CONF_MODEL,
|
|
CONF_SENSORS,
|
|
DATA_DISCOVERY_MANAGER,
|
|
DISCOVERY_INTEGRATION_NAME,
|
|
DISCOVERY_POWER_PROFILES,
|
|
DISCOVERY_SOURCE_ENTITY,
|
|
DOMAIN,
|
|
DUMMY_ENTITY_ID,
|
|
MANUFACTURER_WLED,
|
|
CalculationStrategy,
|
|
)
|
|
from .device_binding import (
|
|
get_config_entry_ids,
|
|
get_non_composite_devices,
|
|
get_related_device_ids,
|
|
)
|
|
from .group_include.filter import (
|
|
CategoryFilter,
|
|
CompositeFilter,
|
|
DomainFilter,
|
|
FilterOperator,
|
|
LambdaFilter,
|
|
NotFilter,
|
|
get_filtered_entity_list,
|
|
)
|
|
from .helpers import get_or_create_unique_id
|
|
from .power_profile.factory import get_power_profile
|
|
from .power_profile.library import ModelInfo, ProfileLibrary
|
|
from .power_profile.power_profile import (
|
|
SUPPORTED_DOMAINS,
|
|
DeviceType,
|
|
DiscoveryBy,
|
|
PowerProfile,
|
|
is_device_type_supported_for_entity,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# Give Home Assistant a moment to settle before the first scan. Discovery walks the whole entity
|
|
# and device registry, which other integrations are still filling while they set up. Starting a
|
|
# little later keeps startup responsive and lets one pass see a more complete picture.
|
|
DISCOVERY_DELAY = timedelta(seconds=10)
|
|
REDISCOVERY_INTERVAL = timedelta(hours=2)
|
|
DISCOVERY_KEY_PREFIX = "pc_"
|
|
|
|
|
|
def device_discovery_key(device_id: str) -> str:
|
|
"""Build the discovery key identifying a device, prefixed to avoid conflicts with other integrations."""
|
|
return f"{DISCOVERY_KEY_PREFIX}{device_id}"
|
|
|
|
|
|
def describe_power_profile(profile: PowerProfile) -> str:
|
|
"""Describe a power profile for logging purposes."""
|
|
return (
|
|
f"{profile.manufacturer}/{profile.model}"
|
|
f"[device_type={profile.device_type}, strategy={profile.calculation_strategy}, "
|
|
f"source={profile.configuration_source}]"
|
|
)
|
|
|
|
|
|
def get_discovery_manager(hass: HomeAssistant) -> DiscoveryManager:
|
|
"""Return the shared discovery manager, creating a throwaway one when not yet set up."""
|
|
try:
|
|
return hass.data[DOMAIN][DATA_DISCOVERY_MANAGER] # type: ignore[no-any-return]
|
|
except KeyError:
|
|
return DiscoveryManager(hass, {})
|
|
|
|
|
|
async def _get_power_profile_by_source(
|
|
hass: HomeAssistant,
|
|
source_entity: SourceEntity,
|
|
discovery_by: DiscoveryBy,
|
|
) -> PowerProfile | None:
|
|
"""Look up a power profile for a source entity, discovered either by entity or by device."""
|
|
discovery_manager = get_discovery_manager(hass)
|
|
model_info = discovery_manager.extract_model_info(source_entity)
|
|
if not model_info:
|
|
return None
|
|
profiles = await discovery_manager.find_power_profiles(model_info, source_entity, discovery_by)
|
|
return profiles[0] if profiles else None
|
|
|
|
|
|
async def get_power_profile_by_source_entity(hass: HomeAssistant, source_entity: SourceEntity) -> PowerProfile | None:
|
|
"""Given a certain entity, lookup the manufacturer and model and return the power profile."""
|
|
if not source_entity.entity_entry:
|
|
return None
|
|
return await _get_power_profile_by_source(hass, source_entity, DiscoveryBy.ENTITY)
|
|
|
|
|
|
async def get_power_profile_by_source_device(hass: HomeAssistant, source_entity: SourceEntity) -> PowerProfile | None:
|
|
"""Look up a device-discovered power profile for a source entity's device."""
|
|
if not source_entity.device_entry or not source_entity.entity_entry:
|
|
return None
|
|
return await _get_power_profile_by_source(hass, source_entity, DiscoveryBy.DEVICE)
|
|
|
|
|
|
class DiscoveryStatus(StrEnum):
|
|
DISABLED = "disabled"
|
|
NOT_STARTED = "not_started"
|
|
IN_PROGRESS = "in_progress"
|
|
FINISHED = "finished"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DiscoveryCandidate:
|
|
"""Normalized source considered during a discovery run."""
|
|
|
|
source_entity: SourceEntity
|
|
discovery_type: DiscoveryBy
|
|
integration_domains: frozenset[str]
|
|
|
|
@property
|
|
def log_identifier(self) -> str:
|
|
"""Label used as prefix for all log messages about this candidate."""
|
|
return self.source_entity.log_identifier
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class DiscoveryStats:
|
|
"""Counters describing the outcome of a single discovery phase.
|
|
|
|
Logged as a summary after every phase, so an issue report shows at which step a device
|
|
dropped out of discovery, without having to walk through thousands of individual log lines.
|
|
"""
|
|
|
|
candidates: int = 0
|
|
low_priority_integration: int = 0
|
|
no_model_info: int = 0
|
|
no_profile_match: int = 0
|
|
already_configured: int = 0
|
|
flows_initiated: int = 0
|
|
errors: int = 0
|
|
|
|
def __str__(self) -> str:
|
|
return ", ".join(f"{key}={value}" for key, value in asdict(self).items())
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DiscoveryMatch:
|
|
"""Information needed to create a discovery flow for a candidate."""
|
|
|
|
power_profiles: list[PowerProfile] | None = None
|
|
extra_data: dict[str, Any] | None = None
|
|
unique_id: str | None = None
|
|
|
|
|
|
class DiscoveryManager:
|
|
"""This class is responsible for scanning the HA instance for entities and their manufacturer / model info
|
|
It checks if any of these devices is supported in the powercalc library
|
|
When entities are found it will dispatch a discovery flow, so the user can add them to their HA instance.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
hass: HomeAssistant,
|
|
ha_config: ConfigType,
|
|
exclude_device_types: list[DeviceType] | None = None,
|
|
exclude_self_usage_profiles: bool = False,
|
|
enabled: bool = True,
|
|
) -> None:
|
|
self.hass = hass
|
|
self.ha_config = ha_config
|
|
self._manually_configured_entities: set[str] | None = None
|
|
self._configured_discovery_keys: set[str] = set()
|
|
self._pending_discovery_keys: set[str] = set()
|
|
self._library: ProfileLibrary | None = None
|
|
self._exclude_device_types = set(exclude_device_types or [])
|
|
self._exclude_self_usage_profiles = exclude_self_usage_profiles
|
|
self._cancel_library_update_interval: CALLBACK_TYPE | None = None
|
|
self._cancel_initial_discovery: CALLBACK_TYPE | None = None
|
|
self._status = DiscoveryStatus.NOT_STARTED if enabled else DiscoveryStatus.DISABLED
|
|
|
|
def setup(self) -> None:
|
|
"""Setup the discovery manager. Schedule the library update interval and initial discovery.
|
|
|
|
The library update is scheduled regardless of the discovery status. Startup deliberately
|
|
loads the library from local storage, so this interval is what keeps it up to date, and
|
|
manually configured sensors depend on the library just as much as discovered ones.
|
|
`start_discovery` is a no-op while discovery is disabled.
|
|
"""
|
|
self._schedule_library_update()
|
|
|
|
if self._status == DiscoveryStatus.DISABLED:
|
|
_LOGGER.debug("Discovery manager is disabled, skipping initial discovery")
|
|
return
|
|
|
|
self._schedule_initial_discovery()
|
|
|
|
def _schedule_library_update(self) -> None:
|
|
"""Refresh the library from the download API every REDISCOVERY_INTERVAL, then rediscover."""
|
|
|
|
async def _rediscover(_: datetime) -> None:
|
|
"""Update the library and rediscover entities."""
|
|
await self.update_library_and_rediscover()
|
|
|
|
if self._cancel_library_update_interval:
|
|
self._cancel_library_update_interval()
|
|
|
|
self._cancel_library_update_interval = async_track_time_interval(
|
|
self.hass,
|
|
_rediscover,
|
|
REDISCOVERY_INTERVAL,
|
|
cancel_on_shutdown=True,
|
|
)
|
|
|
|
def _schedule_initial_discovery(self) -> None:
|
|
"""Start the first discovery run after DISCOVERY_DELAY, instead of during setup."""
|
|
if self._cancel_initial_discovery: # pragma: no cover
|
|
self._cancel_initial_discovery()
|
|
|
|
async def _start_discovery(_: datetime) -> None:
|
|
self._cancel_initial_discovery = None
|
|
# Nothing awaits this task, so handle failures here instead of leaving Home Assistant
|
|
# to report them as an unretrieved task exception.
|
|
try:
|
|
await self.start_discovery()
|
|
except asyncio.CancelledError:
|
|
if not self.hass.is_stopping:
|
|
raise
|
|
_LOGGER.debug("Initial discovery cancelled, Home Assistant is stopping")
|
|
except Exception:
|
|
if self.hass.is_stopping:
|
|
_LOGGER.debug("Initial discovery aborted, Home Assistant is stopping")
|
|
else:
|
|
_LOGGER.exception("Error during initial discovery")
|
|
|
|
_LOGGER.debug("Scheduling initial discovery in %s seconds", DISCOVERY_DELAY.total_seconds())
|
|
self._cancel_initial_discovery = async_call_later(
|
|
self.hass,
|
|
DISCOVERY_DELAY,
|
|
HassJob(_start_discovery, "powercalc initial discovery", cancel_on_shutdown=True),
|
|
)
|
|
|
|
async def update_library_and_rediscover(self) -> None:
|
|
"""Update the library and rediscover entities."""
|
|
library = await self._get_library()
|
|
await library.initialize(prefer_cached=False)
|
|
await self.start_discovery()
|
|
|
|
async def start_discovery(self) -> None:
|
|
"""Start the discovery procedure."""
|
|
if self._status == DiscoveryStatus.DISABLED:
|
|
_LOGGER.debug("Discovery manager is disabled, skipping discovery run")
|
|
return
|
|
if self._status == DiscoveryStatus.IN_PROGRESS:
|
|
_LOGGER.debug("Discovery already in progress, skipping new discovery run")
|
|
return
|
|
self._status = DiscoveryStatus.IN_PROGRESS
|
|
try:
|
|
self.initialize_existing_entries()
|
|
|
|
devices = self.get_devices()
|
|
devices_by_config_entry = self._index_devices_by_config_entry(devices)
|
|
|
|
_LOGGER.debug(
|
|
"Start auto discovery (devices=%d, powercalc_entries=%d, configured_discovery_keys=%d, "
|
|
"excluded_device_types=[%s], exclude_self_usage_profiles=%s)",
|
|
len(devices),
|
|
len(self.hass.config_entries.async_entries(DOMAIN)),
|
|
len(self._configured_discovery_keys),
|
|
",".join(sorted(self._exclude_device_types)),
|
|
self._exclude_self_usage_profiles,
|
|
)
|
|
|
|
_LOGGER.debug("Start entity discovery")
|
|
entity_stats = await self.perform_discovery(self._create_entity_candidates())
|
|
_LOGGER.debug("Done entity discovery (%s)", entity_stats)
|
|
|
|
_LOGGER.debug("Start device discovery")
|
|
device_stats = await self.perform_discovery(self._create_device_candidates(devices))
|
|
_LOGGER.debug("Done device discovery (%s)", device_stats)
|
|
|
|
_LOGGER.debug("Start config entry discovery")
|
|
config_entry_stats = await self.perform_discovery(
|
|
self._create_config_entry_candidates(devices_by_config_entry),
|
|
)
|
|
_LOGGER.debug("Done config entry discovery (%s)", config_entry_stats)
|
|
|
|
_LOGGER.debug(
|
|
"Done auto discovery, initiated %d discovery flow(s)",
|
|
entity_stats.flows_initiated + device_stats.flows_initiated + config_entry_stats.flows_initiated,
|
|
)
|
|
finally:
|
|
self._status = DiscoveryStatus.FINISHED
|
|
|
|
def initialize_existing_entries(self) -> None:
|
|
"""Build a list of config entries which are already setup, to prevent duplicate discovery flows"""
|
|
self._configured_discovery_keys = self._collect_configured_discovery_keys()
|
|
|
|
def _collect_configured_discovery_keys(self, excluded_entry_id: str | None = None) -> set[str]:
|
|
"""Collect discovery keys claimed by existing config entries."""
|
|
keys: set[str] = set()
|
|
for entry in self.hass.config_entries.async_entries(DOMAIN):
|
|
if not entry.unique_id or entry.entry_id == excluded_entry_id:
|
|
continue # pragma: no cover
|
|
keys.update(self._discovery_keys_for_entry(entry))
|
|
return keys
|
|
|
|
def _discovery_keys_for_entry(self, entry: ConfigEntry) -> set[str]:
|
|
"""Return all discovery keys claimed by a config entry.
|
|
|
|
A single physical device can be represented by several device registry entries. HA >=2026.8
|
|
splits devices belonging to multiple config entries into one device per entry, so the entry
|
|
may hold the composite device ID, which no longer resolves to a registered device, or one of
|
|
the split devices after the user resolved the composite device repair. Devices can also be
|
|
registered by several integrations, in which case they share identifiers or connections.
|
|
All of them are the device the user already configured, so none should be discovered again.
|
|
"""
|
|
keys = {entry.unique_id} if entry.unique_id else set()
|
|
|
|
entity_id = entry.data.get(CONF_ENTITY_ID)
|
|
if entity_id and entity_id != DUMMY_ENTITY_ID:
|
|
entity_id = str(entity_id)
|
|
keys.add(entity_id)
|
|
entity_entry = er.async_get(self.hass).async_get(entity_id)
|
|
if entity_entry and entity_entry.device_id:
|
|
keys.add(device_discovery_key(entity_entry.device_id))
|
|
|
|
device_id = entry.data.get(CONF_DEVICE)
|
|
if not device_id:
|
|
return keys
|
|
|
|
for related_device_id in get_related_device_ids(self.hass, str(device_id)):
|
|
keys.add(device_discovery_key(related_device_id))
|
|
return keys
|
|
|
|
def remove_initialized_flow(self, entry: ConfigEntry) -> None:
|
|
"""Remove a flow from the initialized flows."""
|
|
self._pending_discovery_keys.difference_update(self._discovery_keys_for_entry(entry))
|
|
self._configured_discovery_keys = self._collect_configured_discovery_keys(entry.entry_id)
|
|
|
|
async def perform_discovery(
|
|
self,
|
|
candidates: Iterable[DiscoveryCandidate],
|
|
) -> DiscoveryStats:
|
|
"""Discover profiles and create flows for normalized candidates."""
|
|
library = await self._get_library()
|
|
low_priority_domains = library.discovery_low_priority_domains
|
|
stats = DiscoveryStats()
|
|
# Device candidates from low priority integrations go last, so a device represented by
|
|
# several integrations is discovered through the preferred one, and only falls back to
|
|
# the low priority representation when nothing better matched a profile.
|
|
ordered_candidates = sorted(
|
|
candidates,
|
|
key=lambda candidate: (
|
|
candidate.discovery_type == DiscoveryBy.DEVICE
|
|
and bool(candidate.integration_domains & low_priority_domains)
|
|
),
|
|
)
|
|
for candidate in ordered_candidates:
|
|
stats.candidates += 1
|
|
try:
|
|
await self._discover_candidate(candidate, low_priority_domains, stats)
|
|
except Exception:
|
|
stats.errors += 1
|
|
_LOGGER.exception(
|
|
"%s: Error during %s discovery",
|
|
candidate.log_identifier,
|
|
candidate.discovery_type,
|
|
)
|
|
return stats
|
|
|
|
async def _discover_candidate(
|
|
self,
|
|
candidate: DiscoveryCandidate,
|
|
low_priority_domains: set[str],
|
|
stats: DiscoveryStats,
|
|
) -> None:
|
|
"""Discover a single candidate and create a flow for it when it matches a profile."""
|
|
matched_low_priority_domains = candidate.integration_domains & low_priority_domains
|
|
if candidate.discovery_type != DiscoveryBy.DEVICE and matched_low_priority_domains:
|
|
stats.low_priority_integration += 1
|
|
_LOGGER.debug(
|
|
"%s: Integration domain has low discovery priority, skipping discovery (domains=[%s])",
|
|
candidate.log_identifier,
|
|
",".join(sorted(matched_low_priority_domains)),
|
|
)
|
|
return
|
|
|
|
source_entity = candidate.source_entity
|
|
model_info = self.extract_model_info(source_entity)
|
|
if not model_info:
|
|
stats.no_model_info += 1
|
|
return
|
|
|
|
match = await self._find_discovery_match(source_entity, model_info, candidate.discovery_type)
|
|
if not match:
|
|
stats.no_profile_match += 1
|
|
_LOGGER.debug(
|
|
"%s: Model not found in library, skipping discovery (discovery_by=%s)",
|
|
candidate.log_identifier,
|
|
candidate.discovery_type,
|
|
)
|
|
return
|
|
|
|
unique_id = match.unique_id or self.create_unique_id(
|
|
candidate, match.power_profiles[0] if match.power_profiles else None
|
|
)
|
|
|
|
if self._is_already_discovered(candidate, unique_id):
|
|
stats.already_configured += 1
|
|
_LOGGER.debug(
|
|
"%s: Already setup with discovery, skipping new discovery (unique_id=%s)",
|
|
candidate.log_identifier,
|
|
unique_id,
|
|
)
|
|
return
|
|
|
|
await self._init_entity_discovery(
|
|
candidate,
|
|
model_info,
|
|
unique_id,
|
|
match.power_profiles,
|
|
match.extra_data,
|
|
)
|
|
stats.flows_initiated += 1
|
|
|
|
async def _find_discovery_match(
|
|
self,
|
|
source_entity: SourceEntity,
|
|
model_info: ModelInfo,
|
|
discovery_type: DiscoveryBy,
|
|
) -> DiscoveryMatch | None:
|
|
"""Resolve a candidate to either a library profile or a special discovery mode."""
|
|
if source_entity.entity_entry and self.is_wled_light(model_info, source_entity.entity_entry):
|
|
if DeviceType.LIGHT in self._exclude_device_types:
|
|
return None
|
|
unique_id = (
|
|
device_discovery_key(source_entity.device_id)
|
|
if source_entity.device_id
|
|
else get_or_create_unique_id({}, source_entity, None)
|
|
)
|
|
return DiscoveryMatch(extra_data={CONF_MODE: CalculationStrategy.WLED}, unique_id=unique_id)
|
|
|
|
power_profiles = await self.find_power_profiles(model_info, source_entity, discovery_type)
|
|
return DiscoveryMatch(power_profiles=power_profiles) if power_profiles else None
|
|
|
|
def _create_entity_candidates(self) -> Iterator[DiscoveryCandidate]:
|
|
"""Yield normalized entity discovery candidates."""
|
|
for entity_entry in self.get_entities():
|
|
yield DiscoveryCandidate(
|
|
source_entity=create_source_entity(entity_entry.entity_id, self.hass),
|
|
discovery_type=DiscoveryBy.ENTITY,
|
|
integration_domains=frozenset({entity_entry.platform}),
|
|
)
|
|
|
|
def _create_device_candidates(self, devices: Iterable[dr.DeviceEntry]) -> Iterator[DiscoveryCandidate]:
|
|
"""Yield normalized device discovery candidates."""
|
|
for device_entry in devices:
|
|
yield DiscoveryCandidate(
|
|
source_entity=self.create_device_source(device_entry),
|
|
discovery_type=DiscoveryBy.DEVICE,
|
|
integration_domains=self._get_integration_domains(get_config_entry_ids(device_entry)),
|
|
)
|
|
|
|
def _create_config_entry_candidates(
|
|
self,
|
|
devices_by_config_entry: dict[str, list[dr.DeviceEntry]],
|
|
) -> Iterator[DiscoveryCandidate]:
|
|
"""Yield normalized config-entry discovery candidates."""
|
|
for config_entry in self.get_config_entries(devices_by_config_entry):
|
|
yield DiscoveryCandidate(
|
|
source_entity=self.create_config_entry_source(
|
|
config_entry,
|
|
devices_by_config_entry[config_entry.entry_id][0],
|
|
),
|
|
discovery_type=DiscoveryBy.CONFIG_ENTRY,
|
|
integration_domains=frozenset({config_entry.domain}),
|
|
)
|
|
|
|
def _get_integration_domains(self, config_entry_ids: Iterable[str]) -> frozenset[str]:
|
|
"""Resolve config entry IDs to integration domains."""
|
|
return frozenset(
|
|
entry.domain
|
|
for config_entry_id in config_entry_ids
|
|
if (entry := self.hass.config_entries.async_get_entry(config_entry_id)) is not None
|
|
)
|
|
|
|
@staticmethod
|
|
def create_device_source(device_entry: dr.DeviceEntry) -> SourceEntity:
|
|
"""Create SourceEntity for a device."""
|
|
return SourceEntity(
|
|
object_id=device_entry.name_by_user or device_entry.name or "",
|
|
name=device_entry.name,
|
|
entity_id=DUMMY_ENTITY_ID,
|
|
domain=SENSOR_DOMAIN,
|
|
device_entry=device_entry,
|
|
)
|
|
|
|
@staticmethod
|
|
def create_config_entry_source(config_entry: ConfigEntry, device_entry: dr.DeviceEntry) -> SourceEntity:
|
|
"""Create a source representing all devices belonging to a config entry."""
|
|
return SourceEntity(
|
|
object_id=config_entry.entry_id,
|
|
name=config_entry.title,
|
|
entity_id=DUMMY_ENTITY_ID,
|
|
domain=SENSOR_DOMAIN,
|
|
device_entry=device_entry,
|
|
config_entry_id=config_entry.entry_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def create_unique_id(candidate: DiscoveryCandidate, power_profile: PowerProfile | None) -> str:
|
|
"""Generate a unique ID for a discovery candidate."""
|
|
source = candidate.source_entity
|
|
|
|
if candidate.discovery_type == DiscoveryBy.CONFIG_ENTRY:
|
|
config_entry_id = source.config_entry_id or source.object_id
|
|
return device_discovery_key(f"config_entry_{config_entry_id}")
|
|
|
|
if candidate.discovery_type == DiscoveryBy.DEVICE:
|
|
return device_discovery_key(source.device_id or source.object_id)
|
|
|
|
return get_or_create_unique_id({}, source, power_profile)
|
|
|
|
async def find_power_profiles(
|
|
self,
|
|
model_info: ModelInfo,
|
|
source_entity: SourceEntity,
|
|
discovery_type: DiscoveryBy,
|
|
) -> list[PowerProfile]:
|
|
"""Find power profiles for a given entity."""
|
|
library = await self._get_library()
|
|
models = await library.find_models(model_info)
|
|
log_identifier = source_entity.log_identifier
|
|
if not models:
|
|
_LOGGER.debug(
|
|
"%s: No library models found (manufacturer=%s, model=%s, model_id=%s)",
|
|
log_identifier,
|
|
model_info.manufacturer,
|
|
model_info.model,
|
|
model_info.model_id,
|
|
)
|
|
return []
|
|
|
|
_LOGGER.debug(
|
|
"%s: Found %d matching library model(s): [%s]",
|
|
log_identifier,
|
|
len(models),
|
|
",".join(f"{model.manufacturer}/{model.model}" for model in models),
|
|
)
|
|
|
|
power_profiles = []
|
|
for found_model in models:
|
|
model_identifier = f"{found_model.manufacturer}/{found_model.model}"
|
|
metadata = await library.get_model_metadata(found_model)
|
|
if metadata and (
|
|
reason := self._check_discovery_constraints(
|
|
metadata.device_type,
|
|
metadata.discovery_by,
|
|
source_entity,
|
|
discovery_type,
|
|
)
|
|
):
|
|
_LOGGER.debug("%s: Skipping model %s, %s", log_identifier, model_identifier, reason)
|
|
continue
|
|
|
|
profile = await get_power_profile(
|
|
self.hass,
|
|
{},
|
|
source_entity,
|
|
model_info=found_model,
|
|
process_variables=False,
|
|
model_resolved=True,
|
|
)
|
|
if not profile:
|
|
_LOGGER.debug("%s: Could not load profile %s, skipping model", log_identifier, model_identifier)
|
|
continue
|
|
if reason := self._check_profile(profile, source_entity, discovery_type):
|
|
_LOGGER.debug("%s: Skipping profile %s, %s", log_identifier, describe_power_profile(profile), reason)
|
|
continue
|
|
power_profiles.append(profile)
|
|
|
|
return power_profiles
|
|
|
|
def _check_profile(
|
|
self,
|
|
profile: PowerProfile,
|
|
source_entity: SourceEntity,
|
|
discovery_type: DiscoveryBy,
|
|
) -> str | None:
|
|
"""Return the reason a loaded profile is not eligible for a discovery candidate, None when it is."""
|
|
reason = self._check_discovery_constraints(
|
|
profile.device_type,
|
|
profile.discovery_by,
|
|
source_entity,
|
|
discovery_type,
|
|
)
|
|
if reason:
|
|
return reason
|
|
if self._exclude_self_usage_profiles and profile.only_self_usage:
|
|
return "self usage profiles are excluded by configuration"
|
|
|
|
entity_entry = source_entity.entity_entry
|
|
if (
|
|
discovery_type == DiscoveryBy.ENTITY
|
|
and entity_entry
|
|
and profile.compatible_integrations
|
|
and entity_entry.platform not in profile.compatible_integrations
|
|
):
|
|
return (
|
|
f"integration {entity_entry.platform} is not compatible with the profile "
|
|
f"(compatible_integrations=[{','.join(profile.compatible_integrations or [])}])"
|
|
)
|
|
return None
|
|
|
|
def _check_discovery_constraints(
|
|
self,
|
|
device_type: DeviceType | None,
|
|
profile_discovery_type: DiscoveryBy,
|
|
source_entity: SourceEntity,
|
|
requested_discovery_type: DiscoveryBy,
|
|
) -> str | None:
|
|
"""Return the reason a candidate violates the constraints shared by indexed metadata and loaded profiles.
|
|
|
|
Called with indexed metadata before building a profile, to skip models which can never match.
|
|
The index is only authoritative for the properties checked here. Everything it does not carry,
|
|
such as `only_self_usage` and `compatible_integrations`, is still checked on the profile
|
|
itself, so the metadata pass can only save work, never change the outcome.
|
|
"""
|
|
if profile_discovery_type != requested_discovery_type:
|
|
return (
|
|
f"profile is discovered by {profile_discovery_type}, "
|
|
f"candidate is discovered by {requested_discovery_type}"
|
|
)
|
|
if device_type in self._exclude_device_types:
|
|
return f"device type {device_type} is excluded by configuration"
|
|
if (
|
|
requested_discovery_type == DiscoveryBy.ENTITY
|
|
and source_entity.entity_entry is not None
|
|
and not is_device_type_supported_for_entity(device_type, source_entity.entity_entry)
|
|
):
|
|
return f"device type {device_type} is not supported for entity domain {source_entity.entity_entry.domain}"
|
|
return None
|
|
|
|
@staticmethod
|
|
def is_wled_light(model_info: ModelInfo, entity_entry: er.RegistryEntry) -> bool:
|
|
"""Check if the entity is a WLED light."""
|
|
return (
|
|
model_info.manufacturer == MANUFACTURER_WLED
|
|
and entity_entry.domain == LIGHT_DOMAIN
|
|
and not re.search("master|segment", str(entity_entry.original_name), flags=re.IGNORECASE)
|
|
and not re.search("master|segment", str(entity_entry.entity_id), flags=re.IGNORECASE)
|
|
)
|
|
|
|
def get_entities(self) -> list[er.RegistryEntry]:
|
|
"""Get all entities from entity registry which qualifies for discovery."""
|
|
|
|
def _check_already_configured(entity: er.RegistryEntry) -> bool:
|
|
has_user_config = self._is_user_configured(entity.entity_id)
|
|
if has_user_config:
|
|
_LOGGER.debug(
|
|
"%s: Entity is manually configured, skipping auto configuration",
|
|
entity.entity_id,
|
|
)
|
|
return has_user_config
|
|
|
|
entity_filter = CompositeFilter(
|
|
[
|
|
CategoryFilter(
|
|
[
|
|
EntityCategory.CONFIG,
|
|
EntityCategory.DIAGNOSTIC,
|
|
],
|
|
),
|
|
LambdaFilter(_check_already_configured),
|
|
LambdaFilter(lambda entity: entity.device_id is None),
|
|
LambdaFilter(lambda entity: entity.platform == "mqtt" and "segment" in entity.entity_id),
|
|
LambdaFilter(lambda entity: entity.platform in ["powercalc", "switch_as_x"]),
|
|
NotFilter(DomainFilter(SUPPORTED_DOMAINS)),
|
|
],
|
|
FilterOperator.OR,
|
|
)
|
|
return get_filtered_entity_list(self.hass, NotFilter(entity_filter))
|
|
|
|
def get_devices(self) -> list[dr.DeviceEntry]:
|
|
"""Fetch device entries."""
|
|
return get_non_composite_devices(self.hass)
|
|
|
|
@staticmethod
|
|
def _index_devices_by_config_entry(
|
|
devices: Iterable[dr.DeviceEntry],
|
|
) -> dict[str, list[dr.DeviceEntry]]:
|
|
"""Group devices by their owning config entry without rescanning the registry."""
|
|
devices_by_config_entry: defaultdict[str, list[dr.DeviceEntry]] = defaultdict(list)
|
|
for device in devices:
|
|
for config_entry_id in get_config_entry_ids(device):
|
|
devices_by_config_entry[config_entry_id].append(device)
|
|
return dict(devices_by_config_entry)
|
|
|
|
def get_config_entries(
|
|
self,
|
|
devices_by_config_entry: dict[str, list[dr.DeviceEntry]] | None = None,
|
|
) -> list[ConfigEntry]:
|
|
"""Fetch config entries which have at least one non-composite device."""
|
|
if devices_by_config_entry is None:
|
|
devices_by_config_entry = self._index_devices_by_config_entry(self.get_devices())
|
|
return [
|
|
entry
|
|
for entry in self.hass.config_entries.async_entries()
|
|
if entry.domain != DOMAIN and entry.entry_id in devices_by_config_entry
|
|
]
|
|
|
|
def enable(self) -> None:
|
|
"""Enable the discovery."""
|
|
self._status = DiscoveryStatus.NOT_STARTED
|
|
|
|
async def disable(self) -> None:
|
|
"""Disable the discovery.
|
|
|
|
The library update interval is deliberately kept running: manually configured sensors
|
|
still need up to date profiles. `start_discovery` returns early while disabled.
|
|
"""
|
|
if self._cancel_initial_discovery:
|
|
self._cancel_initial_discovery()
|
|
self._cancel_initial_discovery = None
|
|
self._status = DiscoveryStatus.DISABLED
|
|
self._configured_discovery_keys.clear()
|
|
self._pending_discovery_keys.clear()
|
|
flows = self.hass.config_entries.flow.async_progress_by_handler(DOMAIN)
|
|
for flow in flows:
|
|
if flow["context"]["source"] != SOURCE_INTEGRATION_DISCOVERY:
|
|
continue # pragma: no cover
|
|
self.hass.config_entries.flow.async_abort(flow["flow_id"])
|
|
|
|
def extract_model_info(self, source_entity: SourceEntity) -> ModelInfo | None:
|
|
"""Try to fetch manufacturer and model from the known device information."""
|
|
device_entry = source_entity.device_entry
|
|
if not device_entry:
|
|
return None
|
|
|
|
log_identifier = source_entity.log_identifier
|
|
device_id = device_entry.id
|
|
model_info = self.get_model_information_from_device(device_entry)
|
|
|
|
if not model_info:
|
|
_LOGGER.debug(
|
|
"%s: Cannot autodiscover model, manufacturer or model unknown from device registry (device_id=%s)",
|
|
log_identifier,
|
|
device_id,
|
|
)
|
|
return None
|
|
|
|
# Make sure we don't have a literal / in model_id,
|
|
# so we don't get issues with sublut directory matching down the road
|
|
# See github #658
|
|
if "/" in model_info.model:
|
|
model_info = ModelInfo(
|
|
model_info.manufacturer,
|
|
model_info.model.replace("/", "#slash#"),
|
|
model_info.model_id,
|
|
)
|
|
|
|
_LOGGER.debug(
|
|
"%s: Found model information on device (manufacturer=%s, model=%s, model_id=%s, device_id=%s)",
|
|
log_identifier,
|
|
model_info.manufacturer,
|
|
model_info.model,
|
|
model_info.model_id,
|
|
device_id,
|
|
)
|
|
return model_info
|
|
|
|
@staticmethod
|
|
def get_model_information_from_device(device_entry: dr.DeviceEntry) -> ModelInfo | None:
|
|
"""See if we have enough information in device registry to automatically set up the power sensor."""
|
|
if device_entry.manufacturer is None or device_entry.model is None:
|
|
return None
|
|
|
|
# Strip whitespace: some integrations include trailing spaces
|
|
# see https://github.com/home-assistant/core/pull/166187
|
|
manufacturer = str(device_entry.manufacturer).strip()
|
|
model = str(device_entry.model).strip()
|
|
model_id = (
|
|
str(device_entry.model_id).strip() if hasattr(device_entry, "model_id") and device_entry.model_id else None
|
|
)
|
|
|
|
if len(manufacturer) == 0 or len(model) == 0:
|
|
return None
|
|
|
|
return ModelInfo(manufacturer, model, model_id)
|
|
|
|
async def _init_entity_discovery(
|
|
self,
|
|
candidate: DiscoveryCandidate,
|
|
model_info: ModelInfo,
|
|
unique_id: str,
|
|
power_profiles: list[PowerProfile] | None,
|
|
extra_discovery_data: dict[str, Any] | None,
|
|
) -> None:
|
|
"""Dispatch the discovery flow for a given entity."""
|
|
source_entity = candidate.source_entity
|
|
integration_name = await self._get_integration_name(source_entity)
|
|
|
|
discovery_data: dict[str, Any] = {
|
|
CONF_ENTITY_ID: source_entity.entity_id,
|
|
DISCOVERY_INTEGRATION_NAME: integration_name,
|
|
DISCOVERY_SOURCE_ENTITY: source_entity,
|
|
CONF_UNIQUE_ID: unique_id,
|
|
}
|
|
|
|
if power_profiles:
|
|
discovery_data[DISCOVERY_POWER_PROFILES] = power_profiles
|
|
if len(power_profiles) == 1:
|
|
power_profile = power_profiles[0]
|
|
discovery_data[CONF_MANUFACTURER] = power_profile.manufacturer
|
|
discovery_data[CONF_MODEL] = power_profile.model
|
|
|
|
discovery_data.setdefault(CONF_MANUFACTURER, model_info.manufacturer)
|
|
discovery_data.setdefault(CONF_MODEL, model_info.model or model_info.model_id)
|
|
|
|
if extra_discovery_data:
|
|
discovery_data.update(extra_discovery_data)
|
|
|
|
self._pending_discovery_keys.add(unique_id)
|
|
if not source_entity.is_dummy:
|
|
self._pending_discovery_keys.add(source_entity.entity_id)
|
|
|
|
_LOGGER.debug(
|
|
"%s: Initiating discovery flow, discovery_by=%s, unique_id=%s, integration=%s, integration_domains=[%s], "
|
|
"device_id=%s, source_domain=%s, manufacturer=%s, model=%s, model_id=%s, profiles=[%s], extra_data=%s",
|
|
candidate.log_identifier,
|
|
candidate.discovery_type,
|
|
unique_id,
|
|
integration_name,
|
|
",".join(sorted(candidate.integration_domains)),
|
|
source_entity.device_id,
|
|
source_entity.domain,
|
|
discovery_data[CONF_MANUFACTURER],
|
|
discovery_data[CONF_MODEL],
|
|
model_info.model_id,
|
|
", ".join(describe_power_profile(profile) for profile in power_profiles or []),
|
|
extra_discovery_data,
|
|
)
|
|
|
|
discovery_flow.async_create_flow(
|
|
self.hass,
|
|
DOMAIN,
|
|
context={"source": SOURCE_INTEGRATION_DISCOVERY},
|
|
data=discovery_data,
|
|
)
|
|
|
|
async def _get_integration_name(self, source_entity: SourceEntity) -> str | None:
|
|
"""Return the display name of the integration which owns the discovery source."""
|
|
config_entry_id = source_entity.config_entry_id
|
|
if config_entry_id is None and source_entity.entity_entry:
|
|
config_entry_id = source_entity.entity_entry.config_entry_id
|
|
if config_entry_id is None and source_entity.device_entry:
|
|
config_entry_id = next(iter(get_config_entry_ids(source_entity.device_entry)), None)
|
|
if config_entry_id is None:
|
|
return None
|
|
|
|
config_entry = self.hass.config_entries.async_get_entry(config_entry_id)
|
|
if config_entry is None:
|
|
return None
|
|
|
|
try:
|
|
integration = await async_get_integration(self.hass, config_entry.domain)
|
|
except IntegrationNotFound:
|
|
_LOGGER.debug("Unable to resolve integration name for domain %s", config_entry.domain)
|
|
return None
|
|
return integration.name
|
|
|
|
@property
|
|
def status(self) -> DiscoveryStatus:
|
|
"""Get the discovery status"""
|
|
return self._status
|
|
|
|
def _is_user_configured(self, entity_id: str) -> bool:
|
|
"""Check if user have setup powercalc sensors for a given entity_id.
|
|
Either with the YAML or GUI method.
|
|
"""
|
|
# Explicit None check: when nothing is configured manually the result is an empty set,
|
|
# which would otherwise be reloaded for every single entity.
|
|
if self._manually_configured_entities is None:
|
|
self._manually_configured_entities = self._load_manually_configured_entities()
|
|
|
|
return entity_id in self._manually_configured_entities
|
|
|
|
def _load_manually_configured_entities(self) -> set[str]:
|
|
"""Looks at the YAML and GUI config entries for all the configured entity_id's."""
|
|
entities: set[str] = set()
|
|
|
|
# Find entity ids in yaml config (Legacy)
|
|
if SENSOR_DOMAIN in self.ha_config: # pragma: no cover
|
|
sensor_config = self.ha_config.get(SENSOR_DOMAIN)
|
|
platform_entries = [
|
|
item for item in sensor_config or {} if isinstance(item, dict) and item.get(CONF_PLATFORM) == DOMAIN
|
|
]
|
|
for entry in platform_entries:
|
|
entities.update(self._iter_entity_ids(entry))
|
|
|
|
# Find entity ids in yaml config (New)
|
|
domain_config: ConfigType = self.ha_config.get(DOMAIN, {})
|
|
if CONF_SENSORS in domain_config:
|
|
sensors = domain_config[CONF_SENSORS]
|
|
for sensor_config in sensors:
|
|
entities.update(self._iter_entity_ids(sensor_config))
|
|
|
|
# Add entities from existing config entries
|
|
entities.update(
|
|
entity_id
|
|
for entry in self.hass.config_entries.async_entries(DOMAIN)
|
|
if entry.source == SOURCE_USER
|
|
if isinstance((entity_id := entry.data.get(CONF_ENTITY_ID)), str)
|
|
)
|
|
return entities
|
|
|
|
def _iter_entity_ids(self, value: object) -> Iterator[str]:
|
|
"""Yield entity IDs from nested Powercalc YAML configuration."""
|
|
if isinstance(value, dict):
|
|
for key, nested_value in value.items():
|
|
if key == CONF_ENTITY_ID:
|
|
yield from self._iter_configured_entity_ids(nested_value)
|
|
else:
|
|
yield from self._iter_entity_ids(nested_value)
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
yield from self._iter_entity_ids(item)
|
|
|
|
@staticmethod
|
|
def _iter_configured_entity_ids(value: object) -> Iterator[str]:
|
|
"""Yield the entity IDs of a single `entity_id` configuration option."""
|
|
if isinstance(value, str):
|
|
yield value
|
|
elif isinstance(value, list):
|
|
yield from (item for item in value if isinstance(item, str))
|
|
|
|
def _is_already_discovered(self, candidate: DiscoveryCandidate, unique_id: str) -> bool:
|
|
"""Prevent duplicate discovery flows."""
|
|
source_entity = candidate.source_entity
|
|
unique_ids_to_check = {
|
|
key for key in (unique_id, source_entity.entity_id, source_entity.unique_id) if key is not None
|
|
}
|
|
if unique_id.startswith(DISCOVERY_KEY_PREFIX):
|
|
unique_ids_to_check.add(unique_id.removeprefix(DISCOVERY_KEY_PREFIX))
|
|
|
|
if candidate.discovery_type == DiscoveryBy.DEVICE and source_entity.device_id:
|
|
unique_ids_to_check.update(get_related_device_ids(self.hass, source_entity.device_id))
|
|
|
|
unique_ids_to_check.update({device_discovery_key(uid) for uid in unique_ids_to_check})
|
|
|
|
return not (
|
|
unique_ids_to_check.isdisjoint(self._configured_discovery_keys)
|
|
and unique_ids_to_check.isdisjoint(self._pending_discovery_keys)
|
|
)
|
|
|
|
async def _get_library(self) -> ProfileLibrary:
|
|
"""Get the powercalc library instance."""
|
|
if not self._library:
|
|
self._library = await ProfileLibrary.factory(self.hass)
|
|
return self._library
|