from __future__ import annotations import logging from collections.abc import Sequence from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo from app.actuators.models import ( ActuatorRecord, BehaviorMode, BehaviorPattern, BehaviorPrediction, BehaviorState, BehaviorStatus, ExecutionEvent, RelatedAutomation, ) 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.models import HaEntitySummary 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, "activation_ready": False, "activation_reason": ( "Freigabe gesperrt: Noch kein geeigneter Kontext erkannt." ), "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, "activation_ready": False, "activation_reason": ( "Freigabe gesperrt: Noch keine historischen " "Aktorhandlungen gefunden." ), "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, ) trusted_actions = sum( 1 for pattern in patterns if pattern.source in {"user", "automation"} ) 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." ) ) activation_ready = ( status is BehaviorStatus.TRAINED and trusted_actions >= self._settings.min_behavior_actions ) activation_reason = ( "Freigabe bereit: Genügend eindeutig zugeordnete Handlungen gelernt." if activation_ready else ( "Freigabe gesperrt: " f"{max(0, self._settings.min_behavior_actions - trusted_actions)} " "eindeutig zugeordnete Handlungen fehlen." ) ) behavior = record.behavior.model_copy( update={ "status": status, "sample_count": len(patterns), "high_confidence_sample_count": trusted_actions, "activation_ready": activation_ready, "activation_reason": activation_reason, "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, *, context_state_overrides: dict[str, str | None] | None = None, context_changed_at_overrides: dict[str, datetime | None] | None = None, current_entities: Sequence[HaEntitySummary] | None = None, ) -> ActuatorRecord: record = self._store.get(actuator_entity_id) now = datetime.now(timezone.utc) if current_entities is None: try: current_entities = 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}", } ), ) entities = {entity.entity_id: entity for entity in current_entities} 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 } selected_context_ids = { entity_id for entity_id in ( [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] ) if entity_id } for entity_id, state in (context_state_overrides or {}).items(): if entity_id in selected_context_ids and state is not None: current_context[entity_id] = state for entity_id, changed_at in (context_changed_at_overrides or {}).items(): if entity_id in current_context: current_context_changed_at[entity_id] = changed_at or now 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, ) if prediction is not None: prediction = prediction.model_copy( update={ "execution_reason": self._prediction_execution_reason( record, actuator.state, prediction, now, ) } ) 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, prediction.target_state, ) ): 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, "execution_reason": ( f"Ausgeführt mit {prediction.confidence:.0%} Sicherheit." ), } ), "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 record_feedback( self, actuator_entity_id: str, *, correct: bool, expected_state: str | None = None, ) -> ActuatorRecord: record = self._store.get(actuator_entity_id) now = datetime.now(timezone.utc) 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("Aktor ist aktuell nicht in Home Assistant verfügbar.") context_ids = [ entity_id for entity_id in [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] if entity_id ] current_context = { entity_id: entities[entity_id].state for entity_id in context_ids if entity_id in entities and entities[entity_id].state is not None } prediction = record.behavior.prediction patterns = list(record.behavior.patterns) reason = "Nutzerfeedback gespeichert." if correct and prediction is not None: local = now.astimezone(ZoneInfo(self._settings.timezone)) patterns.append( BehaviorPattern( target_state=prediction.target_state, minute_of_day=local.hour * 60 + local.minute, weekday=local.weekday(), context_states={ entity_id: state for entity_id, state in current_context.items() if state is not None }, source="user_feedback", weight=1.0, observed_at=now, ) ) reason = "Vorhersage wurde vom Nutzer als korrekt bestätigt." else: target = prediction.target_state if prediction is not None else None if target: patterns = [ pattern.model_copy(update={"weight": 0.1}) if pattern.target_state == target and _pattern_context_matches(pattern, current_context) else pattern for pattern in patterns ] if expected_state: local = now.astimezone(ZoneInfo(self._settings.timezone)) patterns.append( BehaviorPattern( target_state=expected_state, minute_of_day=local.hour * 60 + local.minute, weekday=local.weekday(), context_states={ entity_id: state for entity_id, state in current_context.items() if state is not None }, source="user_correction", weight=1.0, observed_at=now, ) ) reason = "Vorhersage wurde vom Nutzer als falsch markiert." behavior = record.behavior.model_copy( update={ "patterns": patterns[-_MAX_PATTERNS:], "prediction": ( prediction.model_copy(update={"execution_reason": reason}) if prediction is not None else None ), "reason": reason, "last_trained_at": now, } ) return self._save_behavior(record, behavior) def refresh_related_automations(self, actuator_entity_id: str) -> ActuatorRecord: record = self._store.get(actuator_entity_id) related = [ RelatedAutomation( entity_id=item.entity_id, config_id=item.config_id, friendly_name=item.friendly_name, enabled=item.enabled, ) for item in self._ha_reader.find_automations_for_entity( actuator_entity_id ) ] behavior = record.behavior.model_copy( update={"related_automations": related} ) return self._save_behavior(record, behavior) def set_automation_enabled( self, actuator_entity_id: str, automation_entity_id: str, *, enabled: bool, ) -> ActuatorRecord: record = self.refresh_related_automations(actuator_entity_id) if automation_entity_id not in { item.entity_id for item in record.behavior.related_automations }: raise ValueError( "Die Automation ist diesem Aktor nicht eindeutig zugeordnet." ) self._ha_reader.call_service( "automation", "turn_on" if enabled else "turn_off", {"entity_id": automation_entity_id}, ) related = [ item.model_copy(update={"enabled": enabled}) if item.entity_id == automation_entity_id else item for item in record.behavior.related_automations ] paused = [ entity_id for entity_id in record.behavior.paused_automation_entity_ids if entity_id != automation_entity_id ] behavior = record.behavior.model_copy( update={ "related_automations": related, "paused_automation_entity_ids": paused, } ) return self._save_behavior(record, behavior) def set_active( self, actuator_entity_id: str, *, active: bool, pause_matching_automations: bool = False, restore_paused_automations: bool = False, ) -> ActuatorRecord: record = self.refresh_related_automations(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 not record.behavior.activation_ready: raise ValueError(record.behavior.activation_reason) mode = BehaviorMode.ACTIVE approved_at = now behavior = record.behavior.model_copy( update={ "mode": mode, "approved_at": approved_at, "reason": ( "Autonomes Lernen und Schalten wurde ausdrücklich freigegeben." ), } ) record = self._save_behavior(record, behavior) if pause_matching_automations: paused: list[str] = [] try: for automation in record.behavior.related_automations: if not automation.enabled: continue self._ha_reader.call_service( "automation", "turn_off", {"entity_id": automation.entity_id}, ) paused.append(automation.entity_id) except (HaClientError, ValueError): for entity_id in paused: try: self._ha_reader.call_service( "automation", "turn_on", {"entity_id": entity_id}, ) except (HaClientError, ValueError): logger.exception( "Failed to restore automation %s after handoff error", entity_id, ) rollback = record.behavior.model_copy( update={ "mode": BehaviorMode.SHADOW, "approved_at": None, "reason": ( "Übernahme fehlgeschlagen; SillyHome bleibt im " "Shadow-Modus." ), } ) self._save_behavior(record, rollback) raise related = [ automation.model_copy(update={"enabled": False}) if automation.entity_id in paused else automation for automation in record.behavior.related_automations ] behavior = record.behavior.model_copy( update={ "related_automations": related, "paused_automation_entity_ids": paused, "reason": ( "SillyHome steuert aktiv; passende HA-Automationen " "wurden pausiert." ), } ) return self._save_behavior(record, behavior) return record else: if restore_paused_automations: for entity_id in record.behavior.paused_automation_entity_ids: self._ha_reader.call_service( "automation", "turn_on", {"entity_id": entity_id}, ) mode = BehaviorMode.SHADOW approved_at = None reason = ( "Shadow-Modus aktiv; pausierte HA-Automationen wurden fortgesetzt." if restore_paused_automations else "Shadow-Modus aktiv; Vorhersagen werden nicht ausgeführt." ) behavior = record.behavior.model_copy( update={ "mode": mode, "approved_at": approved_at, "related_automations": [ automation.model_copy(update={"enabled": True}) if ( restore_paused_automations and automation.entity_id in record.behavior.paused_automation_entity_ids ) else automation for automation in record.behavior.related_automations ], "paused_automation_entity_ids": ( [] if restore_paused_automations else record.behavior.paused_automation_entity_ids ), "reason": reason, } ) return self._save_behavior(record, behavior) def _prediction_execution_reason( self, record: ActuatorRecord, current_state: str | None, prediction: BehaviorPrediction, now: datetime, ) -> str: if record.behavior.mode is not BehaviorMode.ACTIVE: return "Nicht ausgeführt: SillyHome ist im Shadow-Modus." if prediction.confidence < self._settings.prediction_confidence: return ( "Nicht ausgeführt: Sicherheit liegt unter der " f"Schaltschwelle von {self._settings.prediction_confidence:.0%}." ) if current_state == prediction.target_state: return "Nicht ausgeführt: Zielzustand ist bereits erreicht." if not self._cooldown_elapsed( record.behavior, now, prediction.target_state, ): return "Nicht ausgeführt: Sicherheits-Cooldown ist noch aktiv." return "Ausführung ist freigegeben." 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, ) 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, target_state: str, ) -> bool: if behavior.last_executed_at is None: return True last_event = behavior.execution_events[-1] if behavior.execution_events else None if last_event is not None and last_event.target_state != target_state: return True return (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 handle_state_change( self, entity_id: str, new_state: dict[str, object] | None, *, current_entities: Sequence[HaEntitySummary] | None = None, ) -> None: """Wird bei jedem HA-State-Change aufgerufen und löst sofortige Vorhersage aus. - Wenn entity_id ein Aktor ist: evaluate() direkt. - Wenn entity_id ein Kontext-Entity ist: alle betroffenen Aktoren evaluieren. - Wenn current_entities gesetzt ist, kommt die Auswertung direkt aus dem WebSocket-State-Cache statt aus einer frischen REST-Abfrage. """ # Aktor direkt evaluieren for record in self._store.list(): if record.actuator_entity_id == entity_id: try: self.evaluate(record.actuator_entity_id, current_entities=current_entities) except Exception: logger.exception("Event-basierte Vorhersage fehlgeschlagen für %s", record.actuator_entity_id) return event_state = _event_state(new_state) event_changed_at = _event_changed_at(new_state) or datetime.now(timezone.utc) # Kontext-Entity: alle Aktoren finden, die diesen Kontext nutzen affected_actuators = [ record.actuator_entity_id for record in self._store.list() if ( record.assignment.selected_numeric_entity_id == entity_id or entity_id in record.assignment.selected_context_entity_ids ) ] for actuator_entity_id in affected_actuators: try: self.evaluate( actuator_entity_id, context_state_overrides={entity_id: event_state}, context_changed_at_overrides={entity_id: event_changed_at}, current_entities=current_entities, ) except Exception: logger.exception("Event-basierte Vorhersage fehlgeschlagen für %s", actuator_entity_id) def _event_state(new_state: dict[str, object] | None) -> str | None: if not isinstance(new_state, dict): return None state = new_state.get("state") return state if isinstance(state, str) else None def _event_changed_at(new_state: dict[str, object] | None) -> datetime | None: if not isinstance(new_state, dict): return None value = new_state.get("last_changed") or new_state.get("last_updated") if not isinstance(value, str): return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed 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", 1.0 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 _pattern_context_matches( pattern: BehaviorPattern, current_context: dict[str, str | None], ) -> bool: comparable = [ (entity_id, expected) for entity_id, expected in pattern.context_states.items() if entity_id in current_context ] if not comparable: return False return all(current_context[entity_id] == expected for entity_id, expected in comparable) 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)