feat(02-06): wire all Phase 2 modules into asyncio REPL entry point
- Create src/led_sync/main.py with AppState, handle_command, repl, async_main, main - Implements all D-13 REPL commands: play/pause/resume/seek/stop/load/add/save/status/beat/help/quit - Create src/led_sync/__main__.py for python -m led_sync invocation - Add led-sync script entry point to pyproject.toml - Wires AudioPlayer, ChoreographyScheduler, BeatDetector, ESP32Transport
This commit is contained in:
Claude
@@ -15,6 +15,9 @@ dependencies = [
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project.scripts]
|
||||
led-sync = "led_sync.main:main"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/led_sync"]
|
||||
|
||||
|
||||
3
src/led_sync/__main__.py
Normal file
3
src/led_sync/__main__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from led_sync.main import main
|
||||
|
||||
main()
|
||||
251
src/led_sync/main.py
Normal file
251
src/led_sync/main.py
Normal file
@@ -0,0 +1,251 @@
|
||||
"""
|
||||
LED Sync Studio — CLI entry point (Phase 2).
|
||||
Wires all modules: audio playback, beat detection, choreography scheduling, UDP transport.
|
||||
|
||||
Usage:
|
||||
uv run python -m led_sync <esp32_ip> [--port 4210] [--beat-device <device_name>]
|
||||
|
||||
Phase 3 Textual TUI will replace this REPL while reusing the same module instances.
|
||||
Design (D-13/D-14): minimal error handling, asyncio REPL, not a production interface.
|
||||
"""
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
|
||||
from led_sync.models import ChoreoEvent, ChoreoFile
|
||||
from led_sync.transport import create_esp32_transport
|
||||
from led_sync.audio import AudioPlayer, BeatDetector
|
||||
from led_sync.scheduler import ChoreographyScheduler
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.WARNING,
|
||||
format="%(levelname)s %(name)s: %(message)s",
|
||||
)
|
||||
|
||||
|
||||
class AppState:
|
||||
"""Holds all runtime state — designed for Phase 3 TUI integration."""
|
||||
|
||||
def __init__(self):
|
||||
self.transport = None
|
||||
self.player = AudioPlayer()
|
||||
self.scheduler: ChoreographyScheduler | None = None
|
||||
self.detector: BeatDetector | None = None
|
||||
self.choreo: ChoreoFile | None = None
|
||||
self.current_file: str | None = None
|
||||
|
||||
|
||||
async def handle_command(line: str, state: AppState) -> bool:
|
||||
"""Handle one REPL command. Returns False to exit."""
|
||||
parts = line.strip().split()
|
||||
if not parts:
|
||||
return True
|
||||
cmd = parts[0].lower()
|
||||
|
||||
if cmd == "quit":
|
||||
return False
|
||||
|
||||
elif cmd == "play":
|
||||
if len(parts) < 2:
|
||||
print("Usage: play <audio_file>")
|
||||
return True
|
||||
file_path = parts[1]
|
||||
state.current_file = file_path
|
||||
state.player.play(file_path)
|
||||
if state.choreo and state.scheduler:
|
||||
state.scheduler.play(state.choreo.events)
|
||||
print(f"Playing: {file_path}")
|
||||
|
||||
elif cmd == "pause":
|
||||
state.player.pause()
|
||||
if state.scheduler:
|
||||
state.scheduler.pause()
|
||||
print(f"Paused at {state.player.position_seconds:.2f}s")
|
||||
|
||||
elif cmd == "resume":
|
||||
state.player.resume()
|
||||
if state.scheduler:
|
||||
state.scheduler.resume()
|
||||
print("Resumed")
|
||||
|
||||
elif cmd == "seek":
|
||||
if len(parts) < 2:
|
||||
print("Usage: seek <seconds>")
|
||||
return True
|
||||
try:
|
||||
secs = float(parts[1])
|
||||
except ValueError:
|
||||
print("seek: invalid number")
|
||||
return True
|
||||
state.player.seek(secs)
|
||||
if state.scheduler and state.choreo:
|
||||
state.scheduler.play(state.choreo.events, seek_seconds=secs)
|
||||
print(f"Seeked to {secs:.2f}s")
|
||||
|
||||
elif cmd == "stop":
|
||||
state.player.stop()
|
||||
if state.scheduler:
|
||||
state.scheduler.stop()
|
||||
print("Stopped")
|
||||
|
||||
elif cmd == "load":
|
||||
if len(parts) < 2:
|
||||
print("Usage: load <choreo.json>")
|
||||
return True
|
||||
path = parts[1]
|
||||
try:
|
||||
state.choreo = ChoreoFile.load(path)
|
||||
print(f"Loaded {path}: {len(state.choreo.events)} events, song={state.choreo.song_path}")
|
||||
except Exception as e:
|
||||
print(f"Error loading {path}: {e}")
|
||||
|
||||
elif cmd == "add":
|
||||
# add <timestamp> <zone> <animation> [params_json]
|
||||
if len(parts) < 4:
|
||||
print("Usage: add <timestamp> <zone> <animation> [params_json]")
|
||||
return True
|
||||
try:
|
||||
ts = float(parts[1])
|
||||
zone = parts[2]
|
||||
animation = parts[3]
|
||||
params = json.loads(parts[4]) if len(parts) > 4 else {}
|
||||
except (ValueError, json.JSONDecodeError) as e:
|
||||
print(f"add: parse error: {e}")
|
||||
return True
|
||||
if state.choreo is None:
|
||||
state.choreo = ChoreoFile(song_path=state.current_file or "")
|
||||
try:
|
||||
event = ChoreoEvent(timestamp=ts, zone=zone, animation=animation, params=params)
|
||||
state.choreo.add_event(event)
|
||||
print(f"Added: t={ts:.2f}s zone={zone} animation={animation}")
|
||||
except Exception as e:
|
||||
print(f"add: validation error: {e}")
|
||||
|
||||
elif cmd == "save":
|
||||
if len(parts) < 2:
|
||||
print("Usage: save <file>")
|
||||
return True
|
||||
if state.choreo is None:
|
||||
print("No choreography loaded. Use 'load' first or 'add' some events.")
|
||||
return True
|
||||
path = parts[1]
|
||||
try:
|
||||
state.choreo.save(path)
|
||||
print(f"Saved to {path}")
|
||||
except Exception as e:
|
||||
print(f"save error: {e}")
|
||||
|
||||
elif cmd == "status":
|
||||
esp = state.transport
|
||||
connected = esp.connected if esp else False
|
||||
last_status = esp.last_status if esp else None
|
||||
pos = state.player.position_seconds
|
||||
playing = state.player.is_playing
|
||||
beat_count = state.detector.beat_count if state.detector else 0
|
||||
bpm = state.detector.bpm if state.detector else 0.0
|
||||
events = len(state.choreo.events) if state.choreo else 0
|
||||
print(
|
||||
f"ESP32: {'connected' if connected else 'disconnected'} | "
|
||||
f"Status: {last_status}\n"
|
||||
f"Player: {'playing' if playing else 'stopped/paused'} @ {pos:.2f}s\n"
|
||||
f"Beats: {beat_count} | BPM: {bpm:.1f}\n"
|
||||
f"Choreo: {events} events loaded"
|
||||
)
|
||||
|
||||
elif cmd == "beat":
|
||||
sub = parts[1].lower() if len(parts) > 1 else ""
|
||||
device = parts[2] if len(parts) > 2 else None
|
||||
if sub == "on":
|
||||
if state.detector is None:
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
print(f"[BEAT] t={beat_time:.3f}s BPM={state.detector.bpm:.1f}")
|
||||
|
||||
state.detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
try:
|
||||
state.detector.start(device=device)
|
||||
print(f"Beat detection started (device={device or 'default'})")
|
||||
except RuntimeError as e:
|
||||
print(f"Beat detection failed: {e}")
|
||||
elif sub == "off":
|
||||
if state.detector:
|
||||
state.detector.stop()
|
||||
print("Beat detection stopped")
|
||||
else:
|
||||
print("Usage: beat on [device_name] | beat off")
|
||||
|
||||
elif cmd == "help":
|
||||
print(
|
||||
"Commands:\n"
|
||||
" play <file> — play audio file (+ choreo if loaded)\n"
|
||||
" pause — pause playback\n"
|
||||
" resume — resume playback\n"
|
||||
" seek <seconds> — seek to position\n"
|
||||
" stop — stop playback\n"
|
||||
" load <choreo.json> — load choreography file\n"
|
||||
" add <ts> <zone> <anim> [params_json] — add choreo event\n"
|
||||
" save <file> — save choreography\n"
|
||||
" beat on [device] — start beat detection\n"
|
||||
" beat off — stop beat detection\n"
|
||||
" status — show ESP32 state, position, beats\n"
|
||||
" quit — exit"
|
||||
)
|
||||
|
||||
else:
|
||||
print(f"Unknown command: {cmd!r}. Type 'help' for commands.")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def repl(state: AppState) -> None:
|
||||
"""Asyncio REPL loop (D-13). Uses run_in_executor for non-blocking input."""
|
||||
loop = asyncio.get_running_loop()
|
||||
print("LED Sync Studio CLI — type 'help' for commands, 'quit' to exit")
|
||||
while True:
|
||||
try:
|
||||
line = await loop.run_in_executor(None, input, "> ")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
break
|
||||
if not await handle_command(line, state):
|
||||
break
|
||||
|
||||
|
||||
async def async_main(esp32_host: str, esp32_port: int, beat_device: str | None) -> None:
|
||||
state = AppState()
|
||||
|
||||
# Connect transport (D-10/D-11/D-12)
|
||||
print(f"Connecting to ESP32 at {esp32_host}:{esp32_port}...")
|
||||
state.transport = await create_esp32_transport(esp32_host, esp32_port)
|
||||
state.scheduler = ChoreographyScheduler(transport=state.transport)
|
||||
print(
|
||||
f"Transport ready. ESP32 "
|
||||
f"{'connected' if state.transport.connected else 'not yet responding (will retry on status ping)'}"
|
||||
)
|
||||
|
||||
try:
|
||||
await repl(state)
|
||||
finally:
|
||||
# Cleanup
|
||||
state.player.stop()
|
||||
if state.scheduler:
|
||||
state.scheduler.stop()
|
||||
if state.detector:
|
||||
state.detector.stop()
|
||||
print("Goodbye.")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="LED Sync Studio — Phase 2 CLI")
|
||||
parser.add_argument("esp32_host", help="ESP32 IP address (e.g. 192.168.1.50)")
|
||||
parser.add_argument("--port", type=int, default=4210, help="ESP32 UDP port (default: 4210)")
|
||||
parser.add_argument("--beat-device", help="sounddevice input device name for beat detection")
|
||||
args = parser.parse_args()
|
||||
|
||||
asyncio.run(async_main(args.esp32_host, args.port, args.beat_device))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user