from __future__ import annotations import hashlib import logging import re from collections.abc import Iterable from datetime import datetime, timedelta, timezone from app.actuators.models import ( ActuatorRecord, AssignmentCandidate, AssignmentSelection, AssignmentSource, LifecycleAuditEntry, LifecycleStatus, ManualOverride, ModelLifecycleState, ReconciliationState, SensorWeightGroup, model_id_for_actuator, ) from app.actuators.store import ActuatorStore from app.config import Settings from app.ha.discovery import DiscoveredEntity, EntityRole, discover_entities from app.ha.history import EntityHistorySeries, NumericHistoryPoint from app.ha.models import HaEntitySummary from app.ha.reader import HaReader from app.ml.feature_store import FeatureVector from app.ml.registry.model_registry import ModelRegistry from app.ml.retraining import retrain_model from app.ml.training import TrainedArtifact logger = logging.getLogger(__name__) _TOKEN_PATTERN = re.compile(r"[a-z0-9]+", re.IGNORECASE) _STOPWORDS = frozenset( { "actuator", "battery", "bin", "binary", "brightness", "current", "door", "energy", "entity", "humidity", "illuminance", "led", "lidl", "light", "licht", "lichtschalter", "monitoring", "power", "sensor", "state", "switch", "temperature", "value", } ) _GENERIC_AREA_NAMES = frozenset({"energie", "monitoring", "power", "strom", "system", "technik"}) _NUMERIC_AUTO_ACCEPT_SCORE = 0.82 _NUMERIC_AUTO_ACCEPT_MIN_SCORE = 0.5 _NUMERIC_MIN_MARGIN = 0.18 _CONTEXT_AUTO_ACCEPT_SCORE = 0.78 _CONTEXT_AUTO_ACCEPT_MIN_SCORE = 0.3 _MAX_CONTEXT_SELECTIONS = 5 _AUDIT_LIMIT = 20 _MANUAL_CONTEXT_DOMAINS = frozenset({ "binary_sensor", "climate", "cover", "device_tracker", "fan", "humidifier", "input_boolean", "input_number", "input_select", "light", "media_player", "person", "remote", "scene", "sensor", "sun", "switch", "weather", }) _CONTEXT_SUGGESTION_LIMIT = 500 _OUTDOOR_TOKENS = frozenset({"aussen", "außen", "outdoor", "garten", "terrasse", "balkon"}) _DIAGNOSTIC_TOKENS = frozenset({ "basic", "battery", "bytes", "connect", "count", "data", "diagnostic", "firmware", "gesehen", "heat", "inbytes", "interface", "last", "linkquality", "knoten", "knotens", "mqtt", "node", "outbytes", "pfsense", "reason", "restart", "rssi", "signal", "ssid", "status", "overheat", "overheating", "overload", "uptime", "vpn", "uberhitzung", "ueberhitzung", "ueberlast", "überhitzung", "überlast", "wifi", "zuletzt", }) _AUTO_CONTEXT_CLASSES = frozenset({ "door", "garage_door", "illuminance", "motion", "occupancy", "opening", "presence", "window", }) _PRESENCE_TOKENS = frozenset({ "besetzt", "occupied", "occupancy", "presence", "prasenz", "praesenz", "motion", "bewegung", "bewegungsmelder", }) _MAILBOX_TOKENS = frozenset({"briefkasten", "mailbox", "post"}) _CABINET_TOKENS = frozenset({"schrank", "cabinet"}) _PV_TOKENS = frozenset({ "pv", "solar", "photovoltaik", "akku", "batterie", "battery", "einspeisung", "wechselrichter", "inverter", "netzbezug", "grid", "verbrauch", }) class ActuatorReconciliationService: def __init__( self, *, ha_reader: HaReader, store: ActuatorStore, registry: ModelRegistry, settings: Settings, ) -> None: self._ha_reader = ha_reader self._store = store self._registry = registry self._settings = settings def list_configured(self) -> list[ActuatorRecord]: return self._store.list() def configure_actuator(self, actuator_entity_id: str, *, enabled: bool = True) -> ActuatorRecord: self._store.configure(actuator_entity_id, enabled=enabled) return self.reconcile_actuator(actuator_entity_id, trigger="configuration") def get_actuator(self, actuator_entity_id: str) -> ActuatorRecord: return self._store.get(actuator_entity_id) def suggest_context_options( self, actuator_entity_id: str, *, limit: int = _CONTEXT_SUGGESTION_LIMIT, ) -> list[HaEntitySummary]: entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()} discovered = {entity.entity_id: entity for entity in discover_entities(list(entities.values()))} actuator = entities.get(actuator_entity_id) if actuator is None: raise KeyError("Aktuator-Konfiguration nicht gefunden.") selected_ids = _selected_context_ids(self._store.get(actuator_entity_id)) ranked: list[tuple[float, str, HaEntitySummary]] = [] for entity in entities.values(): if entity.entity_id == actuator_entity_id or entity.domain not in _MANUAL_CONTEXT_DOMAINS: continue role = _manual_context_role(entity, discovered.get(entity.entity_id)) score, _ = _score_candidate( actuator, entity, role, context=role is not EntityRole.MEASUREMENT, ) selected = entity.entity_id in selected_ids if selected: score = max(score, 1.0) if not selected and _is_diagnostic_context(entity): continue if not selected and not _has_context_relationship(actuator, entity): score = max(score, 0.01) ranked.append((score, _context_sort_group(entity), entity)) ranked.sort( key=lambda item: ( -item[0], item[1], item[2].area_name or "", item[2].friendly_name or item[2].entity_id, item[2].entity_id, ) ) return [entity for _, _, entity in ranked[:limit]] def delete_actuator(self, actuator_entity_id: str) -> None: model_id = model_id_for_actuator(actuator_entity_id) self._registry.archive(model_id) self._store.delete(actuator_entity_id) def set_manual_assignment( self, actuator_entity_id: str, *, numeric_entity_id: str | None, context_entity_ids: list[str], note: str | None = None, ) -> ActuatorRecord: now = datetime.now(timezone.utc) record = self._store.get(actuator_entity_id) entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()} actuator = entities.get(actuator_entity_id) if actuator is None: raise KeyError("Aktuator-Konfiguration nicht gefunden.") selected_context_ids = list(dict.fromkeys(context_entity_ids)) selected_ids = [ entity_id for entity_id in [numeric_entity_id, *selected_context_ids] if entity_id ] missing = [entity_id for entity_id in selected_ids if entity_id not in entities] if missing: raise ValueError(f"Unbekannte Home-Assistant-Entity: {', '.join(missing)}") if actuator_entity_id in selected_ids: raise ValueError("Der Aktor selbst kann nicht als Kontextsensor verwendet werden.") override = ManualOverride( numeric_entity_id=numeric_entity_id, context_entity_ids=selected_context_ids, sensor_weights=record.manual_override.sensor_weights if record.manual_override else {}, sensor_weight_groups=( record.manual_override.sensor_weight_groups if record.manual_override else [] ), updated_at=now, note=note, ) assignment = self._manual_assignment(override) lifecycle = self._reconcile_lifecycle( actuator=actuator, assignment=assignment, lifecycle=record.lifecycle.model_copy(update={"last_reconciled_at": now}), now=now, ) updated = record.model_copy( update={ "assignment": assignment, "manual_override": override, "numeric_candidates": _apply_weight_overrides( _merge_manual_candidates( record.numeric_candidates, entities, [numeric_entity_id] if numeric_entity_id else [], role=EntityRole.MEASUREMENT, ), override, ), "context_candidates": _apply_weight_overrides( _merge_manual_candidates( record.context_candidates, entities, selected_context_ids, role=EntityRole.CONTEXT, ), override, ), "lifecycle": lifecycle, "updated_at": now, } ) return self._store.upsert(updated) def set_weight_overrides( self, actuator_entity_id: str, *, sensor_weights: dict[str, float], sensor_weight_groups: list[SensorWeightGroup], note: str | None = None, ) -> ActuatorRecord: now = datetime.now(timezone.utc) record = self._store.get(actuator_entity_id) selected_ids = { entity_id for entity_id in [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] if entity_id } selected_ids.update(sensor_weights) for group in sensor_weight_groups: selected_ids.update(group.entity_ids) entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()} missing = [entity_id for entity_id in selected_ids if entity_id not in entities] if missing: raise ValueError(f"Unbekannte Home-Assistant-Entity: {', '.join(sorted(missing))}") previous = record.manual_override override = ManualOverride( numeric_entity_id=( previous.numeric_entity_id if previous is not None else record.assignment.selected_numeric_entity_id ), context_entity_ids=( previous.context_entity_ids if previous is not None else record.assignment.selected_context_entity_ids ), sensor_weights={entity_id: round(weight, 4) for entity_id, weight in sensor_weights.items()}, sensor_weight_groups=sensor_weight_groups, updated_at=now, note=note, ) updated = record.model_copy( update={ "manual_override": override, "numeric_candidates": _apply_weight_overrides( record.numeric_candidates, override, ), "context_candidates": _apply_weight_overrides( record.context_candidates, override, ), "updated_at": now, } ) return self._store.upsert(updated) def reconcile_all(self, trigger: str = "manual") -> ReconciliationState: state = self._store.load_reconciliation_state().model_copy( update={ "running": True, "last_started_at": datetime.now(timezone.utc), "last_trigger": trigger, } ) self._store.save_reconciliation_state(state) records = self._store.list() for record in records: self.reconcile_actuator(record.actuator_entity_id, trigger=trigger) self._archive_orphan_models({model_id_for_actuator(record.actuator_entity_id) for record in records}) refreshed = self._store.list() summary = ReconciliationState( last_started_at=state.last_started_at, last_completed_at=datetime.now(timezone.utc), last_trigger=trigger, running=False, configured_actuators=len(refreshed), review_required=sum(1 for record in refreshed if record.assignment.review_required), trained_models=sum( 1 for record in refreshed if record.lifecycle.status is LifecycleStatus.TRAINED ), last_summary=( f"{len(refreshed)} Aktuatoren geprüft, " f"{sum(1 for record in refreshed if record.assignment.review_required)} " "mit niedriger Zuordnungssicherheit." ), ) self._store.save_reconciliation_state(summary) return summary def reconcile_actuator(self, actuator_entity_id: str, trigger: str = "manual") -> ActuatorRecord: now = datetime.now(timezone.utc) record = self._store.get(actuator_entity_id) entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()} discovered = {entity.entity_id: entity for entity in self._ha_reader.discover()} actuator = entities.get(actuator_entity_id) descriptor = discovered.get(actuator_entity_id) lifecycle = record.lifecycle.model_copy(update={"last_reconciled_at": now}) if not record.enabled: lifecycle = self._archive_state( lifecycle, "Aktuator ist deaktiviert; Modell bleibt archiviert.", now=now, ) updated = record.model_copy( update={ "assignment": AssignmentSelection( selected_numeric_entity_id=None, selected_context_entity_ids=[], source=AssignmentSource.NONE, confidence=0.0, review_required=False, reason="Aktuator ist deaktiviert.", ), "numeric_candidates": [], "context_candidates": [], "lifecycle": lifecycle, "updated_at": now, } ) return self._store.upsert(updated) if actuator is None or descriptor is None or descriptor.role is not EntityRole.ACTUATOR: lifecycle = self._archive_state( lifecycle, "Aktuator ist in Home Assistant nicht mehr als Aktor vorhanden.", now=now, status=LifecycleStatus.ORPHANED, ) updated = record.model_copy( update={ "assignment": AssignmentSelection( selected_numeric_entity_id=None, selected_context_entity_ids=[], source=AssignmentSource.NONE, confidence=0.0, review_required=True, reason="Aktuator fehlt oder ist kein unterstützter Aktor mehr.", ), "numeric_candidates": [], "context_candidates": [], "lifecycle": lifecycle, "updated_at": now, } ) return self._store.upsert(updated) numeric_candidates = self._rank_candidates( actuator=actuator, candidates=_filter_candidates(entities, discovered, {EntityRole.MEASUREMENT}), context=False, ) context_candidates = self._rank_candidates( actuator=actuator, candidates=_filter_candidates( entities, discovered, {EntityRole.BINARY_CONTEXT, EntityRole.CONTEXT}, ), context=True, ) if record.manual_override is not None: numeric_candidates = _apply_weight_overrides(numeric_candidates, record.manual_override) context_candidates = _apply_weight_overrides(context_candidates, record.manual_override) assignment = ( self._manual_assignment(record.manual_override) if record.manual_override is not None else self._select_assignment( actuator=actuator, numeric_candidates=numeric_candidates, context_candidates=context_candidates, ) ) lifecycle = self._reconcile_lifecycle( actuator=actuator, assignment=assignment, lifecycle=lifecycle, now=now, ) updated = record.model_copy( update={ "assignment": assignment, "manual_override": record.manual_override, "numeric_candidates": numeric_candidates, "context_candidates": context_candidates, "lifecycle": lifecycle, "updated_at": now, } ) self._store.upsert(updated) logger.info( "Actuator %s reconciled via %s -> %s", actuator_entity_id, trigger, lifecycle.status, ) return updated @staticmethod def _manual_assignment(override: ManualOverride) -> AssignmentSelection: selected_context_ids = list(dict.fromkeys(override.context_entity_ids)) selected_count = len(selected_context_ids) + (1 if override.numeric_entity_id else 0) return AssignmentSelection( selected_numeric_entity_id=override.numeric_entity_id, selected_context_entity_ids=selected_context_ids, source=AssignmentSource.MANUAL, confidence=1.0 if selected_count else 0.0, review_required=selected_count == 0, reason=( f"Manuell festgelegt: {selected_count} Kontext-Entity(s) werden verwendet." if selected_count else "Manuelle Zuordnung enthält noch keine Kontext-Entities." ), ) def _select_assignment( self, *, actuator: HaEntitySummary, numeric_candidates: list[AssignmentCandidate], context_candidates: list[AssignmentCandidate], ) -> AssignmentSelection: top_numeric = next( (candidate for candidate in numeric_candidates if candidate.auto_accepted), None, ) accepted_contexts = [ candidate for candidate in context_candidates if candidate.auto_accepted ][: _MAX_CONTEXT_SELECTIONS] top_contexts = [candidate.entity_id for candidate in accepted_contexts] if top_numeric is None: if accepted_contexts: return AssignmentSelection( selected_numeric_entity_id=None, selected_context_entity_ids=top_contexts, source=AssignmentSource.AUTOMATIC, confidence=max(candidate.confidence for candidate in accepted_contexts), review_required=False, reason=( "Passender Schaltkontext automatisch erkannt. Für diese " "Verhaltensvorhersage ist kein numerischer Sensor erforderlich." ), ) return AssignmentSelection( selected_numeric_entity_id=None, selected_context_entity_ids=top_contexts, source=AssignmentSource.NONE, confidence=0.0, review_required=True, reason=( f"Für {display_name(actuator)} ist noch kein nutzbarer numerischer " "Kontext verfügbar. Die Zuordnung wird automatisch erneut geprüft." ), ) return AssignmentSelection( selected_numeric_entity_id=top_numeric.entity_id, selected_context_entity_ids=top_contexts, source=AssignmentSource.AUTOMATIC, confidence=top_numeric.confidence, review_required=not top_numeric.auto_accepted, reason=( "Kontext automatisch und eindeutig zugeordnet." if top_numeric.auto_accepted else "Besten verfügbaren Kontext automatisch mit niedriger Sicherheit zugeordnet." ), ) def _reconcile_lifecycle( self, *, actuator: HaEntitySummary, assignment: AssignmentSelection, lifecycle: ModelLifecycleState, now: datetime, ) -> ModelLifecycleState: model_id = lifecycle.model_id if assignment.selected_numeric_entity_id is None: return self._archive_state( lifecycle, "Ohne numerische Sensorzuordnung wird kein Modell aktiv gehalten.", now=now, ) sensor_id = assignment.selected_numeric_entity_id series = self._read_history(sensor_id, now) points = series.points if series is not None else [] if len(points) < self._settings.min_training_points: return self._with_audit( lifecycle.model_copy( update={ "status": LifecycleStatus.PENDING_HISTORY, "last_reconciled_at": now, "reason": ( f"{len(points)} von mindestens {self._settings.min_training_points} " f"Messpunkten für {sensor_id} vorhanden." ), "next_action": "Historie wird automatisch weiter gesammelt.", "last_history_point_count": len(points), } ), action="history_wait", reason=( f"Training für {display_name(actuator)} verschoben: zu wenig numerische Historie." ), now=now, ) signature = _history_signature(sensor_id, points) artifact = self._registry.get_optional(model_id) needs_retrain = artifact is None retrain_reason = "Noch kein Modell vorhanden." if artifact is not None: valid, reason = _artifact_valid_for_sensor(artifact, sensor_id) if not valid: self._registry.archive(model_id) needs_retrain = True retrain_reason = reason elif lifecycle.last_history_signature != signature: needs_retrain = True retrain_reason = "Historie hat sich seit dem letzten Training materiell geändert." elif lifecycle.last_trained_at is None or ( now - lifecycle.last_trained_at ) >= timedelta(hours=self._settings.retrain_stale_hours): needs_retrain = True retrain_reason = "Modell gilt als veraltet und wird präventiv neu trainiert." if needs_retrain: vectors = [FeatureVector(sensor_id=sensor_id, values={"value": point.value}) for point in points] result = retrain_model(self._registry, model_id, vectors) return self._with_audit( lifecycle.model_copy( update={ "status": LifecycleStatus.TRAINED, "last_reconciled_at": now, "last_trained_at": now, "last_history_signature": signature, "last_history_point_count": len(points), "reason": retrain_reason, "next_action": "Neue Daten automatisch überwachen und nachtrainieren.", } ), action="retrained" if result.replaced else "trained", reason=f"{retrain_reason} Modell {model_id} aktualisiert.", now=now, ) return self._with_audit( lifecycle.model_copy( update={ "status": LifecycleStatus.TRAINED, "last_reconciled_at": now, "last_history_signature": signature, "last_history_point_count": len(points), "reason": "Modell ist aktuell und passt zur automatischen Kontextzuordnung.", "next_action": "Neue Historie automatisch auswerten.", } ), action="kept", reason=f"Modell {model_id} blieb unverändert.", now=now, ) def _read_history(self, sensor_id: str, now: datetime) -> EntityHistorySeries | None: start = now - timedelta(days=self._settings.history_days) history = list(self._ha_reader.read_history([sensor_id], start, now)) for series in history: if series.entity_id == sensor_id: return series return None def _archive_orphan_models(self, configured_model_ids: set[str]) -> None: for artifact in self._registry.list_models(): if not artifact.artifact_id.startswith("actuator."): continue if artifact.artifact_id not in configured_model_ids: self._registry.archive(artifact.artifact_id) def _archive_state( self, lifecycle: ModelLifecycleState, reason: str, *, now: datetime, status: LifecycleStatus = LifecycleStatus.ARCHIVED, ) -> ModelLifecycleState: self._registry.archive(lifecycle.model_id) return self._with_audit( lifecycle.model_copy( update={ "status": status, "last_reconciled_at": now, "reason": reason, "next_action": "Bei neuen Home-Assistant-Daten automatisch erneut zuordnen.", } ), action="archived", reason=reason, now=now, ) def _rank_candidates( self, *, actuator: HaEntitySummary, candidates: Iterable[tuple[HaEntitySummary, DiscoveredEntity]], context: bool, ) -> list[AssignmentCandidate]: scored: list[AssignmentCandidate] = [] all_scores: list[float] = [] for entity, discovered in candidates: score, evidence = _score_candidate(actuator, entity, discovered.role, context=context) if score <= 0: continue all_scores.append(score) scored.append( AssignmentCandidate( entity_id=entity.entity_id, domain=entity.domain, role=discovered.role, device_class=entity.device_class, state_class=entity.state_class, unit_of_measurement=entity.unit_of_measurement, friendly_name=entity.friendly_name, area_name=entity.area_name, device_name=entity.device_name, score=score, confidence=0.0, evidence=evidence, ) ) if not scored: return [] highest = max(all_scores) sorted_candidates = sorted(scored, key=lambda item: (-item.score, item.entity_id)) second_score = sorted_candidates[1].score if len(sorted_candidates) > 1 else 0.0 for index, candidate in enumerate(sorted_candidates): confidence = candidate.score / highest if highest else 0.0 margin = candidate.score - second_score if index == 0 else 0.0 auto_score = _CONTEXT_AUTO_ACCEPT_SCORE if context else _NUMERIC_AUTO_ACCEPT_SCORE minimum_score = ( _CONTEXT_AUTO_ACCEPT_MIN_SCORE if context else _NUMERIC_AUTO_ACCEPT_MIN_SCORE ) can_auto_accept_context = ( not context or _eligible_for_auto_context(actuator, candidate) ) auto_accepted = ( can_auto_accept_context and candidate.score >= minimum_score and confidence >= auto_score and (context or margin >= _NUMERIC_MIN_MARGIN) ) sorted_candidates[index] = candidate.model_copy( update={ "confidence": round(confidence, 4), "auto_accepted": auto_accepted, } ) return sorted_candidates @staticmethod def _with_audit( lifecycle: ModelLifecycleState, *, action: str, reason: str, now: datetime, ) -> ModelLifecycleState: audit = list(lifecycle.audit) entry = LifecycleAuditEntry(at=now, action=action, reason=reason) if not audit or audit[-1].action != action or audit[-1].reason != reason: audit.append(entry) if len(audit) > _AUDIT_LIMIT: audit = audit[-_AUDIT_LIMIT:] return lifecycle.model_copy(update={"audit": audit}) def display_name(entity: HaEntitySummary) -> str: return entity.friendly_name or entity.device_name or entity.entity_id def _filter_candidates( entities: dict[str, HaEntitySummary], discovered: dict[str, DiscoveredEntity], roles: set[EntityRole], ) -> list[tuple[HaEntitySummary, DiscoveredEntity]]: result: list[tuple[HaEntitySummary, DiscoveredEntity]] = [] for entity_id, summary in entities.items(): candidate = discovered.get(entity_id) if candidate is None or candidate.role not in roles: continue result.append((summary, candidate)) return result def _selected_context_ids(record: ActuatorRecord) -> set[str]: result = set(record.assignment.selected_context_entity_ids) if record.assignment.selected_numeric_entity_id: result.add(record.assignment.selected_numeric_entity_id) if record.manual_override is not None: result.update(record.manual_override.context_entity_ids) if record.manual_override.numeric_entity_id: result.add(record.manual_override.numeric_entity_id) return result def _manual_context_role( entity: HaEntitySummary, discovered: DiscoveredEntity | None, ) -> EntityRole: if discovered is not None and discovered.role is not EntityRole.UNSUPPORTED: return discovered.role if entity.domain == "sensor": return EntityRole.MEASUREMENT if entity.domain == "binary_sensor": return EntityRole.BINARY_CONTEXT return EntityRole.CONTEXT def _context_sort_group(entity: HaEntitySummary) -> str: device_class = entity.device_class or "" text = " ".join( value.lower().replace("_", " ") for value in [entity.entity_id, entity.friendly_name, entity.area_name, entity.device_name] if value ) if device_class in {"motion", "occupancy", "presence"}: return "01_presence" if device_class in {"illuminance"}: return "02_brightness" if device_class in {"door", "garage_door", "opening", "window"}: return "03_opening" if device_class in {"humidity", "moisture"}: return "04_humidity" if device_class in {"temperature"}: return "05_temperature" if any(token in text for token in {"pv", "solar", "akku", "batterie", "battery", "einspeisung"}): return "06_pv_battery" if device_class in {"power", "energy", "current", "voltage"}: return "07_power" if entity.domain in {"weather"}: return "08_weather" if entity.domain in {"fan", "humidifier"}: return "09_ventilation" if entity.domain in {"climate"}: return "10_heating" if entity.domain in {"cover"}: return "11_cover" if entity.domain in {"light", "switch"}: return "12_states" if entity.domain.startswith("input_"): return "13_helper" if entity.domain in {"person", "device_tracker"}: return "14_people" return f"20_{entity.domain}_{device_class}" def _is_diagnostic_context(entity: HaEntitySummary) -> bool: tokens = _metadata_tokens(entity, include_stopwords=True) return bool(tokens.intersection(_DIAGNOSTIC_TOKENS)) def _has_context_relationship(actuator: HaEntitySummary, entity: HaEntitySummary) -> bool: if ( actuator.area_name and entity.area_name and actuator.area_name == entity.area_name and actuator.area_name.lower() not in _GENERIC_AREA_NAMES ): return True if actuator.device_id and entity.device_id and actuator.device_id == entity.device_id: return True if actuator.device_name and entity.device_name and actuator.device_name == entity.device_name: return True if _metadata_tokens(actuator).intersection(_metadata_tokens(entity)): return True actuator_tokens = _metadata_tokens(actuator, include_stopwords=True) entity_tokens = _metadata_tokens(entity, include_stopwords=True) if _is_mailbox_reset_candidate(actuator_tokens, entity_tokens, entity): return True if actuator.domain in {"fan", "humidifier"} and ( _is_presence_context(entity) or entity.device_class in {"humidity", "moisture"} ): return True if actuator.domain in {"climate", "cover", "fan", "humidifier", "light", "switch"} and ( entity_tokens.intersection(_PV_TOKENS) ): return True entity_tokens = _metadata_tokens(entity, include_stopwords=True) return bool( entity_tokens.intersection(_OUTDOOR_TOKENS) and entity.device_class in {"illuminance", "humidity", "temperature"} ) def _eligible_for_auto_context( actuator: HaEntitySummary, candidate: AssignmentCandidate, ) -> bool: device_class = candidate.device_class or "" if device_class in _AUTO_CONTEXT_CLASSES: return True if actuator.domain in {"fan", "humidifier"} and device_class in { "humidity", "moisture", "temperature", }: return True if actuator.domain in {"fan", "humidifier", "light", "switch"} and _is_presence_candidate(candidate): return True actuator_tokens = _metadata_tokens(actuator, include_stopwords=True) candidate_tokens = _candidate_tokens(candidate, include_stopwords=True) if _is_mailbox_reset_candidate(actuator_tokens, candidate_tokens, candidate): return True if ( actuator.device_name and candidate.device_name and actuator.device_name == candidate.device_name and candidate.domain in {"light", "switch"} ): return True return False def _score_candidate( actuator: HaEntitySummary, entity: HaEntitySummary, role: EntityRole, *, context: bool, ) -> tuple[float, list[str]]: evidence: list[str] = [] score = 0.0 actuator_tokens = _metadata_tokens(actuator) entity_tokens = _metadata_tokens(entity) full_entity_tokens = _metadata_tokens(entity, include_stopwords=True) overlap = sorted(actuator_tokens.intersection(entity_tokens)) if overlap: score += min(0.4, 0.1 * len(overlap)) evidence.append(f"Gemeinsame Tokens: {', '.join(overlap[:4])}") if ( actuator.area_name and entity.area_name and actuator.area_name == entity.area_name and actuator.area_name.lower() not in _GENERIC_AREA_NAMES ): score += 0.35 evidence.append(f"Gleicher Bereich: {actuator.area_name}") if actuator.device_id and entity.device_id and actuator.device_id == entity.device_id: score += 0.2 evidence.append("Gleiche Home-Assistant-Geräte-ID") if actuator.device_name and entity.device_name and actuator.device_name == entity.device_name: score += 0.15 evidence.append(f"Gleicher Gerätename: {actuator.device_name}") if actuator.friendly_name and entity.friendly_name and actuator.friendly_name == entity.friendly_name: score += 0.1 evidence.append("Gleicher Friendly Name") preferred_device_classes = _preferred_device_classes(actuator.domain, context=context) if entity.device_class in preferred_device_classes: score += 0.2 evidence.append(f"Passende device_class: {entity.device_class}") if context and actuator.domain in {"fan", "humidifier"} and entity.device_class in { "humidity", "moisture", }: score += 0.3 evidence.append("Luftfeuchtigkeit ist primärer Kontext für Lüftung.") if not context and actuator.domain in {"fan", "humidifier"} and entity.device_class in { "humidity", "moisture", }: score += 0.3 evidence.append("Luftfeuchtigkeit ist primärer Messwert für Lüftung.") if context and actuator.domain in {"fan", "humidifier", "light", "switch"} and _is_presence_context(entity): score += 0.3 evidence.append("Anwesenheit/Belegung ist primärer Schaltkontext.") if context and _is_mailbox_reset_candidate( _metadata_tokens(actuator, include_stopwords=True), _metadata_tokens(entity, include_stopwords=True), entity, ): score += 0.45 evidence.append("Briefkasten-Reset passt zur Schrank-/Entnahme-Tür.") if full_entity_tokens.intersection(_PV_TOKENS): score += 0.12 if context else 0.18 evidence.append("PV-/Akku-/Verbrauchswert ist als Energiemanagement-Kontext relevant.") if not context and actuator.domain == "light" and entity.device_class == "illuminance": score += 0.2 evidence.append("Beleuchtungsstärke wird für Lichtaktoren bevorzugt.") if not context and entity.unit_of_measurement is not None: score += 0.05 evidence.append(f"Numerische Einheit vorhanden: {entity.unit_of_measurement}") if context and role is EntityRole.BINARY_CONTEXT: score += 0.05 evidence.append("Binärer Kontextsensor bevorzugt für Zusatzkontext.") if entity_tokens.intersection(_OUTDOOR_TOKENS) and entity.device_class in { "illuminance", "humidity", "temperature", }: score += 0.1 evidence.append("Außenmesswert ist oft als übergreifender Kontext relevant.") return round(min(score, 1.0), 4), evidence def _merge_manual_candidates( candidates: list[AssignmentCandidate], entities: dict[str, HaEntitySummary], selected_entity_ids: list[str], *, role: EntityRole, ) -> list[AssignmentCandidate]: by_id = {candidate.entity_id: candidate for candidate in candidates} for entity_id in selected_entity_ids: existing = by_id.get(entity_id) if existing is not None: evidence = [ item for item in existing.evidence if item != "Manuell vom Nutzer als relevant festgelegt." ] by_id[entity_id] = existing.model_copy( update={ "auto_accepted": True, "confidence": 1.0, "evidence": [ *evidence, "Manuell vom Nutzer als relevant festgelegt.", ], } ) continue entity = entities.get(entity_id) if entity is None: continue by_id[entity_id] = AssignmentCandidate( entity_id=entity.entity_id, domain=entity.domain, role=role, device_class=entity.device_class, state_class=entity.state_class, unit_of_measurement=entity.unit_of_measurement, friendly_name=entity.friendly_name, area_name=entity.area_name, device_name=entity.device_name, score=1.0, confidence=1.0, auto_accepted=True, evidence=["Manuell vom Nutzer als relevant festgelegt."], ) return sorted(by_id.values(), key=lambda item: (-item.confidence, item.entity_id)) def _apply_weight_overrides( candidates: list[AssignmentCandidate], override: ManualOverride, ) -> list[AssignmentCandidate]: if not override.sensor_weights and not override.sensor_weight_groups: return candidates group_weights: dict[str, float] = {} for group in override.sensor_weight_groups: for entity_id in group.entity_ids: group_weights[entity_id] = max(group_weights.get(entity_id, 0.0), group.weight) weighted: list[AssignmentCandidate] = [] for candidate in candidates: explicit = override.sensor_weights.get(candidate.entity_id) group_weight = group_weights.get(candidate.entity_id) manual_weight = explicit if explicit is not None else group_weight effective_weight = manual_weight if manual_weight is not None else 1.0 evidence = [ item for item in candidate.evidence if not item.startswith("Manuelle Gewichtung:") ] if manual_weight is not None: evidence.append(f"Manuelle Gewichtung: {round(manual_weight * 100)} %.") weighted.append( candidate.model_copy( update={ "manual_weight": manual_weight, "effective_weight": round(effective_weight, 4), "evidence": evidence, } ) ) return sorted(weighted, key=lambda item: (-item.confidence * item.effective_weight, item.entity_id)) def _preferred_device_classes(domain: str, *, context: bool) -> frozenset[str]: if context: mapping = { "climate": {"humidity", "illuminance", "occupancy", "presence", "temperature", "window"}, "cover": {"illuminance", "motion", "occupancy", "presence", "wind_speed"}, "fan": {"humidity", "moisture", "occupancy", "presence", "temperature"}, "humidifier": {"humidity", "moisture", "temperature"}, "light": {"door", "garage_door", "illuminance", "motion", "occupancy", "opening", "presence", "window"}, "media_player": {"occupancy", "presence"}, "switch": {"door", "garage_door", "motion", "occupancy", "opening", "presence", "window"}, } return frozenset( mapping.get( domain, {"door", "garage_door", "motion", "occupancy", "opening", "presence"}, ) ) mapping = { "climate": {"temperature", "humidity"}, "cover": {"illuminance", "temperature", "wind_speed"}, "fan": {"temperature", "humidity", "moisture"}, "humidifier": {"humidity", "moisture", "temperature"}, "light": {"illuminance"}, "media_player": {"power", "energy"}, "remote": {"battery"}, "switch": {"power", "energy", "current"}, "valve": {"temperature", "pressure", "humidity"}, } return frozenset(mapping.get(domain, {"power", "energy", "temperature"})) def _metadata_tokens(entity: HaEntitySummary, *, include_stopwords: bool = False) -> set[str]: raw_values = [ entity.entity_id, entity.friendly_name, entity.area_name, entity.device_name, ] tokens: set[str] = set() for value in raw_values: if value is None: continue for token in _TOKEN_PATTERN.findall(_normalize_text(value)): if (len(token) < 3 and token != "wc") or (not include_stopwords and token in _STOPWORDS): continue tokens.add(token) return _expand_room_tokens(tokens) def _candidate_tokens( candidate: AssignmentCandidate, *, include_stopwords: bool = False, ) -> set[str]: raw_values = [ candidate.entity_id, candidate.friendly_name, candidate.area_name, candidate.device_name, ] tokens: set[str] = set() for value in raw_values: if value is None: continue for token in _TOKEN_PATTERN.findall(_normalize_text(value)): if (len(token) < 3 and token != "wc") or (not include_stopwords and token in _STOPWORDS): continue tokens.add(token) return _expand_room_tokens(tokens) def _expand_room_tokens(tokens: set[str]) -> set[str]: expanded = set(tokens) if "gaste" in expanded: expanded.add("gaeste") if {"gaste", "wc"}.issubset(expanded) or {"gaeste", "wc"}.issubset(expanded): expanded.add("gaestewc") if {"gaeste", "zimmer"}.issubset(expanded): expanded.add("gaestezimmer") return expanded def _normalize_text(value: str) -> str: return ( value.lower() .replace("_", " ") .replace("ä", "ae") .replace("ö", "oe") .replace("ü", "ue") .replace("ß", "ss") ) def _is_presence_context(entity: HaEntitySummary) -> bool: if entity.device_class in {"motion", "occupancy", "presence"}: return True return bool(_metadata_tokens(entity, include_stopwords=True).intersection(_PRESENCE_TOKENS)) def _is_presence_candidate(candidate: AssignmentCandidate) -> bool: if candidate.device_class in {"motion", "occupancy", "presence"}: return True return bool(_candidate_tokens(candidate, include_stopwords=True).intersection(_PRESENCE_TOKENS)) def _is_mailbox_reset_candidate( actuator_tokens: set[str], context_tokens: set[str], entity: HaEntitySummary | AssignmentCandidate, ) -> bool: if not actuator_tokens.intersection(_MAILBOX_TOKENS): return False if not context_tokens.intersection(_CABINET_TOKENS): return False return entity.domain == "binary_sensor" and entity.device_class in { "door", "garage_door", "opening", } def _history_signature(sensor_id: str, points: list[NumericHistoryPoint]) -> str: digest = hashlib.sha256() digest.update(sensor_id.encode("utf-8")) for point in points: digest.update(point.timestamp.isoformat().encode("utf-8")) digest.update(f"{point.value:.6f}".encode("utf-8")) return digest.hexdigest() def _artifact_valid_for_sensor(artifact: TrainedArtifact, sensor_id: str) -> tuple[bool, str]: if sensor_id not in artifact.supported_sensors: return False, "Vorhandenes Modell passt nicht mehr zur aktuellen Sensorzuordnung." feature_models = artifact.feature_models.get(sensor_id, {}) if "value" not in feature_models: return False, "Vorhandenes Modell enthält kein numerisches Trainingsmerkmal 'value'." return True, "Modell ist kompatibel."