Add adaptive learning and model rollback
Some checks failed
quality / test (3.11) (push) Has been cancelled
quality / test (3.13) (push) Has been cancelled

This commit is contained in:
2026-06-17 18:41:03 +02:00
parent 0101596e93
commit 2ec2c64cba
11 changed files with 465 additions and 4 deletions

View File

@@ -7,6 +7,7 @@ from pathlib import Path
from fastapi.testclient import TestClient
from app.actuators.lifecycle import ActuatorReconciliationService
from app.actuators.models import ModelSnapshot
from app.actuators.store import ActuatorStore
from app.behavior.engine import BehaviorEngine
from app.config import Settings
@@ -113,6 +114,7 @@ def _install_service(tmp_path: Path) -> None:
unit_of_measurement="lx",
friendly_name="Abstellkammer Helligkeit",
area_name="Abstellkammer",
state="12",
),
HaEntitySummary(
entity_id="binary_sensor.abstellkammer_motion",
@@ -120,6 +122,7 @@ def _install_service(tmp_path: Path) -> None:
device_class="motion",
friendly_name="Abstellkammer Bewegung",
area_name="Abstellkammer",
state="off",
),
HaEntitySummary(
entity_id="sensor.pfsense_interface_vpn_inbytes",
@@ -300,6 +303,54 @@ def test_safety_profile_can_block_actuator_manually(tmp_path: Path) -> None:
assert payload["behavior"]["safety"]["cooldown_seconds"] == 120
def test_feedback_adapts_sensor_weights_and_model_can_rollback(tmp_path: Path) -> None:
with TestClient(app) as client:
_install_service(tmp_path)
client.post(
"/v1/actuators",
json={"actuator_entity_id": "light.abstellkammer"},
)
record = app.state.actuator_store.get("light.abstellkammer")
version_id = "model-test"
snapshot = ModelSnapshot(
version_id=version_id,
sample_count=1,
high_confidence_sample_count=1,
average_confidence=0.9,
patterns=[],
reason="Test-Snapshot",
)
app.state.actuator_store.upsert(
record.model_copy(
update={
"behavior": record.behavior.model_copy(
update={
"model_snapshots": [snapshot],
"active_model_version": "model-current",
"sample_count": 2,
}
)
}
)
)
feedback = client.post(
"/v1/actuators/light.abstellkammer/feedback",
json={"correct": False, "expected_state": "off"},
)
rollback = client.post(
"/v1/actuators/light.abstellkammer/model/rollback",
json={"version_id": version_id},
)
assert feedback.status_code == 200
feedback_payload = feedback.json()
assert feedback_payload["behavior"]["adaptive_weight_updates"]
assert feedback_payload["manual_override"]["sensor_weights"]
assert rollback.status_code == 200
assert rollback.json()["behavior"]["active_model_version"] == version_id
def test_summary_is_lightweight_and_uses_cached_entity_metadata(tmp_path: Path) -> None:
with TestClient(app) as client:
_install_service(tmp_path)