alexei.dolgolyov
669ae20824
Add an opt-out `normalize` flag to the four magnitude value sources (ha_entity, http, system_metrics, game_event). get_value() stays in [0,1] for every source (the normalized scalar-bus invariant), so all existing consumers — brightness sinks, gradient_map, template `name` bindings, and color-strip bindable floats — are unaffected. - normalize=False is a clamp-passthrough: skip the min/max rescale and clamp the raw reading into [0,1] (for sources already reporting a 0..1 fraction). The un-normalized magnitude stays available via get_raw_value() / template raw[name] / automations. - Add get_raw_value() + a raw channel to GameEventValueStream (it had none). game_event's flag is model/stream-level only (no CRUD schema). - Finite-safe clamp01() util; harden the composite-layer brightness multiply (latent negative-wrap / >=1 skip) with it. - Preview WebSocket: tolerate non-numeric raw, generalize raw-range. - Frontend: settings-toggle slider per HA/system_metrics/http editor with min/max grey-out; toggle hidden for fixed-mapping percent metrics. en/ru/zh locale keys. - Additive optional field (default True) — JSON round-trip, no migration. Tests: store create/update round-trip, clamp-passthrough, live normalize flip, game_event raw channel + build_stream forwarding, and finite-safe clamp01.
320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""Tests for ValueSourceStore — CRUD for all source types."""
|
|
|
|
import pytest
|
|
|
|
from ledgrab.storage.value_source import (
|
|
AdaptiveValueSource,
|
|
AnimatedValueSource,
|
|
AudioValueSource,
|
|
DaylightValueSource,
|
|
HAEntityValueSource,
|
|
HTTPValueSource,
|
|
StaticValueSource,
|
|
SystemMetricsValueSource,
|
|
ValueSource,
|
|
)
|
|
from ledgrab.storage.value_source_store import ValueSourceStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_db) -> ValueSourceStore:
|
|
return ValueSourceStore(tmp_db)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ValueSource model round-trips
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceModels:
|
|
def test_static_round_trip(self):
|
|
data = {
|
|
"id": "vs_1",
|
|
"name": "Static",
|
|
"source_type": "static",
|
|
"value": 0.75,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, StaticValueSource)
|
|
assert src.value == 0.75
|
|
|
|
restored = ValueSource.from_dict(src.to_dict())
|
|
assert restored.value == 0.75
|
|
|
|
def test_animated_round_trip(self):
|
|
data = {
|
|
"id": "vs_2",
|
|
"name": "Wave",
|
|
"source_type": "animated",
|
|
"waveform": "triangle",
|
|
"speed": 30.0,
|
|
"min_value": 0.2,
|
|
"max_value": 0.8,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AnimatedValueSource)
|
|
assert src.waveform == "triangle"
|
|
assert src.speed == 30.0
|
|
|
|
def test_audio_round_trip(self):
|
|
data = {
|
|
"id": "vs_3",
|
|
"name": "Audio",
|
|
"source_type": "audio",
|
|
"audio_source_id": "as_1",
|
|
"mode": "peak",
|
|
"sensitivity": 2.0,
|
|
"smoothing": 0.5,
|
|
"auto_gain": True,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AudioValueSource)
|
|
assert src.mode == "peak"
|
|
assert src.auto_gain is True
|
|
|
|
def test_adaptive_time_round_trip(self):
|
|
data = {
|
|
"id": "vs_4",
|
|
"name": "Time",
|
|
"source_type": "adaptive_time",
|
|
"schedule": [
|
|
{"time": "00:00", "value": 0.1},
|
|
{"time": "12:00", "value": 1.0},
|
|
],
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AdaptiveValueSource)
|
|
assert len(src.schedule) == 2
|
|
|
|
def test_adaptive_scene_round_trip(self):
|
|
data = {
|
|
"id": "vs_5",
|
|
"name": "Scene",
|
|
"source_type": "adaptive_scene",
|
|
"picture_source_id": "ps_1",
|
|
"scene_behavior": "match",
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, AdaptiveValueSource)
|
|
assert src.scene_behavior == "match"
|
|
|
|
def test_daylight_round_trip(self):
|
|
data = {
|
|
"id": "vs_6",
|
|
"name": "Daylight",
|
|
"source_type": "daylight",
|
|
"speed": 2.0,
|
|
"use_real_time": True,
|
|
"latitude": 55.0,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, DaylightValueSource)
|
|
assert src.use_real_time is True
|
|
assert src.latitude == 55.0
|
|
|
|
def test_unknown_type_defaults_to_static(self):
|
|
data = {
|
|
"id": "vs_u",
|
|
"name": "Unknown",
|
|
"source_type": "unknown_future",
|
|
"value": 0.5,
|
|
"created_at": "2025-01-01T00:00:00+00:00",
|
|
"updated_at": "2025-01-01T00:00:00+00:00",
|
|
}
|
|
src = ValueSource.from_dict(data)
|
|
assert isinstance(src, StaticValueSource)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ValueSourceStore CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceStoreCRUD:
|
|
def test_create_static(self, store):
|
|
s = store.create_source(name="S1", source_type="static", value=0.5)
|
|
assert s.id.startswith("vs_")
|
|
assert isinstance(s, StaticValueSource)
|
|
assert s.value == 0.5
|
|
assert store.count() == 1
|
|
|
|
def test_create_animated(self, store):
|
|
s = store.create_source(
|
|
name="A1",
|
|
source_type="animated",
|
|
waveform="sawtooth",
|
|
speed=20.0,
|
|
)
|
|
assert isinstance(s, AnimatedValueSource)
|
|
assert s.waveform == "sawtooth"
|
|
assert s.speed == 20.0
|
|
|
|
def test_create_audio(self, store):
|
|
s = store.create_source(
|
|
name="Au1",
|
|
source_type="audio",
|
|
audio_source_id="as_1",
|
|
mode="beat",
|
|
)
|
|
assert isinstance(s, AudioValueSource)
|
|
assert s.mode == "beat"
|
|
|
|
def test_create_adaptive_time(self, store):
|
|
schedule = [
|
|
{"time": "08:00", "value": 0.5},
|
|
{"time": "20:00", "value": 1.0},
|
|
]
|
|
s = store.create_source(
|
|
name="AT",
|
|
source_type="adaptive_time",
|
|
schedule=schedule,
|
|
)
|
|
assert isinstance(s, AdaptiveValueSource)
|
|
assert len(s.schedule) == 2
|
|
|
|
def test_create_adaptive_time_insufficient_schedule(self, store):
|
|
with pytest.raises(ValueError, match="at least 2 points"):
|
|
store.create_source(
|
|
name="Bad",
|
|
source_type="adaptive_time",
|
|
schedule=[{"time": "12:00", "value": 0.5}],
|
|
)
|
|
|
|
def test_create_daylight(self, store):
|
|
s = store.create_source(
|
|
name="DL",
|
|
source_type="daylight",
|
|
speed=2.0,
|
|
use_real_time=True,
|
|
latitude=48.0,
|
|
)
|
|
assert isinstance(s, DaylightValueSource)
|
|
assert s.use_real_time is True
|
|
|
|
def test_create_invalid_type(self, store):
|
|
with pytest.raises(ValueError, match="Invalid source type"):
|
|
store.create_source(name="Bad", source_type="invalid")
|
|
|
|
def test_get_all(self, store):
|
|
store.create_source("A", "static")
|
|
store.create_source("B", "static")
|
|
assert len(store.get_all_sources()) == 2
|
|
|
|
def test_get(self, store):
|
|
created = store.create_source("Get", "static", value=0.3)
|
|
got = store.get_source(created.id)
|
|
assert got.name == "Get"
|
|
|
|
def test_delete(self, store):
|
|
s = store.create_source("Del", "static")
|
|
store.delete_source(s.id)
|
|
assert store.count() == 0
|
|
|
|
def test_update_static(self, store):
|
|
s = store.create_source("Stat", "static", value=0.5)
|
|
updated = store.update_source(s.id, value=0.9)
|
|
assert isinstance(updated, StaticValueSource)
|
|
assert updated.value == 0.9
|
|
|
|
def test_update_name(self, store):
|
|
s = store.create_source("Old", "static")
|
|
updated = store.update_source(s.id, name="New")
|
|
assert updated.name == "New"
|
|
|
|
def test_update_animated_fields(self, store):
|
|
s = store.create_source("Anim", "animated", waveform="sine", speed=10.0)
|
|
updated = store.update_source(s.id, waveform="square", speed=30.0)
|
|
assert isinstance(updated, AnimatedValueSource)
|
|
assert updated.waveform == "square"
|
|
assert updated.speed == 30.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# normalize flag — full store create/update path (factory builder + applier)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceNormalizeStoreRoundTrip:
|
|
"""The ``normalize`` flag must survive the real write path —
|
|
create_source -> CREATE_BUILDERS and update_source -> UPDATE_APPLIERS —
|
|
not just dataclass from_dict/to_dict. A dropped builder/applier line would
|
|
silently revert/ignore the flag while every dataclass-level test stayed
|
|
green, so these exercise the factory layer end-to-end (game_event is
|
|
intentionally excluded — it has no store/CRUD path).
|
|
"""
|
|
|
|
_CASES = [
|
|
("ha_entity", {"ha_source_id": "ha1", "entity_id": "sensor.temp"}, HAEntityValueSource),
|
|
("http", {"http_endpoint_id": "ep1"}, HTTPValueSource),
|
|
("system_metrics", {"metric": "cpu_load"}, SystemMetricsValueSource),
|
|
]
|
|
|
|
@pytest.mark.parametrize("source_type,kwargs,cls", _CASES)
|
|
def test_create_normalize_false_persists(self, store, source_type, kwargs, cls):
|
|
s = store.create_source(
|
|
name=f"{source_type}_n", source_type=source_type, normalize=False, **kwargs
|
|
)
|
|
assert isinstance(s, cls)
|
|
assert s.normalize is False
|
|
# Re-read exercises the SQLite blob persistence round-trip as well.
|
|
assert store.get_source(s.id).normalize is False
|
|
|
|
@pytest.mark.parametrize("source_type,kwargs,cls", _CASES)
|
|
def test_create_normalize_defaults_true(self, store, source_type, kwargs, cls):
|
|
s = store.create_source(name=f"{source_type}_d", source_type=source_type, **kwargs)
|
|
assert s.normalize is True
|
|
|
|
@pytest.mark.parametrize("source_type,kwargs,cls", _CASES)
|
|
def test_update_normalize_flip(self, store, source_type, kwargs, cls):
|
|
s = store.create_source(
|
|
name=f"{source_type}_u", source_type=source_type, normalize=False, **kwargs
|
|
)
|
|
assert store.update_source(s.id, normalize=True).normalize is True
|
|
assert store.update_source(s.id, normalize=False).normalize is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Name uniqueness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourceNameUniqueness:
|
|
def test_duplicate_name(self, store):
|
|
store.create_source("Same", "static")
|
|
with pytest.raises(ValueError, match="already exists"):
|
|
store.create_source("Same", "animated")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValueSourcePersistence:
|
|
def test_persist_and_reload(self, tmp_path):
|
|
from ledgrab.storage.database import Database
|
|
|
|
db = Database(tmp_path / "vs_persist.db")
|
|
s1 = ValueSourceStore(db)
|
|
src = s1.create_source("Persist", "static", value=0.42)
|
|
sid = src.id
|
|
|
|
s2 = ValueSourceStore(db)
|
|
loaded = s2.get_source(sid)
|
|
assert loaded.name == "Persist"
|
|
assert isinstance(loaded, StaticValueSource)
|
|
assert loaded.value == 0.42
|
|
db.close()
|