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", "licht", "lichtschalter", "monitoring", "power", "sensor", "state", "switch", "temperature", "value", } ) _GENERIC_AREA_NAMES = frozenset({"monitoring", "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", "light", "person", "sensor", "switch", "weather", }) _CONTEXT_SUGGESTION_LIMIT = 120 _OUTDOOR_TOKENS = frozenset({"aussen", "außen", "outdoor", "garten", "terrasse", "balkon"}) _DIAGNOSTIC_TOKENS = frozenset({ "basic", "battery", "connect", "count", "diagnostic", "firmware", "gesehen", "last", "linkquality", "knoten", "knotens", "mqtt", "node", "reason", "restart", "rssi", "signal", "ssid", "status", "uptime", "wifi", "zuletzt", }) 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 self._ha_reader.discover()} 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) or not _has_context_relationship(actuator, entity) ): continue if not selected and score < 0.1: continue 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, 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": _merge_manual_candidates( record.numeric_candidates, entities, [numeric_entity_id] if numeric_entity_id else [], role=EntityRole.MEASUREMENT, ), "context_candidates": _merge_manual_candidates( record.context_candidates, entities, selected_context_ids, role=EntityRole.CONTEXT, ), "lifecycle": lifecycle, "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, ) 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 ) auto_accepted = ( 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 "" 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 {"power", "energy", "current", "voltage"}: return "05_power" if entity.domain in {"light", "switch"}: return "06_states" 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 entity_tokens = _metadata_tokens(entity, include_stopwords=True) return bool( entity_tokens.intersection(_OUTDOOR_TOKENS) and entity.device_class in {"illuminance", "humidity", "temperature"} ) 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 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 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: by_id[entity_id] = existing.model_copy( update={ "auto_accepted": True, "confidence": 1.0, "evidence": [ *existing.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 _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, *, 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(value.lower().replace("_", " ")): if len(token) < 3 or (not include_stopwords and 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."