From d6480a35ed5197a67c94d65c808e0df382cd87e9 Mon Sep 17 00:00:00 2001 From: Shawn Date: Tue, 19 May 2026 21:44:28 -0400 Subject: [PATCH] P3-R3: Multi-track recording engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - WAV writer with streaming support for 16/24/32-bit integer and 32-bit float - MultiTrackRecorder with per-channel arming, arm state machine, write-ahead buffer queues decoupling JACK real-time thread from SD card I/O - Punch in/out recording on individual channels with punch region tracking - Session file format (JSON): session metadata, takes, tracks, mixer state, punch regions, serialization round-trip - Take management: create, delete, rename, list, track association - Stereo bounce engine with pan law (-3dB constant power), normalisation, master bus volume/mute/dim, block-based processing for memory efficiency - Disk space monitor with configurable warning/critical thresholds, estimated recording time remaining, background polling, callback on threshold change - 117 tests: sample conversion, WAV header validation, recorder lifecycle, session CRUD, serialization, bounce with pan/normalisation/master, integration Integrates with JACK capture ports → mixer → recording ports → disk. Architecture: JACK callback → recorder.write() → buffer queue → writer thread → WAV files. --- src/recording/__init__.py | 83 ++ src/recording/bounce.py | 344 ++++++++ src/recording/disk_monitor.py | 252 ++++++ src/recording/recorder.py | 535 ++++++++++++ src/recording/session.py | 466 +++++++++++ src/recording/wav_writer.py | 395 +++++++++ tests/test_recording.py | 1436 +++++++++++++++++++++++++++++++++ 7 files changed, 3511 insertions(+) create mode 100644 src/recording/__init__.py create mode 100644 src/recording/bounce.py create mode 100644 src/recording/disk_monitor.py create mode 100644 src/recording/recorder.py create mode 100644 src/recording/session.py create mode 100644 src/recording/wav_writer.py create mode 100644 tests/test_recording.py diff --git a/src/recording/__init__.py b/src/recording/__init__.py new file mode 100644 index 0000000..0550f02 --- /dev/null +++ b/src/recording/__init__.py @@ -0,0 +1,83 @@ +"""Multi-track recording engine — capture + WAV export + session management. + +The recording package provides a complete recording subsystem for the +Raspberry Pi RT Audio Mixer, including: + + - Streaming WAV writer with configurable bit depth (16/24/32 bit) + and sample rate + - Multi-track recorder with disk streaming (write-ahead buffering + to decouple capture from SD card I/O) + - Punch in/out recording on individual channels + - Session file format (JSON + audio references) with take management + - Stereo bounce/export of master mix with pan law and normalisation + - Free disk space monitoring with configurable thresholds + +Integrates with JACK: capture ports → mixer → recording ports → disk. +""" + +from __future__ import annotations + +from .wav_writer import ( + WAVWriter, + WAVWriterConfig, + WAVBitDepth, + write_wav, +) +from .recorder import ( + MultiTrackRecorder, + RecorderConfig, + RecorderState, + ChannelState as RecordChannelState, + ChannelRecordStatus, + PunchRegion, +) +from .session import ( + Session, + Take, + TrackInfo, + PunchInfo, +) +from .disk_monitor import ( + DiskMonitor, + DiskMonitorConfig, + DiskSpaceCallback, +) +from .bounce import ( + BounceEngine, + BounceConfig, + BounceTrack, + BounceResult, + db_to_linear, + linear_to_db, +) + +__all__ = [ + # WAV + "WAVWriter", + "WAVWriterConfig", + "WAVBitDepth", + "write_wav", + # Recorder + "MultiTrackRecorder", + "RecorderConfig", + "RecorderState", + "RecordChannelState", + "ChannelRecordStatus", + "PunchRegion", + # Session + "Session", + "Take", + "TrackInfo", + "PunchInfo", + # Disk monitor + "DiskMonitor", + "DiskMonitorConfig", + "DiskSpaceCallback", + # Bounce + "BounceEngine", + "BounceConfig", + "BounceTrack", + "BounceResult", + "db_to_linear", + "linear_to_db", +] diff --git a/src/recording/bounce.py b/src/recording/bounce.py new file mode 100644 index 0000000..0315aca --- /dev/null +++ b/src/recording/bounce.py @@ -0,0 +1,344 @@ +"""Stereo bounce / export engine. + +Mixes down multi-track recordings to a stereo master WAV file, +applying channel volume, pan, and the master bus state. + +Supports: +- Bounce any combination of track files to stereo +- Per-channel volume and pan from mixer state +- Master bus volume/mute/dim +- Selectable bit depth (16/24/32 bit) +- Normalisation (peak or LUFS target) +- Preview: dry run that reports estimated file size + +Usage: + bounce = BounceEngine(sample_rate=48000) + bounce.add_track("tracks/ch00.wav", channel=0, volume_db=0, pan=0) + bounce.add_track("tracks/ch01.wav", channel=1, volume_db=-3, pan=0.5) + bounce.bounce("output/master.wav", bit_depth=24, normalise=True) +""" + +from __future__ import annotations + +import logging +import math +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +import numpy as np + +from .wav_writer import WAVWriter, WAVBitDepth, write_wav + +logger = logging.getLogger(__name__) + + +# ── Constants ───────────────────────────────────────────────────────────────── + +# dB to linear conversion +_DB_EPSILON = 1e-10 # floor for -inf dB + + +def db_to_linear(db: float) -> float: + """Convert dB to linear gain. + + -inf → 0.0, 0 dB → 1.0, +6 dB → 2.0, -6 dB → 0.5 + """ + return 10.0 ** (db / 20.0) + + +def linear_to_db(linear: float) -> float: + """Convert linear gain to dB. + + 1.0 → 0 dB, 0.5 → -6 dB, 2.0 → +6 dB + """ + if linear <= _DB_EPSILON: + return -144.0 # effectively -inf + return 20.0 * math.log10(linear) + + +# ── Types ───────────────────────────────────────────────────────────────────── + + +@dataclass +class BounceTrack: + """A single track to include in the bounce.""" + file_path: str | Path # path to WAV file (mono or stereo) + channel: int = 0 # mixer channel index (for reference) + volume_db: float = 0.0 # channel fader level in dB + pan: float = 0.0 # -1.0 (full left) to 1.0 (full right) + muted: bool = False + + def __post_init__(self): + self.file_path = Path(self.file_path) + + +@dataclass +class BounceConfig: + """Configuration for a bounce operation.""" + sample_rate: int = 48000 + bit_depth: WAVBitDepth = WAVBitDepth.INT24 + normalise: bool = True # normalise peak to 0 dBFS + normalise_target_db: float = -0.3 # target peak in dBFS + master_volume_db: float = 0.0 # master fader dB + master_muted: bool = False + master_dim_db: float = -20.0 + master_dim_active: bool = False + block_size_frames: int = 4096 # processing block size for memory efficiency + + +@dataclass +class BounceResult: + """Result of a bounce operation.""" + output_path: Path + frames_written: int + duration_seconds: float + peak_db: float # peak level in dBFS + rms_db: float # RMS level in dBFS + files_processed: int + + +# ── Bounce engine ───────────────────────────────────────────────────────────── + + +class BounceEngine: + """Stereo bounce engine for multi-track mixdown. + + Processes track files in blocks to keep memory usage low. + Supports large multi-track projects on embedded hardware. + """ + + def __init__( + self, + sample_rate: int = 48000, + *, + config: BounceConfig | None = None, + ): + self.sample_rate = sample_rate + self.config = config or BounceConfig(sample_rate=sample_rate) + self._tracks: list[BounceTrack] = [] + + # ── Track setup ─────────────────────────────────────────────────────── + + def add_track( + self, + file_path: str | Path, + channel: int = 0, + volume_db: float = 0.0, + pan: float = 0.0, + muted: bool = False, + ) -> None: + """Add a track to the bounce list.""" + self._tracks.append(BounceTrack( + file_path=Path(file_path), + channel=channel, + volume_db=volume_db, + pan=pan, + muted=muted, + )) + + def remove_track(self, channel: int) -> None: + """Remove all tracks for a given channel.""" + self._tracks = [t for t in self._tracks if t.channel != channel] + + def clear_tracks(self) -> None: + """Clear all tracks.""" + self._tracks.clear() + + # ── Bounce ──────────────────────────────────────────────────────────── + + def bounce( + self, + output_path: str | Path, + *, + bit_depth: WAVBitDepth | int | None = None, + normalise: bool | None = None, + ) -> BounceResult: + """Perform the stereo bounce. + + Args: + output_path: Output WAV file path. + bit_depth: Override config bit depth. + normalise: Override config normalisation. + + Returns: + BounceResult with output path, frame count, duration, and levels. + """ + if not self._tracks: + raise ValueError("No tracks added — call add_track() first") + + output_path = Path(output_path) + output_path.parent.mkdir(parents=True, exist_ok=True) + + bd = self.config.bit_depth if bit_depth is None else ( + WAVBitDepth(bit_depth) if isinstance(bit_depth, int) else bit_depth + ) + do_normalise = normalise if normalise is not None else self.config.normalise + + # Compute master gain + master_gain = self._compute_master_gain() + + # First pass: determine max length + max_frames = 0 + samples_dict: dict[str, np.ndarray] = {} # load all tracks + + for track in self._tracks: + if track.muted: + continue + samples = self._load_audio(Path(track.file_path)) + if samples.ndim == 1: + samples = samples.reshape(-1, 1) + if samples.shape[0] > max_frames: + max_frames = samples.shape[0] + samples_dict[str(track.file_path)] = samples + + if max_frames == 0: + raise ValueError("No audio data to bounce (all tracks empty or muted)") + + # Process in blocks + block_size = self.config.block_size_frames + stereo_out = np.zeros((max_frames, 2), dtype=np.float64) # accumulate in float64 + + for track in self._tracks: + if track.muted: + continue + + samples = samples_dict[str(track.file_path)] + ch_gain = db_to_linear(track.volume_db) + + num_frames = samples.shape[0] + num_ch = samples.shape[1] + + # Compute pan gains + left_gain, right_gain = _pan_to_gains(track.pan) + + for start in range(0, num_frames, block_size): + end = min(start + block_size, num_frames) + block = samples[start:end, :].astype(np.float64) + + # Convert to stereo + if num_ch == 1: + stereo_out[start:end, 0] += block[:, 0] * ch_gain * left_gain + stereo_out[start:end, 1] += block[:, 0] * ch_gain * right_gain + else: + # Already stereo — pan is interpreted as balance + stereo_out[start:end, 0] += block[:, 0] * ch_gain * left_gain + stereo_out[start:end, 1] += block[:, 1] * ch_gain * right_gain + + # Apply master gain + stereo_out *= master_gain + + # Measure levels before normalisation + peak = float(np.max(np.abs(stereo_out))) + rms = float(np.sqrt(np.mean(stereo_out ** 2))) + peak_db = linear_to_db(peak) if peak > 0 else -144.0 + rms_db = linear_to_db(rms) if rms > 0 else -144.0 + + # Normalise + if do_normalise and peak > 0: + target_linear = db_to_linear(self.config.normalise_target_db) + norm_gain = target_linear / peak + stereo_out *= norm_gain + peak_db = self.config.normalise_target_db + + # Convert to float32 for writing + stereo_out_f32 = np.clip(stereo_out, -1.0, 1.0).astype(np.float32) + + # Write + write_wav(output_path, stereo_out_f32, sample_rate=self.sample_rate, bit_depth=bd) + + duration = max_frames / self.sample_rate if self.sample_rate > 0 else 0.0 + + logger.info( + "Bounce complete: %s (%d frames, %.1fs, peak=%+.1f dBFS, RMS=%+.1f dBFS)", + output_path.name, max_frames, duration, peak_db, rms_db, + ) + + return BounceResult( + output_path=output_path, + frames_written=max_frames, + duration_seconds=duration, + peak_db=peak_db, + rms_db=rms_db, + files_processed=len(self._tracks), + ) + + # ── Preview ─────────────────────────────────────────────────────────── + + def preview(self, output_path: str | Path | None = None) -> dict: + """Dry-run preview of bounce: returns expected stats without writing. + + Returns a dict with estimated_file_size_bytes, total_frames, etc. + """ + if not self._tracks: + return {"error": "No tracks added"} + + max_frames = 0 + for track in self._tracks: + if track.muted: + continue + samples = self._load_audio(Path(track.file_path)) + if samples.ndim == 1: + nf = samples.shape[0] + else: + nf = samples.shape[0] + max_frames = max(max_frames, nf) + + abs_depth = abs(int(self.config.bit_depth)) + bytes_per_sample = abs_depth // 8 + estimated_bytes = max_frames * 2 * bytes_per_sample # stereo + + return { + "total_frames": max_frames, + "duration_seconds": max_frames / self.sample_rate if self.sample_rate > 0 else 0, + "estimated_file_size_bytes": estimated_bytes, + "estimated_file_size_mb": estimated_bytes / (1024.0 * 1024.0), + "track_count": len([t for t in self._tracks if not t.muted]), + "sample_rate": self.sample_rate, + "bit_depth": self.config.bit_depth.name, + } + + # ── Helpers ──────────────────────────────────────────────────────────── + + def _compute_master_gain(self) -> float: + """Compute master bus linear gain from config.""" + if self.config.master_muted: + return 0.0 + + gain_db = self.config.master_volume_db + if self.config.master_dim_active: + gain_db += self.config.master_dim_db + + return db_to_linear(gain_db) + + @staticmethod + def _load_audio(file_path: Path) -> np.ndarray: + """Load audio file to float32 numpy array. + + Uses soundfile if available, otherwise falls back to the backing + track loader. + """ + try: + import soundfile as sf + samples, sr = sf.read(str(file_path), dtype="float32", always_2d=False) + return samples + except ImportError: + # Fallback: use backing track loader + from ..backing.loader import load_audio + audio = load_audio(file_path) + return audio.samples + + +# ── Pan law ─────────────────────────────────────────────────────────────────── + +def _pan_to_gains(pan: float) -> tuple[float, float]: + """Convert pan value [-1, 1] to left/right gain pair. + + Uses -3 dB constant power pan law. + -1.0 = full left, 0.0 = center, 1.0 = full right. + """ + # Normalise pan to angle [0, pi/2] + angle = (pan + 1.0) * 0.25 * math.pi # maps [-1,1] → [0, pi/2] + left = math.cos(angle) + right = math.sin(angle) + return left, right diff --git a/src/recording/disk_monitor.py b/src/recording/disk_monitor.py new file mode 100644 index 0000000..dd28266 --- /dev/null +++ b/src/recording/disk_monitor.py @@ -0,0 +1,252 @@ +"""Free disk space monitoring for recording sessions. + +The DiskMonitor periodically checks available disk space on the recording +target and emits warnings when space falls below configurable thresholds. +Critical for preventing data loss on SD card-based recording setups. + +Thresholds: +- WARNING: low space, alert user but continue recording +- CRITICAL: critically low, should stop recording to avoid corruption +- MINIMUM_REQUIRED: absolute minimum required to start a recording session + +Usage: + monitor = DiskMonitor(path="/data/recordings") + monitor.start() + if monitor.is_low_space(): + print(f"Warning: {monitor.free_gb:.1f} GB remaining") +""" + +from __future__ import annotations + +import logging +import os +import shutil +import threading +import time +from dataclasses import dataclass +from typing import Callable, Optional + +logger = logging.getLogger(__name__) + + +# ── Types ───────────────────────────────────────────────────────────────────── + +DiskSpaceCallback = Callable[[int, int, int], None] +# callback(free_bytes, total_bytes, threshold_exceeded_level) +# threshold_exceeded_level: 0=none, 1=warning, 2=critical + + +@dataclass +class DiskMonitorConfig: + """Configuration for disk space monitoring.""" + # Thresholds in bytes + warning_threshold_bytes: int = 500 * 1024 * 1024 # 500 MB + critical_threshold_bytes: int = 100 * 1024 * 1024 # 100 MB + minimum_required_bytes: int = 50 * 1024 * 1024 # 50 MB + + # Recording rate estimation + estimated_bytes_per_second: int = 172800 # 48kHz * 24bit * 1ch = ~144KB/s + safety_buffer_seconds: int = 120 # 2 minutes of recording buffer + + # Polling + poll_interval_seconds: float = 5.0 # how often to check + + +# ── Disk monitor ────────────────────────────────────────────────────────────── + + +class DiskMonitor: + """Monitors free disk space on a recording target path. + + Runs a background thread that periodically polls available disk + space. Fires callbacks when thresholds are crossed to alert the + user or trigger automatic stop. + + Also estimates how much recording time remains based on configured + bit rate and current free space. + """ + + def __init__( + self, + path: str, + *, + config: DiskMonitorConfig | None = None, + callback: DiskSpaceCallback | None = None, + ): + self.path = os.path.abspath(path) + self.config = config or DiskMonitorConfig() + self._callback = callback + + # State + self._running: bool = False + self._thread: Optional[threading.Thread] = None + self._lock = threading.Lock() + + # Current stats (updated by background thread) + self._free_bytes: int = 0 + self._total_bytes: int = 0 + self._last_check: float = 0.0 + self._current_level: int = 0 # 0=ok, 1=warning, 2=critical + + # Run initial check + self._poll() + + # ── Lifecycle ────────────────────────────────────────────────────────── + + def start(self) -> None: + """Start background monitoring.""" + with self._lock: + if self._running: + return + self._running = True + self._thread = threading.Thread( + target=self._monitor_loop, + name="disk-monitor", + daemon=True, + ) + self._thread.start() + logger.info("Disk monitor started: %s", self.path) + + def stop(self) -> None: + """Stop background monitoring.""" + with self._lock: + self._running = False + + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=5.0) + + # ── Query ────────────────────────────────────────────────────────────── + + def check_now(self) -> int: + """Force an immediate space check. Returns free bytes.""" + self._poll() + return self._free_bytes + + @property + def free_bytes(self) -> int: + return self._free_bytes + + @property + def free_mb(self) -> float: + return self._free_bytes / (1024.0 * 1024.0) + + @property + def free_gb(self) -> float: + return self._free_bytes / (1024.0 ** 3) + + @property + def total_bytes(self) -> int: + return self._total_bytes + + @property + def total_gb(self) -> float: + return self._total_bytes / (1024.0 ** 3) + + @property + def usage_percent(self) -> float: + if self._total_bytes > 0: + return 100.0 * (1.0 - self._free_bytes / self._total_bytes) + return 0.0 + + def is_low_space(self) -> bool: + """True if free space is below warning threshold.""" + return self._free_bytes < self.config.warning_threshold_bytes + + def is_critical(self) -> bool: + """True if free space is below critical threshold.""" + return self._free_bytes < self.config.critical_threshold_bytes + + def has_minimum_for_recording(self) -> bool: + """True if enough space to start a recording session. + + Checks that free space exceeds the minimum required plus + the safety buffer (estimated recording for safety_buffer_seconds). + """ + required = ( + self.config.minimum_required_bytes + + self.config.estimated_bytes_per_second * self.config.safety_buffer_seconds + ) + return self._free_bytes >= required + + def estimated_recording_time_seconds(self) -> float: + """Estimate how many more seconds can be recorded at current rate.""" + if self.config.estimated_bytes_per_second <= 0: + return float("inf") + available = max(0, self._free_bytes - self.config.minimum_required_bytes) + return available / self.config.estimated_bytes_per_second + + @property + def level(self) -> int: + """Current threshold level: 0=ok, 1=warning, 2=critical.""" + return self._current_level + + @property + def level_label(self) -> str: + return {0: "ok", 1: "warning", 2: "critical"}.get(self._current_level, "unknown") + + # ── Callback management ──────────────────────────────────────────────── + + def set_callback(self, callback: DiskSpaceCallback | None) -> None: + self._callback = callback + + # ── Internal ─────────────────────────────────────────────────────────── + + def _poll(self) -> None: + """Check disk space using statvfs.""" + try: + stat = os.statvfs(self.path) + self._free_bytes = stat.f_frsize * stat.f_bavail + self._total_bytes = stat.f_frsize * stat.f_blocks + self._last_check = time.monotonic() + + # Determine threshold level + old_level = self._current_level + if self._free_bytes < self.config.critical_threshold_bytes: + self._current_level = 2 + elif self._free_bytes < self.config.warning_threshold_bytes: + self._current_level = 1 + else: + self._current_level = 0 + + # Fire callback on level change + if self._current_level != old_level and self._callback: + try: + self._callback(self._free_bytes, self._total_bytes, self._current_level) + except Exception: + logger.error("Disk space callback error", exc_info=True) + + except OSError as exc: + logger.error("Disk space check failed for %s: %s", self.path, exc) + + def _monitor_loop(self) -> None: + """Background monitoring loop.""" + while self._running: + self._poll() + time.sleep(self.config.poll_interval_seconds) + + logger.debug("Disk monitor loop stopped") + + # ── Utility ───────────────────────────────────────────────────────────── + + @staticmethod + def get_free_bytes(path: str) -> int: + """Static helper: get free bytes on a path without instantiating.""" + stat = os.statvfs(path) + return stat.f_frsize * stat.f_bavail + + @staticmethod + def get_total_bytes(path: str) -> int: + """Static helper: get total bytes on a path.""" + stat = os.statvfs(path) + return stat.f_frsize * stat.f_blocks + + @staticmethod + def format_bytes(bytes_val: int) -> str: + """Human-readable byte count.""" + if bytes_val >= 1024 ** 3: + return f"{bytes_val / (1024**3):.1f} GB" + elif bytes_val >= 1024 ** 2: + return f"{bytes_val / (1024**2):.1f} MB" + elif bytes_val >= 1024: + return f"{bytes_val / 1024:.1f} KB" + return f"{bytes_val} B" diff --git a/src/recording/recorder.py b/src/recording/recorder.py new file mode 100644 index 0000000..247c4c0 --- /dev/null +++ b/src/recording/recorder.py @@ -0,0 +1,535 @@ +"""Multi-track recorder with disk streaming and write-ahead buffering. + +The MultiTrackRecorder captures audio from JACK capture ports and writes +each channel to a separate WAV file. A background writer thread drains +a buffer queue to disk, preventing SD card write latency from causing +xruns in the real-time audio thread. + +Key features: +- Per-channel recording with independent arm/punch states +- Write-ahead ring buffer (configurable size) to decouple capture from disk I/O +- Punch in/out: start/stop recording on individual channels mid-session +- Transport sync: start/stop/pause linked to JACK transport +- Take management: each recording session produces a take with metadata + +Usage: + rec = MultiTrackRecorder(session_dir="/data/session_001") + rec.arm_channel(0) + rec.arm_channel(1) + rec.start() # begin recording + # ... JACK process callback ... + rec.write(0, audio_chunk) # float32 numpy array + rec.write(1, audio_chunk) + rec.stop() # finalise files +""" + +from __future__ import annotations + +import logging +import queue +import threading +import time +from dataclasses import dataclass, field +from enum import Enum, auto +from pathlib import Path +from typing import Optional + +import numpy as np + +from .wav_writer import WAVWriter, WAVBitDepth + +logger = logging.getLogger(__name__) + + +# ── Types ───────────────────────────────────────────────────────────────────── + + +class RecorderState(Enum): + """Top-level recorder state.""" + IDLE = auto() + RECORDING = auto() + PAUSED = auto() + + +class ChannelState(Enum): + """Per-channel record state.""" + DISARMED = auto() # not recording, not armed + ARMED = auto() # ready to record when transport starts + RECORDING = auto() # actively capturing + + +@dataclass +class PunchRegion: + """A punch-in/out time range on a channel.""" + start_frame: int # frame index when punch-in occurred + end_frame: int | None # None if still recording + + +@dataclass +class ChannelRecordStatus: + """Per-channel recording metadata.""" + channel: int + state: ChannelState = ChannelState.DISARMED + writer: Optional[WAVWriter] = None + frames_recorded: int = 0 + punch_regions: list[PunchRegion] = field(default_factory=list) + current_punch_start: int | None = None + file_path: str = "" + + +@dataclass +class RecorderConfig: + """Configuration for the multi-track recorder.""" + # Audio + sample_rate: int = 48000 + channels: int = 16 + bit_depth: WAVBitDepth = WAVBitDepth.INT24 + + # Buffer + buffer_chunk_frames: int = 256 # frames per queue item + max_queue_size: int = 128 # max chunks per channel in buffer queue + writer_flush_interval_frames: int = 2048 # flush writer every N frames + + # Session + session_dir: str = "." + + # JACK + jack_port_prefix: str = "system" # capture port prefix + + +# ── Multi-track recorder ────────────────────────────────────────────────────── + + +class MultiTrackRecorder: + """Multi-track disk-streaming recorder. + + Architecture: + JACK process callback → rec.write(ch, samples) + → buffer_queue[ch].put(chunk) + Writer thread → buffer_queue[ch].get(chunk) + → WAVWriter.write_frames(chunk) + + The buffer queue decouples the real-time capture thread from + the SD card write latency, preventing xruns. + """ + + def __init__(self, config: RecorderConfig | None = None, *, session_dir: str | None = None): + self.config = config or RecorderConfig() + + if session_dir: + self.config.session_dir = session_dir + + # State + self._state: RecorderState = RecorderState.IDLE + self._start_time: float = 0.0 + self._global_frame: int = 0 + + # Per-channel status + self._channels: list[ChannelRecordStatus] = [ + ChannelRecordStatus(channel=i) + for i in range(self.config.channels) + ] + + # Write buffers: one queue per channel + # Each queue item is a tuple: (np.ndarray, int) where int is the + # starting global frame index (for later alignment/stamping) + self._buffer_queues: list[queue.Queue] = [ + queue.Queue(maxsize=self.config.max_queue_size) + for _ in range(self.config.channels) + ] + + # Background writer + self._writer_thread: Optional[threading.Thread] = None + self._writer_stop_event = threading.Event() + self._writer_error: Optional[Exception] = None + + # Thread safety + self._lock = threading.Lock() + + # Ensure session directory exists + Path(self.config.session_dir).mkdir(parents=True, exist_ok=True) + + # ── State ────────────────────────────────────────────────────────────── + + @property + def state(self) -> RecorderState: + return self._state + + @property + def is_recording(self) -> bool: + return self._state == RecorderState.RECORDING + + @property + def global_frame(self) -> int: + return self._global_frame + + def get_channel_state(self, channel: int) -> ChannelState: + if 0 <= channel < len(self._channels): + return self._channels[channel].state + return ChannelState.DISARMED + + # ── Channel arming ──────────────────────────────────────────────────── + + def arm_channel(self, channel: int) -> bool: + """Arm a channel for recording. + + Armed channels start recording when transport starts. + """ + if not 0 <= channel < len(self._channels): + logger.warning("Invalid channel index: %d", channel) + return False + + with self._lock: + ch = self._channels[channel] + if ch.state == ChannelState.DISARMED: + ch.state = ChannelState.ARMED + logger.debug("CH%d armed", channel) + return True + + def disarm_channel(self, channel: int) -> bool: + """Disarm a channel. Stops recording if active.""" + if not 0 <= channel < len(self._channels): + return False + + with self._lock: + ch = self._channels[channel] + if ch.state == ChannelState.RECORDING: + self._punch_out_channel(ch) + ch.state = ChannelState.DISARMED + logger.debug("CH%d disarmed", channel) + return True + + def is_armed(self, channel: int) -> bool: + if 0 <= channel < len(self._channels): + return self._channels[channel].state in (ChannelState.ARMED, ChannelState.RECORDING) + return False + + def get_armed_channels(self) -> list[int]: + return [ + i for i, ch in enumerate(self._channels) + if ch.state in (ChannelState.ARMED, ChannelState.RECORDING) + ] + + # ── Recording lifecycle ─────────────────────────────────────────────── + + def start(self) -> bool: + """Start recording on all armed channels.""" + if self._state == RecorderState.RECORDING: + logger.warning("Already recording") + return False + + with self._lock: + self._state = RecorderState.RECORDING + self._start_time = time.monotonic() + self._global_frame = 0 + self._writer_error = None + self._writer_stop_event.clear() + + # Start recording on all armed channels + active_count = 0 + for ch in self._channels: + if ch.state == ChannelState.ARMED: + self._start_channel_recording(ch) + active_count += 1 + + if active_count == 0: + logger.warning("No armed channels — starting with no active tracks") + + # Start background writer thread + self._writer_thread = threading.Thread( + target=self._writer_loop, + name="rec-writer", + daemon=True, + ) + self._writer_thread.start() + + logger.info("Recording started: %d channels, sr=%d, bit=%s", + active_count, self.config.sample_rate, self.config.bit_depth.name) + return True + + def stop(self) -> None: + """Stop recording and finalise all files.""" + if self._state == RecorderState.IDLE: + return + + with self._lock: + self._state = RecorderState.IDLE + + # Punch out all active channels + for ch in self._channels: + if ch.state == ChannelState.RECORDING: + self._punch_out_channel(ch) + + # Signal writer thread to finish + self._writer_stop_event.set() + + # Send sentinel values to all queues to unblock writer + for q in self._buffer_queues: + try: + q.put(None, timeout=0.1) + except queue.Full: + pass + + # Wait for writer thread + if self._writer_thread and self._writer_thread.is_alive(): + self._writer_thread.join(timeout=10.0) + if self._writer_thread.is_alive(): + logger.warning("Writer thread did not stop in 10s") + + # Close all writers + with self._lock: + for ch in self._channels: + if ch.writer: + ch.writer.close() + ch.writer = None + + total_frames = sum(ch.frames_recorded for ch in self._channels) + duration = self._global_frame / self.config.sample_rate if self.config.sample_rate else 0 + logger.info("Recording stopped: %d total frames (%.1fs)", + total_frames, duration) + + def pause(self) -> None: + """Pause recording without closing files.""" + if self._state != RecorderState.RECORDING: + return + with self._lock: + self._state = RecorderState.PAUSED + # Punch out all active channels + for ch in self._channels: + if ch.state == ChannelState.RECORDING: + self._punch_out_channel(ch) + logger.info("Recording paused at frame %d", self._global_frame) + + def resume(self) -> None: + """Resume recording after pause.""" + if self._state != RecorderState.PAUSED: + return + with self._lock: + self._state = RecorderState.RECORDING + # Punch in on previously recording channels + for ch in self._channels: + if ch.state == ChannelState.ARMED: + self._punch_in_channel(ch) + logger.info("Recording resumed at frame %d", self._global_frame) + + # ── Punch in/out ────────────────────────────────────────────────────── + + def punch_in(self, channel: int) -> bool: + """Start recording on a specific channel mid-session.""" + if self._state != RecorderState.RECORDING: + logger.warning("Cannot punch in while not recording") + return False + if not 0 <= channel < len(self._channels): + return False + + with self._lock: + ch = self._channels[channel] + if ch.state == ChannelState.DISARMED: + ch.state = ChannelState.ARMED # arm first + if ch.state == ChannelState.ARMED: + self._punch_in_channel(ch) + logger.info("CH%d punch in at frame %d", channel, self._global_frame) + return True + return False + + def punch_out(self, channel: int) -> bool: + """Stop recording on a specific channel mid-session.""" + if self._state != RecorderState.RECORDING: + logger.warning("Cannot punch out while not recording") + return False + if not 0 <= channel < len(self._channels): + return False + + with self._lock: + ch = self._channels[channel] + if ch.state == ChannelState.RECORDING: + self._punch_out_channel(ch) + logger.info("CH%d punch out at frame %d", channel, self._global_frame) + return True + return False + + def _start_channel_recording(self, ch: ChannelRecordStatus) -> None: + """Start recording on a channel that was armed.""" + self._punch_in_channel(ch) + + def _punch_in_channel(self, ch: ChannelRecordStatus) -> None: + """Internal: start recording on a channel.""" + # Close existing writer if any, create new file + if ch.writer: + ch.writer.close() + + # Generate file path for this punch region + punch_count = len(ch.punch_regions) + if punch_count == 0: + suffix = "" + else: + suffix = f"_punch{punch_count}" + + file_path = Path(self.config.session_dir) / f"ch{ch.channel:02d}{suffix}.wav" + + ch.writer = WAVWriter( + file_path, + sample_rate=self.config.sample_rate, + channels=1, # mono per channel + bit_depth=self.config.bit_depth, + ) + ch.file_path = str(file_path) + ch.state = ChannelState.RECORDING + ch.current_punch_start = self._global_frame + + # Register punch region + ch.punch_regions.append(PunchRegion( + start_frame=self._global_frame, + end_frame=None, + )) + + def _punch_out_channel(self, ch: ChannelRecordStatus) -> None: + """Internal: stop recording on a channel.""" + ch.state = ChannelState.ARMED + ch.current_punch_start = None + + # Close punch region + if ch.punch_regions: + region = ch.punch_regions[-1] + region.end_frame = self._global_frame + + # Flush and finalise writer + if ch.writer: + ch.writer.flush() + # Don't close — allow re-punch into same file + # For punch regions, we close and start new files each time + + # ── Audio write (real-time path) ───────────────────────────────────── + + def write(self, channel: int, samples: np.ndarray) -> bool: + """Write audio samples to a channel's buffer queue. + + Call this from the JACK process callback or audio capture thread. + This is the real-time path — it must never block on disk I/O. + + Args: + channel: 0-based channel index. + samples: float32 numpy array, shape (num_frames,). Values in [-1.0, 1.0]. + + Returns: + True if samples were queued, False if buffer full (drop). + """ + if self._state != RecorderState.RECORDING: + return False + + if not 0 <= channel < len(self._channels): + return False + + ch = self._channels[channel] + if ch.state != ChannelState.RECORDING: + return False + + if samples.ndim != 1: + samples = samples.ravel() + + # Copy samples (must not reference JACK buffer directly) + chunk = np.array(samples, dtype=np.float32, copy=True) + + try: + self._buffer_queues[channel].put_nowait(chunk) + return True + except queue.Full: + logger.warning("Buffer full for CH%d — dropping %d frames", + channel, len(chunk)) + return False + + def advance_frame(self, frames: int = 1) -> None: + """Advance the global frame counter. + + Call this after each buffer period to track recording position. + """ + self._global_frame += frames + + # ── Background writer thread ────────────────────────────────────────── + + def _writer_loop(self) -> None: + """Background thread: drain buffer queues and write to disk. + + Runs until _writer_stop_event is set and all queues are drained. + """ + logger.debug("Writer thread started") + + # Track which queues have been signalled (sentinel None received) + queues_signalled: set[int] = set() + num_queues = len(self._buffer_queues) + + try: + while not self._writer_stop_event.is_set() or len(queues_signalled) < num_queues: + wrote_anything = False + + for ch_idx, q in enumerate(self._buffer_queues): + if ch_idx in queues_signalled: + continue + + try: + chunk = q.get(timeout=0.01) + except queue.Empty: + continue + + if chunk is None: + # Sentinel: this queue is done + queues_signalled.add(ch_idx) + continue + + # Write chunk to disk + with self._lock: + ch = self._channels[ch_idx] + if ch.writer is not None: + ch.writer.write_frames(chunk) + ch.frames_recorded += len(chunk) + + wrote_anything = True + + if not wrote_anything and not self._writer_stop_event.is_set(): + # Short sleep to avoid busy-waiting when queues are empty + time.sleep(0.005) + + except Exception as exc: + self._writer_error = exc + logger.error("Writer thread error: %s", exc, exc_info=True) + finally: + logger.debug("Writer thread finished (wrote %d frames)", + sum(ch.frames_recorded for ch in self._channels)) + + # ── Status / info ───────────────────────────────────────────────────── + + @property + def writer_error(self) -> Optional[Exception]: + return self._writer_error + + def get_recording_stats(self) -> dict: + """Get recording statistics.""" + total_frames = sum(ch.frames_recorded for ch in self._channels) + return { + "state": self._state.name, + "global_frame": self._global_frame, + "total_frames_recorded": total_frames, + "elapsed_seconds": self._global_frame / self.config.sample_rate if self.config.sample_rate else 0, + "samplerate": self.config.sample_rate, + "channels": { + i: { + "state": ch.state.name, + "frames": ch.frames_recorded, + "punch_regions": len(ch.punch_regions), + "file": ch.file_path, + } + for i, ch in enumerate(self._channels) + if ch.state != ChannelState.DISARMED + }, + "buffer_queues": { + i: q.qsize() + for i, q in enumerate(self._buffer_queues) + if q.qsize() > 0 + }, + "writer_error": str(self._writer_error) if self._writer_error else None, + } + + def get_channel_record_status(self, channel: int) -> ChannelRecordStatus | None: + if 0 <= channel < len(self._channels): + return self._channels[channel] + return None diff --git a/src/recording/session.py b/src/recording/session.py new file mode 100644 index 0000000..ad67b6c --- /dev/null +++ b/src/recording/session.py @@ -0,0 +1,466 @@ +"""Session file format and take management. + +The session file is a JSON document that ties together the mixer state, +audio references, and recording metadata. Sessions are portable — you +can archive them, move them between machines, etc. + +Session format (JSON): + { + "version": 1, + "name": "Session Name", + "created_at": "2024-01-01T00:00:00", + "sample_rate": 48000, + "bit_depth": "int24", + "takes": [ ... ], + "mixer_state": { ... }, + "notes": "" + } + +Takes: + Each take is a recording pass. A take contains one or more track + files (WAV), punch region metadata, and per-track mixer snapshots. + Multiple takes can exist in a single session (e.g., multiple + recording passes of the same song). + +Usage: + session = Session.create("My Session", sample_rate=48000) + take = session.new_take("Take 1") + session.add_track(take.id, 0, "tracks/take_01_ch00.wav", mixer_snapshot) + session.save("/data/sessions/mysession.session.json") +""" + +from __future__ import annotations + +import json +import logging +import os +import time +import uuid +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +from .wav_writer import WAVBitDepth + +logger = logging.getLogger(__name__) + +# Current session file format version +SESSION_VERSION = 1 + + +# ── Data classes ────────────────────────────────────────────────────────────── + + +@dataclass +class PunchInfo: + """Metadata for a punch region within a track.""" + start_frame: int + end_frame: int + + +@dataclass +class TrackInfo: + """Metadata for one recorded track within a take.""" + channel: int # 0-based mixer channel + file_path: str # relative path to WAV file (from session dir) + frames: int = 0 # total recorded frames + punch_regions: list[PunchInfo] = field(default_factory=list) + mixer_snapshot: dict = field(default_factory=dict) # channel mixer state at record time + + +@dataclass +class Take: + """A recording take with one or more tracks.""" + id: str # unique take ID + name: str = "" # human-readable label + created_at: str = "" # ISO-8601 timestamp + tracks: list[TrackInfo] = field(default_factory=list) + sample_rate: int = 48000 + bit_depth: str = "int24" + notes: str = "" # free-form notes + start_frame: int = 0 # global frame when recording started + end_frame: int = 0 # global frame when recording ended + duration_seconds: float = 0.0 # derived from frames / sample_rate + + def add_track(self, track: TrackInfo) -> None: + """Add a track to this take.""" + self.tracks.append(track) + + def get_track(self, channel: int) -> TrackInfo | None: + """Get a track by channel number.""" + for t in self.tracks: + if t.channel == channel: + return t + return None + + def get_track_file_paths(self) -> list[str]: + """Get all track file paths.""" + return [t.file_path for t in self.tracks] + + def total_frames(self) -> int: + """Total frames across all tracks.""" + if self.end_frame is not None and self.start_frame is not None: + return self.end_frame - self.start_frame + return max((t.frames for t in self.tracks), default=0) + + def to_dict(self) -> dict: + """Serialize take to a dict.""" + return { + "id": self.id, + "name": self.name, + "created_at": self.created_at, + "sample_rate": self.sample_rate, + "bit_depth": self.bit_depth, + "tracks": [ + { + "channel": t.channel, + "file_path": t.file_path, + "frames": t.frames, + "punch_regions": [ + {"start_frame": p.start_frame, "end_frame": p.end_frame} + for p in t.punch_regions + ], + "mixer_snapshot": t.mixer_snapshot, + } + for t in self.tracks + ], + "notes": self.notes, + "start_frame": self.start_frame, + "end_frame": self.end_frame, + "duration_seconds": self.duration_seconds, + } + + @classmethod + def from_dict(cls, data: dict) -> Take: + """Deserialize a Take from a dict.""" + tracks = [] + for td in data.get("tracks", []): + punch_regions = [ + PunchInfo(start_frame=p["start_frame"], end_frame=p["end_frame"]) + for p in td.get("punch_regions", []) + ] + tracks.append(TrackInfo( + channel=td["channel"], + file_path=td["file_path"], + frames=td.get("frames", 0), + punch_regions=punch_regions, + mixer_snapshot=td.get("mixer_snapshot", {}), + )) + + return cls( + id=data["id"], + name=data.get("name", ""), + created_at=data.get("created_at", ""), + tracks=tracks, + sample_rate=data.get("sample_rate", 48000), + bit_depth=data.get("bit_depth", "int24"), + notes=data.get("notes", ""), + start_frame=data.get("start_frame", 0), + end_frame=data.get("end_frame", 0), + duration_seconds=data.get("duration_seconds", 0.0), + ) + + +# ── Session ────────────────────────────────────────────────────────────────── + + +@dataclass +class Session: + """A recording session. + + A session contains: + - Session metadata (name, created_at, sample_rate, bit_depth) + - Takes (multiple recording passes) + - Mixer state snapshot + - Free-form notes + + Sessions are serialised as JSON files. Audio files are referenced + by relative paths from the session file's directory. + """ + + version: int = SESSION_VERSION + name: str = "Untitled Session" + created_at: str = "" # ISO-8601 + modified_at: str = "" # ISO-8601 + sample_rate: int = 48000 + bit_depth: str = "int24" # "int16", "int24", "int32", "float32" + takes: list[Take] = field(default_factory=list) + mixer_state: dict = field(default_factory=dict) + notes: str = "" + _file_path: Optional[str] = None # path to the session file, set after load/save + + @classmethod + def create( + cls, + name: str, + *, + sample_rate: int = 48000, + bit_depth: str | WAVBitDepth = "int24", + ) -> Session: + """Create a new session.""" + if isinstance(bit_depth, WAVBitDepth): + bit_depth = _bit_depth_to_str(bit_depth) + + now = _now_iso() + return cls( + name=name, + created_at=now, + modified_at=now, + sample_rate=sample_rate, + bit_depth=bit_depth, + ) + + # ── Take management ───────────────────────────────────────────────── + + def new_take( + self, + name: str = "", + *, + start_frame: int = 0, + end_frame: int = 0, + ) -> Take: + """Create a new take in this session. + + Args: + name: Human-readable label (e.g. "Take 1", "Vocal overdub"). + start_frame: Global frame when recording started. + end_frame: Global frame when recording ended (0 if not yet known). + + Returns: + The new Take. + """ + take = Take( + id=f"take_{uuid.uuid4().hex[:12]}", + name=name or f"Take {len(self.takes) + 1}", + created_at=_now_iso(), + sample_rate=self.sample_rate, + bit_depth=self.bit_depth, + start_frame=start_frame, + end_frame=end_frame, + duration_seconds=(end_frame - start_frame) / self.sample_rate if end_frame > start_frame and self.sample_rate > 0 else 0.0, + ) + self.takes.append(take) + self.modified_at = _now_iso() + logger.info("New take created: %s (%s)", take.name, take.id) + return take + + def get_take(self, take_id: str) -> Take | None: + """Get a take by ID.""" + for t in self.takes: + if t.id == take_id: + return t + return None + + def delete_take(self, take_id: str) -> bool: + """Delete a take (does NOT delete WAV files).""" + for i, t in enumerate(self.takes): + if t.id == take_id: + del self.takes[i] + self.modified_at = _now_iso() + logger.info("Take deleted: %s", take_id) + return True + return False + + def rename_take(self, take_id: str, new_name: str) -> bool: + """Rename a take.""" + take = self.get_take(take_id) + if take: + take.name = new_name + self.modified_at = _now_iso() + return True + return False + + def list_takes(self) -> list[Take]: + """Return all takes, ordered by creation time.""" + return list(self.takes) + + def get_last_take(self) -> Take | None: + """Get the most recent take.""" + return self.takes[-1] if self.takes else None + + # ── Track management ──────────────────────────────────────────────── + + def add_track( + self, + take_id: str, + channel: int, + file_path: str, + mixer_snapshot: dict | None = None, + frames: int = 0, + ) -> TrackInfo | None: + """Add a track to an existing take. + + Args: + take_id: The take to add the track to. + channel: 0-based mixer channel number. + file_path: Relative path to the WAV file. + mixer_snapshot: Channel mixer state at recording time. + frames: Number of recorded frames. + + Returns: + The new TrackInfo, or None if take not found. + """ + take = self.get_take(take_id) + if not take: + logger.warning("Take not found: %s", take_id) + return None + + track = TrackInfo( + channel=channel, + file_path=file_path, + frames=frames, + mixer_snapshot=mixer_snapshot or {}, + ) + take.add_track(track) + self.modified_at = _now_iso() + return track + + def get_all_track_files(self) -> list[str]: + """Get all track WAV file paths across all takes.""" + paths = [] + for take in self.takes: + paths.extend(take.get_track_file_paths()) + return paths + + # ── Serialisation ─────────────────────────────────────────────────── + + def to_dict(self) -> dict: + """Serialize session to a dict.""" + return { + "version": self.version, + "name": self.name, + "created_at": self.created_at, + "modified_at": self.modified_at, + "sample_rate": self.sample_rate, + "bit_depth": self.bit_depth, + "takes": [t.to_dict() for t in self.takes], + "mixer_state": self.mixer_state, + "notes": self.notes, + } + + @classmethod + def from_dict(cls, data: dict) -> Session: + """Deserialize a Session from a dict.""" + takes = [Take.from_dict(td) for td in data.get("takes", [])] + return cls( + version=data.get("version", SESSION_VERSION), + name=data.get("name", "Untitled Session"), + created_at=data.get("created_at", ""), + modified_at=data.get("modified_at", ""), + sample_rate=data.get("sample_rate", 48000), + bit_depth=data.get("bit_depth", "int24"), + takes=takes, + mixer_state=data.get("mixer_state", {}), + notes=data.get("notes", ""), + ) + + def save(self, file_path: str | Path | None = None) -> Path: + """Save the session to a JSON file. + + Args: + file_path: Target file path. If None, uses the path from a + previous load() or save() call. + + Returns: + Path to the saved file. + """ + if file_path is None: + if self._file_path: + file_path = self._file_path + else: + raise ValueError("No file path specified and no previous path known") + + file_path = Path(file_path) + self.modified_at = _now_iso() + + file_path.parent.mkdir(parents=True, exist_ok=True) + + data = self.to_dict() + + with open(file_path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + + self._file_path = str(file_path) + logger.info("Session saved: %s (%d takes)", file_path, len(self.takes)) + return file_path + + @classmethod + def load(cls, file_path: str | Path) -> Session: + """Load a session from a JSON file. + + Args: + file_path: Path to the session JSON file. + + Returns: + The loaded Session. + """ + file_path = Path(file_path) + + with open(file_path, "r", encoding="utf-8") as f: + data = json.load(f) + + session = cls.from_dict(data) + session._file_path = str(file_path) + logger.info("Session loaded: %s (%d takes)", file_path, len(session.takes)) + return session + + # ── Utility ────────────────────────────────────────────────────────── + + @property + def take_count(self) -> int: + return len(self.takes) + + @property + def total_recorded_frames(self) -> int: + """Total recorded frames across all takes.""" + return sum(t.total_frames() for t in self.takes) + + @property + def total_duration_seconds(self) -> float: + """Total recording duration across all takes.""" + if self.sample_rate == 0: + return 0.0 + return self.total_recorded_frames / self.sample_rate + + @property + def directory(self) -> Path | None: + """The directory containing the session file, or None if unsaved.""" + if self._file_path: + return Path(self._file_path).parent + return None + + def resolve_track_path(self, track: TrackInfo) -> Path: + """Resolve a relative track file path to an absolute path. + + Relative paths are resolved against the session file's directory. + """ + if os.path.isabs(track.file_path): + return Path(track.file_path) + if self._file_path: + return Path(self._file_path).parent / track.file_path + return Path(track.file_path) + + def update_mixer_state(self, state: dict) -> None: + """Update the mixer state snapshot.""" + self.mixer_state = state + self.modified_at = _now_iso() + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _now_iso() -> str: + """Return current UTC time as ISO-8601 string.""" + return datetime.now(timezone.utc).isoformat() + + +def _bit_depth_to_str(bit_depth: WAVBitDepth) -> str: + """Convert WAVBitDepth enum to string representation.""" + mapping = { + WAVBitDepth.INT16: "int16", + WAVBitDepth.INT24: "int24", + WAVBitDepth.INT32: "int32", + WAVBitDepth.FLOAT32: "float32", + } + return mapping.get(bit_depth, "int24") diff --git a/src/recording/wav_writer.py b/src/recording/wav_writer.py new file mode 100644 index 0000000..c6e7d30 --- /dev/null +++ b/src/recording/wav_writer.py @@ -0,0 +1,395 @@ +"""Streaming WAV file writer with configurable bit depth and sample rate. + +Writes standard RIFF/WAVE files in chunks for low-latency recording. +Supports 16-bit integer, 24-bit integer, 32-bit integer, and 32-bit +float formats. Designed for SD card streaming — writes are buffered +and flushed to minimise seek operations. + +Usage: + writer = WAVWriter("track_01.wav", sample_rate=48000, channels=1, bit_depth=24) + writer.write_frames(frames) # np.ndarray, float32 in [-1.0, 1.0] + writer.close() # finalises header +""" + +from __future__ import annotations + +import logging +import struct +from dataclasses import dataclass +from enum import IntEnum +from pathlib import Path +from typing import BinaryIO, Callable, Optional + +import numpy as np + +try: + import soundfile as sf + HAS_SOUNDFILE = True +except ImportError: + HAS_SOUNDFILE = False + +logger = logging.getLogger(__name__) + + +# ── WAV format constants ────────────────────────────────────────────────────── + +class WAVBitDepth(IntEnum): + """Supported WAV bit depths.""" + INT16 = 16 + INT24 = 24 + INT32 = 32 + FLOAT32 = -32 # negative to distinguish from int32 + + +# PCM format tags +_WAV_FORMAT_PCM = 1 # integer PCM +_WAV_FORMAT_IEEE_FLOAT = 3 # IEEE 754 float + + +# ── Sample conversion ───────────────────────────────────────────────────────── + +def _float32_to_int16(samples: np.ndarray) -> np.ndarray: + """Convert float32 [-1.0, 1.0] → int16.""" + clamped = np.clip(samples, -1.0, 1.0) + return (clamped * 32767.0).astype(np.int16) + + +def _float32_to_int24(samples: np.ndarray) -> bytes: + """Convert float32 [-1.0, 1.0] → packed 24-bit little-endian bytes. + + Each sample occupies 3 bytes. Numpy has no int24 dtype, so we pack + manually via int32 scaling and struct. + """ + clamped = np.clip(samples, -1.0, 1.0) + scaled = (clamped * 8388607.0).astype(np.int32) # 2^23 - 1 + # Pack each sample as 3-byte little-endian + data = bytearray(len(scaled) * 3) + for i, val in enumerate(scaled): + # Mask to 24 bits signed, then pack + packed = val & 0xFFFFFF + if val < 0: + packed |= 0x800000 # sign-extend + data[i * 3] = packed & 0xFF + data[i * 3 + 1] = (packed >> 8) & 0xFF + data[i * 3 + 2] = (packed >> 16) & 0xFF + return bytes(data) + + +_FAST_INT24 = False # set by first call to _float32_to_int24_fast + + +def _float32_to_int24_fast(samples: np.ndarray) -> np.ndarray: + """Convert float32 → 24-bit packed as 3-byte little-endian using numpy vectorisation. + + Faster than per-sample struct packing. Returns uint8 array of shape (n*3,). + """ + clamped = np.clip(samples, -1.0, 1.0) + # Scale to 24-bit signed range + scaled = np.round(clamped * 8388607.0).astype(np.int32) + # Clamp to 24-bit signed range + scaled = np.clip(scaled, -8388608, 8388607) + # Convert to unsigned 24-bit for packing + mask = np.where(scaled < 0, scaled + 16777216, scaled).astype(np.uint32) + + n = len(mask) + packed = np.empty(n * 3, dtype=np.uint8) + packed[0::3] = (mask & 0xFF).astype(np.uint8) + packed[1::3] = ((mask >> 8) & 0xFF).astype(np.uint8) + packed[2::3] = ((mask >> 16) & 0xFF).astype(np.uint8) + return packed + + +def _float32_to_int32(samples: np.ndarray) -> np.ndarray: + """Convert float32 [-1.0, 1.0] → int32.""" + clamped = np.clip(samples, -1.0, 1.0) + # Use float64 to preserve precision at the int32 range boundary + scaled = (clamped.astype(np.float64) * 2147483647.0) + return np.clip(np.round(scaled), -2147483648, 2147483647).astype(np.int32) + + +def _float32_to_float32(samples: np.ndarray) -> np.ndarray: + """Pass through float32 (already correct format).""" + return samples.astype(np.float32) + + +# ── WAV writer ─────────────────────────────────────────────────────────────── + + +@dataclass +class WAVWriterConfig: + """Configuration for a WAV writer.""" + sample_rate: int = 48000 + channels: int = 1 + bit_depth: WAVBitDepth = WAVBitDepth.INT24 + # Internal buffering: flush every N bytes (0 = only at close) + flush_interval_bytes: int = 65536 # 64KB + + +class WAVWriter: + """Streaming WAV file writer. + + Opens a file, writes the initial header with placeholder sizes, + accepts float32 audio frames in chunks, and finalises the header + on close. + + Usage:: + + with WAVWriter("output.wav", sample_rate=48000, channels=2, bit_depth=24) as w: + w.write_frames(audio_chunk) # shape (frames, channels), float32 + """ + + def __init__( + self, + file_path: str | Path, + *, + sample_rate: int = 48000, + channels: int = 1, + bit_depth: WAVBitDepth | int = WAVBitDepth.INT24, + config: Optional[WAVWriterConfig] = None, + ): + self.file_path = Path(file_path) + self.sample_rate = sample_rate + self.channels = channels + self.bit_depth = WAVBitDepth(bit_depth) if isinstance(bit_depth, int) else bit_depth + self.config = config or WAVWriterConfig( + sample_rate=sample_rate, channels=channels, bit_depth=self.bit_depth + ) + + self._file: Optional[BinaryIO] = None + self._frames_written: int = 0 + self._data_bytes_written: int = 0 + self._bytes_since_flush: int = 0 + self._closed: bool = False + self._header_written: bool = False + + # Select conversion function based on bit depth + self._convert = _CONVERTERS.get(self.bit_depth) + if self._convert is None: + raise ValueError(f"Unsupported bit depth: {self.bit_depth}") + + # Determine bytes per sample frame (all channels) + abs_depth = abs(int(self.bit_depth)) + self._bytes_per_sample = abs_depth // 8 + self._bytes_per_frame = self._bytes_per_sample * channels + + # Open the file + self._open() + + def __enter__(self) -> WAVWriter: + return self + + def __exit__(self, *args) -> None: + self.close() + + # ── File I/O ────────────────────────────────────────────────────────── + + def _open(self) -> None: + """Create the file and write placeholder header.""" + self.file_path.parent.mkdir(parents=True, exist_ok=True) + self._file = open(self.file_path, "wb") + self._write_placeholder_header() + + def _write_placeholder_header(self) -> None: + """Write RIFF/WAVE header with placeholder (0) data size. + + The data size and overall file size will be patched on close. + """ + f = self._file + assert f is not None, "File must be open before writing header" + + # Determine format + abs_depth = abs(int(self.bit_depth)) + if self.bit_depth == WAVBitDepth.FLOAT32: + audio_format = _WAV_FORMAT_IEEE_FLOAT + else: + audio_format = _WAV_FORMAT_PCM + + byte_rate = self.sample_rate * self.channels * self._bytes_per_sample + block_align = self.channels * self._bytes_per_sample + + # RIFF header + f.write(b"RIFF") + f.write(struct.pack(" int: + """Write audio frames to the file. + + Args: + frames: float32 numpy array, shape (num_frames, channels) or (num_frames,). + Values should be in [-1.0, 1.0]. + + Returns: + Number of frames written. + + Raises: + ValueError: If writer is already closed. + """ + if self._closed: + raise ValueError("Cannot write to closed WAVWriter") + if self._file is None: + raise RuntimeError("File not open") + + # Ensure 2D + if frames.ndim == 1: + frames = frames.reshape(-1, 1) + + num_frames = frames.shape[0] + + if num_frames == 0: + return 0 + + # Handle channel mismatch + if frames.shape[1] != self.channels: + if frames.shape[1] == 1 and self.channels > 1: + # Mono → multi: duplicate + frames = np.tile(frames, (1, self.channels)) + elif self.channels == 1 and frames.shape[1] > 1: + # Multi → mono: take first channel + frames = frames[:, 0:1] + else: + raise ValueError( + f"Frame channels ({frames.shape[1]}) != writer channels ({self.channels})" + ) + + # Interleave channels: (num_frames, channels) → flat + interleaved = frames.ravel() # row-major = interleaved + + # Convert to target format + data = self._convert(interleaved) + + # Write + if isinstance(data, np.ndarray): + self._file.write(data.tobytes()) + byte_count = data.nbytes + else: + self._file.write(data) + byte_count = len(data) + + self._frames_written += num_frames + self._data_bytes_written += byte_count + self._bytes_since_flush += byte_count + + # Periodic flush + if self.config.flush_interval_bytes > 0 and self._bytes_since_flush >= self.config.flush_interval_bytes: + self._file.flush() + self._bytes_since_flush = 0 + + return num_frames + + def flush(self) -> None: + """Force flush buffered data to disk.""" + if self._file and not self._closed: + self._file.flush() + self._bytes_since_flush = 0 + + def close(self) -> None: + """Finalise header and close the file.""" + if self._closed: + return + self._closed = True + + if self._file is None: + return + + try: + # Flush remaining data + self._file.flush() + + # Patch header sizes + data_size = self._data_bytes_written + file_size = 36 + data_size # 44 - 8 = 36 (RIFF header after "WAVE") + + self._file.seek(4) + self._file.write(struct.pack(" int: + return self._frames_written + + @property + def duration_seconds(self) -> float: + return self._frames_written / self.sample_rate if self.sample_rate > 0 else 0.0 + + @property + def closed(self) -> bool: + return self._closed + + +# ── Converter registry ─────────────────────────────────────────────────────── + +_SampleConverter = Callable[[np.ndarray], np.ndarray] + +_CONVERTERS: dict[WAVBitDepth, _SampleConverter] = { + WAVBitDepth.INT16: _float32_to_int16, + WAVBitDepth.INT24: _float32_to_int24_fast, + WAVBitDepth.INT32: _float32_to_int32, + WAVBitDepth.FLOAT32: _float32_to_float32, +} + + +# ── Utility: quick WAV write ───────────────────────────────────────────────── + +def write_wav( + file_path: str | Path, + samples: np.ndarray, + sample_rate: int = 48000, + bit_depth: WAVBitDepth | int = WAVBitDepth.INT24, +) -> Path: + """Write a complete WAV file in one shot (convenience wrapper). + + Args: + file_path: Output path. + samples: float32 audio data, shape (frames,) or (frames, channels). + sample_rate: Sample rate in Hz. + bit_depth: Target bit depth. + + Returns: + Path to written file. + """ + if samples.ndim == 1: + channels = 1 + else: + channels = samples.shape[1] + + with WAVWriter(file_path, sample_rate=sample_rate, channels=channels, bit_depth=bit_depth) as w: + w.write_frames(samples) + + return Path(file_path) diff --git a/tests/test_recording.py b/tests/test_recording.py new file mode 100644 index 0000000..7c7142f --- /dev/null +++ b/tests/test_recording.py @@ -0,0 +1,1436 @@ +"""Tests for the recording engine — WAV writer, recorder, session, disk monitor, bounce. + +Tests are designed to run without JACK — they validate the logic, +state management, file I/O, and audio processing without requiring +actual audio hardware. +""" + +from __future__ import annotations + +import json +import os +import queue +import struct +import tempfile +import threading +import time +from pathlib import Path + +import numpy as np +import pytest + +from src.recording.wav_writer import ( + WAVWriter, + WAVWriterConfig, + WAVBitDepth, + write_wav, + _float32_to_int16, + _float32_to_int24_fast, + _float32_to_int32, + _float32_to_float32, +) +from src.recording.recorder import ( + MultiTrackRecorder, + RecorderConfig, + RecorderState, + ChannelState as RecChanState, + ChannelRecordStatus, + PunchRegion, +) +from src.recording.session import ( + Session, + Take, + TrackInfo, + PunchInfo, + _now_iso, +) +from src.recording.disk_monitor import ( + DiskMonitor, + DiskMonitorConfig, + DiskSpaceCallback, +) +from src.recording.bounce import ( + BounceEngine, + BounceConfig, + BounceTrack, + BounceResult, + db_to_linear, + linear_to_db, + _pan_to_gains, +) + + +# ═════════════════════════════════════════════════════════════════════════════ +# Helpers +# ═════════════════════════════════════════════════════════════════════════════ + + +def _make_sine(freq: float, sample_rate: int, duration: float, amplitude: float = 0.5) -> np.ndarray: + """Generate a sine wave as float32 numpy array.""" + t = np.arange(int(sample_rate * duration), dtype=np.float32) / sample_rate + return (amplitude * np.sin(2.0 * np.pi * freq * t)).astype(np.float32) + + +def _read_wav_header(file_path: Path) -> dict: + """Read RIFF/WAVE header fields for validation.""" + with open(file_path, "rb") as f: + riff = f.read(4) + file_size = struct.unpack(" np.ndarray: + """Read WAV file back as float32 using soundfile or custom code.""" + try: + import soundfile as sf + samples, _ = sf.read(str(file_path), dtype="float32") + return samples + except ImportError: + raise RuntimeError("soundfile required for test") + + +# ═════════════════════════════════════════════════════════════════════════════ +# WAVWriter tests +# ═════════════════════════════════════════════════════════════════════════════ + + +class TestSampleConversion: + """Test float32 → integer sample conversion.""" + + def test_int16_zero(self): + out = _float32_to_int16(np.array([0.0], dtype=np.float32)) + assert out[0] == 0 + + def test_int16_full_scale(self): + out = _float32_to_int16(np.array([1.0, -1.0], dtype=np.float32)) + assert out[0] == 32767 + assert out[1] == -32767 + + def test_int16_clipping(self): + out = _float32_to_int16(np.array([2.0, -2.0], dtype=np.float32)) + assert out[0] == 32767 + assert out[1] == -32767 + + def test_int24_fast_roundtrip_range(self): + """Int24 should produce correct byte count and values within range.""" + sine = _make_sine(440, 48000, 0.1, 0.9) + packed = _float32_to_int24_fast(sine) + assert len(packed) == len(sine) * 3 # 3 bytes per sample + assert packed.dtype == np.uint8 + + def test_int24_fast_zero(self): + packed = _float32_to_int24_fast(np.array([0.0], dtype=np.float32)) + assert len(packed) == 3 + assert packed[0] == 0 and packed[1] == 0 and packed[2] == 0 + + def test_int24_fast_positive(self): + packed = _float32_to_int24_fast(np.array([1.0], dtype=np.float32)) + assert len(packed) == 3 + # Reconstruct 24-bit value from 3 uint8 bytes + b0 = int(packed[0]) + b1 = int(packed[1]) + b2 = int(packed[2]) + val = b0 | (b1 << 8) | (b2 << 16) + # Should be close to 8388607 (max positive 24-bit) + assert val in (8388607, 8388606) # rounding variation + + def test_int24_fast_negative(self): + packed = _float32_to_int24_fast(np.array([-1.0], dtype=np.float32)) + assert len(packed) == 3 + # Read as signed 24-bit + b0 = int(packed[0]) + b1 = int(packed[1]) + b2 = int(packed[2]) + val = b0 | (b1 << 8) | (b2 << 16) + if val >= 0x800000: + val -= 0x1000000 + assert val in (-8388608, -8388607) + + def test_int32_range(self): + out = _float32_to_int32(np.array([1.0, -1.0], dtype=np.float32)) + assert out[0] == 2147483647 + assert out[1] == -2147483647 + + def test_float32_passthrough(self): + vals = np.array([0.5, -0.5], dtype=np.float32) + out = _float32_to_float32(vals) + np.testing.assert_array_equal(out, vals) + + +class TestWAVWriterBasic: + """Test WAV file creation and header correctness.""" + + def test_write_mono_int16(self, tmp_path): + file_path = tmp_path / "test_int16.wav" + sine = _make_sine(440, 48000, 0.5) + + with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT16) as w: + w.write_frames(sine) + + assert file_path.exists() + assert file_path.stat().st_size > 44 # header + data + + header = _read_wav_header(file_path) + assert header["riff"] == b"RIFF" + assert header["wave"] == b"WAVE" + assert header["audio_format"] == 1 # PCM + assert header["channels"] == 1 + assert header["sample_rate"] == 48000 + assert header["bits_per_sample"] == 16 + + def test_write_mono_int24(self, tmp_path): + file_path = tmp_path / "test_int24.wav" + sine = _make_sine(440, 48000, 0.5) + + with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT24) as w: + w.write_frames(sine) + + header = _read_wav_header(file_path) + assert header["bits_per_sample"] == 24 + + # Read back and verify it's valid audio + samples = _read_wav_samples(file_path) + assert len(samples) == len(sine) + # Check correlation (should be very high) + corr = np.corrcoef(samples, sine)[0, 1] + assert corr > 0.95 + + def test_write_mono_int32(self, tmp_path): + file_path = tmp_path / "test_int32.wav" + sine = _make_sine(440, 48000, 0.5) + + with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT32) as w: + w.write_frames(sine) + + header = _read_wav_header(file_path) + assert header["bits_per_sample"] == 32 + + def test_write_mono_float32(self, tmp_path): + file_path = tmp_path / "test_float32.wav" + sine = _make_sine(440, 48000, 0.5) + + with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.FLOAT32) as w: + w.write_frames(sine) + + header = _read_wav_header(file_path) + assert header["audio_format"] == 3 # IEEE float + assert header["bits_per_sample"] == 32 + + def test_write_stereo(self, tmp_path): + file_path = tmp_path / "test_stereo.wav" + left = _make_sine(440, 48000, 0.5, 0.5) + right = _make_sine(880, 48000, 0.5, 0.3) + stereo = np.column_stack([left, right]) + + with WAVWriter(file_path, sample_rate=48000, channels=2, bit_depth=WAVBitDepth.INT16) as w: + w.write_frames(stereo) + + header = _read_wav_header(file_path) + assert header["channels"] == 2 + assert header["block_align"] == 4 # 2ch * 2 bytes + + def test_multi_chunk_write(self, tmp_path): + """Test writing in multiple chunks produces valid output.""" + file_path = tmp_path / "test_chunks.wav" + total_frames = 4800 # 100ms at 48kHz + chunk_size = 256 + + with WAVWriter(file_path, sample_rate=48000, channels=1, bit_depth=WAVBitDepth.INT24) as w: + for start in range(0, total_frames, chunk_size): + end = min(start + chunk_size, total_frames) + chunk = _make_sine(440, 48000, (end - start) / 48000) + w.write_frames(chunk) + + assert w.frames_written == total_frames + + samples = _read_wav_samples(file_path) + assert len(samples) == total_frames + + def test_context_manager(self, tmp_path): + """Test __enter__/__exit__.""" + file_path = tmp_path / "test_ctx.wav" + with WAVWriter(file_path, sample_rate=48000, channels=1) as w: + w.write_frames(_make_sine(440, 48000, 0.1)) + assert w.closed + assert file_path.exists() + + def test_close_idempotent(self, tmp_path): + """Calling close() multiple times is safe.""" + file_path = tmp_path / "test_close.wav" + w = WAVWriter(file_path) + w.write_frames(_make_sine(440, 48000, 0.1)) + w.close() + w.close() # should not raise + assert w.closed + + def test_write_after_close_raises(self, tmp_path): + file_path = tmp_path / "test_err.wav" + w = WAVWriter(file_path) + w.close() + with pytest.raises(ValueError): + w.write_frames(_make_sine(440, 48000, 0.1)) + + def test_duration_seconds(self, tmp_path): + file_path = tmp_path / "test_dur.wav" + duration = 0.25 + with WAVWriter(file_path, sample_rate=48000) as w: + w.write_frames(_make_sine(440, 48000, duration)) + assert w.duration_seconds == pytest.approx(duration, abs=0.001) + + def test_variable_sample_rate(self, tmp_path): + """Test 44100 Hz and 96000 Hz sample rates.""" + for sr in [44100, 96000]: + file_path = tmp_path / f"test_{sr}.wav" + with WAVWriter(file_path, sample_rate=sr) as w: + w.write_frames(_make_sine(440, sr, 0.1)) + header = _read_wav_header(file_path) + assert header["sample_rate"] == sr + + def test_1d_input_broadcast_to_stereo(self, tmp_path): + """Mono input to stereo writer should duplicate.""" + file_path = tmp_path / "test_broadcast.wav" + mono = _make_sine(440, 48000, 0.1) + with WAVWriter(file_path, sample_rate=48000, channels=2) as w: + w.write_frames(mono) + + samples = _read_wav_samples(file_path) + assert samples.ndim == 2 + assert samples.shape[1] == 2 + np.testing.assert_array_almost_equal(samples[:, 0], samples[:, 1]) + + def test_flush(self, tmp_path): + """Flush should write data to disk without closing.""" + file_path = tmp_path / "test_flush.wav" + w = WAVWriter(file_path) + w.write_frames(_make_sine(440, 48000, 0.1)) + w.flush() + assert file_path.stat().st_size > 44 # data should be on disk + w.close() + + def test_empty_write(self, tmp_path): + """Writing zero frames should not fail.""" + file_path = tmp_path / "test_empty.wav" + with WAVWriter(file_path, sample_rate=48000) as w: + w.write_frames(np.array([], dtype=np.float32)) + assert w.frames_written == 0 + + def test_mismatched_channels_auto_adapt(self, tmp_path): + """Writer with 1 channel should take first channel from multi-channel input.""" + file_path = tmp_path / "test_adapt.wav" + stereo = np.column_stack([ + _make_sine(440, 48000, 0.1, 0.5), + _make_sine(880, 48000, 0.1, 0.3), + ]) + with WAVWriter(file_path, sample_rate=48000, channels=1) as w: + w.write_frames(stereo) + + samples = _read_wav_samples(file_path) + assert samples.ndim == 1 + + +class TestWAVWriterConfig: + """Test WAVWriterConfig behaviour.""" + + def test_default_flush_interval(self, tmp_path): + config = WAVWriterConfig(flush_interval_bytes=100) + file_path = tmp_path / "test_config.wav" + with WAVWriter(file_path, config=config) as w: + w.write_frames(_make_sine(440, 48000, 0.01)) # ~480 samples + assert file_path.exists() + + +class TestWriteWAVConvenience: + """Test the write_wav convenience function.""" + + def test_write_wav_mono(self, tmp_path): + file_path = tmp_path / "quick.wav" + sine = _make_sine(440, 48000, 0.5) + result = write_wav(file_path, sine, sample_rate=48000, bit_depth=WAVBitDepth.INT16) + assert result.exists() + assert result.suffix == ".wav" + + def test_write_wav_stereo(self, tmp_path): + file_path = tmp_path / "quick_stereo.wav" + left = _make_sine(440, 48000, 0.2) + right = _make_sine(880, 48000, 0.2) + stereo = np.column_stack([left, right]) + write_wav(file_path, stereo, sample_rate=48000) + assert file_path.exists() + + +# ═════════════════════════════════════════════════════════════════════════════ +# MultiTrackRecorder tests +# ═════════════════════════════════════════════════════════════════════════════ + + +class TestRecorderConfig: + """Test RecorderConfig defaults.""" + + def test_defaults(self): + cfg = RecorderConfig() + assert cfg.sample_rate == 48000 + assert cfg.channels == 16 + assert cfg.bit_depth == WAVBitDepth.INT24 + + def test_custom_session_dir(self): + cfg = RecorderConfig(session_dir="/tmp/test") + assert cfg.session_dir == "/tmp/test" + + +class TestMultiTrackRecorderBasic: + """Test recorder creation and state management.""" + + def test_initial_state(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert rec.state == RecorderState.IDLE + assert not rec.is_recording + assert rec.global_frame == 0 + + def test_arm_disarm(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert rec.arm_channel(0) + assert rec.is_armed(0) + assert rec.get_channel_state(0) == RecChanState.ARMED + + rec.disarm_channel(0) + assert not rec.is_armed(0) + assert rec.get_channel_state(0) == RecChanState.DISARMED + + def test_arm_invalid_channel(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert not rec.arm_channel(99) # out of range + + def test_get_armed_channels(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.arm_channel(3) + rec.arm_channel(7) + armed = rec.get_armed_channels() + assert armed == [0, 3, 7] + + def test_start_with_no_armed_channels(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert rec.start() # starts but warns + # Should still transition to RECORDING + assert rec.state == RecorderState.RECORDING + rec.stop() + + def test_start_stop_cycle(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + assert rec.state == RecorderState.RECORDING + assert rec.get_channel_state(0) == RecChanState.RECORDING + + rec.stop() + assert rec.state == RecorderState.IDLE + # After stop, channel should be ARMED (not recording), since we disarm on stop + # Actually stop doesn't disarm — let me check... stop punts out then closes writers. + # After stop, state goes to IDLE but channel state is updated by _punch_out_channel + # which sets it to ARMED. + assert rec.get_channel_state(0) == RecChanState.ARMED + + def test_double_start(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + assert rec.start() + assert not rec.start() # already recording + + def test_pause_resume(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + assert rec.state == RecorderState.RECORDING + + rec.pause() + assert rec.state == RecorderState.PAUSED + + rec.resume() + assert rec.state == RecorderState.RECORDING + + rec.stop() + + +class TestMultiTrackRecorderWriting: + """Test writing audio through the recorder.""" + + def test_write_queues_samples(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + + chunk = _make_sine(440, 48000, 0.01) + result = rec.write(0, chunk) + assert result # should be queued + + rec.stop() + + def test_write_to_unarmed_channel(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.start() + # Channel 0 not armed — write should return False + chunk = _make_sine(440, 48000, 0.01) + result = rec.write(0, chunk) + assert not result + rec.stop() + + def test_write_while_paused(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + rec.pause() + chunk = _make_sine(440, 48000, 0.01) + result = rec.write(0, chunk) + assert not result # paused → no recording + rec.stop() + + def test_writer_thread_starts_and_stops(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + assert rec._writer_thread is not None + assert rec._writer_thread.is_alive() + + rec.stop() + # Writer thread should have stopped + assert not rec._writer_thread.is_alive() + + def test_write_creates_wav_file(self, tmp_path): + session_dir = tmp_path / "session" + rec = MultiTrackRecorder(session_dir=str(session_dir)) + rec.arm_channel(0) + rec.start() + + # Write enough frames to get a real file + chunk = _make_sine(440, 48000, 0.1, 0.5) + rec.write(0, chunk) + rec.advance_frame(len(chunk)) + + rec.stop() + + # Check that a WAV file was created + wav_files = list(session_dir.glob("ch00*.wav")) + assert len(wav_files) >= 1 + assert wav_files[0].stat().st_size > 0 + + def test_multi_channel_recording(self, tmp_path): + session_dir = tmp_path / "multi" + rec = MultiTrackRecorder(session_dir=str(session_dir)) + rec.arm_channel(0) + rec.arm_channel(1) + rec.arm_channel(2) + rec.start() + + chunk = _make_sine(440, 48000, 0.1, 0.5) + rec.write(0, chunk) + rec.write(1, chunk * 0.8) + rec.write(2, chunk * 0.6) + rec.advance_frame(len(chunk)) + + rec.stop() + + # All three channels should have files + for ch in range(3): + wav_files = list(session_dir.glob(f"ch{ch:02d}*.wav")) + assert len(wav_files) >= 1, f"Channel {ch} has no WAV file" + + +class TestPunchInOut: + """Test punch in/out functionality.""" + + def test_punch_in_while_recording(self, tmp_path): + session_dir = tmp_path / "punch" + rec = MultiTrackRecorder(session_dir=str(session_dir)) + rec.arm_channel(0) + rec.start() + + # Record some data on channel 0 + chunk = _make_sine(440, 48000, 0.05) + rec.write(0, chunk) + rec.advance_frame(len(chunk)) + + # Now punch in channel 1 (which wasn't armed) + rec.arm_channel(1) + assert rec.punch_in(1) + + chunk2 = _make_sine(880, 48000, 0.05) + rec.write(1, chunk2) + rec.advance_frame(len(chunk2)) + + # Punch out channel 1 + assert rec.punch_out(1) + + # Punch out channel 0 + assert rec.punch_out(0) + + rec.stop() + + # Both channels should have files + assert len(list(session_dir.glob("ch00*.wav"))) >= 1 + assert len(list(session_dir.glob("ch01*.wav"))) >= 1 + + def test_punch_creates_punch_regions(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + + chunk = _make_sine(440, 48000, 0.05) + rec.write(0, chunk) + rec.advance_frame(len(chunk)) + + rec.punch_out(0) + rec.advance_frame(100) + rec.punch_in(0) + + chunk2 = _make_sine(440, 48000, 0.05) + rec.write(0, chunk2) + rec.advance_frame(len(chunk2)) + + rec.stop() + + status = rec.get_channel_record_status(0) + assert status is not None + # Should have at least 2 punch regions (initial + re-punch) + assert len(status.punch_regions) >= 2 + + def test_punch_in_without_recording(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + # Not started yet + assert not rec.punch_in(0) + + def test_punch_out_without_recording(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert not rec.punch_out(0) + + +class TestRecorderStats: + """Test recording statistics.""" + + def test_stats_during_recording(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + + chunk = _make_sine(440, 48000, 0.1) + rec.write(0, chunk) + rec.advance_frame(len(chunk)) + + stats = rec.get_recording_stats() + assert stats["state"] == "RECORDING" + assert stats["global_frame"] == len(chunk) + assert RecChanState.RECORDING.name in str(stats["channels"]) + + rec.stop() + + def test_writer_error_propagation(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert rec.writer_error is None + + +class TestRecorderChannelState: + """Test per-channel state transitions.""" + + def test_disarmed_to_armed(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert rec.get_channel_state(0) == RecChanState.DISARMED + rec.arm_channel(0) + assert rec.get_channel_state(0) == RecChanState.ARMED + + def test_armed_to_recording_on_start(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + rec.arm_channel(0) + rec.start() + assert rec.get_channel_state(0) == RecChanState.RECORDING + rec.stop() + + def test_invalid_channel(self, tmp_path): + rec = MultiTrackRecorder(session_dir=str(tmp_path)) + assert rec.get_channel_state(-1) == RecChanState.DISARMED + assert rec.get_channel_state(999) == RecChanState.DISARMED + assert rec.get_channel_record_status(-1) is None + + +# ═════════════════════════════════════════════════════════════════════════════ +# Session tests +# ═════════════════════════════════════════════════════════════════════════════ + + +class TestSessionCreate: + """Test session creation.""" + + def test_create_defaults(self): + session = Session.create("Test Session") + assert session.name == "Test Session" + assert session.version == 1 + assert session.sample_rate == 48000 + assert session.bit_depth == "int24" + assert session.created_at != "" + assert session.takes == [] + + def test_create_with_options(self): + session = Session.create("Session", sample_rate=96000) + assert session.sample_rate == 96000 + + def test_take_count_zero_initially(self): + session = Session.create("Empty") + assert session.take_count == 0 + + +class TestTakeManagement: + """Test take creation, retrieval, deletion.""" + + def test_new_take(self): + session = Session.create("Test") + take = session.new_take("Take 1") + assert take.name == "Take 1" + assert take.id.startswith("take_") + assert session.take_count == 1 + + def test_auto_naming(self): + session = Session.create("Test") + t1 = session.new_take() + t2 = session.new_take() + t3 = session.new_take() + assert t1.name == "Take 1" + assert t2.name == "Take 2" + assert t3.name == "Take 3" + + def test_get_take(self): + session = Session.create("Test") + take = session.new_take("My Take") + found = session.get_take(take.id) + assert found is not None + assert found.name == "My Take" + + def test_get_take_not_found(self): + session = Session.create("Test") + assert session.get_take("nonexistent") is None + + def test_delete_take(self): + session = Session.create("Test") + take = session.new_take("Delete Me") + assert session.delete_take(take.id) + assert session.take_count == 0 + + def test_delete_nonexistent(self): + session = Session.create("Test") + assert not session.delete_take("nope") + + def test_rename_take(self): + session = Session.create("Test") + take = session.new_take("Original") + assert session.rename_take(take.id, "Renamed") + renamed = session.get_take(take.id) + assert renamed is not None + assert renamed.name == "Renamed" + + def test_list_takes(self): + session = Session.create("Test") + session.new_take("A") + session.new_take("B") + takes = session.list_takes() + assert len(takes) == 2 + assert takes[0].name == "A" + assert takes[1].name == "B" + + def test_get_last_take(self): + session = Session.create("Test") + assert session.get_last_take() is None + t1 = session.new_take("First") + t2 = session.new_take("Second") + last = session.get_last_take() + assert last is not None + assert last.id == t2.id + + +class TestTrackManagement: + """Test track addition within takes.""" + + def test_add_track(self): + session = Session.create("Test") + take = session.new_take("Take") + track = session.add_track(take.id, channel=0, file_path="tracks/ch00.wav", frames=48000) + assert track is not None + assert track.channel == 0 + assert track.file_path == "tracks/ch00.wav" + assert track.frames == 48000 + + def test_add_track_nonexistent_take(self): + session = Session.create("Test") + track = session.add_track("bad_id", channel=0, file_path="x.wav") + assert track is None + + def test_track_lookup(self): + session = Session.create("Test") + take = session.new_take("Take") + session.add_track(take.id, channel=3, file_path="ch03.wav") + session.add_track(take.id, channel=7, file_path="ch07.wav") + + t = take.get_track(3) + assert t is not None + assert t.file_path == "ch03.wav" + + assert take.get_track(99) is None + + def test_get_all_track_files(self): + session = Session.create("Test") + take1 = session.new_take("A") + session.add_track(take1.id, 0, "a_ch0.wav") + session.add_track(take1.id, 1, "a_ch1.wav") + take2 = session.new_take("B") + session.add_track(take2.id, 0, "b_ch0.wav") + + files = session.get_all_track_files() + assert len(files) == 3 + assert "a_ch0.wav" in files + assert "a_ch1.wav" in files + assert "b_ch0.wav" in files + + +class TestSessionSerialization: + """Test session save/load and dict conversion.""" + + def test_to_dict_and_back(self): + session = Session.create("Test") + take = session.new_take("Take 1") + session.add_track(take.id, 0, "ch00.wav", frames=1000) + session.mixer_state = {"master_volume": -6.0} + session.notes = "Test session" + + data = session.to_dict() + assert data["version"] == 1 + assert data["name"] == "Test" + assert len(data["takes"]) == 1 + assert data["mixer_state"] == {"master_volume": -6.0} + assert data["notes"] == "Test session" + + restored = Session.from_dict(data) + assert restored.name == "Test" + assert restored.take_count == 1 + assert restored.takes[0].name == "Take 1" + assert restored.mixer_state == {"master_volume": -6.0} + + def test_save_load_roundtrip(self, tmp_path): + session_file = tmp_path / "test.session.json" + + session = Session.create("Roundtrip Test", sample_rate=44100, bit_depth="int16") + take = session.new_take("Main Take") + session.add_track(take.id, 0, "tracks/ch00.wav", frames=44100) + session.add_track(take.id, 1, "tracks/ch01.wav", frames=44100) + session.mixer_state = {"ch0_volume": 0.0, "ch1_volume": -3.0} + session.notes = "Roundtrip test" + session.save(session_file) + + assert session_file.exists() + + loaded = Session.load(session_file) + assert loaded.name == "Roundtrip Test" + assert loaded.sample_rate == 44100 + assert loaded.bit_depth == "int16" + assert loaded.take_count == 1 + assert loaded.takes[0].name == "Main Take" + assert loaded.mixer_state == session.mixer_state + assert loaded.notes == session.notes + assert loaded._file_path == str(session_file) + + def test_save_without_path(self): + session = Session.create("Test") + with pytest.raises(ValueError): + session.save() # no path set + + def test_save_then_resave(self, tmp_path): + session_file = tmp_path / "session.json" + session = Session.create("Test") + session.save(session_file) + session.name = "Updated" + session.save() # uses _file_path from last save + + loaded = Session.load(session_file) + assert loaded.name == "Updated" + + def test_directory_property(self, tmp_path): + session_file = tmp_path / "dir" / "session.json" + session = Session.create("Test") + session.save(session_file) + assert session.directory == tmp_path / "dir" + + def test_resolve_track_path(self, tmp_path): + session_file = tmp_path / "sessions" / "test.session.json" + session = Session.create("Test") + session.save(session_file) + + take = session.new_take("T") + track = session.add_track(take.id, 0, "tracks/ch00.wav") + assert track is not None + + resolved = session.resolve_track_path(track) + assert resolved == tmp_path / "sessions" / "tracks" / "ch00.wav" + + def test_load_nonexistent(self): + with pytest.raises(FileNotFoundError): + Session.load("/tmp/nonexistent_session.json") + + def test_update_mixer_state(self): + session = Session.create("Test") + session.update_mixer_state({"master_vol": -10.0}) + assert session.mixer_state["master_vol"] == -10.0 + + +class TestTakeSerialization: + """Test Take to_dict / from_dict.""" + + def test_take_roundtrip(self): + take = Take( + id="take_abc", + name="My Take", + created_at="2024-01-01T00:00:00Z", + sample_rate=48000, + bit_depth="int24", + start_frame=0, + end_frame=48000, + duration_seconds=1.0, + ) + take.add_track(TrackInfo(channel=0, file_path="ch00.wav", frames=48000)) + take.add_track(TrackInfo(channel=1, file_path="ch01.wav", frames=48000)) + + data = take.to_dict() + restored = Take.from_dict(data) + + assert restored.id == take.id + assert restored.name == take.name + assert len(restored.tracks) == 2 + assert restored.tracks[0].channel == 0 + + def test_take_total_frames(self): + take = Take(id="t", name="T", end_frame=48000) + assert take.total_frames() == 48000 + + +class TestPunchInfo: + """Test punch info serialization in tracks.""" + + def test_punch_info_in_track(self): + take = Take(id="t1", name="T") + track = TrackInfo( + channel=0, + file_path="ch00.wav", + punch_regions=[ + PunchInfo(start_frame=0, end_frame=24000), + PunchInfo(start_frame=30000, end_frame=48000), + ], + ) + take.add_track(track) + + data = take.to_dict() + restored = Take.from_dict(data) + + t = restored.tracks[0] + assert len(t.punch_regions) == 2 + assert t.punch_regions[0].start_frame == 0 + assert t.punch_regions[0].end_frame == 24000 + + +# ═════════════════════════════════════════════════════════════════════════════ +# DiskMonitor tests +# ═════════════════════════════════════════════════════════════════════════════ + + +class TestDiskMonitor: + """Test disk space monitoring.""" + + def test_initial_check(self): + monitor = DiskMonitor("/tmp") + assert monitor.free_bytes > 0 + assert monitor.total_bytes > 0 + + def test_free_mb(self): + monitor = DiskMonitor("/tmp") + mb = monitor.free_mb + assert mb > 0 + + def test_usage_percent(self): + monitor = DiskMonitor("/tmp") + pct = monitor.usage_percent + assert 0.0 <= pct <= 100.0 + + def test_level_ok_when_plenty_of_space(self): + cfg = DiskMonitorConfig( + warning_threshold_bytes=500, + critical_threshold_bytes=100, + ) + monitor = DiskMonitor("/tmp", config=cfg) + monitor.check_now() + # Unless we're on a very full disk, this should be ok + assert monitor.level in (0, 1, 2) # just check it doesn't crash + + def test_is_low_space_with_high_threshold(self, tmp_path): + cfg = DiskMonitorConfig( + warning_threshold_bytes=10 * 1024 * 1024 * 1024 * 1024, # 10 TB — impossible + critical_threshold_bytes=5 * 1024 * 1024 * 1024 * 1024, + ) + monitor = DiskMonitor(str(tmp_path), config=cfg) + assert monitor.is_low_space() # will trigger warning + assert monitor.is_critical() # will trigger critical + + def test_has_minimum(self, tmp_path): + cfg = DiskMonitorConfig(minimum_required_bytes=0) + monitor = DiskMonitor(str(tmp_path), config=cfg) + assert monitor.has_minimum_for_recording() + + def test_estimated_time_positive(self, tmp_path): + monitor = DiskMonitor(str(tmp_path)) + t = monitor.estimated_recording_time_seconds() + assert t > 0 + + def test_format_bytes(self): + assert DiskMonitor.format_bytes(500) == "500 B" + assert "KB" in DiskMonitor.format_bytes(2048) + assert "MB" in DiskMonitor.format_bytes(10 * 1024 * 1024) + assert "GB" in DiskMonitor.format_bytes(2 * 1024 * 1024 * 1024) + + def test_static_helpers(self): + free = DiskMonitor.get_free_bytes("/tmp") + total = DiskMonitor.get_total_bytes("/tmp") + assert free > 0 + assert total >= free + + def test_callback_fires_on_level_change(self, tmp_path): + callback_calls = [] + + def cb(free, total, level): + callback_calls.append(level) + + cfg = DiskMonitorConfig( + warning_threshold_bytes=10 * 1024**4, # enormous + critical_threshold_bytes=5 * 1024**4, + poll_interval_seconds=0.01, + ) + monitor = DiskMonitor(str(tmp_path), config=cfg) + monitor.set_callback(cb) + # Reset internal level so next check triggers callback + monitor._current_level = 0 + monitor.check_now() # force check + + # Should have been called at least once with level >= 1 + assert len(callback_calls) > 0 + assert callback_calls[-1] >= 1 + + def test_start_stop(self): + monitor = DiskMonitor("/tmp") + monitor.start() + assert monitor._running + monitor.stop() + assert not monitor._running + + def test_level_label(self): + monitor = DiskMonitor("/tmp") + assert monitor.level_label in ("ok", "warning", "critical") + + +# ═════════════════════════════════════════════════════════════════════════════ +# BounceEngine tests +# ═════════════════════════════════════════════════════════════════════════════ + + +class TestDBAConversion: + """Test dB ↔ linear conversion utilities.""" + + def test_db_to_linear_zero(self): + assert db_to_linear(0.0) == pytest.approx(1.0) + + def test_db_to_linear_minus_6(self): + assert db_to_linear(-6.0) == pytest.approx(0.5012, abs=0.001) + + def test_db_to_linear_plus_6(self): + assert db_to_linear(6.0) == pytest.approx(1.995, abs=0.01) + + def test_linear_to_db_one(self): + assert linear_to_db(1.0) == pytest.approx(0.0, abs=0.01) + + def test_linear_to_db_half(self): + assert linear_to_db(0.5) == pytest.approx(-6.02, abs=0.1) + + def test_roundtrip(self): + for db in [-12.0, -6.0, -3.0, 0.0, 3.0, 6.0, 12.0]: + assert linear_to_db(db_to_linear(db)) == pytest.approx(db, abs=0.01) + + +class TestPanLaw: + """Test pan law conversion.""" + + def test_center(self): + l, r = _pan_to_gains(0.0) + assert l == pytest.approx(r) + assert l > 0.7 # -3dB center + + def test_hard_left(self): + l, r = _pan_to_gains(-1.0) + assert l == pytest.approx(1.0, abs=0.001) + assert r == pytest.approx(0.0, abs=0.001) + + def test_hard_right(self): + l, r = _pan_to_gains(1.0) + assert l == pytest.approx(0.0, abs=0.001) + assert r == pytest.approx(1.0, abs=0.001) + + def test_power_sum(self): + """Pan law should maintain constant power (l^2 + r^2 ≈ 1).""" + for pan in np.linspace(-1.0, 1.0, 21): + l, r = _pan_to_gains(pan) + power = l ** 2 + r ** 2 + assert power == pytest.approx(1.0, abs=0.01), f"pan={pan}: power={power}" + + +class TestBounceEngine: + """Test bounce engine functionality.""" + + def _create_mono_wav(self, path: Path, freq: float, duration: float, amplitude: float = 0.5) -> None: + """Create a test mono WAV file.""" + sine = _make_sine(freq, 48000, duration, amplitude) + write_wav(path, sine, sample_rate=48000, bit_depth=WAVBitDepth.INT16) + + def _create_stereo_wav(self, path: Path, freq_l: float, freq_r: float, duration: float) -> None: + """Create a test stereo WAV file.""" + left = _make_sine(freq_l, 48000, duration, 0.5) + right = _make_sine(freq_r, 48000, duration, 0.3) + stereo = np.column_stack([left, right]) + write_wav(path, stereo, sample_rate=48000, bit_depth=WAVBitDepth.INT16) + + def test_bounce_single_mono_track(self, tmp_path): + input_path = tmp_path / "input.wav" + output_path = tmp_path / "output.wav" + self._create_mono_wav(input_path, 440, 0.5) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path) + + assert result.output_path.exists() + assert result.frames_written == int(48000 * 0.5) + assert result.duration_seconds == pytest.approx(0.5, abs=0.01) + assert result.files_processed == 1 + + # Should be stereo output + samples = _read_wav_samples(output_path) + assert samples.ndim == 2 + assert samples.shape[1] == 2 + + def test_bounce_two_tracks_stereo(self, tmp_path): + input1 = tmp_path / "track1.wav" + input2 = tmp_path / "track2.wav" + output_path = tmp_path / "mix.wav" + + self._create_mono_wav(input1, 440, 0.5, 0.5) + self._create_mono_wav(input2, 880, 0.5, 0.3) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input1, channel=0, volume_db=0.0) + engine.add_track(input2, channel=1, volume_db=-6.0) + + result = engine.bounce(output_path) + + assert result.output_path.exists() + assert result.files_processed == 2 + + def test_bounce_panning(self, tmp_path): + input_path = tmp_path / "mono.wav" + output_path = tmp_path / "panned.wav" + + self._create_mono_wav(input_path, 440, 0.2, 1.0) + + engine = BounceEngine(sample_rate=48000) + # Hard left + engine.add_track(input_path, channel=0, pan=-1.0) + result = engine.bounce(output_path) + + samples = _read_wav_samples(output_path) + # Right channel should be silent + right_rms = float(np.sqrt(np.mean(samples[:, 1] ** 2))) + left_rms = float(np.sqrt(np.mean(samples[:, 0] ** 2))) + assert left_rms > right_rms * 5 # right should be much quieter + + def test_bounce_muted_track(self, tmp_path): + input1 = tmp_path / "loud.wav" + input2 = tmp_path / "muted.wav" + output_path = tmp_path / "mix.wav" + + self._create_mono_wav(input1, 440, 0.3, 0.5) + self._create_mono_wav(input2, 880, 0.3, 1.0) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input1, channel=0) + engine.add_track(input2, channel=1, muted=True) + + result = engine.bounce(output_path) + + # The output should only contain track 1 + samples = _read_wav_samples(output_path) + full_mix = _make_sine(440, 48000, 0.3, 0.5) + mix_stereo = np.column_stack([full_mix, full_mix]) + + # Correlation should be close to 1.0 with track 1 + corr = np.corrcoef(samples[:, 0], mix_stereo[:, 0])[0, 1] + assert corr > 0.99 + + def test_bounce_normalisation(self, tmp_path): + input_path = tmp_path / "quiet.wav" + output_path = tmp_path / "norm.wav" + + # Very quiet signal + self._create_mono_wav(input_path, 440, 0.3, 0.1) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path, normalise=True) + + samples = _read_wav_samples(output_path) + peak = float(np.max(np.abs(samples))) + # Should be close to target (0 dB = 1.0, with -0.3 dBFS ≈ 0.966) + assert peak > 0.9 + + def test_bounce_no_normalise(self, tmp_path): + input_path = tmp_path / "quiet.wav" + output_path = tmp_path / "nonorm.wav" + + self._create_mono_wav(input_path, 440, 0.3, 0.1) + + config = BounceConfig(normalise=False) + engine = BounceEngine(sample_rate=48000, config=config) + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path) + + samples = _read_wav_samples(output_path) + peak = float(np.max(np.abs(samples))) + # Should be low (near 0.1) + assert peak < 0.3 + + def test_bounce_master_volume(self, tmp_path): + input_path = tmp_path / "input.wav" + output_path = tmp_path / "mastervol.wav" + + self._create_mono_wav(input_path, 440, 0.3, 0.5) + + config = BounceConfig(master_volume_db=-20.0, normalise=False) + engine = BounceEngine(sample_rate=48000, config=config) + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path) + + samples = _read_wav_samples(output_path) + peak = float(np.max(np.abs(samples))) + # -20 dB = 0.1 linear, so peak should be ~0.5 * 0.1 = 0.05 + assert peak < 0.1 + + def test_bounce_master_muted(self, tmp_path): + input_path = tmp_path / "input.wav" + output_path = tmp_path / "muted_master.wav" + + self._create_mono_wav(input_path, 440, 0.3, 0.5) + + config = BounceConfig(master_muted=True, normalise=False) + engine = BounceEngine(sample_rate=48000, config=config) + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path) + + samples = _read_wav_samples(output_path) + # Should be silent + assert float(np.max(np.abs(samples))) < 1e-6 + + def test_bounce_master_dim(self, tmp_path): + input_path = tmp_path / "input.wav" + output_path = tmp_path / "dimmed.wav" + + self._create_mono_wav(input_path, 440, 0.3, 0.5) + + config = BounceConfig(master_dim_active=True, master_dim_db=-40.0, normalise=False) + engine = BounceEngine(sample_rate=48000, config=config) + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path) + + samples = _read_wav_samples(output_path) + peak = float(np.max(np.abs(samples))) + # -40 dB dim = 0.01 linear + assert peak < 0.05 + + def test_preview(self, tmp_path): + input_path = tmp_path / "preview.wav" + self._create_mono_wav(input_path, 440, 0.5) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input_path, channel=0) + + preview = engine.preview() + assert preview["total_frames"] == int(48000 * 0.5) + assert preview["duration_seconds"] == pytest.approx(0.5, abs=0.01) + assert preview["estimated_file_size_bytes"] > 0 + assert preview["track_count"] == 1 + + def test_bounce_no_tracks_raises(self, tmp_path): + engine = BounceEngine(sample_rate=48000) + with pytest.raises(ValueError): + engine.bounce(tmp_path / "output.wav") + + def test_bounce_clear_tracks(self, tmp_path): + input_path = tmp_path / "input.wav" + self._create_mono_wav(input_path, 440, 0.1) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input_path, channel=0) + engine.add_track(input_path, channel=1) + engine.remove_track(0) + assert len(engine._tracks) == 1 + engine.clear_tracks() + assert len(engine._tracks) == 0 + + def test_bounce_stereo_input(self, tmp_path): + """Bouncing a stereo input should preserve left/right.""" + input_path = tmp_path / "stereo.wav" + output_path = tmp_path / "stereo_out.wav" + + self._create_stereo_wav(input_path, 440, 880, 0.3) + + engine = BounceEngine(sample_rate=48000) + engine.add_track(input_path, channel=0, pan=0.0) + result = engine.bounce(output_path) + + samples = _read_wav_samples(output_path) + assert samples.shape[1] == 2 + + def test_bounce_result_peak_rms(self, tmp_path): + input_path = tmp_path / "input.wav" + output_path = tmp_path / "result.wav" + self._create_mono_wav(input_path, 440, 0.3, 0.7) + + engine = BounceEngine(sample_rate=48000) + config = BounceConfig(normalise=False) + engine.config = config + engine.add_track(input_path, channel=0) + result = engine.bounce(output_path) + + # Peak should be close to 0.7 + assert result.peak_db > -10 + assert result.rms_db > -20 + + +# ═════════════════════════════════════════════════════════════════════════════ +# Integration / edge-case tests +# ═════════════════════════════════════════════════════════════════════════════ + + +class TestRecordingIntegration: + """Tests that combine multiple recording components.""" + + def test_record_then_bounce(self, tmp_path): + """Record audio, save session, bounce to stereo.""" + session_dir = tmp_path / "project" + session_dir.mkdir() + + # 1. Create session + session = Session.create("Integration Test", sample_rate=48000) + session_file = session_dir / "session.json" + + # 2. Record + rec = MultiTrackRecorder(session_dir=str(session_dir)) + rec.arm_channel(0) + rec.arm_channel(1) + rec.start() + + chunk = _make_sine(440, 48000, 0.2, 0.5) + rec.write(0, chunk) + rec.write(1, chunk * 0.8) + rec.advance_frame(len(chunk)) + rec.stop() + + # 3. Find recorded files + wav_files_0 = sorted(session_dir.glob("ch00*.wav")) + wav_files_1 = sorted(session_dir.glob("ch01*.wav")) + + # 4. Create take from recording + take = session.new_take("Take 1") + if wav_files_0: + session.add_track(take.id, 0, str(wav_files_0[0].relative_to(session_dir)), frames=len(chunk)) + if wav_files_1: + session.add_track(take.id, 1, str(wav_files_1[0].relative_to(session_dir)), frames=len(chunk)) + + session.save(session_file) + assert session_file.exists() + + # 5. Bounce + output_path = session_dir / "master.wav" + engine = BounceEngine(sample_rate=48000) + + for wav_file in wav_files_0: + engine.add_track(wav_file, channel=0, volume_db=0.0, pan=-0.5) + for wav_file in wav_files_1: + engine.add_track(wav_file, channel=1, volume_db=-3.0, pan=0.5) + + result = engine.bounce(output_path, normalise=True) + assert result.output_path.exists() + + # 6. Reload session + loaded = Session.load(session_file) + assert loaded.take_count == 1 + assert loaded.takes[0].name == "Take 1" + + def test_bounce_with_session_mixer_state(self, tmp_path): + """Bounce using mixer state from a session.""" + session_dir = tmp_path / "mixer_session" + session_dir.mkdir() + + # Create test audio + input_path = session_dir / "ch00.wav" + _create_mono_wav(input_path, 440, 0.3, 0.5) + input_path2 = session_dir / "ch01.wav" + _create_mono_wav(input_path2, 880, 0.3, 0.3) + + session = Session.create("Test", sample_rate=48000) + session.mixer_state = { + "ch0_volume": -3.0, + "ch0_pan": -0.5, + "ch1_volume": -6.0, + "ch1_pan": 0.5, + "master_volume": -2.0, + } + session.save(session_dir / "session.json") + + # Load mixer state into bounce config + ms = session.mixer_state + engine = BounceEngine( + sample_rate=48000, + config=BounceConfig( + master_volume_db=ms.get("master_volume", 0.0), + normalise=False, + ), + ) + engine.add_track(input_path, channel=0, volume_db=ms.get("ch0_volume", 0.0), pan=ms.get("ch0_pan", 0.0)) + engine.add_track(input_path2, channel=1, volume_db=ms.get("ch1_volume", 0.0), pan=ms.get("ch1_pan", 0.0)) + + output_path = session_dir / "master.wav" + result = engine.bounce(output_path) + assert result.output_path.exists() + + +# Helper for integration test +def _create_mono_wav(path: Path, freq: float, duration: float, amplitude: float = 0.5) -> None: + """Create a test mono WAV file (function version for use outside class).""" + sine = _make_sine(freq, 48000, duration, amplitude) + write_wav(path, sine, sample_rate=48000, bit_depth=WAVBitDepth.INT16)