1264 lines
47 KiB
Python
1264 lines
47 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,
|
|
SensorWeightGroup,
|
|
model_id_for_actuator,
|
|
)
|
|
from app.actuators.store import ActuatorStore
|
|
from app.config import Settings
|
|
from app.ha.discovery import DiscoveredEntity, EntityRole, discover_entities
|
|
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",
|
|
"led",
|
|
"lidl",
|
|
"light",
|
|
"licht",
|
|
"lichtschalter",
|
|
"monitoring",
|
|
"power",
|
|
"sensor",
|
|
"state",
|
|
"switch",
|
|
"temperature",
|
|
"value",
|
|
}
|
|
)
|
|
_GENERIC_AREA_NAMES = frozenset({"energie", "monitoring", "power", "strom", "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",
|
|
"input_boolean",
|
|
"input_number",
|
|
"input_select",
|
|
"light",
|
|
"media_player",
|
|
"person",
|
|
"remote",
|
|
"scene",
|
|
"sensor",
|
|
"sun",
|
|
"switch",
|
|
"weather",
|
|
})
|
|
_CONTEXT_SUGGESTION_LIMIT = 500
|
|
_OUTDOOR_TOKENS = frozenset({"aussen", "außen", "outdoor", "garten", "terrasse", "balkon"})
|
|
_DIAGNOSTIC_TOKENS = frozenset({
|
|
"basic",
|
|
"battery",
|
|
"bytes",
|
|
"connect",
|
|
"count",
|
|
"data",
|
|
"diagnostic",
|
|
"firmware",
|
|
"gesehen",
|
|
"heat",
|
|
"inbytes",
|
|
"interface",
|
|
"last",
|
|
"linkquality",
|
|
"knoten",
|
|
"knotens",
|
|
"mqtt",
|
|
"node",
|
|
"outbytes",
|
|
"pfsense",
|
|
"reason",
|
|
"restart",
|
|
"rssi",
|
|
"signal",
|
|
"ssid",
|
|
"status",
|
|
"overheat",
|
|
"overheating",
|
|
"overload",
|
|
"uptime",
|
|
"vpn",
|
|
"uberhitzung",
|
|
"ueberhitzung",
|
|
"ueberlast",
|
|
"überhitzung",
|
|
"überlast",
|
|
"wifi",
|
|
"zuletzt",
|
|
})
|
|
_AUTO_CONTEXT_CLASSES = frozenset({
|
|
"door",
|
|
"garage_door",
|
|
"illuminance",
|
|
"motion",
|
|
"occupancy",
|
|
"opening",
|
|
"presence",
|
|
"window",
|
|
})
|
|
_PRESENCE_TOKENS = frozenset({
|
|
"besetzt",
|
|
"occupied",
|
|
"occupancy",
|
|
"presence",
|
|
"prasenz",
|
|
"praesenz",
|
|
"motion",
|
|
"bewegung",
|
|
"bewegungsmelder",
|
|
})
|
|
_MAILBOX_TOKENS = frozenset({"briefkasten", "mailbox", "post"})
|
|
_CABINET_TOKENS = frozenset({"schrank", "cabinet"})
|
|
_PV_TOKENS = frozenset({
|
|
"pv",
|
|
"solar",
|
|
"photovoltaik",
|
|
"akku",
|
|
"batterie",
|
|
"battery",
|
|
"einspeisung",
|
|
"wechselrichter",
|
|
"inverter",
|
|
"netzbezug",
|
|
"grid",
|
|
"verbrauch",
|
|
})
|
|
|
|
|
|
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 discover_entities(list(entities.values()))}
|
|
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 _is_diagnostic_context(entity):
|
|
continue
|
|
if not selected and not _has_context_relationship(actuator, entity):
|
|
score = max(score, 0.01)
|
|
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,
|
|
sensor_weights=record.manual_override.sensor_weights if record.manual_override else {},
|
|
sensor_weight_groups=(
|
|
record.manual_override.sensor_weight_groups if record.manual_override else []
|
|
),
|
|
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": _apply_weight_overrides(
|
|
_merge_manual_candidates(
|
|
record.numeric_candidates,
|
|
entities,
|
|
[numeric_entity_id] if numeric_entity_id else [],
|
|
role=EntityRole.MEASUREMENT,
|
|
),
|
|
override,
|
|
),
|
|
"context_candidates": _apply_weight_overrides(
|
|
_merge_manual_candidates(
|
|
record.context_candidates,
|
|
entities,
|
|
selected_context_ids,
|
|
role=EntityRole.CONTEXT,
|
|
),
|
|
override,
|
|
),
|
|
"lifecycle": lifecycle,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
return self._store.upsert(updated)
|
|
|
|
def set_weight_overrides(
|
|
self,
|
|
actuator_entity_id: str,
|
|
*,
|
|
sensor_weights: dict[str, float],
|
|
sensor_weight_groups: list[SensorWeightGroup],
|
|
note: str | None = None,
|
|
) -> ActuatorRecord:
|
|
now = datetime.now(timezone.utc)
|
|
record = self._store.get(actuator_entity_id)
|
|
selected_ids = {
|
|
entity_id
|
|
for entity_id in [
|
|
record.assignment.selected_numeric_entity_id,
|
|
*record.assignment.selected_context_entity_ids,
|
|
]
|
|
if entity_id
|
|
}
|
|
selected_ids.update(sensor_weights)
|
|
for group in sensor_weight_groups:
|
|
selected_ids.update(group.entity_ids)
|
|
entities = {entity.entity_id: entity for entity in self._ha_reader.read_entities()}
|
|
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(sorted(missing))}")
|
|
|
|
previous = record.manual_override
|
|
override = ManualOverride(
|
|
numeric_entity_id=(
|
|
previous.numeric_entity_id
|
|
if previous is not None
|
|
else record.assignment.selected_numeric_entity_id
|
|
),
|
|
context_entity_ids=(
|
|
previous.context_entity_ids
|
|
if previous is not None
|
|
else record.assignment.selected_context_entity_ids
|
|
),
|
|
sensor_weights={entity_id: round(weight, 4) for entity_id, weight in sensor_weights.items()},
|
|
sensor_weight_groups=sensor_weight_groups,
|
|
updated_at=now,
|
|
note=note,
|
|
)
|
|
updated = record.model_copy(
|
|
update={
|
|
"manual_override": override,
|
|
"numeric_candidates": _apply_weight_overrides(
|
|
record.numeric_candidates,
|
|
override,
|
|
),
|
|
"context_candidates": _apply_weight_overrides(
|
|
record.context_candidates,
|
|
override,
|
|
),
|
|
"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,
|
|
)
|
|
if record.manual_override is not None:
|
|
numeric_candidates = _apply_weight_overrides(numeric_candidates, record.manual_override)
|
|
context_candidates = _apply_weight_overrides(context_candidates, record.manual_override)
|
|
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 not None
|
|
and actuator.domain in {"light", "switch"}
|
|
and any(
|
|
(candidate.device_class or "") in {"door", "garage_door", "opening", "window"}
|
|
for candidate in 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=(
|
|
"Tür-/Öffnungskontext automatisch erkannt. Für diese "
|
|
"direkte Schaltlogik ist kein Helligkeitssensor erforderlich."
|
|
),
|
|
)
|
|
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
|
|
)
|
|
can_auto_accept_context = (
|
|
not context or _eligible_for_auto_context(actuator, candidate)
|
|
)
|
|
auto_accepted = (
|
|
can_auto_accept_context
|
|
and 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 ""
|
|
text = " ".join(
|
|
value.lower().replace("_", " ")
|
|
for value in [entity.entity_id, entity.friendly_name, entity.area_name, entity.device_name]
|
|
if value
|
|
)
|
|
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 {"temperature"}:
|
|
return "05_temperature"
|
|
if any(token in text for token in {"pv", "solar", "akku", "batterie", "battery", "einspeisung"}):
|
|
return "06_pv_battery"
|
|
if device_class in {"power", "energy", "current", "voltage"}:
|
|
return "07_power"
|
|
if entity.domain in {"weather"}:
|
|
return "08_weather"
|
|
if entity.domain in {"fan", "humidifier"}:
|
|
return "09_ventilation"
|
|
if entity.domain in {"climate"}:
|
|
return "10_heating"
|
|
if entity.domain in {"cover"}:
|
|
return "11_cover"
|
|
if entity.domain in {"light", "switch"}:
|
|
return "12_states"
|
|
if entity.domain.startswith("input_"):
|
|
return "13_helper"
|
|
if entity.domain in {"person", "device_tracker"}:
|
|
return "14_people"
|
|
return f"20_{entity.domain}_{device_class}"
|
|
|
|
|
|
def _is_diagnostic_context(entity: HaEntitySummary) -> bool:
|
|
tokens = _metadata_tokens(entity, include_stopwords=True)
|
|
return bool(tokens.intersection(_DIAGNOSTIC_TOKENS))
|
|
|
|
|
|
def _has_context_relationship(actuator: HaEntitySummary, entity: HaEntitySummary) -> bool:
|
|
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
|
|
):
|
|
return True
|
|
if actuator.device_id and entity.device_id and actuator.device_id == entity.device_id:
|
|
return True
|
|
if actuator.device_name and entity.device_name and actuator.device_name == entity.device_name:
|
|
return True
|
|
if _metadata_tokens(actuator).intersection(_metadata_tokens(entity)):
|
|
return True
|
|
actuator_tokens = _metadata_tokens(actuator, include_stopwords=True)
|
|
entity_tokens = _metadata_tokens(entity, include_stopwords=True)
|
|
if _is_mailbox_reset_candidate(actuator_tokens, entity_tokens, entity):
|
|
return True
|
|
if actuator.domain in {"fan", "humidifier"} and (
|
|
_is_presence_context(entity) or entity.device_class in {"humidity", "moisture"}
|
|
):
|
|
return True
|
|
if actuator.domain in {"climate", "cover", "fan", "humidifier", "light", "switch"} and (
|
|
entity_tokens.intersection(_PV_TOKENS)
|
|
):
|
|
return True
|
|
entity_tokens = _metadata_tokens(entity, include_stopwords=True)
|
|
return bool(
|
|
entity_tokens.intersection(_OUTDOOR_TOKENS)
|
|
and entity.device_class in {"illuminance", "humidity", "temperature"}
|
|
)
|
|
|
|
|
|
def _eligible_for_auto_context(
|
|
actuator: HaEntitySummary,
|
|
candidate: AssignmentCandidate,
|
|
) -> bool:
|
|
device_class = candidate.device_class or ""
|
|
if device_class in _AUTO_CONTEXT_CLASSES:
|
|
return True
|
|
if actuator.domain in {"fan", "humidifier"} and device_class in {
|
|
"humidity",
|
|
"moisture",
|
|
"temperature",
|
|
}:
|
|
return True
|
|
if actuator.domain in {"fan", "humidifier", "light", "switch"} and _is_presence_candidate(candidate):
|
|
return True
|
|
actuator_tokens = _metadata_tokens(actuator, include_stopwords=True)
|
|
candidate_tokens = _candidate_tokens(candidate, include_stopwords=True)
|
|
if _is_mailbox_reset_candidate(actuator_tokens, candidate_tokens, candidate):
|
|
return True
|
|
if (
|
|
actuator.device_name
|
|
and candidate.device_name
|
|
and actuator.device_name == candidate.device_name
|
|
and candidate.domain in {"light", "switch"}
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
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)
|
|
full_entity_tokens = _metadata_tokens(entity, include_stopwords=True)
|
|
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 context and actuator.domain in {"fan", "humidifier"} and entity.device_class in {
|
|
"humidity",
|
|
"moisture",
|
|
}:
|
|
score += 0.3
|
|
evidence.append("Luftfeuchtigkeit ist primärer Kontext für Lüftung.")
|
|
if not context and actuator.domain in {"fan", "humidifier"} and entity.device_class in {
|
|
"humidity",
|
|
"moisture",
|
|
}:
|
|
score += 0.3
|
|
evidence.append("Luftfeuchtigkeit ist primärer Messwert für Lüftung.")
|
|
if context and actuator.domain in {"fan", "humidifier", "light", "switch"} and _is_presence_context(entity):
|
|
score += 0.3
|
|
evidence.append("Anwesenheit/Belegung ist primärer Schaltkontext.")
|
|
if context and _is_mailbox_reset_candidate(
|
|
_metadata_tokens(actuator, include_stopwords=True),
|
|
_metadata_tokens(entity, include_stopwords=True),
|
|
entity,
|
|
):
|
|
score += 0.45
|
|
evidence.append("Briefkasten-Reset passt zur Schrank-/Entnahme-Tür.")
|
|
if full_entity_tokens.intersection(_PV_TOKENS):
|
|
score += 0.12 if context else 0.18
|
|
evidence.append("PV-/Akku-/Verbrauchswert ist als Energiemanagement-Kontext relevant.")
|
|
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.")
|
|
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:
|
|
evidence = [
|
|
item
|
|
for item in existing.evidence
|
|
if item != "Manuell vom Nutzer als relevant festgelegt."
|
|
]
|
|
by_id[entity_id] = existing.model_copy(
|
|
update={
|
|
"auto_accepted": True,
|
|
"confidence": 1.0,
|
|
"evidence": [
|
|
*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 _apply_weight_overrides(
|
|
candidates: list[AssignmentCandidate],
|
|
override: ManualOverride,
|
|
) -> list[AssignmentCandidate]:
|
|
if not override.sensor_weights and not override.sensor_weight_groups:
|
|
return candidates
|
|
group_weights: dict[str, float] = {}
|
|
for group in override.sensor_weight_groups:
|
|
for entity_id in group.entity_ids:
|
|
group_weights[entity_id] = max(group_weights.get(entity_id, 0.0), group.weight)
|
|
weighted: list[AssignmentCandidate] = []
|
|
for candidate in candidates:
|
|
explicit = override.sensor_weights.get(candidate.entity_id)
|
|
group_weight = group_weights.get(candidate.entity_id)
|
|
manual_weight = explicit if explicit is not None else group_weight
|
|
effective_weight = manual_weight if manual_weight is not None else 1.0
|
|
evidence = [
|
|
item
|
|
for item in candidate.evidence
|
|
if not item.startswith("Manuelle Gewichtung:")
|
|
]
|
|
if manual_weight is not None:
|
|
evidence.append(f"Manuelle Gewichtung: {round(manual_weight * 100)} %.")
|
|
weighted.append(
|
|
candidate.model_copy(
|
|
update={
|
|
"manual_weight": manual_weight,
|
|
"effective_weight": round(effective_weight, 4),
|
|
"evidence": evidence,
|
|
}
|
|
)
|
|
)
|
|
return sorted(weighted, key=lambda item: (-item.confidence * item.effective_weight, item.entity_id))
|
|
|
|
|
|
def _preferred_device_classes(domain: str, *, context: bool) -> frozenset[str]:
|
|
if context:
|
|
mapping = {
|
|
"climate": {"humidity", "illuminance", "occupancy", "presence", "temperature", "window"},
|
|
"cover": {"illuminance", "motion", "occupancy", "presence", "wind_speed"},
|
|
"fan": {"humidity", "moisture", "occupancy", "presence", "temperature"},
|
|
"humidifier": {"humidity", "moisture", "temperature"},
|
|
"light": {"door", "garage_door", "illuminance", "motion", "occupancy", "opening", "presence", "window"},
|
|
"media_player": {"occupancy", "presence"},
|
|
"switch": {"door", "garage_door", "motion", "occupancy", "opening", "presence", "window"},
|
|
}
|
|
return frozenset(
|
|
mapping.get(
|
|
domain,
|
|
{"door", "garage_door", "motion", "occupancy", "opening", "presence"},
|
|
)
|
|
)
|
|
mapping = {
|
|
"climate": {"temperature", "humidity"},
|
|
"cover": {"illuminance", "temperature", "wind_speed"},
|
|
"fan": {"temperature", "humidity", "moisture"},
|
|
"humidifier": {"humidity", "moisture", "temperature"},
|
|
"light": {"illuminance"},
|
|
"media_player": {"power", "energy"},
|
|
"remote": {"battery"},
|
|
"switch": {"power", "energy", "current"},
|
|
"valve": {"temperature", "pressure", "humidity"},
|
|
}
|
|
return frozenset(mapping.get(domain, {"power", "energy", "temperature"}))
|
|
|
|
|
|
def _metadata_tokens(entity: HaEntitySummary, *, include_stopwords: bool = False) -> 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(_normalize_text(value)):
|
|
if (len(token) < 3 and token != "wc") or (not include_stopwords and token in _STOPWORDS):
|
|
continue
|
|
tokens.add(token)
|
|
return _expand_room_tokens(tokens)
|
|
|
|
|
|
def _candidate_tokens(
|
|
candidate: AssignmentCandidate,
|
|
*,
|
|
include_stopwords: bool = False,
|
|
) -> set[str]:
|
|
raw_values = [
|
|
candidate.entity_id,
|
|
candidate.friendly_name,
|
|
candidate.area_name,
|
|
candidate.device_name,
|
|
]
|
|
tokens: set[str] = set()
|
|
for value in raw_values:
|
|
if value is None:
|
|
continue
|
|
for token in _TOKEN_PATTERN.findall(_normalize_text(value)):
|
|
if (len(token) < 3 and token != "wc") or (not include_stopwords and token in _STOPWORDS):
|
|
continue
|
|
tokens.add(token)
|
|
return _expand_room_tokens(tokens)
|
|
|
|
|
|
def _expand_room_tokens(tokens: set[str]) -> set[str]:
|
|
expanded = set(tokens)
|
|
if "gaste" in expanded:
|
|
expanded.add("gaeste")
|
|
if {"gaste", "wc"}.issubset(expanded) or {"gaeste", "wc"}.issubset(expanded):
|
|
expanded.add("gaestewc")
|
|
if {"gaeste", "zimmer"}.issubset(expanded):
|
|
expanded.add("gaestezimmer")
|
|
return expanded
|
|
|
|
|
|
def _normalize_text(value: str) -> str:
|
|
return (
|
|
value.lower()
|
|
.replace("_", " ")
|
|
.replace("ä", "ae")
|
|
.replace("ö", "oe")
|
|
.replace("ü", "ue")
|
|
.replace("ß", "ss")
|
|
)
|
|
|
|
|
|
def _is_presence_context(entity: HaEntitySummary) -> bool:
|
|
if entity.device_class in {"motion", "occupancy", "presence"}:
|
|
return True
|
|
return bool(_metadata_tokens(entity, include_stopwords=True).intersection(_PRESENCE_TOKENS))
|
|
|
|
|
|
def _is_presence_candidate(candidate: AssignmentCandidate) -> bool:
|
|
if candidate.device_class in {"motion", "occupancy", "presence"}:
|
|
return True
|
|
return bool(_candidate_tokens(candidate, include_stopwords=True).intersection(_PRESENCE_TOKENS))
|
|
|
|
|
|
def _is_mailbox_reset_candidate(
|
|
actuator_tokens: set[str],
|
|
context_tokens: set[str],
|
|
entity: HaEntitySummary | AssignmentCandidate,
|
|
) -> bool:
|
|
if not actuator_tokens.intersection(_MAILBOX_TOKENS):
|
|
return False
|
|
if not context_tokens.intersection(_CABINET_TOKENS):
|
|
return False
|
|
return entity.domain == "binary_sensor" and entity.device_class in {
|
|
"door",
|
|
"garage_door",
|
|
"opening",
|
|
}
|
|
|
|
|
|
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."
|