Files
sillyhome-next-dev/app/behavior/engine.py
Otto b3cf68eade
Some checks failed
quality / test (3.11) (push) Has been cancelled
quality / test (3.13) (push) Has been cancelled
CONTROL-001: add safe HA automation handoff
2026-06-14 16:21:57 +02:00

795 lines
30 KiB
Python

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,
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.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) -> 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,
)
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 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 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 _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)