shawn
d6480a35ed
Lint & Validate / lint (push) Has been cancelled
- 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.
1437 lines
51 KiB
Python
1437 lines
51 KiB
Python
"""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("<I", f.read(4))[0]
|
|
wave = f.read(4)
|
|
fmt_chunk = f.read(4)
|
|
fmt_size = struct.unpack("<I", f.read(4))[0]
|
|
audio_fmt = struct.unpack("<H", f.read(2))[0]
|
|
channels = struct.unpack("<H", f.read(2))[0]
|
|
sample_rate = struct.unpack("<I", f.read(4))[0]
|
|
byte_rate = struct.unpack("<I", f.read(4))[0]
|
|
block_align = struct.unpack("<H", f.read(2))[0]
|
|
bits = struct.unpack("<H", f.read(2))[0]
|
|
return {
|
|
"riff": riff,
|
|
"file_size": file_size,
|
|
"wave": wave,
|
|
"audio_format": audio_fmt,
|
|
"channels": channels,
|
|
"sample_rate": sample_rate,
|
|
"byte_rate": byte_rate,
|
|
"block_align": block_align,
|
|
"bits_per_sample": bits,
|
|
}
|
|
|
|
|
|
def _read_wav_samples(file_path: Path) -> 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)
|