113 lines
3.9 KiB
Python
113 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from collections.abc import Callable
|
|
|
|
from app.core.decision import DecisionEngineV2
|
|
from app.core.handoff import HandoffMatrix
|
|
from app.core.models import (
|
|
AuditEvent,
|
|
ControlProfile,
|
|
Decision,
|
|
EntityState,
|
|
LearningProfile,
|
|
StateEvent,
|
|
)
|
|
from app.core.stores import FutureStores
|
|
|
|
|
|
class RoutingIndex:
|
|
def affected_actuators(self, trigger_entity_id: str, learning: dict[str, LearningProfile]) -> list[str]:
|
|
result = [
|
|
actuator_id
|
|
for actuator_id, profile in learning.items()
|
|
if any(pattern.trigger_entity_id == trigger_entity_id for pattern in profile.patterns)
|
|
]
|
|
return sorted(set(result))
|
|
|
|
|
|
class EventCore:
|
|
def __init__(self, stores: FutureStores) -> None:
|
|
self._stores = stores
|
|
self._router = RoutingIndex()
|
|
self._decision = DecisionEngineV2()
|
|
self._handoff = HandoffMatrix()
|
|
|
|
def process_state_event(
|
|
self,
|
|
event: StateEvent,
|
|
*,
|
|
execute: Callable[[Decision], bool] | None = None,
|
|
) -> list[AuditEvent]:
|
|
runtime = self._stores.runtime()
|
|
learning = self._stores.learning()
|
|
control = self._stores.control()
|
|
runtime.entities[event.entity_id] = EntityState(
|
|
entity_id=event.entity_id,
|
|
domain=event.entity_id.split(".", 1)[0],
|
|
state=event.new_state,
|
|
changed_at=event.changed_at,
|
|
area_name=event.attributes.get("area_name"),
|
|
device_id=event.attributes.get("device_id"),
|
|
friendly_name=event.attributes.get("friendly_name"),
|
|
)
|
|
audit: list[AuditEvent] = [
|
|
AuditEvent(
|
|
event_id=_event_id("state"),
|
|
kind="state",
|
|
entity_id=event.entity_id,
|
|
message=f"{event.entity_id} -> {event.new_state}",
|
|
)
|
|
]
|
|
for actuator_id in self._router.affected_actuators(event.entity_id, learning.profiles):
|
|
learning_profile = learning.profiles[actuator_id]
|
|
control_profile = control.profiles.get(
|
|
actuator_id,
|
|
ControlProfile(actuator_entity_id=actuator_id),
|
|
)
|
|
control_profile = control_profile.model_copy(
|
|
update={"handoff_mode": self._handoff.classify(control_profile)}
|
|
)
|
|
control.profiles[actuator_id] = control_profile
|
|
decision = self._decision.decide(
|
|
actuator_entity_id=actuator_id,
|
|
trigger_entity_id=event.entity_id,
|
|
runtime=runtime,
|
|
learning=learning_profile,
|
|
control=control_profile,
|
|
)
|
|
if callable(execute) and decision.allowed and not decision.dry_run:
|
|
decision = _execute_decision(decision, execute)
|
|
audit.append(
|
|
AuditEvent(
|
|
event_id=_event_id("decision"),
|
|
kind="decision",
|
|
entity_id=actuator_id,
|
|
message=decision.reason,
|
|
decision=decision,
|
|
)
|
|
)
|
|
runtime.audit = [*runtime.audit, *audit][-200:]
|
|
self._stores.save_runtime(runtime)
|
|
self._stores.save_control(control)
|
|
return audit
|
|
|
|
|
|
def _event_id(prefix: str) -> str:
|
|
return f"{prefix}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S%f')}"
|
|
|
|
|
|
def _execute_decision(decision: Decision, execute: Callable[[Decision], bool]) -> Decision:
|
|
try:
|
|
result = execute(decision)
|
|
except Exception as exc:
|
|
return decision.model_copy(
|
|
update={
|
|
"executed": False,
|
|
"allowed": False,
|
|
"reason": f"Ausfuehrung fehlgeschlagen: {exc}",
|
|
"blockers": [*decision.blockers, str(exc)],
|
|
}
|
|
)
|
|
return decision.model_copy(update={"executed": bool(result)})
|