feat(sequences): multi-lane playback and per-lane manual beats
- Add sequence_playback with beat and time advance, zone targeting fixes - Per-lane manual beat routing in beat_driver_route (parallel lanes) - Sequence API, editor JS, fix sequence model filename, tests Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -1,62 +1,80 @@
|
||||
from models.squence import Sequence
|
||||
from models.sequence import Sequence
|
||||
import os
|
||||
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
_PROJECT_DB = os.path.normpath(os.path.join(_HERE, "..", "..", "db", "sequence.json"))
|
||||
|
||||
|
||||
def test_sequence():
|
||||
"""Test Sequence model CRUD operations."""
|
||||
# Clean up any existing test file
|
||||
if os.path.exists("Sequence.json"):
|
||||
os.remove("Sequence.json")
|
||||
|
||||
if os.path.exists(_PROJECT_DB):
|
||||
os.remove(_PROJECT_DB)
|
||||
|
||||
sequences = Sequence()
|
||||
|
||||
|
||||
print("Testing create sequence")
|
||||
sequence_id = sequences.create("test_group", ["preset1", "preset2"])
|
||||
sequence_id = sequences.create("1")
|
||||
print(f"Created sequence with ID: {sequence_id}")
|
||||
assert sequence_id is not None
|
||||
assert sequence_id in sequences
|
||||
|
||||
|
||||
print("\nTesting read sequence")
|
||||
sequence = sequences.read(sequence_id)
|
||||
print(f"Read: {sequence}")
|
||||
assert sequence is not None
|
||||
assert sequence["group_name"] == "test_group"
|
||||
assert len(sequence["presets"]) == 2
|
||||
assert "sequence_duration" in sequence
|
||||
assert "sequence_loop" in sequence
|
||||
|
||||
assert sequence["profile_id"] == "1"
|
||||
assert sequence["steps"] == []
|
||||
assert sequence["lanes"] == [[]]
|
||||
assert sequence.get("lanes_group_ids") == [[]]
|
||||
assert sequence.get("advance_mode") == "time"
|
||||
assert sequence["step_duration_ms"] == 3000
|
||||
assert sequence["loop"] is True
|
||||
assert sequence.get("sequence_transition") == 500
|
||||
|
||||
print("\nTesting update sequence")
|
||||
update_data = {
|
||||
"group_name": "updated_group",
|
||||
"presets": ["preset3", "preset4", "preset5"],
|
||||
"sequence_duration": 5000,
|
||||
"sequence_transition": 1000,
|
||||
"sequence_loop": True,
|
||||
"sequence_repeat_count": 3
|
||||
"name": "updated_seq",
|
||||
"steps": [
|
||||
{"preset_id": "5", "group_ids": ["1"], "beats": 2},
|
||||
{"preset_id": "6", "group_ids": [], "beats": 4},
|
||||
],
|
||||
"lanes_group_ids": [["1"]],
|
||||
"step_duration_ms": 5000,
|
||||
"loop": True,
|
||||
"advance_mode": "beats",
|
||||
}
|
||||
result = sequences.update(sequence_id, update_data)
|
||||
assert result is True
|
||||
updated = sequences.read(sequence_id)
|
||||
assert updated["group_name"] == "updated_group"
|
||||
assert len(updated["presets"]) == 3
|
||||
assert updated["sequence_duration"] == 5000
|
||||
assert updated["sequence_loop"] is True
|
||||
|
||||
assert updated["name"] == "updated_seq"
|
||||
assert len(updated["steps"]) == 2
|
||||
assert updated["steps"][0]["preset_id"] == "5"
|
||||
assert updated["steps"][0]["group_ids"] == ["1"]
|
||||
assert updated["steps"][0].get("beats") == 2
|
||||
assert isinstance(updated.get("lanes"), list)
|
||||
assert len(updated["lanes"]) == 1
|
||||
assert len(updated["lanes"][0]) == 2
|
||||
assert updated["lanes"][0][0]["beats"] == 2
|
||||
assert updated.get("advance_mode") == "beats"
|
||||
assert updated["step_duration_ms"] == 5000
|
||||
assert updated["loop"] is True
|
||||
|
||||
print("\nTesting list sequences")
|
||||
sequence_list = sequences.list()
|
||||
print(f"Sequence list: {sequence_list}")
|
||||
assert sequence_id in sequence_list
|
||||
|
||||
|
||||
print("\nTesting delete sequence")
|
||||
deleted = sequences.delete(sequence_id)
|
||||
assert deleted is True
|
||||
assert sequence_id not in sequences
|
||||
|
||||
|
||||
print("\nTesting read after delete")
|
||||
sequence = sequences.read(sequence_id)
|
||||
assert sequence is None
|
||||
|
||||
|
||||
print("\nAll sequence tests passed!")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
test_sequence()
|
||||
|
||||
@@ -123,7 +123,7 @@ def server(monkeypatch, tmp_path_factory):
|
||||
import models.pallet as models_pallet # noqa: E402
|
||||
import models.scene as models_scene # noqa: E402
|
||||
import models.pattern as models_pattern # noqa: E402
|
||||
import models.squence as models_sequence # noqa: E402
|
||||
import models.sequence as models_sequence # noqa: E402
|
||||
import models.device as models_device # noqa: E402
|
||||
|
||||
for cls in (
|
||||
@@ -527,21 +527,24 @@ def test_groups_sequences_scenes_palettes_patterns_endpoints(server):
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Sequences.
|
||||
unique_seq_group_name = f"pytest-seq-group-{uuid.uuid4().hex[:8]}"
|
||||
unique_seq_name = f"pytest-seq-{uuid.uuid4().hex[:8]}"
|
||||
resp = c.post(
|
||||
f"{base_url}/sequences",
|
||||
json={"group_name": unique_seq_group_name, "presets": []},
|
||||
json={
|
||||
"name": unique_seq_name,
|
||||
"steps": [{"preset_id": "1", "group_ids": []}],
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
sequences_list = c.get(f"{base_url}/sequences").json()
|
||||
seq_id = _find_id_by_field(sequences_list, "group_name", unique_seq_group_name)
|
||||
seq_id = _find_id_by_field(sequences_list, "name", unique_seq_name)
|
||||
|
||||
resp = c.get(f"{base_url}/sequences/{seq_id}")
|
||||
assert resp.status_code == 200
|
||||
|
||||
resp = c.put(f"{base_url}/sequences/{seq_id}", json={"sequence_duration": 1234})
|
||||
resp = c.put(f"{base_url}/sequences/{seq_id}", json={"step_duration_ms": 1234})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["sequence_duration"] == 1234
|
||||
assert resp.json()["step_duration_ms"] == 1234
|
||||
|
||||
resp = c.delete(f"{base_url}/sequences/{seq_id}")
|
||||
assert resp.status_code == 200
|
||||
|
||||
Reference in New Issue
Block a user