feat(02-01): implement ChoreoEvent and ChoreoFile Pydantic v2 models
- ChoreoEvent: validated zone (schrank/wand/all), animation (8 names), timestamp (>= 0) - ChoreoFile: save/load .choreo.json roundtrip, events always sorted by timestamp - add_event() inserts and re-sorts by timestamp (D-02) - Exports VALID_ZONES and VALID_ANIMATIONS constants - 17 tests pass: validators, roundtrip, sorting, constants
This commit is contained in:
96
src/led_sync/models.py
Normal file
96
src/led_sync/models.py
Normal file
@@ -0,0 +1,96 @@
|
|||||||
|
"""Choreography data model for LED Sync Studio.
|
||||||
|
|
||||||
|
Implements ChoreoEvent and ChoreoFile Pydantic v2 models per CONTEXT.md D-01/D-02/D-03.
|
||||||
|
|
||||||
|
Valid zones: schrank, wand, all (from docs/protocol.md)
|
||||||
|
Valid animations: chase, pulse, rainbow, strobe, color_wash, breathe, sparkle, gradient_sweep
|
||||||
|
|
||||||
|
File extension convention: .choreo.json (D-03)
|
||||||
|
Events are always sorted by timestamp (D-02).
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from pydantic import BaseModel, field_validator
|
||||||
|
|
||||||
|
VALID_ZONES: set[str] = {"schrank", "wand", "all"}
|
||||||
|
|
||||||
|
VALID_ANIMATIONS: set[str] = {
|
||||||
|
"chase",
|
||||||
|
"pulse",
|
||||||
|
"rainbow",
|
||||||
|
"strobe",
|
||||||
|
"color_wash",
|
||||||
|
"breathe",
|
||||||
|
"sparkle",
|
||||||
|
"gradient_sweep",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class ChoreoEvent(BaseModel):
|
||||||
|
"""A single choreography event: at `timestamp` seconds, trigger `animation` on `zone`."""
|
||||||
|
|
||||||
|
timestamp: float # seconds from song start; must be >= 0
|
||||||
|
zone: str # "schrank" | "wand" | "all"
|
||||||
|
animation: str # must match ESP32 animation names from docs/protocol.md
|
||||||
|
params: dict # speed, intensity, colors, direction — forwarded as-is to ESP32
|
||||||
|
loop: Optional[int] = None # None = play once; N = repeat N times
|
||||||
|
duration: Optional[float] = None # seconds; None = run until next event
|
||||||
|
|
||||||
|
@field_validator("timestamp")
|
||||||
|
@classmethod
|
||||||
|
def validate_timestamp(cls, v: float) -> float:
|
||||||
|
if v < 0:
|
||||||
|
raise ValueError("timestamp must be >= 0")
|
||||||
|
return v
|
||||||
|
|
||||||
|
@field_validator("zone")
|
||||||
|
@classmethod
|
||||||
|
def validate_zone(cls, v: str) -> str:
|
||||||
|
if v not in VALID_ZONES:
|
||||||
|
raise ValueError(
|
||||||
|
f"Invalid zone: {v!r}. Must be one of {sorted(VALID_ZONES)}"
|
||||||
|
)
|
||||||
|
return v
|
||||||
|
|
||||||
|
@field_validator("animation")
|
||||||
|
@classmethod
|
||||||
|
def validate_animation(cls, v: str) -> str:
|
||||||
|
if v not in VALID_ANIMATIONS:
|
||||||
|
raise ValueError(
|
||||||
|
f"Unknown animation: {v!r}. Must be one of {sorted(VALID_ANIMATIONS)}"
|
||||||
|
)
|
||||||
|
return v
|
||||||
|
|
||||||
|
|
||||||
|
class ChoreoFile(BaseModel):
|
||||||
|
"""A choreography file: song path, optional BPM, and a sorted list of events.
|
||||||
|
|
||||||
|
Events are always kept sorted by timestamp (D-02).
|
||||||
|
Persist with .choreo.json extension (D-03).
|
||||||
|
"""
|
||||||
|
|
||||||
|
song_path: str
|
||||||
|
bpm: Optional[float] = None
|
||||||
|
events: list[ChoreoEvent] = []
|
||||||
|
|
||||||
|
def model_post_init(self, __context: object) -> None:
|
||||||
|
"""Sort events by timestamp on initialization (D-02)."""
|
||||||
|
self.events = sorted(self.events, key=lambda e: e.timestamp)
|
||||||
|
|
||||||
|
def add_event(self, event: ChoreoEvent) -> None:
|
||||||
|
"""Insert an event and re-sort by timestamp."""
|
||||||
|
self.events.append(event)
|
||||||
|
self.events.sort(key=lambda e: e.timestamp)
|
||||||
|
|
||||||
|
def save(self, path: str | Path) -> None:
|
||||||
|
"""Write choreography to a .choreo.json file."""
|
||||||
|
Path(path).write_text(self.model_dump_json(indent=2))
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def load(cls, path: str | Path) -> "ChoreoFile":
|
||||||
|
"""Load a choreography from a .choreo.json file. Events will be sorted."""
|
||||||
|
return cls.model_validate_json(Path(path).read_text())
|
||||||
212
tests/test_models.py
Normal file
212
tests/test_models.py
Normal file
@@ -0,0 +1,212 @@
|
|||||||
|
"""TDD: Tests for ChoreoEvent and ChoreoFile Pydantic v2 models.
|
||||||
|
|
||||||
|
Tests cover:
|
||||||
|
- Field validation (zone, animation, timestamp)
|
||||||
|
- File save/load roundtrip
|
||||||
|
- Event sorting by timestamp
|
||||||
|
- add_event() re-sorts
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
|
||||||
|
# --- RED: these imports will fail until models.py exists ---
|
||||||
|
def import_models():
|
||||||
|
from led_sync.models import ChoreoEvent, ChoreoFile, VALID_ZONES, VALID_ANIMATIONS
|
||||||
|
return ChoreoEvent, ChoreoFile, VALID_ZONES, VALID_ANIMATIONS
|
||||||
|
|
||||||
|
|
||||||
|
class TestChoreoEventValidation:
|
||||||
|
def test_valid_event_creation(self):
|
||||||
|
ChoreoEvent, _, _, _ = import_models()
|
||||||
|
e = ChoreoEvent(
|
||||||
|
timestamp=1.5,
|
||||||
|
zone="wand",
|
||||||
|
animation="chase",
|
||||||
|
params={"speed": 0.5},
|
||||||
|
)
|
||||||
|
assert e.timestamp == 1.5
|
||||||
|
assert e.zone == "wand"
|
||||||
|
assert e.animation == "chase"
|
||||||
|
assert e.params == {"speed": 0.5}
|
||||||
|
assert e.loop is None
|
||||||
|
assert e.duration is None
|
||||||
|
|
||||||
|
def test_all_valid_zones_accepted(self):
|
||||||
|
ChoreoEvent, _, VALID_ZONES, _ = import_models()
|
||||||
|
for zone in VALID_ZONES:
|
||||||
|
e = ChoreoEvent(timestamp=0, zone=zone, animation="pulse", params={})
|
||||||
|
assert e.zone == zone
|
||||||
|
|
||||||
|
def test_invalid_zone_rejected(self):
|
||||||
|
ChoreoEvent, _, _, _ = import_models()
|
||||||
|
with pytest.raises(ValidationError) as exc_info:
|
||||||
|
ChoreoEvent(timestamp=1.0, zone="bedroom", animation="chase", params={})
|
||||||
|
assert "Invalid zone" in str(exc_info.value)
|
||||||
|
|
||||||
|
def test_all_valid_animations_accepted(self):
|
||||||
|
ChoreoEvent, _, _, VALID_ANIMATIONS = import_models()
|
||||||
|
for anim in VALID_ANIMATIONS:
|
||||||
|
e = ChoreoEvent(timestamp=0, zone="wand", animation=anim, params={})
|
||||||
|
assert e.animation == anim
|
||||||
|
|
||||||
|
def test_invalid_animation_rejected(self):
|
||||||
|
ChoreoEvent, _, _, _ = import_models()
|
||||||
|
with pytest.raises(ValidationError) as exc_info:
|
||||||
|
ChoreoEvent(timestamp=1.0, zone="wand", animation="disco", params={})
|
||||||
|
assert "Unknown animation" in str(exc_info.value)
|
||||||
|
|
||||||
|
def test_negative_timestamp_rejected(self):
|
||||||
|
ChoreoEvent, _, _, _ = import_models()
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
ChoreoEvent(timestamp=-1.0, zone="wand", animation="chase", params={})
|
||||||
|
|
||||||
|
def test_zero_timestamp_accepted(self):
|
||||||
|
ChoreoEvent, _, _, _ = import_models()
|
||||||
|
e = ChoreoEvent(timestamp=0.0, zone="schrank", animation="breathe", params={})
|
||||||
|
assert e.timestamp == 0.0
|
||||||
|
|
||||||
|
def test_loop_and_duration_optional(self):
|
||||||
|
ChoreoEvent, _, _, _ = import_models()
|
||||||
|
e = ChoreoEvent(
|
||||||
|
timestamp=2.0,
|
||||||
|
zone="all",
|
||||||
|
animation="rainbow",
|
||||||
|
params={},
|
||||||
|
loop=3,
|
||||||
|
duration=5.0,
|
||||||
|
)
|
||||||
|
assert e.loop == 3
|
||||||
|
assert e.duration == 5.0
|
||||||
|
|
||||||
|
|
||||||
|
class TestChoreoFileBasics:
|
||||||
|
def test_empty_choreofile_creation(self):
|
||||||
|
_, ChoreoFile, _, _ = import_models()
|
||||||
|
cf = ChoreoFile(song_path="test.mp3")
|
||||||
|
assert cf.song_path == "test.mp3"
|
||||||
|
assert cf.bpm is None
|
||||||
|
assert cf.events == []
|
||||||
|
|
||||||
|
def test_choreofile_with_bpm(self):
|
||||||
|
_, ChoreoFile, _, _ = import_models()
|
||||||
|
cf = ChoreoFile(song_path="test.mp3", bpm=120.0)
|
||||||
|
assert cf.bpm == 120.0
|
||||||
|
|
||||||
|
|
||||||
|
class TestChoreoFileSortingAndAdd:
|
||||||
|
def test_events_sorted_on_load(self):
|
||||||
|
ChoreoEvent, ChoreoFile, _, _ = import_models()
|
||||||
|
e1 = ChoreoEvent(timestamp=3.0, zone="wand", animation="chase", params={})
|
||||||
|
e2 = ChoreoEvent(timestamp=1.0, zone="wand", animation="pulse", params={})
|
||||||
|
e3 = ChoreoEvent(timestamp=2.0, zone="schrank", animation="rainbow", params={})
|
||||||
|
cf = ChoreoFile(song_path="test.mp3", events=[e1, e2, e3])
|
||||||
|
timestamps = [e.timestamp for e in cf.events]
|
||||||
|
assert timestamps == sorted(timestamps), "Events must be sorted by timestamp"
|
||||||
|
|
||||||
|
def test_add_event_maintains_sort(self):
|
||||||
|
ChoreoEvent, ChoreoFile, _, _ = import_models()
|
||||||
|
cf = ChoreoFile(song_path="test.mp3")
|
||||||
|
cf.add_event(ChoreoEvent(timestamp=5.0, zone="wand", animation="strobe", params={}))
|
||||||
|
cf.add_event(ChoreoEvent(timestamp=1.0, zone="wand", animation="chase", params={}))
|
||||||
|
cf.add_event(ChoreoEvent(timestamp=3.0, zone="all", animation="pulse", params={}))
|
||||||
|
timestamps = [e.timestamp for e in cf.events]
|
||||||
|
assert timestamps == [1.0, 3.0, 5.0]
|
||||||
|
|
||||||
|
|
||||||
|
class TestChoreoFileRoundtrip:
|
||||||
|
def test_save_creates_json_file(self):
|
||||||
|
ChoreoEvent, ChoreoFile, _, _ = import_models()
|
||||||
|
cf = ChoreoFile(song_path="my_song.mp3", bpm=128.0)
|
||||||
|
e = ChoreoEvent(timestamp=1.5, zone="wand", animation="chase", params={"speed": 0.5})
|
||||||
|
cf.add_event(e)
|
||||||
|
|
||||||
|
with tempfile.NamedTemporaryFile(suffix=".choreo.json", delete=False) as f:
|
||||||
|
tmp = Path(f.name)
|
||||||
|
|
||||||
|
try:
|
||||||
|
cf.save(tmp)
|
||||||
|
assert tmp.exists()
|
||||||
|
data = json.loads(tmp.read_text())
|
||||||
|
assert data["song_path"] == "my_song.mp3"
|
||||||
|
assert data["bpm"] == 128.0
|
||||||
|
assert len(data["events"]) == 1
|
||||||
|
assert data["events"][0]["timestamp"] == 1.5
|
||||||
|
finally:
|
||||||
|
tmp.unlink(missing_ok=True)
|
||||||
|
|
||||||
|
def test_load_roundtrip_preserves_all_fields(self):
|
||||||
|
ChoreoEvent, ChoreoFile, _, _ = import_models()
|
||||||
|
e1 = ChoreoEvent(timestamp=1.5, zone="wand", animation="chase", params={"speed": 0.5})
|
||||||
|
e2 = ChoreoEvent(
|
||||||
|
timestamp=0.5,
|
||||||
|
zone="schrank",
|
||||||
|
animation="strobe",
|
||||||
|
params={"intensity": 0.8},
|
||||||
|
loop=2,
|
||||||
|
duration=3.0,
|
||||||
|
)
|
||||||
|
e3 = ChoreoEvent(timestamp=3.0, zone="all", animation="rainbow", params={})
|
||||||
|
original = ChoreoFile(song_path="test.mp3", bpm=120.0, events=[e1, e2, e3])
|
||||||
|
|
||||||
|
with tempfile.NamedTemporaryFile(suffix=".choreo.json", delete=False) as f:
|
||||||
|
tmp = Path(f.name)
|
||||||
|
|
||||||
|
try:
|
||||||
|
original.save(tmp)
|
||||||
|
loaded = ChoreoFile.load(tmp)
|
||||||
|
assert loaded.song_path == original.song_path
|
||||||
|
assert loaded.bpm == original.bpm
|
||||||
|
assert len(loaded.events) == 3
|
||||||
|
# Events should be sorted: 0.5, 1.5, 3.0
|
||||||
|
assert loaded.events[0].timestamp == 0.5
|
||||||
|
assert loaded.events[1].timestamp == 1.5
|
||||||
|
assert loaded.events[2].timestamp == 3.0
|
||||||
|
assert loaded.events[0].loop == 2
|
||||||
|
assert loaded.events[0].duration == 3.0
|
||||||
|
finally:
|
||||||
|
tmp.unlink(missing_ok=True)
|
||||||
|
|
||||||
|
def test_load_sorts_events_by_timestamp(self):
|
||||||
|
"""Loading a file with unsorted events should produce sorted output."""
|
||||||
|
_, ChoreoFile, _, _ = import_models()
|
||||||
|
# Write raw JSON with unsorted events
|
||||||
|
raw = {
|
||||||
|
"song_path": "test.mp3",
|
||||||
|
"bpm": None,
|
||||||
|
"events": [
|
||||||
|
{"timestamp": 3.0, "zone": "wand", "animation": "chase", "params": {}, "loop": None, "duration": None},
|
||||||
|
{"timestamp": 1.0, "zone": "wand", "animation": "pulse", "params": {}, "loop": None, "duration": None},
|
||||||
|
{"timestamp": 2.0, "zone": "all", "animation": "rainbow", "params": {}, "loop": None, "duration": None},
|
||||||
|
],
|
||||||
|
}
|
||||||
|
with tempfile.NamedTemporaryFile(
|
||||||
|
suffix=".choreo.json", mode="w", delete=False
|
||||||
|
) as f:
|
||||||
|
json.dump(raw, f)
|
||||||
|
tmp = Path(f.name)
|
||||||
|
|
||||||
|
try:
|
||||||
|
loaded = ChoreoFile.load(tmp)
|
||||||
|
timestamps = [e.timestamp for e in loaded.events]
|
||||||
|
assert timestamps == [1.0, 2.0, 3.0]
|
||||||
|
finally:
|
||||||
|
tmp.unlink(missing_ok=True)
|
||||||
|
|
||||||
|
|
||||||
|
class TestExportedConstants:
|
||||||
|
def test_valid_zones_exported(self):
|
||||||
|
_, _, VALID_ZONES, _ = import_models()
|
||||||
|
assert "schrank" in VALID_ZONES
|
||||||
|
assert "wand" in VALID_ZONES
|
||||||
|
assert "all" in VALID_ZONES
|
||||||
|
assert len(VALID_ZONES) == 3
|
||||||
|
|
||||||
|
def test_valid_animations_exported(self):
|
||||||
|
_, _, _, VALID_ANIMATIONS = import_models()
|
||||||
|
expected = {"chase", "pulse", "rainbow", "strobe", "color_wash", "breathe", "sparkle", "gradient_sweep"}
|
||||||
|
assert VALID_ANIMATIONS == expected
|
||||||
Reference in New Issue
Block a user