178 lines
4.4 KiB
Python
178 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from enum import StrEnum
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.ha.models import HaEntitySummary
|
|
|
|
|
|
class EntityRole(StrEnum):
|
|
MEASUREMENT = "measurement"
|
|
BINARY_CONTEXT = "binary_context"
|
|
CONTEXT = "context"
|
|
ACTUATOR = "actuator"
|
|
UNSUPPORTED = "unsupported"
|
|
|
|
|
|
class DiscoveredEntity(BaseModel):
|
|
entity_id: str
|
|
domain: str
|
|
device_class: str | None = None
|
|
state_class: str | None = None
|
|
unit_of_measurement: str | None = None
|
|
role: EntityRole
|
|
learnable: bool
|
|
reason: str
|
|
|
|
|
|
_MEASUREMENT_CLASSES = frozenset({
|
|
"apparent_power",
|
|
"atmospheric_pressure",
|
|
"battery",
|
|
"carbon_dioxide",
|
|
"carbon_monoxide",
|
|
"current",
|
|
"distance",
|
|
"duration",
|
|
"energy",
|
|
"frequency",
|
|
"gas",
|
|
"humidity",
|
|
"illuminance",
|
|
"moisture",
|
|
"monetary",
|
|
"nitrogen_dioxide",
|
|
"nitrogen_monoxide",
|
|
"nitrous_oxide",
|
|
"ozone",
|
|
"pm1",
|
|
"pm10",
|
|
"pm25",
|
|
"power",
|
|
"precipitation",
|
|
"pressure",
|
|
"reactive_power",
|
|
"signal_strength",
|
|
"sound_pressure",
|
|
"speed",
|
|
"sulphur_dioxide",
|
|
"temperature",
|
|
"volatile_organic_compounds",
|
|
"voltage",
|
|
"volume",
|
|
"volume_flow_rate",
|
|
"water",
|
|
"weight",
|
|
"wind_speed",
|
|
})
|
|
_BINARY_CONTEXT_CLASSES = frozenset({
|
|
"door",
|
|
"garage_door",
|
|
"lock",
|
|
"motion",
|
|
"occupancy",
|
|
"opening",
|
|
"presence",
|
|
"problem",
|
|
"safety",
|
|
"smoke",
|
|
"sound",
|
|
"vibration",
|
|
"window",
|
|
})
|
|
_ACTUATOR_DOMAINS = frozenset({
|
|
"cover",
|
|
"fan",
|
|
"humidifier",
|
|
"light",
|
|
"switch",
|
|
})
|
|
_CONTEXT_DOMAINS = frozenset({"device_tracker", "person", "sun", "weather", "zone"})
|
|
_LEARNABLE_CONTEXT_DOMAINS = frozenset({"device_tracker", "person", "weather"})
|
|
_NUMERIC_STATE_CLASSES = frozenset({"measurement", "total", "total_increasing"})
|
|
|
|
|
|
def classify_entity(entity: HaEntitySummary) -> DiscoveredEntity:
|
|
if entity.domain == "sensor" and (
|
|
entity.state_class in _NUMERIC_STATE_CLASSES
|
|
or entity.device_class in _MEASUREMENT_CLASSES
|
|
or entity.unit_of_measurement is not None
|
|
):
|
|
return _result(
|
|
entity,
|
|
EntityRole.MEASUREMENT,
|
|
learnable=True,
|
|
reason="Numerischer Messsensor für Zeitreihen und Training.",
|
|
)
|
|
|
|
if entity.domain == "binary_sensor" and entity.device_class in _BINARY_CONTEXT_CLASSES:
|
|
return _result(
|
|
entity,
|
|
EntityRole.BINARY_CONTEXT,
|
|
learnable=True,
|
|
reason="Binärer Kontextsensor für Zustands- und Anwesenheitsmuster.",
|
|
)
|
|
|
|
if entity.domain in _CONTEXT_DOMAINS:
|
|
learnable = entity.domain in _LEARNABLE_CONTEXT_DOMAINS
|
|
return _result(
|
|
entity,
|
|
EntityRole.CONTEXT,
|
|
learnable=learnable,
|
|
reason=(
|
|
"Kontextquelle für Training und Erklärungen."
|
|
if learnable
|
|
else "Kontextquelle ohne direkte Trainingsfreigabe."
|
|
),
|
|
)
|
|
|
|
if entity.domain in _ACTUATOR_DOMAINS:
|
|
return _result(
|
|
entity,
|
|
EntityRole.ACTUATOR,
|
|
learnable=False,
|
|
reason="Aktor ist ein mögliches Automationsziel, aber kein Trainingssensor.",
|
|
)
|
|
|
|
return _result(
|
|
entity,
|
|
EntityRole.UNSUPPORTED,
|
|
learnable=False,
|
|
reason="Entity-Typ ist noch nicht für Lernen oder Automationen klassifiziert.",
|
|
)
|
|
|
|
|
|
def discover_entities(
|
|
entities: list[HaEntitySummary],
|
|
domains: set[str] | None = None,
|
|
learnable: bool | None = None,
|
|
) -> list[DiscoveredEntity]:
|
|
normalized_domains = {domain.strip().lower() for domain in domains or set() if domain.strip()}
|
|
discovered = [classify_entity(entity) for entity in entities]
|
|
return [
|
|
entity
|
|
for entity in discovered
|
|
if (not normalized_domains or entity.domain in normalized_domains)
|
|
and (learnable is None or entity.learnable is learnable)
|
|
]
|
|
|
|
|
|
def _result(
|
|
entity: HaEntitySummary,
|
|
role: EntityRole,
|
|
*,
|
|
learnable: bool,
|
|
reason: str,
|
|
) -> DiscoveredEntity:
|
|
return DiscoveredEntity(
|
|
entity_id=entity.entity_id,
|
|
domain=entity.domain,
|
|
device_class=entity.device_class,
|
|
state_class=entity.state_class,
|
|
unit_of_measurement=entity.unit_of_measurement,
|
|
role=role,
|
|
learnable=learnable,
|
|
reason=reason,
|
|
)
|