Otto/sillyhome-next-dev
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat: add manual context assignment and fix actuator discovery

Browse Source
This commit is contained in:
Otto
2026-06-14 23:15:00 +02:00
parent d87d3abc00
commit 09e14689a3
11 changed files with 380 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
app/actuators/lifecycle.py
View File

@@ -13,6 +13,7 @@ from app.actuators.models import (
AssignmentSource,
LifecycleAuditEntry,
LifecycleStatus,
ManualOverride,
ModelLifecycleState,
ReconciliationState,
model_id_for_actuator,
@@ -47,6 +48,7 @@ _STOPWORDS = frozenset(
"light",
"licht",
"lichtschalter",
"monitoring",
"power",
"sensor",
"state",
@@ -55,6 +57,7 @@ _STOPWORDS = frozenset(
"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
@@ -93,6 +96,67 @@ class ActuatorReconciliationService:
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={
@@ -198,10 +262,14 @@ class ActuatorReconciliationService:
),
context=True,
)
assignment = self._select_assignment(
actuator=actuator,
numeric_candidates=numeric_candidates,
context_candidates=context_candidates,
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,
@@ -212,7 +280,7 @@ class ActuatorReconciliationService:
updated = record.model_copy(
update={
"assignment": assignment,
"manual_override": None,
"manual_override": record.manual_override,
"numeric_candidates": numeric_candidates,
"context_candidates": context_candidates,
"lifecycle": lifecycle,
@@ -228,6 +296,23 @@ class ActuatorReconciliationService:
)
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,
*,
@@ -522,7 +607,12 @@ def _score_candidate(
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:
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:
@@ -550,6 +640,49 @@ def _score_candidate(
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"})