from __future__ import annotations import logging from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo from app.actuators.models import ( ActuatorRecord, BehaviorMode, BehaviorPattern, BehaviorPrediction, BehaviorState, BehaviorStatus, ExecutionEvent, ) from app.actuators.store import ActuatorStore from app.config import Settings from app.ha.exceptions import HaClientError from app.ha.history import LogbookEntry, StateHistoryPoint, StateHistorySeries from app.ha.reader import HaReader _MAX_PATTERNS = 500 _MAX_EXECUTION_EVENTS = 100 _ACTION_LOGBOOK_TOLERANCE = timedelta(seconds=10) _CONTEXT_TRIGGER_TOLERANCE = timedelta(seconds=3) _OWN_ACTION_TOLERANCE = timedelta(seconds=20) _SAFE_ACTIVE_DOMAINS = frozenset({"cover", "fan", "humidifier", "light", "switch"}) _AUTOMATION_CONTEXT_DOMAINS = frozenset({"automation", "script"}) logger = logging.getLogger(__name__) class BehaviorEngine: def __init__( self, *, ha_reader: HaReader, store: ActuatorStore, settings: Settings, ) -> None: self._ha_reader = ha_reader self._store = store self._settings = settings def train_all(self) -> list[ActuatorRecord]: results: list[ActuatorRecord] = [] for record in self._store.list(): try: results.append(self.train(record.actuator_entity_id)) except Exception: logger.exception("Behavior training failed for %s", record.actuator_entity_id) results.append(record) return results def train(self, actuator_entity_id: str) -> ActuatorRecord: record = self._store.get(actuator_entity_id) now = datetime.now(timezone.utc) raw_context_ids = list( dict.fromkeys( [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] ) ) context_ids = [ entity_id for entity_id in raw_context_ids if isinstance(entity_id, str) ] if not context_ids: return self._save_behavior( record, record.behavior.model_copy( update={ "status": BehaviorStatus.COLLECTING, "last_trained_at": now, "reason": "Noch kein geeigneter Kontext für Verhaltenslernen vorhanden.", } ), ) start = now - timedelta(days=self._settings.history_days) history_ids = [actuator_entity_id, *context_ids] try: history = { series.entity_id: series for series in self._ha_reader.read_state_history(history_ids, start, now) } except (HaClientError, ValueError) as exc: logger.warning("Behavior history unavailable for %s: %s", actuator_entity_id, exc) return self._save_behavior( record, record.behavior.model_copy( update={ "status": BehaviorStatus.BLOCKED, "last_trained_at": now, "reason": f"Home-Assistant-Historie konnte nicht gelesen werden: {exc}", } ), ) actuator_history = history.get(actuator_entity_id) if actuator_history is None or len(actuator_history.points) < 2: return self._save_behavior( record, record.behavior.model_copy( update={ "status": BehaviorStatus.COLLECTING, "sample_count": 0, "high_confidence_sample_count": 0, "patterns": [], "last_trained_at": now, "reason": "Noch keine historischen Aktorhandlungen gefunden.", } ), ) try: logbook = list(self._ha_reader.read_logbook(actuator_entity_id, start, now)) except (HaClientError, ValueError) as exc: logger.warning("Logbook unavailable for %s: %s", actuator_entity_id, exc) logbook = [] patterns = self._build_patterns( actuator_history=actuator_history, context_history=history, context_ids=context_ids, logbook=logbook, own_executions=record.behavior.execution_events, ) high_confidence = sum(1 for pattern in patterns if pattern.source == "user") status = ( BehaviorStatus.TRAINED if len(patterns) >= self._settings.min_behavior_actions else BehaviorStatus.COLLECTING ) reason = ( f"{len(patterns)} Handlungen mit automatisch erfasstem Kontext gelernt." if status is BehaviorStatus.TRAINED else ( f"{len(patterns)} von mindestens {self._settings.min_behavior_actions} " "benötigten Handlungen gelernt." ) ) behavior = record.behavior.model_copy( update={ "status": status, "sample_count": len(patterns), "high_confidence_sample_count": high_confidence, "patterns": patterns[-_MAX_PATTERNS:], "last_trained_at": now, "reason": reason, } ) return self._save_behavior(record, behavior) def evaluate_all(self) -> list[ActuatorRecord]: results: list[ActuatorRecord] = [] for record in self._store.list(): try: results.append(self.evaluate(record.actuator_entity_id)) except Exception: logger.exception("Behavior evaluation failed for %s", record.actuator_entity_id) results.append(record) return results def evaluate(self, actuator_entity_id: str) -> ActuatorRecord: record = self._store.get(actuator_entity_id) now = datetime.now(timezone.utc) try: entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()} except HaClientError as exc: logger.warning("Current HA state unavailable for %s: %s", actuator_entity_id, exc) return self._save_behavior( record, record.behavior.model_copy( update={ "last_evaluated_at": now, "prediction": None, "reason": f"Aktueller Home-Assistant-Zustand ist nicht verfügbar: {exc}", } ), ) actuator = entities.get(actuator_entity_id) if actuator is None: return self._save_behavior( record, record.behavior.model_copy( update={ "last_evaluated_at": now, "prediction": None, "reason": "Aktor ist aktuell nicht in Home Assistant verfügbar.", } ), ) current_context = { entity_id: entities[entity_id].state for entity_id in ( [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] ) if entity_id and entity_id in entities and entities[entity_id].state is not None } current_context_changed_at = { entity_id: entities[entity_id].last_changed for entity_id in current_context } prediction = predict_behavior( record.behavior.patterns, current_context=current_context, current_context_changed_at=current_context_changed_at, now=now, min_support=self._settings.min_behavior_actions, window_minutes=self._settings.prediction_window_minutes, causal_window_seconds=self._settings.prediction_interval_seconds * 2, timezone_name=self._settings.timezone, ) behavior = record.behavior.model_copy( update={ "last_evaluated_at": now, "prediction": prediction, "reason": ( prediction.reason if prediction is not None else "Aktuell ist kein gelerntes Handlungsmuster fällig." ), } ) if ( prediction is not None and behavior.mode is BehaviorMode.ACTIVE and prediction.confidence >= self._settings.prediction_confidence and actuator.state != prediction.target_state and self._cooldown_elapsed(behavior, now) ): domain = actuator_entity_id.split(".", 1)[0] service = service_for_state(domain, prediction.target_state) if service is not None: try: self._ha_reader.call_service( domain, service, {"entity_id": actuator_entity_id}, ) except (HaClientError, ValueError) as exc: logger.error( "Predicted action failed for %s: %s", actuator_entity_id, exc, ) behavior = behavior.model_copy( update={ "reason": f"Vorhersage wurde aus Sicherheitsgründen nicht ausgeführt: {exc}" } ) return self._save_behavior(record, behavior) event = ExecutionEvent( target_state=prediction.target_state, executed_at=now, ) behavior = behavior.model_copy( update={ "prediction": prediction.model_copy(update={"executed": True}), "last_executed_at": now, "execution_events": [ *behavior.execution_events, event, ][-_MAX_EXECUTION_EVENTS:], "reason": ( f"Vorhersage mit {prediction.confidence:.0%} Sicherheit ausgeführt." ), } ) else: behavior = behavior.model_copy( update={ "reason": ( f"Der vorhergesagte Zustand {prediction.target_state!r} " "ist für autonomes Schalten nicht freigegeben." ) } ) return self._save_behavior(record, behavior) def set_active(self, actuator_entity_id: str, *, active: bool) -> ActuatorRecord: record = self._store.get(actuator_entity_id) now = datetime.now(timezone.utc) if active: domain = actuator_entity_id.split(".", 1)[0] if domain not in _SAFE_ACTIVE_DOMAINS: raise ValueError( f"Automatisches Schalten ist für die Domain {domain} nicht freigegeben." ) if record.behavior.status is not BehaviorStatus.TRAINED: raise ValueError("Das Verhaltensmodell hat noch nicht genügend Handlungen gelernt.") if ( record.behavior.high_confidence_sample_count < self._settings.min_behavior_actions ): raise ValueError( "Für die Freigabe fehlen noch eindeutig dir zugeordnete Handlungen. " "Bediene den Aktor einige Male über Home Assistant." ) mode = BehaviorMode.ACTIVE approved_at = now reason = "Autonomes Lernen und Schalten wurde ausdrücklich freigegeben." else: mode = BehaviorMode.SHADOW approved_at = None reason = "Shadow-Modus aktiv; Vorhersagen werden nicht ausgeführt." behavior = record.behavior.model_copy( update={ "mode": mode, "approved_at": approved_at, "reason": reason, } ) return self._save_behavior(record, behavior) def _build_patterns( self, *, actuator_history: StateHistorySeries, context_history: dict[str, StateHistorySeries], context_ids: list[str], logbook: list[LogbookEntry], own_executions: list[ExecutionEvent], ) -> list[BehaviorPattern]: patterns: list[BehaviorPattern] = [] previous_state = actuator_history.points[0].state for point in actuator_history.points[1:]: if point.state == previous_state: continue previous_state = point.state if _matches_own_execution(point, own_executions): continue source, weight = _action_source(point, logbook) trigger = _recent_context_transition( context_history, context_ids, point.timestamp, ) if source == "automation" and trigger is None: continue contexts = { entity_id: state for entity_id in context_ids if (state := _state_at(context_history.get(entity_id), point.timestamp)) is not None } local = point.timestamp.astimezone(ZoneInfo(self._settings.timezone)) patterns.append( BehaviorPattern( target_state=point.state, minute_of_day=local.hour * 60 + local.minute, weekday=local.weekday(), context_states=contexts, trigger_entity_id=trigger[0] if trigger else None, trigger_from_state=trigger[1] if trigger else None, trigger_to_state=trigger[2] if trigger else None, source=source, weight=weight, observed_at=point.timestamp, ) ) return patterns def _cooldown_elapsed(self, behavior: BehaviorState, now: datetime) -> bool: return behavior.last_executed_at is None or ( now - behavior.last_executed_at ) >= timedelta(seconds=self._settings.execution_cooldown_seconds) def _save_behavior( self, record: ActuatorRecord, behavior: BehaviorState, ) -> ActuatorRecord: updated = record.model_copy( update={ "behavior": behavior, "updated_at": datetime.now(timezone.utc), } ) return self._store.upsert(updated) def predict_behavior( patterns: list[BehaviorPattern], *, current_context: dict[str, str | None], now: datetime, min_support: int, window_minutes: int, current_context_changed_at: dict[str, datetime | None] | None = None, causal_window_seconds: int = 120, timezone_name: str = "Europe/Berlin", ) -> BehaviorPrediction | None: if not patterns: return None local = now.astimezone(ZoneInfo(timezone_name)) minute_of_day = local.hour * 60 + local.minute changed_at = current_context_changed_at or {} by_state: dict[str, list[float]] = {} causal_support_by_state: dict[str, int] = {} for pattern in patterns: if pattern.trigger_entity_id and pattern.trigger_to_state: trigger_changed_at = changed_at.get(pattern.trigger_entity_id) trigger_age = ( (now - trigger_changed_at).total_seconds() if trigger_changed_at is not None else None ) if not ( current_context.get(pattern.trigger_entity_id) == pattern.trigger_to_state and trigger_age is not None and 0 <= trigger_age <= causal_window_seconds ): continue comparable = [ (entity_id, expected) for entity_id, expected in pattern.context_states.items() if entity_id in current_context ] context_score = ( sum( current_context[entity_id] == expected for entity_id, expected in comparable ) / len(comparable) if comparable else 0.5 ) score = pattern.weight * (0.85 + 0.15 * context_score) by_state.setdefault(pattern.target_state, []).append(score) causal_support_by_state[pattern.target_state] = ( causal_support_by_state.get(pattern.target_state, 0) + 1 ) continue distance = _circular_minute_distance(minute_of_day, pattern.minute_of_day) if distance > window_minutes: continue time_score = 1.0 - (distance / max(window_minutes, 1)) weekday_score = ( 1.0 if local.weekday() == pattern.weekday else 0.5 if (local.weekday() >= 5) == (pattern.weekday >= 5) else 0.0 ) comparable = [ (entity_id, expected) for entity_id, expected in pattern.context_states.items() if entity_id in current_context ] context_score = ( sum(current_context[entity_id] == expected for entity_id, expected in comparable) / len(comparable) if comparable else 0.5 ) score = pattern.weight * ( 0.45 * time_score + 0.45 * context_score + 0.10 * weekday_score ) by_state.setdefault(pattern.target_state, []).append(score) if not by_state: return None target_state, scores = max( by_state.items(), key=lambda item: (sum(item[1]), len(item[1]), item[0]), ) support = len(scores) causal_support = causal_support_by_state.get(target_state, 0) confidence = min(1.0, (sum(scores) / support) * min(1.0, support / min_support)) if confidence <= 0: return None return BehaviorPrediction( target_state=target_state, confidence=round(confidence, 4), generated_at=now, matching_patterns=support, reason=( ( f"{causal_support} historische Handlungen folgten demselben " "frischen Sensorwechsel." ) if causal_support else f"{support} ähnliche Handlungsmuster passen zu Zeit und aktuellem Kontext." ), ) def service_for_state(domain: str, target_state: str) -> str | None: if domain in {"fan", "humidifier", "light", "switch"}: return {"on": "turn_on", "off": "turn_off"}.get(target_state) if domain == "cover": return {"open": "open_cover", "closed": "close_cover"}.get(target_state) return None def _state_at(series: StateHistorySeries | None, timestamp: datetime) -> str | None: if series is None: return None state: str | None = None for point in series.points: if point.timestamp > timestamp: break state = point.state return state def _action_source( point: StateHistoryPoint, logbook: list[LogbookEntry], ) -> tuple[str, float]: nearest = min( logbook, key=lambda item: abs(item.timestamp - point.timestamp), default=None, ) if nearest is None or abs(nearest.timestamp - point.timestamp) > _ACTION_LOGBOOK_TOLERANCE: return "physical_or_unknown", 0.7 if nearest.context_user_id: return "user", 1.0 if nearest.context_domain in _AUTOMATION_CONTEXT_DOMAINS: return "automation", 0.1 return "physical_or_unknown", 0.7 def _matches_own_execution( point: StateHistoryPoint, own_executions: list[ExecutionEvent], ) -> bool: return any( event.target_state == point.state and abs(event.executed_at - point.timestamp) <= _OWN_ACTION_TOLERANCE for event in own_executions ) def _recent_context_transition( history: dict[str, StateHistorySeries], context_ids: list[str], timestamp: datetime, ) -> tuple[str, str, str] | None: nearest: tuple[timedelta, str, str, str] | None = None for entity_id in context_ids: series = history.get(entity_id) if series is None: continue previous_state: str | None = None for point in series.points: if point.timestamp > timestamp: break if previous_state is not None and point.state != previous_state: age = timestamp - point.timestamp if age <= _CONTEXT_TRIGGER_TOLERANCE and ( nearest is None or age < nearest[0] ): nearest = (age, entity_id, previous_state, point.state) previous_state = point.state if nearest is None: return None return nearest[1], nearest[2], nearest[3] def _circular_minute_distance(left: int, right: int) -> int: direct = abs(left - right) return min(direct, 1440 - direct)