from __future__ import annotations import json import os from datetime import datetime, timezone from pathlib import Path from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from pydantic import BaseModel, Field from app.actuators.cache_db import DashboardCache from app.actuators.lifecycle import ActuatorReconciliationService from app.actuators.models import ( ActuatorRecord, AnomalyEvent, AssignmentCandidate, BehaviorPattern, FeedbackKind, ReconciliationState, SensorWeightGroup, SimulationOutcome, ) from app.actuators.models import JobQueueItem, JobQueueState, JobStatus, SafetyProfile from app.actuators.store import ActuatorStore from app.behavior.engine import BehaviorEngine from app.config import Settings from app.dependencies import get_ha_reader from app.ha.discovery import DiscoveredEntity, EntityRole, discover_entities from app.ha.exceptions import HaClientError from app.ha.models import HaEntitySummary from app.ha.reader import HaReader router = APIRouter(prefix="/v1/actuators", tags=["actuators"]) class ConfigureActuatorRequest(BaseModel): actuator_entity_id: str = Field(pattern=r"^[a-z0-9_]+\.[a-z0-9_]+$") enabled: bool = True class ActivationRequest(BaseModel): active: bool pause_matching_automations: bool = False restore_paused_automations: bool = False class AutomationControlRequest(BaseModel): automation_entity_id: str = Field(pattern=r"^automation\.[a-z0-9_]+$") enabled: bool class ManualAssignmentRequest(BaseModel): numeric_entity_id: str | None = Field(default=None, pattern=r"^[a-z0-9_]+\.[a-z0-9_]+$") context_entity_ids: list[str] = Field(default_factory=list) note: str | None = Field(default=None, max_length=500) class WeightOverrideRequest(BaseModel): sensor_weights: dict[str, float] = Field(default_factory=dict) sensor_weight_groups: list[SensorWeightGroup] = Field(default_factory=list) note: str | None = Field(default=None, max_length=500) class SimulationRequest(BaseModel): sensor_states: dict[str, str] = Field(default_factory=dict) sensor_weights: dict[str, float] = Field(default_factory=dict) state_options: dict[str, list[str]] = Field(default_factory=dict) include_current: bool = True max_results: int = Field(default=8, ge=1, le=20) class FeedbackRequest(BaseModel): correct: bool expected_state: str | None = Field(default=None, max_length=100) kind: FeedbackKind | None = None class DryRunRequest(BaseModel): enabled: bool class BackupPayload(BaseModel): exported_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) records: list[ActuatorRecord] = Field(default_factory=list) reconciliation: ReconciliationState = Field(default_factory=ReconciliationState) jobs: JobQueueState = Field(default_factory=JobQueueState) class RestoreRequest(BaseModel): backup: BackupPayload replace_existing: bool = False class RestoreResult(BaseModel): restored_records: int = 0 skipped_existing: int = 0 restored_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) class SafetyProfileRequest(BaseModel): safety: SafetyProfile class ModelRollbackRequest(BaseModel): version_id: str = Field(min_length=1, max_length=120) class ActuatorSuggestion(BaseModel): entity_id: str domain: str friendly_name: str | None = None area_name: str | None = None device_name: str | None = None confidence: float reason: str related_automation_count: int = 0 likely_context_count: int = 0 class ActuatorSummary(BaseModel): actuator_entity_id: str domain: str friendly_name: str | None = None area_name: str | None = None device_name: str | None = None enabled: bool behavior_mode: str behavior_status: str lifecycle_status: str activation_ready: bool activation_reason: str sample_count: int anomaly_count: int = 0 critical_anomaly_count: int = 0 prediction_target_state: str | None = None prediction_confidence: float | None = None updated_at: str class EntityCacheStatus(BaseModel): available: bool updated_at: str | None = None entity_count: int = 0 class DashboardSystemStatus(BaseModel): api_status: str = "ok" websocket_status: str = "unavailable" websocket_error: str | None = None reconciliation_last_completed_at: str | None = None configured_actuators: int = 0 trained_models: int = 0 review_required: int = 0 performance_budget_ms: int = 3000 job_p95_duration_ms: int | None = None slow_job_count: int = 0 performance_status: str = "unknown" anomaly_count: int = 0 critical_anomaly_count: int = 0 class DashboardDiscoveryGroup(BaseModel): category: str role: str count: int class DashboardOverview(BaseModel): system: DashboardSystemStatus cache: EntityCacheStatus actuators: list[ActuatorSummary] discovery_groups: list[DashboardDiscoveryGroup] jobs: JobQueueState = Field(default_factory=JobQueueState) class AnomalyOverview(BaseModel): actuator_entity_id: str friendly_name: str | None = None anomalies: list[AnomalyEvent] = Field(default_factory=list) class RoomManagementSensor(BaseModel): entity_id: str domain: str role: str category: str friendly_name: str | None = None device_class: str | None = None state: str | None = None confidence: float = Field(default=0.0, ge=0.0, le=1.0) active: bool = False optional: bool = False not_required: bool = False reason: str class RoomManagementAction(BaseModel): action_id: str category: str actuator_entity_id: str title: str when: str then: str why: str confidence: float = Field(default=0.0, ge=0.0, le=1.0) learnable: bool = True sort_key: str = "" class RoomManagementActuator(BaseModel): actuator_entity_id: str friendly_name: str | None = None domain: str behavior_mode: str behavior_status: str lifecycle_status: str sample_count: int = 0 selected_numeric_entity_id: str | None = None selected_context_entity_ids: list[str] = Field(default_factory=list) sensors: list[RoomManagementSensor] = Field(default_factory=list) prediction_rules: list[str] = Field(default_factory=list) suggested_actions: list[RoomManagementAction] = Field(default_factory=list) management_hint: str class RoomManagementGroup(BaseModel): room: str actuator_count: int sensor_count: int = 0 action_count: int = 0 sensors: list[RoomManagementSensor] = Field(default_factory=list) actuators: list[RoomManagementActuator] = Field(default_factory=list) prediction_rules: list[str] = Field(default_factory=list) suggested_actions: list[RoomManagementAction] = Field(default_factory=list) continuous_hint: str = "Wird bei Discovery, Reconciliation und Lernrefresh automatisch neu bewertet." class RoomManagementOverview(BaseModel): rooms: list[RoomManagementGroup] = Field(default_factory=list) unmanaged_actuators: list[ActuatorSuggestion] = Field(default_factory=list) @router.get("/discovery", response_model=list[HaEntitySummary]) def discover_actuators( request: Request, refresh: bool = Query(default=False), ha_reader: HaReader = Depends(get_ha_reader), ) -> list[HaEntitySummary]: cached_entities = [] if refresh else _load_cached_entities(request) if cached_entities: entities = {entity.entity_id: entity for entity in cached_entities} else: job = _start_job( request, kind="discovery", trigger="manual" if refresh else "cache-miss", summary="Home-Assistant-Entities werden gelesen und klassifiziert.", ) try: fresh_entities = list(ha_reader.read_entities()) _save_cached_entities(request, fresh_entities) except Exception as exc: _finish_job(job, request, status=JobStatus.FAILED, summary="Discovery fehlgeschlagen.", error=str(exc)) raise _finish_job(job, request, status=JobStatus.COMPLETED, summary=f"{len(fresh_entities)} Entities klassifiziert.") entities = {entity.entity_id: entity for entity in fresh_entities} discovered = discover_entities(list(entities.values())) actuator_ids = _deduplicate_actuator_ids( [ (entity.entity_id, entity.category) for entity in discovered if entity.role is EntityRole.ACTUATOR ], entities, ) return [entities[entity_id] for entity_id in actuator_ids if entity_id in entities] @router.get("/suggestions", response_model=list[ActuatorSuggestion]) def suggest_actuators( request: Request, ha_reader: HaReader = Depends(get_ha_reader), ) -> list[ActuatorSuggestion]: entities = {entity.entity_id: entity for entity in ha_reader.read_entities()} discovered = {entity.entity_id: entity for entity in discover_entities(list(entities.values()))} configured_ids = {record.actuator_entity_id for record in _service(request).list_configured()} actuator_ids = _deduplicate_actuator_ids( [ (entity.entity_id, entity.category) for entity in discovered.values() if entity.role is EntityRole.ACTUATOR ], entities, ) suggestions: list[ActuatorSuggestion] = [] for entity_id in actuator_ids: if entity_id in configured_ids: continue entity = entities.get(entity_id) if entity is None: continue try: automations = ha_reader.find_automations_for_entity(entity_id) except Exception: automations = [] context_count = _likely_context_count(entity, entities, discovered) if not automations and context_count == 0: continue confidence = 1.0 if automations else min(0.85, 0.35 + context_count * 0.1) reason_parts = [] if automations: reason_parts.append(f"{len(automations)} passende HA-Automation(en)") if context_count: reason_parts.append(f"{context_count} naheliegende Kontext-Entity(s)") suggestions.append( ActuatorSuggestion( entity_id=entity.entity_id, domain=entity.domain, friendly_name=entity.friendly_name, area_name=entity.area_name, device_name=entity.device_name, confidence=round(confidence, 4), reason=", ".join(reason_parts), related_automation_count=len(automations), likely_context_count=context_count, ) ) return sorted( suggestions, key=lambda item: ( -item.related_automation_count, -item.confidence, item.area_name or "", item.friendly_name or item.entity_id, ), )[:30] @router.get("/context-options", response_model=list[HaEntitySummary]) def context_options( request: Request, actuator_entity_id: str | None = Query(default=None, pattern=r"^[a-z0-9_]+\.[a-z0-9_]+$"), ) -> list[HaEntitySummary]: if actuator_entity_id is None: return [] try: return _service(request).suggest_context_options(actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.get("/summary", response_model=list[ActuatorSummary]) def list_configured_summary(request: Request) -> list[ActuatorSummary]: records = _service(request).list_configured() entity_map = _load_cached_entity_map( request, {record.actuator_entity_id for record in records}, ) return [ ActuatorSummary( actuator_entity_id=record.actuator_entity_id, domain=record.actuator_entity_id.split(".", 1)[0], friendly_name=( entity_map[record.actuator_entity_id].friendly_name if record.actuator_entity_id in entity_map else None ), area_name=( entity_map[record.actuator_entity_id].area_name if record.actuator_entity_id in entity_map else None ), device_name=( entity_map[record.actuator_entity_id].device_name if record.actuator_entity_id in entity_map else None ), enabled=record.enabled, behavior_mode=record.behavior.mode.value, behavior_status=record.behavior.status.value, lifecycle_status=record.lifecycle.status.value, activation_ready=record.behavior.activation_ready, activation_reason=record.behavior.activation_reason, sample_count=record.behavior.sample_count, anomaly_count=len([item for item in record.behavior.anomalies if not item.resolved]), critical_anomaly_count=len( [ item for item in record.behavior.anomalies if not item.resolved and item.severity == "critical" ] ), prediction_target_state=( record.behavior.prediction.target_state if record.behavior.prediction is not None else None ), prediction_confidence=( record.behavior.prediction.confidence if record.behavior.prediction is not None else None ), updated_at=record.updated_at.isoformat(), ) for record in records ] @router.get("/dashboard", response_model=DashboardOverview) def dashboard_overview(request: Request) -> DashboardOverview: return _dashboard_overview(request, include_background=True, include_actuators=True) @router.get("/dashboard/start", response_model=DashboardOverview) def dashboard_start(request: Request) -> DashboardOverview: return _dashboard_overview(request, include_background=False, include_actuators=True) @router.get("/dashboard/system", response_model=DashboardOverview) def dashboard_system(request: Request) -> DashboardOverview: return _dashboard_overview(request, include_background=False, include_actuators=False) def _dashboard_overview( request: Request, *, include_background: bool, include_actuators: bool, ) -> DashboardOverview: cache_status = _load_entity_cache_status(request) raw_updated_at = cache_status.get("updated_at") updated_at = raw_updated_at if isinstance(raw_updated_at, str) else None raw_groups = cache_status.get("discovery_groups", []) cached_groups = [ DashboardDiscoveryGroup.model_validate(group) for group in raw_groups if isinstance(group, dict) ] if include_background and isinstance(raw_groups, list) else [] reconciliation = _reconciliation_state_or_default(request) ws_status = getattr(request.app.state, "ws_status", None) actuators = list_configured_summary(request) if include_actuators else [] store = getattr(request.app.state, "actuator_store", None) jobs = ( store.load_job_queue() if include_background and isinstance(store, ActuatorStore) else JobQueueState() ) job_p95_duration_ms, slow_job_count, performance_status = _performance_status(jobs) anomaly_count = sum(record.anomaly_count for record in actuators) critical_anomaly_count = sum(record.critical_anomaly_count for record in actuators) entity_count = cache_status.get("entity_count") if not isinstance(entity_count, int): entity_count = 0 return DashboardOverview( system=DashboardSystemStatus( websocket_status=getattr(ws_status, "status", "unavailable"), websocket_error=getattr(ws_status, "error", None), reconciliation_last_completed_at=( reconciliation.last_completed_at.isoformat() if reconciliation.last_completed_at is not None else None ), configured_actuators=( len(actuators) if include_actuators else reconciliation.configured_actuators ), trained_models=reconciliation.trained_models, review_required=reconciliation.review_required, job_p95_duration_ms=job_p95_duration_ms, slow_job_count=slow_job_count, performance_status=performance_status, anomaly_count=anomaly_count, critical_anomaly_count=critical_anomaly_count, ), cache=EntityCacheStatus( available=bool(entity_count), updated_at=updated_at, entity_count=entity_count, ), actuators=actuators, discovery_groups=cached_groups, jobs=jobs, ) @router.get("/anomalies", response_model=list[AnomalyOverview]) def list_anomalies(request: Request) -> list[AnomalyOverview]: records = _service(request).list_configured() entity_map = _load_cached_entity_map( request, {record.actuator_entity_id for record in records}, ) overview: list[AnomalyOverview] = [] for record in records: active = [item for item in record.behavior.anomalies if not item.resolved] if not active: continue entity = entity_map.get(record.actuator_entity_id) overview.append( AnomalyOverview( actuator_entity_id=record.actuator_entity_id, friendly_name=entity.friendly_name if entity is not None else None, anomalies=active, ) ) return overview @router.get("/settings/rooms", response_model=RoomManagementOverview) def room_management_overview(request: Request) -> RoomManagementOverview: service = _service(request) records = service.list_configured() try: entities = {entity.entity_id: entity for entity in service._ha_reader.read_entities()} except Exception: entities = _load_cached_entity_map( request, { entity_id for record in records for entity_id in [ record.actuator_entity_id, record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, *[candidate.entity_id for candidate in record.numeric_candidates[:8]], *[candidate.entity_id for candidate in record.context_candidates[:12]], ] if entity_id }, ) discovered = {entity.entity_id: entity for entity in discover_entities(list(entities.values()))} configured_ids = {record.actuator_entity_id for record in records} rooms = _build_room_shells(entities, discovered) for record in records: actuator = entities.get(record.actuator_entity_id) room = ( actuator.area_name if actuator is not None and actuator.area_name else _candidate_room(record) ) or "Ohne Raum" selected_context_ids = set(record.assignment.selected_context_entity_ids) selected_numeric_id = record.assignment.selected_numeric_entity_id selected_ids = {selected_numeric_id, *selected_context_ids} - {None} ranked_candidates = _rank_management_candidates(record) has_opening_context = any( candidate.entity_id in selected_context_ids and (candidate.device_class or "") in {"door", "garage_door", "opening", "window"} for candidate in ranked_candidates ) sensors = [ _management_sensor( candidate, entities.get(candidate.entity_id), active=candidate.entity_id in selected_ids, optional=( candidate.role is EntityRole.MEASUREMENT and candidate.entity_id != selected_numeric_id ), not_required=( record.actuator_entity_id.startswith(("light.", "switch.")) and has_opening_context and (candidate.device_class or "") == "illuminance" ), ) for candidate in ranked_candidates[:12] ] actuator_group = RoomManagementActuator( actuator_entity_id=record.actuator_entity_id, friendly_name=actuator.friendly_name if actuator is not None else None, domain=record.actuator_entity_id.split(".", 1)[0], behavior_mode=record.behavior.mode.value, behavior_status=record.behavior.status.value, lifecycle_status=record.lifecycle.status.value, sample_count=record.behavior.sample_count, selected_numeric_entity_id=selected_numeric_id, selected_context_entity_ids=record.assignment.selected_context_entity_ids, sensors=sensors, prediction_rules=_prediction_rule_lines(record, ranked_candidates), suggested_actions=_suggest_room_actions( actuator_id=record.actuator_entity_id, domain=record.actuator_entity_id.split(".", 1)[0], sensors=sensors, configured=True, ), management_hint=_management_hint(record, has_opening_context), ) if room not in rooms: rooms[room] = RoomManagementGroup(room=room, actuator_count=0) rooms[room].actuators.append(actuator_group) rooms[room].actuator_count += 1 rooms[room].prediction_rules = _unique_lines([ *rooms[room].prediction_rules, *actuator_group.prediction_rules, ])[:8] rooms[room].sensors = _merge_room_sensors(rooms[room].sensors, sensors) rooms[room].suggested_actions = _merge_room_actions( rooms[room].suggested_actions, actuator_group.suggested_actions, ) for entity_id, descriptor in discovered.items(): if descriptor.role is not EntityRole.ACTUATOR or entity_id in configured_ids: continue actuator = entities.get(entity_id) if actuator is None: continue room = _entity_room(actuator) if room not in rooms: rooms[room] = RoomManagementGroup(room=room, actuator_count=0) room_sensors = _room_sensors_for_actuator(actuator, rooms[room].sensors) actions = _suggest_room_actions( actuator_id=entity_id, domain=actuator.domain, sensors=room_sensors, configured=False, ) rooms[room].actuators.append( RoomManagementActuator( actuator_entity_id=entity_id, friendly_name=actuator.friendly_name, domain=actuator.domain, behavior_mode="unmanaged", behavior_status="suggested", lifecycle_status="unconfigured", sensors=room_sensors, prediction_rules=[_action_rule_line(action) for action in actions[:5]], suggested_actions=actions, management_hint=( "Noch nicht verwaltet: übernehmen, wenn diese Handlung gelernt oder vorgeschlagen werden soll." ), ) ) rooms[room].actuator_count += 1 rooms[room].prediction_rules = _unique_lines([ *rooms[room].prediction_rules, *[_action_rule_line(action) for action in actions], ])[:8] rooms[room].suggested_actions = _merge_room_actions(rooms[room].suggested_actions, actions) for room in rooms.values(): room.sensors = _merge_room_sensors([], room.sensors) room.sensor_count = len(room.sensors) room.action_count = len(room.suggested_actions) unmanaged = [ suggestion for suggestion in suggest_actuators(request, service._ha_reader) if suggestion.entity_id not in configured_ids ][:10] return RoomManagementOverview( rooms=sorted(rooms.values(), key=lambda item: item.room.lower()), unmanaged_actuators=unmanaged, ) @router.get("/backup/export", response_model=BackupPayload) def export_backup(request: Request) -> BackupPayload: store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Actuator Store nicht initialisiert.", ) return BackupPayload( records=store.list(), reconciliation=store.load_reconciliation_state(), jobs=store.load_job_queue(), ) @router.post("/backup/restore", response_model=RestoreResult) def restore_backup(payload: RestoreRequest, request: Request) -> RestoreResult: store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Actuator Store nicht initialisiert.", ) existing_ids = {record.actuator_entity_id for record in store.list()} restored = 0 skipped = 0 for record in payload.backup.records: if record.actuator_entity_id in existing_ids and not payload.replace_existing: skipped += 1 continue store.upsert(record) restored += 1 store.save_reconciliation_state(payload.backup.reconciliation) store.save_job_queue(payload.backup.jobs) return RestoreResult(restored_records=restored, skipped_existing=skipped) @router.post("/planning/refresh", response_model=list[ActuatorRecord]) def refresh_planning_insights(request: Request) -> list[ActuatorRecord]: return _behavior(request).refresh_planning_insights() @router.get("", response_model=list[ActuatorRecord]) def list_configured(request: Request) -> list[ActuatorRecord]: return _service(request).list_configured() @router.post("", response_model=ActuatorRecord, status_code=201) def configure(payload: ConfigureActuatorRequest, request: Request) -> ActuatorRecord: try: record = _service(request).configure_actuator( payload.actuator_entity_id, enabled=payload.enabled, ) _behavior(request).train(record.actuator_entity_id) return _behavior(request).evaluate(record.actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.get("/{actuator_entity_id}", response_model=ActuatorRecord) def get_actuator(actuator_entity_id: str, request: Request) -> ActuatorRecord: try: return _service(request).get_actuator(actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.get("/{actuator_entity_id}/detail", response_model=ActuatorRecord) def get_actuator_detail(actuator_entity_id: str, request: Request) -> ActuatorRecord: try: record = _service(request).get_actuator(actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc selected_ids = { entity_id for entity_id in [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] if entity_id } compact_snapshots = [ snapshot.model_copy(update={"patterns": []}) for snapshot in record.behavior.model_snapshots[-3:] ] compact_behavior = record.behavior.model_copy( update={ "patterns": [], "model_snapshots": compact_snapshots, } ) return record.model_copy( update={ "behavior": compact_behavior, "numeric_candidates": [ candidate for candidate in record.numeric_candidates if candidate.entity_id in selected_ids ], "context_candidates": [ candidate for candidate in record.context_candidates if candidate.entity_id in selected_ids ], } ) @router.delete("/{actuator_entity_id}", status_code=204) def delete_actuator(actuator_entity_id: str, request: Request) -> None: _service(request).delete_actuator(actuator_entity_id) @router.post("/{actuator_entity_id}/reconcile", response_model=ActuatorRecord) def reconcile_actuator( actuator_entity_id: str, request: Request, ) -> ActuatorRecord: try: _service(request).reconcile_actuator(actuator_entity_id, trigger="manual") _behavior(request).train(actuator_entity_id) return _behavior(request).evaluate(actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/evaluate", response_model=ActuatorRecord) def evaluate_actuator( actuator_entity_id: str, request: Request, ) -> ActuatorRecord: try: return _behavior(request).evaluate(actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/simulate", response_model=list[SimulationOutcome]) def simulate_actuator( actuator_entity_id: str, payload: SimulationRequest, request: Request, ) -> list[SimulationOutcome]: try: _validate_simulation_payload(payload) return _behavior(request).simulate( actuator_entity_id, sensor_states=payload.sensor_states, sensor_weights=payload.sensor_weights, state_options=payload.state_options, include_current=payload.include_current, max_results=payload.max_results, ) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/feedback", response_model=ActuatorRecord) def record_feedback( actuator_entity_id: str, payload: FeedbackRequest, request: Request, ) -> ActuatorRecord: try: return _behavior(request).record_feedback( actuator_entity_id, correct=payload.correct, expected_state=payload.expected_state, kind=payload.kind, ) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/dry-run", response_model=ActuatorRecord) def set_dry_run( actuator_entity_id: str, payload: DryRunRequest, request: Request, ) -> ActuatorRecord: try: return _behavior(request).set_dry_run(actuator_entity_id, enabled=payload.enabled) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/safety", response_model=ActuatorRecord) def set_safety_profile( actuator_entity_id: str, payload: SafetyProfileRequest, request: Request, ) -> ActuatorRecord: try: return _behavior(request).set_safety_profile(actuator_entity_id, profile=payload.safety) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/model/rollback", response_model=ActuatorRecord) def rollback_model( actuator_entity_id: str, payload: ModelRollbackRequest, request: Request, ) -> ActuatorRecord: try: return _behavior(request).rollback_model(actuator_entity_id, version_id=payload.version_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/activation", response_model=ActuatorRecord) def set_activation( actuator_entity_id: str, payload: ActivationRequest, request: Request, ) -> ActuatorRecord: try: return _behavior(request).set_active( actuator_entity_id, active=payload.active, pause_matching_automations=payload.pause_matching_automations, restore_paused_automations=payload.restore_paused_automations, ) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/assignment", response_model=ActuatorRecord) def set_manual_assignment( actuator_entity_id: str, payload: ManualAssignmentRequest, request: Request, ) -> ActuatorRecord: try: record = _service(request).set_manual_assignment( actuator_entity_id, numeric_entity_id=payload.numeric_entity_id, context_entity_ids=payload.context_entity_ids, note=payload.note, ) _behavior(request).train(record.actuator_entity_id) return _behavior(request).evaluate(record.actuator_entity_id) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @router.post("/{actuator_entity_id}/weights", response_model=ActuatorRecord) def set_weight_overrides( actuator_entity_id: str, payload: WeightOverrideRequest, request: Request, ) -> ActuatorRecord: try: _validate_weight_payload(payload) record = _service(request).set_weight_overrides( actuator_entity_id, sensor_weights=payload.sensor_weights, sensor_weight_groups=payload.sensor_weight_groups, note=payload.note, ) return record except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @router.post( "/{actuator_entity_id}/related-automations/refresh", response_model=ActuatorRecord, ) def refresh_related_automations( actuator_entity_id: str, request: Request, ) -> ActuatorRecord: job = _start_job( request, kind="automation_refresh", trigger="manual", target=actuator_entity_id, summary="Passende HA-Automationen werden gesucht.", ) try: record = _behavior(request).refresh_related_automations(actuator_entity_id) _finish_job( job, request, status=JobStatus.COMPLETED, summary=f"{len(record.behavior.related_automations)} Automationen gefunden.", ) return record except KeyError as exc: _finish_job(job, request, status=JobStatus.FAILED, summary="Automation-Refresh fehlgeschlagen.", error=str(exc)) raise HTTPException(status_code=404, detail=str(exc)) from exc except (ValueError, HaClientError) as exc: _finish_job(job, request, status=JobStatus.FAILED, summary="Automation-Refresh fehlgeschlagen.", error=str(exc)) raise HTTPException(status_code=409, detail=str(exc)) from exc @router.post( "/{actuator_entity_id}/related-automations/control", response_model=ActuatorRecord, ) def control_related_automation( actuator_entity_id: str, payload: AutomationControlRequest, request: Request, ) -> ActuatorRecord: try: return _behavior(request).set_automation_enabled( actuator_entity_id, payload.automation_entity_id, enabled=payload.enabled, ) except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except (ValueError, HaClientError) as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc @router.get("/reconciliation/state", response_model=ReconciliationState) def get_reconciliation_state(request: Request) -> ReconciliationState: store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Actuator Store nicht initialisiert.", ) return store.load_reconciliation_state() @router.post("/reconciliation/run", response_model=ReconciliationState) def run_reconciliation( request: Request, trigger: str = Query(default="manual", pattern=r"^[a-z0-9_-]{1,32}$"), ) -> ReconciliationState: reconciliation_job = _start_job( request, kind="reconciliation", trigger=trigger, summary="Kontext, Zuordnung und Modelle werden abgeglichen.", ) training_job: JobQueueItem | None = None evaluation_job: JobQueueItem | None = None try: state = _service(request).reconcile_all(trigger=trigger) _finish_job(reconciliation_job, request, status=JobStatus.COMPLETED, summary=state.last_summary) reconciliation_job = None training_job = _start_job( request, kind="training", trigger=trigger, summary="Gelernte Aktorhandlungen werden aktualisiert.", ) _behavior(request).train_all() _finish_job(training_job, request, status=JobStatus.COMPLETED, summary="Training abgeschlossen.") training_job = None evaluation_job = _start_job( request, kind="evaluation", trigger=trigger, summary="Aktuelle Vorhersagen werden neu berechnet.", ) _behavior(request).evaluate_all() _finish_job(evaluation_job, request, status=JobStatus.COMPLETED, summary="Evaluation abgeschlossen.") evaluation_job = None except Exception as exc: for job in [reconciliation_job, training_job, evaluation_job]: if isinstance(job, JobQueueItem) and job.status is JobStatus.RUNNING: _finish_job(job, request, status=JobStatus.FAILED, summary="Job fehlgeschlagen.", error=str(exc)) raise return state @router.get("/job-queue/state", response_model=JobQueueState) def get_job_queue(request: Request) -> JobQueueState: store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Actuator Store nicht initialisiert.", ) return store.load_job_queue() def _start_job( request: Request, *, kind: str, trigger: str, target: str | None = None, summary: str = "", ) -> JobQueueItem | None: store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): return None return store.start_job(kind=kind, trigger=trigger, target=target, summary=summary) def _finish_job( job: JobQueueItem | None, request: Request, *, status: JobStatus, summary: str, error: str | None = None, ) -> None: if job is None: return store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): return store.finish_job(job.job_id, status=status, summary=summary, error=error) def _performance_status(jobs: JobQueueState) -> tuple[int | None, int, str]: budget_ms = 3000 durations = sorted( job.duration_ms for job in jobs.jobs if job.status is JobStatus.COMPLETED and job.duration_ms is not None ) slow_count = sum(1 for duration in durations if duration >= budget_ms) if durations: index = min(len(durations) - 1, int(round((len(durations) - 1) * 0.95))) p95: int | None = durations[index] status_value = "slow" if slow_count else "ok" else: p95 = None status_value = "unknown" if any(job.status is JobStatus.RUNNING for job in jobs.jobs): status_value = "running" if status_value == "unknown" else status_value return p95, slow_count, status_value def _service(request: Request) -> ActuatorReconciliationService: service = getattr(request.app.state, "actuator_service", None) if not isinstance(service, ActuatorReconciliationService): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Actuator-Reconciliation nicht initialisiert.", ) return service def _behavior(request: Request) -> BehaviorEngine: engine = getattr(request.app.state, "behavior_engine", None) if not isinstance(engine, BehaviorEngine): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Verhaltenslernen ist nicht initialisiert.", ) return engine def _validate_weight_payload(payload: WeightOverrideRequest) -> None: for entity_id, weight in payload.sensor_weights.items(): if "." not in entity_id: raise ValueError(f"Ungültige Entity-ID: {entity_id}") if not 0.0 <= weight <= 1.0: raise ValueError(f"Ungültige Gewichtung für {entity_id}: {weight}") for group in payload.sensor_weight_groups: if not group.entity_ids: raise ValueError(f"Gruppe {group.name} enthält keine Entities.") for entity_id in group.entity_ids: if "." not in entity_id: raise ValueError(f"Ungültige Entity-ID in Gruppe {group.name}: {entity_id}") def _build_room_shells( entities: dict[str, HaEntitySummary], discovered: dict[str, DiscoveredEntity], ) -> dict[str, RoomManagementGroup]: rooms: dict[str, RoomManagementGroup] = {} for entity in entities.values(): room = _entity_room(entity) if room not in rooms: rooms[room] = RoomManagementGroup(room=room, actuator_count=0) descriptor = discovered.get(entity.entity_id) if descriptor is None or descriptor.role is EntityRole.ACTUATOR: continue sensor = _entity_management_sensor(entity, descriptor) if sensor is not None: rooms[room].sensors = _merge_room_sensors(rooms[room].sensors, [sensor]) return rooms def _entity_room(entity: HaEntitySummary) -> str: room = entity.area_name or _room_from_text(entity.friendly_name or entity.device_name or entity.entity_id) return room or "Ohne Raum" def _room_from_text(value: str) -> str | None: normalized = value.replace("_", " ").replace("-", " ").strip() if not normalized: return None known_rooms = { "abstellkammer": "Abstellkammer", "abstellraum": "Abstellkammer", "bad": "Bad", "badezimmer": "Bad", "buro": "Büro", "buero": "Büro", "flur": "Flur", "gaeste wc": "Gäste WC", "gaste wc": "Gäste WC", "keller": "Keller", "kuche": "Küche", "kueche": "Küche", "schlafzimmer": "Schlafzimmer", "terrasse": "Terrasse", "wohnbereich": "Wohnbereich", "wohnzimmer": "Wohnbereich", } lowered = normalized.lower() for token, room in known_rooms.items(): if token in lowered: return room return None def _entity_management_sensor( entity: HaEntitySummary, descriptor: DiscoveredEntity, ) -> RoomManagementSensor | None: if descriptor.role not in {EntityRole.MEASUREMENT, EntityRole.BINARY_CONTEXT, EntityRole.CONTEXT}: return None candidate = AssignmentCandidate( entity_id=entity.entity_id, domain=entity.domain, role=descriptor.role, device_class=entity.device_class, state_class=entity.state_class, unit_of_measurement=entity.unit_of_measurement, friendly_name=entity.friendly_name, area_name=entity.area_name, device_name=entity.device_name, score=0.55, confidence=0.55, evidence=["Gehört laut Home Assistant zu diesem Bereich."], ) return _management_sensor( candidate, entity, active=False, optional=descriptor.role is EntityRole.MEASUREMENT, not_required=False, ) def _room_sensors_for_actuator( actuator: HaEntitySummary, sensors: list[RoomManagementSensor], ) -> list[RoomManagementSensor]: preferred = _preferred_sensor_categories(actuator.domain) ranked = sorted( sensors, key=lambda sensor: ( sensor.category not in preferred, preferred.index(sensor.category) if sensor.category in preferred else 99, -sensor.confidence, sensor.friendly_name or sensor.entity_id, ), ) return ranked[:12] def _preferred_sensor_categories(domain: str) -> list[str]: mapping = { "climate": ["Temperatur", "Luftfeuchtigkeit", "Tür/Fenster", "Präsenz", "Energie"], "cover": ["Helligkeit", "Präsenz", "Tür/Fenster", "Temperatur"], "fan": ["Luftfeuchtigkeit", "Präsenz", "Temperatur", "Tür/Fenster", "Energie"], "humidifier": ["Luftfeuchtigkeit", "Temperatur", "Präsenz"], "light": ["Präsenz", "Tür/Fenster", "Helligkeit", "Zone/Person"], "lock": ["Tür/Fenster", "Präsenz", "Zone/Person"], "siren": ["Sicherheit", "Tür/Fenster", "Präsenz"], "switch": ["Präsenz", "Tür/Fenster", "Energie", "Luftfeuchtigkeit", "Helligkeit"], "valve": ["Wasser", "Luftfeuchtigkeit", "Temperatur", "Tür/Fenster"], } return mapping.get(domain, ["Präsenz", "Tür/Fenster", "Energie", "Kontext"]) def _candidate_room(record: ActuatorRecord) -> str | None: for candidate in [*record.context_candidates, *record.numeric_candidates]: if candidate.area_name: return candidate.area_name return None def _rank_management_candidates(record: ActuatorRecord) -> list[AssignmentCandidate]: selected_ids = { entity_id for entity_id in [ record.assignment.selected_numeric_entity_id, *record.assignment.selected_context_entity_ids, ] if entity_id } candidates = { candidate.entity_id: candidate for candidate in [*record.context_candidates, *record.numeric_candidates] } ranked = sorted( candidates.values(), key=lambda item: ( item.entity_id not in selected_ids, _management_sort_group(item), -item.confidence, -item.score, item.entity_id, ), ) return ranked def _management_sort_group(candidate: AssignmentCandidate) -> str: device_class = candidate.device_class or "" if device_class in {"door", "garage_door", "opening", "window"}: return "01_opening" if device_class in {"motion", "occupancy", "presence"}: return "02_presence" if device_class == "illuminance": return "03_brightness" if device_class in {"humidity", "moisture"}: return "04_humidity" if candidate.role is EntityRole.MEASUREMENT: return "08_measurement" return f"20_{candidate.domain}_{device_class}" def _management_sensor( candidate: AssignmentCandidate, entity: HaEntitySummary | None, *, active: bool, optional: bool, not_required: bool, ) -> RoomManagementSensor: if not_required: reason = "Nicht nötig, weil ein Tür-/Öffnungskontakt die Lichtlogik direkt erklärt." elif active: reason = "Wird aktuell für Lernen und Vorhersage verwendet." elif optional: reason = "Optionaler Messwert; nur verwenden, wenn Helligkeit oder Verbrauch wirklich steuern soll." else: reason = ", ".join(candidate.evidence[:2]) or "Naheliegender Kontext aus Raum, Gerät oder Namen." return RoomManagementSensor( entity_id=candidate.entity_id, domain=candidate.domain, role=candidate.role.value, category=_sensor_category_label(candidate), friendly_name=candidate.friendly_name, device_class=candidate.device_class, state=entity.state if entity is not None else None, confidence=candidate.confidence, active=active, optional=optional, not_required=not_required, reason=reason, ) def _sensor_category_label(candidate: AssignmentCandidate) -> str: device_class = candidate.device_class or "" if device_class in {"door", "garage_door", "opening", "window"}: return "Tür/Fenster" if device_class in {"motion", "occupancy", "presence"}: return "Präsenz" if device_class == "illuminance": return "Helligkeit" if device_class in {"humidity", "moisture"}: return "Luftfeuchtigkeit" if device_class == "temperature": return "Temperatur" if device_class in {"power", "energy", "current", "voltage"}: return "Energie" if device_class in {"gas", "water"} or candidate.unit_of_measurement in {"m3", "L", "l"}: return "Wasser" if device_class in {"problem", "safety", "smoke", "vibration"}: return "Sicherheit" if candidate.domain in {"cover"}: return "Rollo/Cover" if candidate.domain in {"zone", "person", "device_tracker"}: return "Zone/Person" return "Kontext" def _merge_room_sensors( existing: list[RoomManagementSensor], incoming: list[RoomManagementSensor], ) -> list[RoomManagementSensor]: by_id = {sensor.entity_id: sensor for sensor in existing} for sensor in incoming: current = by_id.get(sensor.entity_id) if current is None: by_id[sensor.entity_id] = sensor continue by_id[sensor.entity_id] = current.model_copy( update={ "active": current.active or sensor.active, "optional": current.optional and sensor.optional, "not_required": current.not_required and sensor.not_required, "confidence": max(current.confidence, sensor.confidence), } ) return sorted( by_id.values(), key=lambda item: ( not item.active, item.not_required, item.category, item.friendly_name or item.entity_id, ), )[:18] def _prediction_rule_lines( record: ActuatorRecord, candidates: list[AssignmentCandidate], ) -> list[str]: lines = _pattern_rule_lines(record.behavior.patterns) if lines: return lines[:8] selected_contexts = [ candidate for candidate in candidates if candidate.entity_id in set(record.assignment.selected_context_entity_ids) ] result: list[str] = [] for candidate in selected_contexts: label = candidate.friendly_name or candidate.entity_id device_class = candidate.device_class or "" if device_class in {"door", "garage_door", "opening", "window"}: result.extend([ f"{label} geöffnet -> {record.actuator_entity_id} an.", f"{label} geschlossen -> {record.actuator_entity_id} aus.", ]) elif device_class in {"motion", "occupancy", "presence"}: result.extend([ f"{label} erkannt -> {record.actuator_entity_id} an, bei Licht bevorzugt gedimmt.", f"{label} aus -> {record.actuator_entity_id} verzögert ausschalten.", ]) elif device_class in {"humidity", "moisture"}: result.append(f"{label} hoch -> {record.actuator_entity_id} einschalten, bis Feuchte wieder normal ist.") if record.assignment.selected_numeric_entity_id: result.append( f"{record.assignment.selected_numeric_entity_id} nur als Messwert verwenden, nicht als Pflichtsensor." ) return _unique_lines(result)[:8] or ["Noch keine stabile Vorhersage; erst Kontext prüfen und weiter beobachten."] def _suggest_room_actions( *, actuator_id: str, domain: str, sensors: list[RoomManagementSensor], configured: bool, ) -> list[RoomManagementAction]: sensor_categories = {sensor.category for sensor in sensors} sensor_labels = { sensor.category: sensor.friendly_name or sensor.entity_id for sensor in sensors } confidence_base = 0.78 if configured else 0.58 actions: list[RoomManagementAction] = [] def add(category: str, title: str, when: str, then: str, why: str, confidence: float) -> None: actions.append( RoomManagementAction( action_id=f"{actuator_id}:{category}:{len(actions)}", category=category, actuator_entity_id=actuator_id, title=title, when=when, then=then, why=why, confidence=round(min(1.0, confidence), 4), learnable=True, sort_key=f"{category}:{actuator_id}:{len(actions):02d}", ) ) presence = sensor_labels.get("Präsenz") opening = sensor_labels.get("Tür/Fenster") brightness = sensor_labels.get("Helligkeit") humidity = sensor_labels.get("Luftfeuchtigkeit") temperature = sensor_labels.get("Temperatur") energy = sensor_labels.get("Energie") water = sensor_labels.get("Wasser") safety = sensor_labels.get("Sicherheit") zone = sensor_labels.get("Zone/Person") if domain == "light": if opening: add("licht", "Türlicht", f"{opening} öffnet oder schließt", "Licht passend an/aus schalten.", "Türkontakt erklärt kleine Räume ohne Helligkeitssensor.", confidence_base + 0.12) if presence: when = f"{presence} erkennt Anwesenheit" if brightness: when += f" und {brightness} ist dunkel" add("licht", "Präsenzlicht", when, "Licht gedimmt einschalten und bei Abwesenheit verzögert ausschalten.", "Anwesenheit plus Helligkeit vermeidet unnötiges Licht.", confidence_base + (0.12 if brightness else 0.04)) if zone: add("licht", "Zonenstimmung", f"{zone} wird betreten oder verlassen", "Beim Betreten dimmen, beim Aufstehen heller/weiß stellen und später vorherige Stimmung wiederherstellen.", "Zonen wie Sofa brauchen andere Helligkeit als Durchgang oder Aktivität.", confidence_base) elif domain in {"switch", "input_boolean"}: if energy: add("strom", "Verbrauchssteuerung", f"{energy} zeigt Standby oder Last", "Steckdose/Schalter bei Bedarf schalten oder Standby reduzieren.", "Stromwerte zeigen, ob ein Verbraucher wirklich gebraucht wird.", confidence_base + 0.1) if presence: add("strom", "Anwesenheitsschalter", f"{presence} aus", "Verbraucher verzögert ausschalten.", "Schalter und Steckdosen sollen Räume nicht unnötig versorgen.", confidence_base) if opening: add("schalter", "Kontaktlogik", f"{opening} wechselt", "Schalter passend zum Öffnen/Schließen setzen.", "Kontaktzustände sind direkte, leicht prüfbare Auslöser.", confidence_base) elif domain == "climate": if temperature: add("heizung", "Temperaturregelung", f"{temperature} weicht vom Ziel ab", "Heizung nach Lernprofil anpassen.", "Temperaturverlauf und Anwesenheit erklären Heizbedarf.", confidence_base + 0.12) if opening: add("heizung", "Fenster-Offen-Schutz", f"{opening} offen", "Heizung pausieren oder Sollwert senken.", "Offene Fenster/Türen sollen nicht gegen die Heizung arbeiten.", confidence_base + 0.1) if presence: add("heizung", "Anwesenheitswärme", f"{presence} an/aus", "Komforttemperatur nur bei Nutzung halten.", "Anwesenheit macht Heizprofile einfacher und sparsamer.", confidence_base) elif domain in {"fan", "humidifier"}: if humidity: add("belueftung", "Feuchteführung", f"{humidity} steigt oder bleibt hoch", "Lüftung/Entfeuchtung einschalten, später zurücknehmen.", "Feuchtigkeit ist der wichtigste Kontext für Lüftung.", confidence_base + 0.16) if presence: add("belueftung", "Nutzungsabhängige Lüftung", f"{presence} aktiv", "Lüftung leise/bedarfsgerecht führen.", "Nutzung erklärt Gerüche, Feuchte und Komfort.", confidence_base) elif domain == "cover": if brightness: add("rollo", "Sonnen-/Dunkellogik", f"{brightness} sehr hell oder dunkel", "Rollo passend beschatten oder öffnen.", "Helligkeit steuert Blendung, Wärme und Tageslicht.", confidence_base + 0.12) if presence: add("rollo", "Privatsphäre", f"{presence} und Abend/Dunkelheit", "Rollo für Privatsphäre schließen.", "Anwesenheit und Lichtlage erklären Rollo-Bedarf.", confidence_base) elif domain in {"valve"}: if water or humidity: add("wasser", "Wasser-/Leckschutz", f"{water or humidity} auffällig", "Ventil schließen oder Sperre vorschlagen.", "Wasser- und Feuchtesensoren sind Sicherheitskontext.", confidence_base + 0.14) elif domain in {"lock", "siren"}: if opening or safety: add("sicherheit", "Sicherheitszustand", f"{opening or safety} meldet Änderung", "Sicherheitsaktion vorschlagen, aber nicht ohne Freigabe aktiv ausführen.", "Sicherheitsaktionen brauchen hohe Sicherheit und klare Erklärung.", confidence_base) if not actions: add( domain, "Allgemeine Lernregel", "passende Sensoren in diesem Raum ändern sich", "Aktor im Shadow-Modus beobachten und Vorschläge sammeln.", "Noch fehlen eindeutige Kontextsensoren; Discovery prüft den Raum weiter.", max(0.35, confidence_base - 0.18), ) return sorted(actions, key=lambda item: (-item.confidence, item.sort_key))[:8] def _merge_room_actions( existing: list[RoomManagementAction], incoming: list[RoomManagementAction], ) -> list[RoomManagementAction]: by_key = {action.action_id: action for action in existing} for action in incoming: current = by_key.get(action.action_id) if current is None or action.confidence > current.confidence: by_key[action.action_id] = action return sorted(by_key.values(), key=lambda item: (-item.confidence, item.sort_key))[:18] def _action_rule_line(action: RoomManagementAction) -> str: return f"{action.when} -> {action.then}" def _pattern_rule_lines(patterns: list[BehaviorPattern]) -> list[str]: buckets: dict[tuple[str, tuple[tuple[str, str], ...]], int] = {} attrs: dict[tuple[str, tuple[tuple[str, str], ...]], dict[str, object]] = {} for pattern in patterns[-120:]: context = tuple(sorted(pattern.context_states.items())) key = (pattern.target_state, context) buckets[key] = buckets.get(key, 0) + 1 attrs[key] = pattern.target_attributes ordered = sorted(buckets.items(), key=lambda item: (-item[1], item[0])) lines: list[str] = [] for (target_state, context), count in ordered[:8]: conditions = ", ".join(f"{entity}={state}" for entity, state in context[:3]) if not conditions: conditions = "aktueller Zeit-/Nutzungskontext passt" attr_text = _attribute_text(attrs.get((target_state, context), {})) lines.append(f"{conditions} -> {target_state}{attr_text} ({count}x gelernt).") return lines def _attribute_text(attributes: dict[str, object]) -> str: if not attributes: return "" brightness = attributes.get("brightness") if isinstance(brightness, int | float): percent = round(max(0, min(255, float(brightness))) / 255 * 100) return f", Helligkeit {percent} %" return "" def _management_hint(record: ActuatorRecord, has_opening_context: bool) -> str: if has_opening_context and record.actuator_entity_id.startswith(("light.", "switch.")): return "Direkte Türlogik: kein Helligkeitssensor nötig, Sensor und Aktor reichen." if record.behavior.activation_ready: return "Regeln sind lernbereit; vor Aktivierung Vorhersagen prüfen." if record.assignment.review_required: return "Kontext prüfen: Vorschläge übernehmen oder unpassende Sensoren entfernen." return "Weiter beobachten, bis genug eindeutige Schaltbeispiele vorhanden sind." def _unique_lines(lines: list[str]) -> list[str]: seen: set[str] = set() result: list[str] = [] for line in lines: normalized = line.strip() if not normalized or normalized in seen: continue seen.add(normalized) result.append(normalized) return result def _validate_simulation_payload(payload: SimulationRequest) -> None: for entity_id in [ *payload.sensor_states.keys(), *payload.sensor_weights.keys(), *payload.state_options.keys(), ]: if "." not in entity_id: raise ValueError(f"Ungültige Entity-ID: {entity_id}") for entity_id, weight in payload.sensor_weights.items(): if not 0.0 <= weight <= 1.0: raise ValueError(f"Ungültige Gewichtung für {entity_id}: {weight}") for entity_id, states in payload.state_options.items(): if not states: raise ValueError(f"Keine Zustände für {entity_id} angegeben.") def _reconciliation_state_or_default(request: Request) -> ReconciliationState: store = getattr(request.app.state, "actuator_store", None) if not isinstance(store, ActuatorStore): return ReconciliationState() try: return store.load_reconciliation_state() except ValueError: return ReconciliationState(last_summary="Reconciliation-Status ist unlesbar.") def _entity_cache_path(request: Request) -> Path: store = getattr(request.app.state, "actuator_store", None) settings = getattr(request.app.state, "settings", None) if isinstance(store, ActuatorStore): base_dir = store._root elif isinstance(settings, Settings): base_dir = Path(settings.actuator_store).resolve().parent else: base_dir = Path(".").resolve() return Path(os.getenv("SILLYHOME_ENTITY_CACHE", base_dir / "ha_entity_cache.json")) def _load_cached_entities(request: Request) -> list[HaEntitySummary]: payload = _load_entity_cache_payload(request) raw_entities = payload.get("entities", []) if not isinstance(raw_entities, list): return [] try: return [HaEntitySummary.model_validate(entity) for entity in raw_entities] except ValueError: return [] def _load_cached_entity_map( request: Request, entity_ids: set[str], ) -> dict[str, HaEntitySummary]: if not entity_ids: return {} cache = getattr(request.app.state, "dashboard_cache", None) if isinstance(cache, DashboardCache): cached_result = cache.load_entity_map(entity_ids) if cached_result: return cached_result payload = _load_entity_cache_payload(request) raw_entities = payload.get("entities", []) if not isinstance(raw_entities, list): return {} result: dict[str, HaEntitySummary] = {} for raw_entity in raw_entities: if not isinstance(raw_entity, dict): continue entity_id = raw_entity.get("entity_id") if not isinstance(entity_id, str) or entity_id not in entity_ids: continue try: result[entity_id] = HaEntitySummary.model_validate(raw_entity) except ValueError: continue return result def _load_entity_cache_status(request: Request) -> dict[str, object]: cache = getattr(request.app.state, "dashboard_cache", None) if isinstance(cache, DashboardCache): status_payload = cache.load_status() if status_payload.get("entity_count"): return status_payload path = _entity_cache_path(request) if not path.exists(): return {} try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, TypeError, ValueError): return {} if not isinstance(payload, dict): return {} raw_entities = payload.get("entities", []) return { "updated_at": payload.get("updated_at"), "discovery_groups": payload.get("discovery_groups", []), "entity_count": len(raw_entities) if isinstance(raw_entities, list) else 0, } def _load_entity_cache_payload(request: Request) -> dict[str, object]: cache = getattr(request.app.state, "dashboard_cache", None) if isinstance(cache, DashboardCache): payload = cache.load_entities_payload() if payload.get("entities"): return payload path = _entity_cache_path(request) if not path.exists(): return {} try: payload = json.loads(path.read_text(encoding="utf-8")) return payload if isinstance(payload, dict) else {} except (OSError, TypeError, ValueError): return {} def _save_cached_entities(request: Request, entities: list[HaEntitySummary]) -> None: group_payload = _discovery_group_payload(entities) cache = getattr(request.app.state, "dashboard_cache", None) if isinstance(cache, DashboardCache): cache.save_entities_payload( entities=entities, discovery_groups=group_payload, ) path = _entity_cache_path(request) path.parent.mkdir(parents=True, exist_ok=True) payload = { "updated_at": datetime.now(timezone.utc).isoformat(), "discovery_groups": group_payload, "entities": [entity.model_dump(mode="json") for entity in entities], } temporary = path.with_suffix(".json.tmp") temporary.write_text( json.dumps(payload, ensure_ascii=True, sort_keys=True) + "\n", encoding="utf-8", ) os.replace(temporary, path) def _discovery_group_payload(entities: list[HaEntitySummary]) -> list[dict[str, object]]: group_counts: dict[tuple[str, str], int] = {} for entity in discover_entities(entities): key = (entity.category, entity.role.value) group_counts[key] = group_counts.get(key, 0) + 1 return [ {"category": category, "role": role, "count": count} for (category, role), count in sorted(group_counts.items()) ] def _deduplicate_actuator_ids( discovered: list[tuple[str, str]], entities: dict[str, HaEntitySummary], ) -> list[str]: priority = { "light": 0, "cover_shutter": 1, "heating": 2, "lock": 3, "fan": 4, "switch_socket": 5, "button": 6, "helper": 7, } selected: dict[str, tuple[int, str]] = {} for entity_id, category in discovered: entity = entities.get(entity_id) if entity is None: continue key = _actuator_duplicate_key(entity, category) rank = priority.get(category, 50) current = selected.get(key) if current is None or (rank, entity_id) < current: selected[key] = (rank, entity_id) return sorted(entity_id for _, entity_id in selected.values()) def _actuator_duplicate_key(entity: HaEntitySummary, category: str) -> str: if entity.device_id and category in {"light", "switch_socket", "button"}: return f"device:{entity.device_id}:control" if entity.device_name and category in {"light", "switch_socket", "button"}: return f"device-name:{entity.device_name.lower()}:control" return f"entity:{entity.entity_id}" def _likely_context_count( actuator: HaEntitySummary, entities: dict[str, HaEntitySummary], discovered: dict[str, DiscoveredEntity], ) -> int: actuator_tokens = _tokens(actuator) count = 0 for entity in entities.values(): if entity.entity_id == actuator.entity_id: continue descriptor = discovered.get(entity.entity_id) role = descriptor.role if descriptor is not None else None if role not in { EntityRole.MEASUREMENT, EntityRole.BINARY_CONTEXT, EntityRole.CONTEXT, }: continue if entity.device_class not in { "door", "energy", "garage_door", "humidity", "illuminance", "motion", "occupancy", "opening", "power", "presence", "temperature", "window", }: continue same_area = bool( actuator.area_name and entity.area_name and actuator.area_name == entity.area_name ) same_device = bool( actuator.device_id and entity.device_id and actuator.device_id == entity.device_id ) token_match = bool(actuator_tokens.intersection(_tokens(entity))) if same_area or same_device or token_match: count += 1 return count def _tokens(entity: HaEntitySummary) -> set[str]: values = [ entity.entity_id, entity.friendly_name, entity.area_name, entity.device_name, ] tokens: set[str] = set() for value in values: if not value: continue tokens.update(token for token in value.lower().replace("_", " ").split() if len(token) > 2) return tokens