838 lines
32 KiB
Python
838 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import re
|
|
from collections.abc import Iterable
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.actuators.models import (
|
|
ActuatorRecord,
|
|
AssignmentCandidate,
|
|
AssignmentSelection,
|
|
AssignmentSource,
|
|
LifecycleAuditEntry,
|
|
LifecycleStatus,
|
|
ManualOverride,
|
|
ModelLifecycleState,
|
|
ReconciliationState,
|
|
model_id_for_actuator,
|
|
)
|
|
from app.actuators.store import ActuatorStore
|
|
from app.config import Settings
|
|
from app.ha.discovery import DiscoveredEntity, EntityRole
|
|
from app.ha.history import EntityHistorySeries, NumericHistoryPoint
|
|
from app.ha.models import HaEntitySummary
|
|
from app.ha.reader import HaReader
|
|
from app.ml.feature_store import FeatureVector
|
|
from app.ml.registry.model_registry import ModelRegistry
|
|
from app.ml.retraining import retrain_model
|
|
from app.ml.training import TrainedArtifact
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_TOKEN_PATTERN = re.compile(r"[a-z0-9]+", re.IGNORECASE)
|
|
_STOPWORDS = frozenset(
|
|
{
|
|
"actuator",
|
|
"battery",
|
|
"bin",
|
|
"binary",
|
|
"brightness",
|
|
"current",
|
|
"door",
|
|
"energy",
|
|
"entity",
|
|
"humidity",
|
|
"illuminance",
|
|
"light",
|
|
"licht",
|
|
"lichtschalter",
|
|
"monitoring",
|
|
"power",
|
|
"sensor",
|
|
"state",
|
|
"switch",
|
|
"temperature",
|
|
"value",
|
|
}
|
|
)
|
|
_GENERIC_AREA_NAMES = frozenset({"monitoring", "system", "technik"})
|
|
_NUMERIC_AUTO_ACCEPT_SCORE = 0.82
|
|
_NUMERIC_AUTO_ACCEPT_MIN_SCORE = 0.5
|
|
_NUMERIC_MIN_MARGIN = 0.18
|
|
_CONTEXT_AUTO_ACCEPT_SCORE = 0.78
|
|
_CONTEXT_AUTO_ACCEPT_MIN_SCORE = 0.3
|
|
_MAX_CONTEXT_SELECTIONS = 5
|
|
_AUDIT_LIMIT = 20
|
|
_MANUAL_CONTEXT_DOMAINS = frozenset({
|
|
"binary_sensor",
|
|
"climate",
|
|
"cover",
|
|
"device_tracker",
|
|
"fan",
|
|
"humidifier",
|
|
"light",
|
|
"person",
|
|
"sensor",
|
|
"switch",
|
|
"weather",
|
|
})
|
|
_CONTEXT_SUGGESTION_LIMIT = 120
|
|
|
|
|
|
class ActuatorReconciliationService:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
ha_reader: HaReader,
|
|
store: ActuatorStore,
|
|
registry: ModelRegistry,
|
|
settings: Settings,
|
|
) -> None:
|
|
self._ha_reader = ha_reader
|
|
self._store = store
|
|
self._registry = registry
|
|
self._settings = settings
|
|
|
|
def list_configured(self) -> list[ActuatorRecord]:
|
|
return self._store.list()
|
|
|
|
def configure_actuator(self, actuator_entity_id: str, *, enabled: bool = True) -> ActuatorRecord:
|
|
self._store.configure(actuator_entity_id, enabled=enabled)
|
|
return self.reconcile_actuator(actuator_entity_id, trigger="configuration")
|
|
|
|
def get_actuator(self, actuator_entity_id: str) -> ActuatorRecord:
|
|
return self._store.get(actuator_entity_id)
|
|
|
|
def suggest_context_options(
|
|
self,
|
|
actuator_entity_id: str,
|
|
*,
|
|
limit: int = _CONTEXT_SUGGESTION_LIMIT,
|
|
) -> list[HaEntitySummary]:
|
|
entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()}
|
|
discovered = {entity.entity_id: entity for entity in self._ha_reader.discover()}
|
|
actuator = entities.get(actuator_entity_id)
|
|
if actuator is None:
|
|
raise KeyError("Aktuator-Konfiguration nicht gefunden.")
|
|
selected_ids = _selected_context_ids(self._store.get(actuator_entity_id))
|
|
ranked: list[tuple[float, str, HaEntitySummary]] = []
|
|
for entity in entities.values():
|
|
if entity.entity_id == actuator_entity_id or entity.domain not in _MANUAL_CONTEXT_DOMAINS:
|
|
continue
|
|
role = _manual_context_role(entity, discovered.get(entity.entity_id))
|
|
score, _ = _score_candidate(
|
|
actuator,
|
|
entity,
|
|
role,
|
|
context=role is not EntityRole.MEASUREMENT,
|
|
)
|
|
selected = entity.entity_id in selected_ids
|
|
if selected:
|
|
score = max(score, 1.0)
|
|
if not selected and score < 0.1:
|
|
continue
|
|
ranked.append((score, _context_sort_group(entity), entity))
|
|
ranked.sort(
|
|
key=lambda item: (
|
|
-item[0],
|
|
item[1],
|
|
item[2].area_name or "",
|
|
item[2].friendly_name or item[2].entity_id,
|
|
item[2].entity_id,
|
|
)
|
|
)
|
|
return [entity for _, _, entity in ranked[:limit]]
|
|
|
|
def delete_actuator(self, actuator_entity_id: str) -> None:
|
|
model_id = model_id_for_actuator(actuator_entity_id)
|
|
self._registry.archive(model_id)
|
|
self._store.delete(actuator_entity_id)
|
|
|
|
def set_manual_assignment(
|
|
self,
|
|
actuator_entity_id: str,
|
|
*,
|
|
numeric_entity_id: str | None,
|
|
context_entity_ids: list[str],
|
|
note: str | None = None,
|
|
) -> ActuatorRecord:
|
|
now = datetime.now(timezone.utc)
|
|
record = self._store.get(actuator_entity_id)
|
|
entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()}
|
|
actuator = entities.get(actuator_entity_id)
|
|
if actuator is None:
|
|
raise KeyError("Aktuator-Konfiguration nicht gefunden.")
|
|
selected_context_ids = list(dict.fromkeys(context_entity_ids))
|
|
selected_ids = [
|
|
entity_id
|
|
for entity_id in [numeric_entity_id, *selected_context_ids]
|
|
if entity_id
|
|
]
|
|
missing = [entity_id for entity_id in selected_ids if entity_id not in entities]
|
|
if missing:
|
|
raise ValueError(f"Unbekannte Home-Assistant-Entity: {', '.join(missing)}")
|
|
if actuator_entity_id in selected_ids:
|
|
raise ValueError("Der Aktor selbst kann nicht als Kontextsensor verwendet werden.")
|
|
|
|
override = ManualOverride(
|
|
numeric_entity_id=numeric_entity_id,
|
|
context_entity_ids=selected_context_ids,
|
|
updated_at=now,
|
|
note=note,
|
|
)
|
|
assignment = self._manual_assignment(override)
|
|
lifecycle = self._reconcile_lifecycle(
|
|
actuator=actuator,
|
|
assignment=assignment,
|
|
lifecycle=record.lifecycle.model_copy(update={"last_reconciled_at": now}),
|
|
now=now,
|
|
)
|
|
updated = record.model_copy(
|
|
update={
|
|
"assignment": assignment,
|
|
"manual_override": override,
|
|
"numeric_candidates": _merge_manual_candidates(
|
|
record.numeric_candidates,
|
|
entities,
|
|
[numeric_entity_id] if numeric_entity_id else [],
|
|
role=EntityRole.MEASUREMENT,
|
|
),
|
|
"context_candidates": _merge_manual_candidates(
|
|
record.context_candidates,
|
|
entities,
|
|
selected_context_ids,
|
|
role=EntityRole.CONTEXT,
|
|
),
|
|
"lifecycle": lifecycle,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
return self._store.upsert(updated)
|
|
|
|
def reconcile_all(self, trigger: str = "manual") -> ReconciliationState:
|
|
state = self._store.load_reconciliation_state().model_copy(
|
|
update={
|
|
"running": True,
|
|
"last_started_at": datetime.now(timezone.utc),
|
|
"last_trigger": trigger,
|
|
}
|
|
)
|
|
self._store.save_reconciliation_state(state)
|
|
records = self._store.list()
|
|
for record in records:
|
|
self.reconcile_actuator(record.actuator_entity_id, trigger=trigger)
|
|
self._archive_orphan_models({model_id_for_actuator(record.actuator_entity_id) for record in records})
|
|
refreshed = self._store.list()
|
|
summary = ReconciliationState(
|
|
last_started_at=state.last_started_at,
|
|
last_completed_at=datetime.now(timezone.utc),
|
|
last_trigger=trigger,
|
|
running=False,
|
|
configured_actuators=len(refreshed),
|
|
review_required=sum(1 for record in refreshed if record.assignment.review_required),
|
|
trained_models=sum(
|
|
1 for record in refreshed if record.lifecycle.status is LifecycleStatus.TRAINED
|
|
),
|
|
last_summary=(
|
|
f"{len(refreshed)} Aktuatoren geprüft, "
|
|
f"{sum(1 for record in refreshed if record.assignment.review_required)} "
|
|
"mit niedriger Zuordnungssicherheit."
|
|
),
|
|
)
|
|
self._store.save_reconciliation_state(summary)
|
|
return summary
|
|
|
|
def reconcile_actuator(self, actuator_entity_id: str, trigger: str = "manual") -> ActuatorRecord:
|
|
now = datetime.now(timezone.utc)
|
|
record = self._store.get(actuator_entity_id)
|
|
entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()}
|
|
discovered = {entity.entity_id: entity for entity in self._ha_reader.discover()}
|
|
actuator = entities.get(actuator_entity_id)
|
|
descriptor = discovered.get(actuator_entity_id)
|
|
lifecycle = record.lifecycle.model_copy(update={"last_reconciled_at": now})
|
|
|
|
if not record.enabled:
|
|
lifecycle = self._archive_state(
|
|
lifecycle,
|
|
"Aktuator ist deaktiviert; Modell bleibt archiviert.",
|
|
now=now,
|
|
)
|
|
updated = record.model_copy(
|
|
update={
|
|
"assignment": AssignmentSelection(
|
|
selected_numeric_entity_id=None,
|
|
selected_context_entity_ids=[],
|
|
source=AssignmentSource.NONE,
|
|
confidence=0.0,
|
|
review_required=False,
|
|
reason="Aktuator ist deaktiviert.",
|
|
),
|
|
"numeric_candidates": [],
|
|
"context_candidates": [],
|
|
"lifecycle": lifecycle,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
return self._store.upsert(updated)
|
|
|
|
if actuator is None or descriptor is None or descriptor.role is not EntityRole.ACTUATOR:
|
|
lifecycle = self._archive_state(
|
|
lifecycle,
|
|
"Aktuator ist in Home Assistant nicht mehr als Aktor vorhanden.",
|
|
now=now,
|
|
status=LifecycleStatus.ORPHANED,
|
|
)
|
|
updated = record.model_copy(
|
|
update={
|
|
"assignment": AssignmentSelection(
|
|
selected_numeric_entity_id=None,
|
|
selected_context_entity_ids=[],
|
|
source=AssignmentSource.NONE,
|
|
confidence=0.0,
|
|
review_required=True,
|
|
reason="Aktuator fehlt oder ist kein unterstützter Aktor mehr.",
|
|
),
|
|
"numeric_candidates": [],
|
|
"context_candidates": [],
|
|
"lifecycle": lifecycle,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
return self._store.upsert(updated)
|
|
|
|
numeric_candidates = self._rank_candidates(
|
|
actuator=actuator,
|
|
candidates=_filter_candidates(entities, discovered, {EntityRole.MEASUREMENT}),
|
|
context=False,
|
|
)
|
|
context_candidates = self._rank_candidates(
|
|
actuator=actuator,
|
|
candidates=_filter_candidates(
|
|
entities,
|
|
discovered,
|
|
{EntityRole.BINARY_CONTEXT, EntityRole.CONTEXT},
|
|
),
|
|
context=True,
|
|
)
|
|
assignment = (
|
|
self._manual_assignment(record.manual_override)
|
|
if record.manual_override is not None
|
|
else self._select_assignment(
|
|
actuator=actuator,
|
|
numeric_candidates=numeric_candidates,
|
|
context_candidates=context_candidates,
|
|
)
|
|
)
|
|
lifecycle = self._reconcile_lifecycle(
|
|
actuator=actuator,
|
|
assignment=assignment,
|
|
lifecycle=lifecycle,
|
|
now=now,
|
|
)
|
|
updated = record.model_copy(
|
|
update={
|
|
"assignment": assignment,
|
|
"manual_override": record.manual_override,
|
|
"numeric_candidates": numeric_candidates,
|
|
"context_candidates": context_candidates,
|
|
"lifecycle": lifecycle,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
self._store.upsert(updated)
|
|
logger.info(
|
|
"Actuator %s reconciled via %s -> %s",
|
|
actuator_entity_id,
|
|
trigger,
|
|
lifecycle.status,
|
|
)
|
|
return updated
|
|
|
|
@staticmethod
|
|
def _manual_assignment(override: ManualOverride) -> AssignmentSelection:
|
|
selected_context_ids = list(dict.fromkeys(override.context_entity_ids))
|
|
selected_count = len(selected_context_ids) + (1 if override.numeric_entity_id else 0)
|
|
return AssignmentSelection(
|
|
selected_numeric_entity_id=override.numeric_entity_id,
|
|
selected_context_entity_ids=selected_context_ids,
|
|
source=AssignmentSource.MANUAL,
|
|
confidence=1.0 if selected_count else 0.0,
|
|
review_required=selected_count == 0,
|
|
reason=(
|
|
f"Manuell festgelegt: {selected_count} Kontext-Entity(s) werden verwendet."
|
|
if selected_count
|
|
else "Manuelle Zuordnung enthält noch keine Kontext-Entities."
|
|
),
|
|
)
|
|
|
|
def _select_assignment(
|
|
self,
|
|
*,
|
|
actuator: HaEntitySummary,
|
|
numeric_candidates: list[AssignmentCandidate],
|
|
context_candidates: list[AssignmentCandidate],
|
|
) -> AssignmentSelection:
|
|
top_numeric = next(
|
|
(candidate for candidate in numeric_candidates if candidate.auto_accepted),
|
|
None,
|
|
)
|
|
accepted_contexts = [
|
|
candidate
|
|
for candidate in context_candidates
|
|
if candidate.auto_accepted
|
|
][: _MAX_CONTEXT_SELECTIONS]
|
|
top_contexts = [candidate.entity_id for candidate in accepted_contexts]
|
|
if top_numeric is None:
|
|
if accepted_contexts:
|
|
return AssignmentSelection(
|
|
selected_numeric_entity_id=None,
|
|
selected_context_entity_ids=top_contexts,
|
|
source=AssignmentSource.AUTOMATIC,
|
|
confidence=max(candidate.confidence for candidate in accepted_contexts),
|
|
review_required=False,
|
|
reason=(
|
|
"Passender Schaltkontext automatisch erkannt. Für diese "
|
|
"Verhaltensvorhersage ist kein numerischer Sensor erforderlich."
|
|
),
|
|
)
|
|
return AssignmentSelection(
|
|
selected_numeric_entity_id=None,
|
|
selected_context_entity_ids=top_contexts,
|
|
source=AssignmentSource.NONE,
|
|
confidence=0.0,
|
|
review_required=True,
|
|
reason=(
|
|
f"Für {display_name(actuator)} ist noch kein nutzbarer numerischer "
|
|
"Kontext verfügbar. Die Zuordnung wird automatisch erneut geprüft."
|
|
),
|
|
)
|
|
|
|
return AssignmentSelection(
|
|
selected_numeric_entity_id=top_numeric.entity_id,
|
|
selected_context_entity_ids=top_contexts,
|
|
source=AssignmentSource.AUTOMATIC,
|
|
confidence=top_numeric.confidence,
|
|
review_required=not top_numeric.auto_accepted,
|
|
reason=(
|
|
"Kontext automatisch und eindeutig zugeordnet."
|
|
if top_numeric.auto_accepted
|
|
else "Besten verfügbaren Kontext automatisch mit niedriger Sicherheit zugeordnet."
|
|
),
|
|
)
|
|
|
|
def _reconcile_lifecycle(
|
|
self,
|
|
*,
|
|
actuator: HaEntitySummary,
|
|
assignment: AssignmentSelection,
|
|
lifecycle: ModelLifecycleState,
|
|
now: datetime,
|
|
) -> ModelLifecycleState:
|
|
model_id = lifecycle.model_id
|
|
if assignment.selected_numeric_entity_id is None:
|
|
return self._archive_state(
|
|
lifecycle,
|
|
"Ohne numerische Sensorzuordnung wird kein Modell aktiv gehalten.",
|
|
now=now,
|
|
)
|
|
sensor_id = assignment.selected_numeric_entity_id
|
|
series = self._read_history(sensor_id, now)
|
|
points = series.points if series is not None else []
|
|
if len(points) < self._settings.min_training_points:
|
|
return self._with_audit(
|
|
lifecycle.model_copy(
|
|
update={
|
|
"status": LifecycleStatus.PENDING_HISTORY,
|
|
"last_reconciled_at": now,
|
|
"reason": (
|
|
f"{len(points)} von mindestens {self._settings.min_training_points} "
|
|
f"Messpunkten für {sensor_id} vorhanden."
|
|
),
|
|
"next_action": "Historie wird automatisch weiter gesammelt.",
|
|
"last_history_point_count": len(points),
|
|
}
|
|
),
|
|
action="history_wait",
|
|
reason=(
|
|
f"Training für {display_name(actuator)} verschoben: zu wenig numerische Historie."
|
|
),
|
|
now=now,
|
|
)
|
|
|
|
signature = _history_signature(sensor_id, points)
|
|
artifact = self._registry.get_optional(model_id)
|
|
needs_retrain = artifact is None
|
|
retrain_reason = "Noch kein Modell vorhanden."
|
|
if artifact is not None:
|
|
valid, reason = _artifact_valid_for_sensor(artifact, sensor_id)
|
|
if not valid:
|
|
self._registry.archive(model_id)
|
|
needs_retrain = True
|
|
retrain_reason = reason
|
|
elif lifecycle.last_history_signature != signature:
|
|
needs_retrain = True
|
|
retrain_reason = "Historie hat sich seit dem letzten Training materiell geändert."
|
|
elif lifecycle.last_trained_at is None or (
|
|
now - lifecycle.last_trained_at
|
|
) >= timedelta(hours=self._settings.retrain_stale_hours):
|
|
needs_retrain = True
|
|
retrain_reason = "Modell gilt als veraltet und wird präventiv neu trainiert."
|
|
|
|
if needs_retrain:
|
|
vectors = [FeatureVector(sensor_id=sensor_id, values={"value": point.value}) for point in points]
|
|
result = retrain_model(self._registry, model_id, vectors)
|
|
return self._with_audit(
|
|
lifecycle.model_copy(
|
|
update={
|
|
"status": LifecycleStatus.TRAINED,
|
|
"last_reconciled_at": now,
|
|
"last_trained_at": now,
|
|
"last_history_signature": signature,
|
|
"last_history_point_count": len(points),
|
|
"reason": retrain_reason,
|
|
"next_action": "Neue Daten automatisch überwachen und nachtrainieren.",
|
|
}
|
|
),
|
|
action="retrained" if result.replaced else "trained",
|
|
reason=f"{retrain_reason} Modell {model_id} aktualisiert.",
|
|
now=now,
|
|
)
|
|
|
|
return self._with_audit(
|
|
lifecycle.model_copy(
|
|
update={
|
|
"status": LifecycleStatus.TRAINED,
|
|
"last_reconciled_at": now,
|
|
"last_history_signature": signature,
|
|
"last_history_point_count": len(points),
|
|
"reason": "Modell ist aktuell und passt zur automatischen Kontextzuordnung.",
|
|
"next_action": "Neue Historie automatisch auswerten.",
|
|
}
|
|
),
|
|
action="kept",
|
|
reason=f"Modell {model_id} blieb unverändert.",
|
|
now=now,
|
|
)
|
|
|
|
def _read_history(self, sensor_id: str, now: datetime) -> EntityHistorySeries | None:
|
|
start = now - timedelta(days=self._settings.history_days)
|
|
history = list(self._ha_reader.read_history([sensor_id], start, now))
|
|
for series in history:
|
|
if series.entity_id == sensor_id:
|
|
return series
|
|
return None
|
|
|
|
def _archive_orphan_models(self, configured_model_ids: set[str]) -> None:
|
|
for artifact in self._registry.list_models():
|
|
if not artifact.artifact_id.startswith("actuator."):
|
|
continue
|
|
if artifact.artifact_id not in configured_model_ids:
|
|
self._registry.archive(artifact.artifact_id)
|
|
|
|
def _archive_state(
|
|
self,
|
|
lifecycle: ModelLifecycleState,
|
|
reason: str,
|
|
*,
|
|
now: datetime,
|
|
status: LifecycleStatus = LifecycleStatus.ARCHIVED,
|
|
) -> ModelLifecycleState:
|
|
self._registry.archive(lifecycle.model_id)
|
|
return self._with_audit(
|
|
lifecycle.model_copy(
|
|
update={
|
|
"status": status,
|
|
"last_reconciled_at": now,
|
|
"reason": reason,
|
|
"next_action": "Bei neuen Home-Assistant-Daten automatisch erneut zuordnen.",
|
|
}
|
|
),
|
|
action="archived",
|
|
reason=reason,
|
|
now=now,
|
|
)
|
|
|
|
def _rank_candidates(
|
|
self,
|
|
*,
|
|
actuator: HaEntitySummary,
|
|
candidates: Iterable[tuple[HaEntitySummary, DiscoveredEntity]],
|
|
context: bool,
|
|
) -> list[AssignmentCandidate]:
|
|
scored: list[AssignmentCandidate] = []
|
|
all_scores: list[float] = []
|
|
for entity, discovered in candidates:
|
|
score, evidence = _score_candidate(actuator, entity, discovered.role, context=context)
|
|
if score <= 0:
|
|
continue
|
|
all_scores.append(score)
|
|
scored.append(
|
|
AssignmentCandidate(
|
|
entity_id=entity.entity_id,
|
|
domain=entity.domain,
|
|
role=discovered.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=score,
|
|
confidence=0.0,
|
|
evidence=evidence,
|
|
)
|
|
)
|
|
if not scored:
|
|
return []
|
|
highest = max(all_scores)
|
|
sorted_candidates = sorted(scored, key=lambda item: (-item.score, item.entity_id))
|
|
second_score = sorted_candidates[1].score if len(sorted_candidates) > 1 else 0.0
|
|
for index, candidate in enumerate(sorted_candidates):
|
|
confidence = candidate.score / highest if highest else 0.0
|
|
margin = candidate.score - second_score if index == 0 else 0.0
|
|
auto_score = _CONTEXT_AUTO_ACCEPT_SCORE if context else _NUMERIC_AUTO_ACCEPT_SCORE
|
|
minimum_score = (
|
|
_CONTEXT_AUTO_ACCEPT_MIN_SCORE
|
|
if context
|
|
else _NUMERIC_AUTO_ACCEPT_MIN_SCORE
|
|
)
|
|
auto_accepted = (
|
|
candidate.score >= minimum_score
|
|
and confidence >= auto_score
|
|
and (context or margin >= _NUMERIC_MIN_MARGIN)
|
|
)
|
|
sorted_candidates[index] = candidate.model_copy(
|
|
update={
|
|
"confidence": round(confidence, 4),
|
|
"auto_accepted": auto_accepted,
|
|
}
|
|
)
|
|
return sorted_candidates
|
|
|
|
@staticmethod
|
|
def _with_audit(
|
|
lifecycle: ModelLifecycleState,
|
|
*,
|
|
action: str,
|
|
reason: str,
|
|
now: datetime,
|
|
) -> ModelLifecycleState:
|
|
audit = list(lifecycle.audit)
|
|
entry = LifecycleAuditEntry(at=now, action=action, reason=reason)
|
|
if not audit or audit[-1].action != action or audit[-1].reason != reason:
|
|
audit.append(entry)
|
|
if len(audit) > _AUDIT_LIMIT:
|
|
audit = audit[-_AUDIT_LIMIT:]
|
|
return lifecycle.model_copy(update={"audit": audit})
|
|
|
|
|
|
def display_name(entity: HaEntitySummary) -> str:
|
|
return entity.friendly_name or entity.device_name or entity.entity_id
|
|
|
|
|
|
def _filter_candidates(
|
|
entities: dict[str, HaEntitySummary],
|
|
discovered: dict[str, DiscoveredEntity],
|
|
roles: set[EntityRole],
|
|
) -> list[tuple[HaEntitySummary, DiscoveredEntity]]:
|
|
result: list[tuple[HaEntitySummary, DiscoveredEntity]] = []
|
|
for entity_id, summary in entities.items():
|
|
candidate = discovered.get(entity_id)
|
|
if candidate is None or candidate.role not in roles:
|
|
continue
|
|
result.append((summary, candidate))
|
|
return result
|
|
|
|
|
|
def _selected_context_ids(record: ActuatorRecord) -> set[str]:
|
|
result = set(record.assignment.selected_context_entity_ids)
|
|
if record.assignment.selected_numeric_entity_id:
|
|
result.add(record.assignment.selected_numeric_entity_id)
|
|
if record.manual_override is not None:
|
|
result.update(record.manual_override.context_entity_ids)
|
|
if record.manual_override.numeric_entity_id:
|
|
result.add(record.manual_override.numeric_entity_id)
|
|
return result
|
|
|
|
|
|
def _manual_context_role(
|
|
entity: HaEntitySummary,
|
|
discovered: DiscoveredEntity | None,
|
|
) -> EntityRole:
|
|
if discovered is not None and discovered.role is not EntityRole.UNSUPPORTED:
|
|
return discovered.role
|
|
if entity.domain == "sensor":
|
|
return EntityRole.MEASUREMENT
|
|
if entity.domain == "binary_sensor":
|
|
return EntityRole.BINARY_CONTEXT
|
|
return EntityRole.CONTEXT
|
|
|
|
|
|
def _context_sort_group(entity: HaEntitySummary) -> str:
|
|
device_class = entity.device_class or ""
|
|
if device_class in {"motion", "occupancy", "presence"}:
|
|
return "01_presence"
|
|
if device_class in {"illuminance"}:
|
|
return "02_brightness"
|
|
if device_class in {"door", "garage_door", "opening", "window"}:
|
|
return "03_opening"
|
|
if device_class in {"humidity", "moisture"}:
|
|
return "04_humidity"
|
|
if device_class in {"power", "energy", "current", "voltage"}:
|
|
return "05_power"
|
|
if entity.domain in {"light", "switch"}:
|
|
return "06_states"
|
|
return f"20_{entity.domain}_{device_class}"
|
|
|
|
|
|
def _score_candidate(
|
|
actuator: HaEntitySummary,
|
|
entity: HaEntitySummary,
|
|
role: EntityRole,
|
|
*,
|
|
context: bool,
|
|
) -> tuple[float, list[str]]:
|
|
evidence: list[str] = []
|
|
score = 0.0
|
|
actuator_tokens = _metadata_tokens(actuator)
|
|
entity_tokens = _metadata_tokens(entity)
|
|
overlap = sorted(actuator_tokens.intersection(entity_tokens))
|
|
if overlap:
|
|
score += min(0.4, 0.1 * len(overlap))
|
|
evidence.append(f"Gemeinsame Tokens: {', '.join(overlap[:4])}")
|
|
if (
|
|
actuator.area_name
|
|
and entity.area_name
|
|
and actuator.area_name == entity.area_name
|
|
and actuator.area_name.lower() not in _GENERIC_AREA_NAMES
|
|
):
|
|
score += 0.35
|
|
evidence.append(f"Gleicher Bereich: {actuator.area_name}")
|
|
if actuator.device_id and entity.device_id and actuator.device_id == entity.device_id:
|
|
score += 0.2
|
|
evidence.append("Gleiche Home-Assistant-Geräte-ID")
|
|
if actuator.device_name and entity.device_name and actuator.device_name == entity.device_name:
|
|
score += 0.15
|
|
evidence.append(f"Gleicher Gerätename: {actuator.device_name}")
|
|
if actuator.friendly_name and entity.friendly_name and actuator.friendly_name == entity.friendly_name:
|
|
score += 0.1
|
|
evidence.append("Gleicher Friendly Name")
|
|
preferred_device_classes = _preferred_device_classes(actuator.domain, context=context)
|
|
if entity.device_class in preferred_device_classes:
|
|
score += 0.2
|
|
evidence.append(f"Passende device_class: {entity.device_class}")
|
|
if not context and actuator.domain == "light" and entity.device_class == "illuminance":
|
|
score += 0.2
|
|
evidence.append("Beleuchtungsstärke wird für Lichtaktoren bevorzugt.")
|
|
if not context and entity.unit_of_measurement is not None:
|
|
score += 0.05
|
|
evidence.append(f"Numerische Einheit vorhanden: {entity.unit_of_measurement}")
|
|
if context and role is EntityRole.BINARY_CONTEXT:
|
|
score += 0.05
|
|
evidence.append("Binärer Kontextsensor bevorzugt für Zusatzkontext.")
|
|
outdoor_tokens = {"aussen", "außen", "outdoor", "garten", "terrasse", "balkon"}
|
|
if entity_tokens.intersection(outdoor_tokens) and entity.device_class in {
|
|
"illuminance",
|
|
"humidity",
|
|
"temperature",
|
|
}:
|
|
score += 0.1
|
|
evidence.append("Außenmesswert ist oft als übergreifender Kontext relevant.")
|
|
return round(min(score, 1.0), 4), evidence
|
|
|
|
|
|
def _merge_manual_candidates(
|
|
candidates: list[AssignmentCandidate],
|
|
entities: dict[str, HaEntitySummary],
|
|
selected_entity_ids: list[str],
|
|
*,
|
|
role: EntityRole,
|
|
) -> list[AssignmentCandidate]:
|
|
by_id = {candidate.entity_id: candidate for candidate in candidates}
|
|
for entity_id in selected_entity_ids:
|
|
existing = by_id.get(entity_id)
|
|
if existing is not None:
|
|
by_id[entity_id] = existing.model_copy(
|
|
update={
|
|
"auto_accepted": True,
|
|
"confidence": 1.0,
|
|
"evidence": [
|
|
*existing.evidence,
|
|
"Manuell vom Nutzer als relevant festgelegt.",
|
|
],
|
|
}
|
|
)
|
|
continue
|
|
entity = entities.get(entity_id)
|
|
if entity is None:
|
|
continue
|
|
by_id[entity_id] = AssignmentCandidate(
|
|
entity_id=entity.entity_id,
|
|
domain=entity.domain,
|
|
role=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=1.0,
|
|
confidence=1.0,
|
|
auto_accepted=True,
|
|
evidence=["Manuell vom Nutzer als relevant festgelegt."],
|
|
)
|
|
return sorted(by_id.values(), key=lambda item: (-item.confidence, item.entity_id))
|
|
|
|
|
|
def _preferred_device_classes(domain: str, *, context: bool) -> frozenset[str]:
|
|
if context:
|
|
return frozenset({"door", "garage_door", "motion", "occupancy", "opening", "presence"})
|
|
mapping = {
|
|
"climate": {"temperature", "humidity", "power"},
|
|
"cover": {"illuminance", "temperature", "wind_speed"},
|
|
"fan": {"temperature", "humidity", "power"},
|
|
"humidifier": {"humidity", "temperature", "power"},
|
|
"light": {"illuminance", "power", "energy"},
|
|
"switch": {"power", "energy", "current"},
|
|
"valve": {"temperature", "pressure", "humidity"},
|
|
}
|
|
return frozenset(mapping.get(domain, {"power", "energy", "temperature"}))
|
|
|
|
|
|
def _metadata_tokens(entity: HaEntitySummary) -> set[str]:
|
|
raw_values = [
|
|
entity.entity_id,
|
|
entity.friendly_name,
|
|
entity.area_name,
|
|
entity.device_name,
|
|
]
|
|
tokens: set[str] = set()
|
|
for value in raw_values:
|
|
if value is None:
|
|
continue
|
|
for token in _TOKEN_PATTERN.findall(value.lower().replace("_", " ")):
|
|
if len(token) < 3 or token in _STOPWORDS:
|
|
continue
|
|
tokens.add(token)
|
|
return tokens
|
|
|
|
|
|
def _history_signature(sensor_id: str, points: list[NumericHistoryPoint]) -> str:
|
|
digest = hashlib.sha256()
|
|
digest.update(sensor_id.encode("utf-8"))
|
|
for point in points:
|
|
digest.update(point.timestamp.isoformat().encode("utf-8"))
|
|
digest.update(f"{point.value:.6f}".encode("utf-8"))
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _artifact_valid_for_sensor(artifact: TrainedArtifact, sensor_id: str) -> tuple[bool, str]:
|
|
if sensor_id not in artifact.supported_sensors:
|
|
return False, "Vorhandenes Modell passt nicht mehr zur aktuellen Sensorzuordnung."
|
|
feature_models = artifact.feature_models.get(sensor_id, {})
|
|
if "value" not in feature_models:
|
|
return False, "Vorhandenes Modell enthält kein numerisches Trainingsmerkmal 'value'."
|
|
return True, "Modell ist kompatibel."
|