pino/sillyhome-next
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 47 Wiki Activity

Add production diagnostics and planning features
Some checks failed
quality / test (3.11) (push) Has been cancelled
Details
quality / test (3.13) (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Otto
2026-06-18 11:53:53 +02:00
parent d9dc186f9b
commit 575211f0db
12 changed files with 737 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

356
app/behavior/engine.py
View File

@@ -3,12 +3,15 @@ from __future__ import annotations
import logging
from collections.abc import Sequence
from datetime import datetime, timedelta, timezone
from time import perf_counter
from zoneinfo import ZoneInfo
from app.actuators.models import (
ActuatorRecord,
AdaptiveWeightUpdate,
AgentInsight,
AnomalyEvent,
ActuatorGroup,
AutomationConflict,
BehaviorMode,
BehaviorPattern,
@@ -16,12 +19,16 @@ from app.actuators.models import (
BehaviorState,
BehaviorStatus,
DecisionFactor,
DecisionTrace,
ExecutionEvent,
FeedbackKind,
LatencyMeasurement,
ManualOverride,
ModelSnapshot,
RelatedAutomation,
SafetyProfile,
SafetyStage,
SceneSuggestion,
TimeProfile,
)
from app.actuators.store import ActuatorStore
@@ -35,6 +42,9 @@ _MAX_PATTERNS = 500
_MAX_MODEL_SNAPSHOTS = 3
_MAX_SNAPSHOT_PATTERNS = 120
_MAX_EXECUTION_EVENTS = 100
_MAX_DECISION_TRACES = 30
_MAX_LATENCY_MEASUREMENTS = 50
_MAX_FEEDBACK_LOG = 50
_ACTION_LOGBOOK_TOLERANCE = timedelta(seconds=10)
_CONTEXT_TRIGGER_TOLERANCE = timedelta(seconds=3)
_OWN_ACTION_TOLERANCE = timedelta(seconds=20)
@@ -254,7 +264,11 @@ class BehaviorEngine:
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,
trigger_entity_id: str | None = None,
trigger_state: str | None = None,
event_received_at: datetime | None = None,
) -> ActuatorRecord:
started_perf = perf_counter()
record = self._store.get(actuator_entity_id)
now = datetime.now(timezone.utc)
if current_entities is None:
@@ -344,6 +358,7 @@ class BehaviorEngine:
else:
safety_allowed = False
safety_blockers = ["Keine fällige Vorhersage."]
decision_to_service_ms: int | None = None
decision_factors = _decision_factors_for(record, current_context, prediction)
behavior = record.behavior.model_copy(
update={
@@ -383,12 +398,47 @@ class BehaviorEngine:
domain = actuator_entity_id.split(".", 1)[0]
service = service_for_state(domain, prediction.target_state)
if service is not None:
if record.behavior.dry_run_enabled:
behavior = behavior.model_copy(
update={
"prediction": prediction.model_copy(
update={
"executed": False,
"execution_reason": (
"Dry-run: Aktion wäre ausgeführt worden."
),
}
),
"dry_run_sample_count": record.behavior.dry_run_sample_count + 1,
"reason": (
f"Dry-run hätte {prediction.target_state!r} mit "
f"{prediction.confidence:.0%} Sicherheit ausgeführt."
),
}
)
return self._save_behavior(
record,
_append_decision_trace(
behavior,
trigger_entity_id=trigger_entity_id,
trigger_state=trigger_state,
prediction=prediction,
safety_blockers=safety_blockers,
duration_ms=_elapsed_ms(started_perf),
event_received_at=event_received_at,
decision_to_service_ms=None,
executed=False,
source="event" if event_received_at is not None else "manual",
),
)
try:
service_started_perf = perf_counter()
self._ha_reader.call_service(
domain,
service,
{"entity_id": actuator_entity_id},
)
decision_to_service_ms = _elapsed_ms(service_started_perf)
except (HaClientError, ValueError) as exc:
logger.error(
"Predicted action failed for %s: %s",
@@ -400,7 +450,21 @@ class BehaviorEngine:
"reason": f"Vorhersage wurde aus Sicherheitsgründen nicht ausgeführt: {exc}"
}
)
return self._save_behavior(record, behavior)
return self._save_behavior(
record,
_append_decision_trace(
behavior,
trigger_entity_id=trigger_entity_id,
trigger_state=trigger_state,
prediction=prediction,
safety_blockers=[str(exc)],
duration_ms=_elapsed_ms(started_perf),
event_received_at=event_received_at,
decision_to_service_ms=None,
executed=False,
source="event" if event_received_at is not None else "manual",
),
)
event = ExecutionEvent(
target_state=prediction.target_state,
executed_at=now,
@@ -434,6 +498,20 @@ class BehaviorEngine:
)
}
)
behavior = _append_decision_trace(
behavior,
trigger_entity_id=trigger_entity_id,
trigger_state=trigger_state,
prediction=prediction,
safety_blockers=safety_blockers,
duration_ms=_elapsed_ms(started_perf),
event_received_at=event_received_at,
decision_to_service_ms=(
decision_to_service_ms
),
executed=bool(prediction is not None and behavior.prediction is not None and behavior.prediction.executed),
source="event" if event_received_at is not None else "manual",
)
return self._save_behavior(record, behavior)
def record_feedback(
@@ -442,6 +520,7 @@ class BehaviorEngine:
*,
correct: bool,
expected_state: str | None = None,
kind: FeedbackKind | None = None,
) -> ActuatorRecord:
record = self._store.get(actuator_entity_id)
now = datetime.now(timezone.utc)
@@ -485,6 +564,7 @@ class BehaviorEngine:
reason = "Vorhersage wurde vom Nutzer als korrekt bestätigt."
correct_count = record.behavior.correct_feedback_count + 1
incorrect_count = record.behavior.incorrect_feedback_count
feedback_kind = kind or FeedbackKind.CORRECT
else:
target = prediction.target_state if prediction is not None else None
if target:
@@ -515,11 +595,24 @@ class BehaviorEngine:
reason = "Vorhersage wurde vom Nutzer als falsch markiert."
correct_count = record.behavior.correct_feedback_count
incorrect_count = record.behavior.incorrect_feedback_count + 1
feedback_kind = kind or FeedbackKind.WRONG
if feedback_kind is FeedbackKind.NEVER_AUTOMATE:
safety = record.behavior.safety.model_copy(
update={
"manual_block": True,
"updated_at": now,
"note": "Durch Nutzerfeedback dauerhaft blockiert.",
}
)
else:
safety = record.behavior.safety
adaptive_updates, manual_override = _adapt_sensor_weights(
record,
current_context,
correct=correct,
)
if correct and prediction is not None:
safety = record.behavior.safety
behavior = record.behavior.model_copy(
update={
"patterns": patterns[-_MAX_PATTERNS:],
@@ -532,6 +625,11 @@ class BehaviorEngine:
"last_trained_at": now,
"correct_feedback_count": correct_count,
"incorrect_feedback_count": incorrect_count,
"feedback_log": [
*record.behavior.feedback_log,
feedback_kind,
][-_MAX_FEEDBACK_LOG:],
"safety": safety,
"adaptive_weight_updates": [
*record.behavior.adaptive_weight_updates,
*adaptive_updates,
@@ -557,6 +655,43 @@ class BehaviorEngine:
)
return self._save_behavior(record_for_save, behavior)
def set_dry_run(self, actuator_entity_id: str, *, enabled: bool) -> ActuatorRecord:
record = self._store.get(actuator_entity_id)
now = datetime.now(timezone.utc)
behavior = record.behavior.model_copy(
update={
"dry_run_enabled": enabled,
"dry_run_started_at": now if enabled else record.behavior.dry_run_started_at,
"reason": (
"Dry-run aktiv; freigegebene Aktionen werden protokolliert, aber nicht geschaltet."
if enabled
else "Dry-run beendet."
),
}
)
return self._save_behavior(record, behavior)
def refresh_planning_insights(self) -> list[ActuatorRecord]:
records = self._store.list()
groups = _derive_actuator_groups(records)
scenes = _derive_scene_suggestions(records)
insights_by_actuator = _derive_agent_insights(records)
updated: list[ActuatorRecord] = []
for record in records:
behavior = record.behavior.model_copy(
update={
"actuator_groups": [
group for group in groups if record.actuator_entity_id in group.member_entity_ids
],
"scene_suggestions": [
scene for scene in scenes if record.actuator_entity_id in scene.member_entity_ids
],
"agent_insights": insights_by_actuator.get(record.actuator_entity_id, []),
}
)
updated.append(self._save_behavior(record, behavior))
return updated
def rollback_model(
self,
actuator_entity_id: str,
@@ -966,11 +1101,19 @@ class BehaviorEngine:
- Wenn current_entities gesetzt ist, kommt die Auswertung direkt aus dem
WebSocket-State-Cache statt aus einer frischen REST-Abfrage.
"""
event_received_at = datetime.now(timezone.utc)
records = self._store.list()
# Aktor direkt evaluieren
for record in self._store.list():
for record in records:
if record.actuator_entity_id == entity_id:
try:
self.evaluate(record.actuator_entity_id, current_entities=current_entities)
self.evaluate(
record.actuator_entity_id,
current_entities=current_entities,
trigger_entity_id=entity_id,
trigger_state=_event_state(new_state),
event_received_at=event_received_at,
)
except Exception:
logger.exception("Event-basierte Vorhersage fehlgeschlagen für %s", record.actuator_entity_id)
return
@@ -979,7 +1122,7 @@ class BehaviorEngine:
# Kontext-Entity: alle Aktoren finden, die diesen Kontext nutzen
affected_actuators = [
record.actuator_entity_id
for record in self._store.list()
for record in records
if (
record.assignment.selected_numeric_entity_id == entity_id
or entity_id in record.assignment.selected_context_entity_ids
@@ -992,6 +1135,9 @@ class BehaviorEngine:
context_state_overrides={entity_id: event_state},
context_changed_at_overrides={entity_id: event_changed_at},
current_entities=current_entities,
trigger_entity_id=entity_id,
trigger_state=event_state,
event_received_at=event_received_at,
)
except Exception:
logger.exception("Event-basierte Vorhersage fehlgeschlagen für %s", actuator_entity_id)
@@ -1019,6 +1165,208 @@ def _event_changed_at(new_state: dict[str, object] | None) -> datetime | None:
return parsed
def _elapsed_ms(started_perf: float) -> int:
return max(0, int((perf_counter() - started_perf) * 1000))
def _append_decision_trace(
behavior: BehaviorState,
*,
trigger_entity_id: str | None,
trigger_state: str | None,
prediction: BehaviorPrediction | None,
safety_blockers: list[str],
duration_ms: int,
event_received_at: datetime | None,
decision_to_service_ms: int | None,
executed: bool,
source: str,
) -> BehaviorState:
now = datetime.now(timezone.utc)
blocked = prediction is None or bool(safety_blockers)
trace = DecisionTrace(
trace_id=f"{now.strftime('%Y%m%d%H%M%S%f')}.{trigger_entity_id or 'manual'}",
created_at=now,
trigger_entity_id=trigger_entity_id,
trigger_state=trigger_state,
target_state=prediction.target_state if prediction is not None else None,
confidence=prediction.confidence if prediction is not None else None,
executed=executed,
blocked=blocked,
reason=(
prediction.execution_reason
if prediction is not None
else behavior.reason
),
blockers=safety_blockers if prediction is not None else ["Keine fällige Vorhersage."],
duration_ms=duration_ms,
)
updated = behavior.model_copy(
update={
"decision_timeline": [
*behavior.decision_timeline,
trace,
][-_MAX_DECISION_TRACES:],
}
)
if event_received_at is None:
return updated
return _append_latency_measurement(
updated,
trigger_entity_id=trigger_entity_id,
event_received_at=event_received_at,
event_to_decision_ms=duration_ms,
decision_to_service_ms=decision_to_service_ms,
executed=executed,
source=source,
)
def _append_latency_measurement(
behavior: BehaviorState,
*,
trigger_entity_id: str | None,
event_received_at: datetime | None,
event_to_decision_ms: int | None,
decision_to_service_ms: int | None,
executed: bool,
source: str,
) -> BehaviorState:
if event_received_at is None:
return behavior
now = datetime.now(timezone.utc)
event_to_done_ms = max(0, int((now - event_received_at).total_seconds() * 1000))
measurement = LatencyMeasurement(
measured_at=now,
trigger_entity_id=trigger_entity_id,
event_to_decision_ms=event_to_decision_ms,
decision_to_service_ms=decision_to_service_ms,
event_to_done_ms=event_to_done_ms,
executed=executed,
source=source,
)
return behavior.model_copy(
update={
"latency_measurements": [
*behavior.latency_measurements,
measurement,
][-_MAX_LATENCY_MEASUREMENTS:],
}
)
def _derive_actuator_groups(records: list[ActuatorRecord]) -> list[ActuatorGroup]:
by_area: dict[str, list[str]] = {}
for record in records:
area = _area_hint(record)
if area:
by_area.setdefault(area, []).append(record.actuator_entity_id)
return [
ActuatorGroup(
group_id=_slug(f"area_{area}"),
name=f"Raum {area}",
area_name=area,
member_entity_ids=sorted(entity_ids),
reason="Aktor-Gruppe aus gemeinsamer Raum-/Kontextzuordnung abgeleitet.",
)
for area, entity_ids in sorted(by_area.items())
if len(entity_ids) >= 2
]
def _derive_scene_suggestions(records: list[ActuatorRecord]) -> list[SceneSuggestion]:
scenes: list[SceneSuggestion] = []
by_context: dict[tuple[str, str], list[str]] = {}
for record in records:
for pattern in record.behavior.patterns:
for entity_id, state in pattern.context_states.items():
by_context.setdefault((entity_id, state), []).append(record.actuator_entity_id)
for (entity_id, state), members in sorted(by_context.items()):
unique_members = sorted(set(members))
if len(unique_members) < 2:
continue
scenes.append(
SceneSuggestion(
scene_id=_slug(f"{entity_id}_{state}"),
label=f"{entity_id} ist {state}",
member_entity_ids=unique_members,
confidence=min(1.0, len(members) / max(3, len(unique_members) * 2)),
reason="Mehrere Aktoren reagieren historisch auf denselben Kontext.",
last_seen_at=max(
(
pattern.observed_at
for record in records
for pattern in record.behavior.patterns
if pattern.context_states.get(entity_id) == state
),
default=None,
),
)
)
return scenes[-20:]
def _derive_agent_insights(records: list[ActuatorRecord]) -> dict[str, list[AgentInsight]]:
result: dict[str, list[AgentInsight]] = {}
for record in records:
insights: list[AgentInsight] = []
if record.behavior.automation_conflicts:
insights.append(
AgentInsight(
insight_id=f"{record.actuator_entity_id}.automation_conflict",
severity="warning",
title="Automation-Konflikt prüfen",
detail="Eine passende HA-Automation kann parallel zu SillyHome schalten.",
action="Automation pausieren oder SillyHome im Shadow-Modus lassen.",
)
)
if record.behavior.latency_measurements:
durations = [
item.event_to_done_ms
for item in record.behavior.latency_measurements
if item.event_to_done_ms is not None
]
if durations and max(durations) > 1500:
insights.append(
AgentInsight(
insight_id=f"{record.actuator_entity_id}.latency",
severity="warning",
title="Schalt-Latenz beobachten",
detail=f"Letzte maximale Event-Latenz: {max(durations)} ms.",
action="WebSocket-Status, HA-Servicezeit und Sensor-Routing pruefen.",
)
)
if record.behavior.incorrect_feedback_count > record.behavior.correct_feedback_count:
insights.append(
AgentInsight(
insight_id=f"{record.actuator_entity_id}.feedback",
severity="warning",
title="Viele negative Feedbacks",
detail="Das Modell trifft aktuell mehr falsche als richtige Entscheidungen.",
action="Kontextzuordnung, Gewichtung oder Modell-Rollback pruefen.",
)
)
result[record.actuator_entity_id] = insights[:5]
return result
def _area_hint(record: ActuatorRecord) -> str | None:
for candidate in [*record.context_candidates, *record.numeric_candidates]:
if candidate.area_name:
return candidate.area_name
return None
def _slug(value: str) -> str:
result = []
for char in value.lower():
if char.isalnum():
result.append(char)
elif char in {".", "_", "-", " "}:
result.append("_")
return "".join(result).strip("_")[:64] or "item"
def _confidence_threshold_for(profile: SafetyProfile, target_state: str) -> float:
if target_state == "on" and profile.min_confidence_on is not None:
return profile.min_confidence_on