import pygame import soundfile as sf import sounddevice as sd import numpy as np import asyncio import time import json import threading from dataclasses import dataclass from typing import List, Optional, Tuple # ----------------------------- # Config # ----------------------------- @dataclass class VisualConfig: num_pumpkins: int = 4 pumpkin_width: int = 80 pumpkin_height: int = 150 pumpkin_spacing: int = 50 background_color: Tuple[int, int, int] = (20, 20, 20) pumpkin_color: Tuple[int, int, int] = (200, 100, 0) fire_color: Tuple[int, int, int] = (255, 0, 0) @dataclass class AudioConfig: track_id: str = "4PTG3Z6ehGkBFwjybzWkR8" audio_file: str = "Never Gonna Give You Up.ogg" analysis_file: Optional[str] = None timing_offset: float = 0 beat_skip: int = 1 segment_skip: int = 1 top_percent: float = 0.2 # Only fire segments in top 30% confidence def __post_init__(self): if self.analysis_file is None: self.analysis_file = f"{self.track_id}_analysis.json" @dataclass class Pumpkin: rect: pygame.Rect fire_end_time: Optional[float] = None # store when fire should stop PUMPKIN_FREQS = [440, 523, 587, 659] # Hz for pumpkins 0-3 # ----------------------------- # Audio Handling # ----------------------------- class AudioPlayer: def __init__(self, audio_file: str, volume: float = 1): # volume 0.0-1.0 self.audio_data, self.samplerate = sf.read(audio_file, dtype="float32") if self.audio_data.ndim == 1: self.audio_data = self.audio_data[:, np.newaxis] self.audio_data *= volume # scale volume down self.position = 0 self.start_time: Optional[float] = None self.stream: Optional[sd.OutputStream] = None def audio_callback(self, outdata, frames, time_info, status): if status: print("Stream status:", status) if self.start_time is None: self.start_time = time.time() end_pos = self.position + frames if end_pos >= len(self.audio_data): outdata[:len(self.audio_data)-self.position] = self.audio_data[self.position:] outdata[len(self.audio_data)-self.position:] = 0 raise sd.CallbackStop() else: outdata[:] = self.audio_data[self.position:end_pos] self.position = end_pos def play_blocking(self): with sd.OutputStream( channels=self.audio_data.shape[1], samplerate=self.samplerate, blocksize=2048, callback=self.audio_callback ) as stream: self.stream = stream while self.stream.active: sd.sleep(10) async def play_async(self): await asyncio.to_thread(self.play_blocking) # ----------------------------- # Beep Utility # ----------------------------- def play_tune_for_duration(freq, duration, samplerate=44100): t = np.linspace(0, duration, int(samplerate * duration), endpoint=False) waveform = 0.5 * np.sin(2 * np.pi * freq * t) waveform = waveform.astype(np.float32)[:, np.newaxis] # make 2D for stereo threading.Thread(target=sd.play, args=(waveform, samplerate), daemon=True).start() def play_beep_async(beep_wave, samplerate=44100): threading.Thread(target=sd.play, args=(beep_wave, samplerate), daemon=True).start() # ----------------------------- # Event Scheduler # ----------------------------- class EventScheduler: def __init__(self, analysis_file: str, num_pumpkins: int, beat_skip=1, segment_skip=0, top_percent=1.0, max_segment_duration=0.4): self.num_pumpkins = num_pumpkins self.top_percent = top_percent self.max_segment_duration = max_segment_duration self.events: List[Tuple[float, float, List[int]]] = [] # list of (start, end, pumpkin indices) self.load_analysis(analysis_file, beat_skip, segment_skip) def load_analysis(self, analysis_file: str, beat_skip: int, segment_skip: int): with open(analysis_file) as f: analysis = json.load(f) # Compute confidence threshold confidences = [seg["confidence"] for seg in analysis.get("segments", []) if seg.get("confidence") is not None] if not confidences: threshold = 0.0 else: confidences.sort(reverse=True) cutoff_idx = int(len(confidences) * self.top_percent) threshold = confidences[cutoff_idx - 1] if cutoff_idx > 0 else 0.0 # Store segments above threshold for i, seg in enumerate(analysis.get("segments", [])): if i % segment_skip != 0: continue confidence = seg.get("confidence", 0) if confidence < threshold or seg["duration"] > self.max_segment_duration: continue dominant_pitch = int(np.argmax(seg["pitches"])) pumpkin_idx = dominant_pitch % self.num_pumpkins self.events.append((seg["start"], seg["start"] + seg["duration"], [pumpkin_idx])) # Optional: add bars firing all pumpkins # for bar in analysis.get("bars", []): # start = bar["start"] # end = start + bar.get("duration", 0.1) # self.events.append((start, end, list(range(self.num_pumpkins)))) # Sort events by start time self.events.sort(key=lambda x: x[0]) def get_triggered_events(self, elapsed: float, current_index: int): triggered = [] while current_index < len(self.events) and self.events[current_index][0] <= elapsed: start_time, end_time, indices = self.events[current_index] triggered.append((indices, end_time)) # indices is now a list current_index += 1 return triggered, current_index # ----------------------------- # Renderer # ----------------------------- class PumpkinRenderer: def __init__(self, visual_config: VisualConfig): self.config = visual_config pygame.init() self.screen = pygame.display.set_mode( (self.config.num_pumpkins * self.config.pumpkin_width + (self.config.num_pumpkins + 1) * self.config.pumpkin_spacing, 300) ) pygame.display.set_caption("Pumpkin Fire Debug") self.font = pygame.font.SysFont(None, 24) self.clock = pygame.time.Clock() self.pumpkins: List[Pumpkin] = [] self.init_pumpkins() def init_pumpkins(self): for i in range(self.config.num_pumpkins): x = self.config.pumpkin_spacing + i * (self.config.pumpkin_width + self.config.pumpkin_spacing) y = 100 rect = pygame.Rect(x, y, self.config.pumpkin_width, self.config.pumpkin_height) self.pumpkins.append(Pumpkin(rect)) def render(self, elapsed: float): self.screen.fill(self.config.background_color) for i, p in enumerate(self.pumpkins): color = self.config.pumpkin_color if p.fire_end_time and elapsed < p.fire_end_time: color = self.config.fire_color pygame.draw.rect(self.screen, color, p.rect) txt = self.font.render(f"Pumpkin {i + 1}", True, (255, 255, 255)) self.screen.blit(txt, (p.rect.x, p.rect.y - 25)) pygame.display.flip() self.clock.tick(60) # ----------------------------- # Main Loop # ----------------------------- async def main_loop(audio_cfg: AudioConfig, visual_cfg: VisualConfig): player = AudioPlayer(audio_cfg.audio_file) scheduler = EventScheduler( audio_cfg.analysis_file, visual_cfg.num_pumpkins, audio_cfg.beat_skip, audio_cfg.segment_skip, top_percent=audio_cfg.top_percent ) renderer = PumpkinRenderer(visual_cfg) # Pre-generate beep wave # beep_wave = generate_beep(duration=0.1) audio_task = asyncio.create_task(player.play_async()) event_index = 0 running = True try: while running: elapsed = (time.time() - player.start_time + audio_cfg.timing_offset) if player.start_time else 0 triggered, event_index = scheduler.get_triggered_events(elapsed, event_index) for indices, end_time in triggered: for idx in indices: renderer.pumpkins[idx].fire_end_time = end_time # Play a tone that lasts the same as the fire duration duration = end_time - elapsed if duration > 0: play_tune_for_duration(PUMPKIN_FREQS[idx], duration) renderer.render(elapsed) # Handle quit for event in pygame.event.get(): if event.type == pygame.QUIT: running = False raise sd.CallbackStop() await asyncio.sleep(0.001) finally: pygame.quit() print("Playback finished.") # ----------------------------- # Entry point # ----------------------------- if __name__ == "__main__": visual_cfg = VisualConfig() audio_cfg = AudioConfig() asyncio.run(main_loop(audio_cfg, visual_cfg))