tape_recorder/core/audio_recorder.py

"""
Audio Recorder — Capture audio from sound card input.
"""

import sounddevice as sd
import numpy as np
import soundfile as sf
from typing import Optional, Callable
import threading

from .audio_device import DeviceManager

class AudioRecorder:
    """
    Captures audio from the selected input device.
    Supports continuous ring-buffering for UI updates and saving to file.
    """
    def __init__(self, device_manager: DeviceManager):
        self.device_manager = device_manager
        
        self.stream: Optional[sd.InputStream] = None
        self.is_recording: bool = False
        
        # Buffer for capturing the full recording
        self.recorded_data: list[np.ndarray] = []
        
        # Callbacks
        self.on_audio_data: Optional[Callable[[np.ndarray], None]] = None
        
        self.vu_levels = (-100.0, -100.0)

    def start(self):
        """Start capturing audio."""
        if self.is_recording:
            return
            
        self.recorded_data = []
        
        kwargs = self.device_manager.get_sd_input_kwargs()
        
        self.stream = sd.InputStream(
            callback=self._audio_callback,
            **kwargs
        )
        
        self.is_recording = True
        self.stream.start()

    def stop(self) -> np.ndarray:
        """Stop capturing and return the recorded data."""
        if self.is_recording and self.stream:
            self.stream.stop()
            self.stream.close()
            self.stream = None
            self.is_recording = False
            
        if not self.recorded_data:
            channels = self.device_manager.channels
            return np.zeros((0, channels), dtype=np.float32)
            
        # Concatenate all blocks
        full_data = np.concatenate(self.recorded_data, axis=0)
        return full_data

    def save_to_wav(self, file_path: str):
        """Save the recorded data to a WAV file."""
        data = self.stop()
        if len(data) > 0:
            sf.write(
                file_path, 
                data, 
                self.device_manager.sample_rate, 
                subtype='FLOAT'
            )
            return True
        return False

    def _audio_callback(self, indata: np.ndarray, frames: int, time, status: sd.CallbackFlags):
        """sounddevice callback for incoming audio."""
        if status:
            print(f"Record status: {status}")
            
        # Make a copy of the input data
        chunk = indata.copy()
        
        # Store for full recording
        self.recorded_data.append(chunk)
        
        # Update VU
        self._update_vu_levels(chunk)
        
        # Notify UI/analyzers (e.g. for real-time spectrum)
        if self.on_audio_data:
            self.on_audio_data(chunk)

    def _update_vu_levels(self, chunk: np.ndarray):
        """Calculate RMS levels for VU meters."""
        if chunk.size == 0:
            return
            
        rms = np.sqrt(np.mean(chunk**2, axis=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]))