diff --git a/pyproject.toml b/pyproject.toml index fe4c38c..bd851a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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"] diff --git a/src/led_sync/__main__.py b/src/led_sync/__main__.py new file mode 100644 index 0000000..b5917fa --- /dev/null +++ b/src/led_sync/__main__.py @@ -0,0 +1,3 @@ +from led_sync.main import main + +main() diff --git a/src/led_sync/main.py b/src/led_sync/main.py new file mode 100644 index 0000000..ad5ca08 --- /dev/null +++ b/src/led_sync/main.py @@ -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 [--port 4210] [--beat-device ] + +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 ") + 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 ") + 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 ") + 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 [params_json] + if len(parts) < 4: + print("Usage: add [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 ") + 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 — play audio file (+ choreo if loaded)\n" + " pause — pause playback\n" + " resume — resume playback\n" + " seek — seek to position\n" + " stop — stop playback\n" + " load — load choreography file\n" + " add [params_json] — add choreo event\n" + " save — 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()