""" Audio Player — Play audio files through the EQ engine to the sound card output. """ import sounddevice as sd import numpy as np from typing import Optional, Callable import threading from .file_loader import load_audio, AudioData, normalize_audio from .eq_engine import EQEngine from .tape_processor import TapeProcessor from .compressor import Compressor from .audio_device import DeviceManager class AudioPlayer: """ Handles playback of audio files through a TapeProcessor and EQEngine. """ def __init__(self, eq_engine: EQEngine, tape_processor: TapeProcessor, device_manager: DeviceManager): self.eq_engine = eq_engine self.tape_processor = tape_processor self.compressor = Compressor() self.device_manager = device_manager self.audio_data: Optional[AudioData] = None self.stream: Optional[sd.OutputStream] = None self.current_frame: float = 0.0 self.is_playing: bool = False self.is_looping: bool = False # Callbacks for UI updates self.on_position_update: Optional[Callable[[float], None]] = None self.on_playback_stopped: Optional[Callable[[], None]] = None # VU meter levels (left, right RMS in dB) self.vu_levels = (-100.0, -100.0) self.playback_speed: float = 1.0 self.output_volume: float = 1.0 self._is_restarting: bool = False self._last_raw_chunk: Optional[np.ndarray] = None self._last_processed_chunk: Optional[np.ndarray] = None def load_file(self, file_path: str, normalize: bool = False): """Load an audio file and prepare for playback.""" self.stop() audio = load_audio(file_path) if normalize: audio = normalize_audio(audio) # Convert mono to stereo for output if needed if audio.channels == 1 and self.device_manager.channels == 2: audio.samples = np.repeat(audio.samples, 2, axis=1) audio.channels = 2 self.audio_data = audio self.current_frame = 0.0 def play(self): """Start or resume playback.""" if self.audio_data is None: return if self.is_playing: return out_sr = self.device_manager.sample_rate self.eq_engine.reset_state() self.eq_engine.set_sample_rate(out_sr) self.tape_processor.reset_state() self.tape_processor.set_sample_rate(out_sr) kwargs = self.device_manager.get_sd_output_kwargs() self.stream = sd.OutputStream( callback=self._audio_callback, finished_callback=self._on_finished, **kwargs ) self.is_playing = True self.stream.start() def pause(self): """Pause playback.""" if self.stream: try: self.stream.stop() self.stream.close() except Exception as e: print(f"Error closing stream: {e}") self.stream = None self.is_playing = False def stop(self): """Stop playback and reset position.""" self.pause() self.current_frame = 0.0 self.vu_levels = (-100.0, -100.0) if self.on_position_update: self.on_position_update(0.0) def set_speed(self, speed: float): """Set playback speed. Stream adjusts seamlessly via interpolation.""" self.playback_speed = max(0.1, speed) def seek(self, position_sec: float): """Seek to a specific position in seconds.""" if self.audio_data is None: return frame = position_sec * self.audio_data.sample_rate self.current_frame = max(0.0, min(float(frame), float(self.audio_data.num_samples - 1))) if self.is_playing: self.eq_engine.reset_state() self.tape_processor.reset_state() def _audio_callback(self, outdata: np.ndarray, frames: int, time, status: sd.CallbackFlags): """sounddevice callback for generating output audio.""" if status: print(f"Playback status: {status}") if self.audio_data is None or not self.is_playing: outdata.fill(0) return # Ratio of input samples to advance per output sample ratio = (self.audio_data.sample_rate / self.device_manager.sample_rate) * self.playback_speed # We need to produce `frames` samples out_indices = np.arange(frames) in_indices = self.current_frame + out_indices * ratio # Find which indices are valid valid_mask = in_indices < (self.audio_data.num_samples - 1) valid_count = int(np.sum(valid_mask)) if valid_count == 0: if self.is_looping: self.current_frame = 0.0 self.eq_engine.reset_state() self.tape_processor.reset_state() in_indices = self.current_frame + out_indices * ratio valid_mask = in_indices < (self.audio_data.num_samples - 1) valid_count = int(np.sum(valid_mask)) else: outdata.fill(0) raise sd.CallbackStop() # Get integer part and fractional part in_idx_int = in_indices[:valid_count].astype(np.int32) in_idx_frac = in_indices[:valid_count] - in_idx_int # Broadcast for channels (e.g. 2 channels) in_idx_frac = in_idx_frac[:, np.newaxis].astype(np.float32) c0 = self.audio_data.samples[in_idx_int] c1 = self.audio_data.samples[in_idx_int + 1] raw_chunk = c0 + (c1 - c0) * in_idx_frac # Tape Processing & EQ Chain chunk = self.tape_processor.process_pre_eq(raw_chunk) chunk = self.eq_engine.process(chunk) chunk = self.tape_processor.process_post_eq(chunk) # Dynamic Range Compression chunk = self.compressor.process(chunk) # Apply output volume if self.output_volume != 1.0: chunk = chunk * self.output_volume # Write to output buffer outdata[:valid_count] = chunk self._last_raw_chunk = np.copy(raw_chunk) self._last_processed_chunk = np.copy(chunk) # Pad with zeros if we reached the end if valid_count < frames: outdata[valid_count:] = 0 # Update VU meters self._update_vu_levels(chunk) self.current_frame = float(in_indices[valid_count - 1] + ratio) # Notify UI (safely, could be high rate, UI should throttle) if self.on_position_update: pos_sec = self.current_frame / self.audio_data.sample_rate self.on_position_update(pos_sec) if valid_count < frames and not self.is_looping: raise sd.CallbackStop() def _update_vu_levels(self, chunk: np.ndarray): """Calculate RMS levels for VU meters.""" if chunk.size == 0: return # Calculate RMS per channel rms = np.sqrt(np.mean(chunk**2, axis=0)) # Convert to dB, avoid log(0) db = 20 * np.log10(np.maximum(rms, 1e-5)) if chunk.shape[1] >= 2: self.vu_levels = (float(db[0]), float(db[1])) else: self.vu_levels = (float(db[0]), float(db[0])) def _on_finished(self): """Called when stream stops.""" if self._is_restarting: return self.is_playing = False if self.on_playback_stopped: self.on_playback_stopped()