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, model_id_for_actuator, ) from app.actuators.store import ActuatorStore from app.config import Settings from app.ha.discovery import DiscoveredEntity, EntityRole 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", "light", "power", "sensor", "state", "switch", "temperature", "value", } ) _NUMERIC_AUTO_ACCEPT_SCORE = 0.82 _NUMERIC_MIN_MARGIN = 0.18 _CONTEXT_AUTO_ACCEPT_SCORE = 0.78 _MAX_CONTEXT_SELECTIONS = 3 _AUDIT_LIMIT = 20 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 set_override( self, actuator_entity_id: str, override: ManualOverride | None, ) -> ActuatorRecord: record = self._store.get(actuator_entity_id) updated = record.model_copy( update={ "manual_override": override, "updated_at": datetime.now(timezone.utc), } ) self._store.upsert(updated) return self.reconcile_actuator(actuator_entity_id, trigger="override") 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 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 Prüfbedarf." ), ) 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, ) assignment = self._select_assignment( actuator=actuator, numeric_candidates=numeric_candidates, context_candidates=context_candidates, override=record.manual_override, ) lifecycle = self._reconcile_lifecycle( actuator=actuator, assignment=assignment, lifecycle=lifecycle, now=now, ) updated = record.model_copy( update={ "assignment": assignment, "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 def _select_assignment( self, *, actuator: HaEntitySummary, numeric_candidates: list[AssignmentCandidate], context_candidates: list[AssignmentCandidate], override: ManualOverride | None, ) -> AssignmentSelection: if override is not None: selected_numeric = override.numeric_entity_id selected_contexts = list(dict.fromkeys(override.context_entity_ids)) return AssignmentSelection( selected_numeric_entity_id=selected_numeric, selected_context_entity_ids=selected_contexts, source=AssignmentSource.MANUAL, confidence=1.0 if selected_numeric else 0.6, review_required=False, reason=( "Manuelle Zuordnung überschreibt die automatische Heuristik dauerhaft." ), ) top_numeric = numeric_candidates[0] if numeric_candidates else None top_contexts = [ candidate.entity_id for candidate in context_candidates if candidate.auto_accepted ][: _MAX_CONTEXT_SELECTIONS] if top_numeric is None: return AssignmentSelection( selected_numeric_entity_id=None, selected_context_entity_ids=top_contexts, source=AssignmentSource.NONE, confidence=0.0, review_required=True, reason=f"Kein numerischer Sensor konnte für {display_name(actuator)} bestimmt werden.", ) 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=( "Automatisch akzeptiert." if top_numeric.auto_accepted else "Top-Kandidat gefunden, aber Zuordnung ist noch nicht eindeutig genug." ), ) 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, ) if assignment.review_required and assignment.source is not AssignmentSource.MANUAL: return self._archive_state( lifecycle, "Zuordnung ist nicht eindeutig; Modell wartet auf Review.", now=now, status=LifecycleStatus.REVIEW_REQUIRED, ) 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": "Mehr Historie sammeln und Reconciliation erneut ausführen.", "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": "Automatisch überwachen und bei neuen Daten neu trainieren.", } ), 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 bestätigten Sensorzuordnung.", "next_action": "Auf neue Historie oder Staleness warten.", } ), 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": "Review oder neue Zuordnung erforderlich.", } ), 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 auto_accepted = 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 _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) 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: 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 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.") return round(min(score, 1.0), 4), evidence def _preferred_device_classes(domain: str, *, context: bool) -> frozenset[str]: if context: return frozenset({"door", "garage_door", "motion", "occupancy", "opening", "presence"}) mapping = { "climate": {"temperature", "humidity", "power"}, "cover": {"illuminance", "temperature", "wind_speed"}, "fan": {"temperature", "humidity", "power"}, "humidifier": {"humidity", "temperature", "power"}, "light": {"illuminance", "power", "energy"}, "switch": {"power", "energy", "current"}, "valve": {"temperature", "pressure", "humidity"}, } return frozenset(mapping.get(domain, {"power", "energy", "temperature"})) def _metadata_tokens(entity: HaEntitySummary) -> 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(value.lower().replace("_", " ")): if len(token) < 3 or token in _STOPWORDS: continue tokens.add(token) return tokens 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."