diff --git a/abogen/domain/audio_buffer.py b/abogen/domain/audio_buffer.py new file mode 100644 index 0000000..626f9cf --- /dev/null +++ b/abogen/domain/audio_buffer.py @@ -0,0 +1,165 @@ +"""Audio buffer operations for audiobook generation. + +This module provides core audio buffer manipulation functions including: +- Silence generation +- Audio mixing +- Audio normalization +- Audio buffer resizing +""" + +from __future__ import annotations + +from typing import Optional + +import numpy as np + +# Standard sample rate used throughout the application +SAMPLE_RATE = 24000 + + +def create_silence(duration_seconds: float) -> np.ndarray: + """Create a silence audio buffer. + + Args: + duration_seconds: Duration of silence in seconds. + + Returns: + Numpy array of float32 zeros with length = duration_seconds * SAMPLE_RATE. + Returns empty array if duration is <= 0. + """ + if duration_seconds <= 0: + return np.array([], dtype="float32") + + samples = int(round(duration_seconds * SAMPLE_RATE)) + if samples <= 0: + return np.array([], dtype="float32") + + return np.zeros(samples, dtype="float32") + + +def mix_audio( + target: np.ndarray, + source: np.ndarray, + start_sample: int, + end_sample: Optional[int] = None, +) -> None: + """Mix source audio into target buffer at specified position. + + This performs additive mixing (target += source). The target buffer + is extended if necessary to accommodate the source audio. + + Args: + target: The target audio buffer to mix into (modified in-place). + source: The source audio buffer to mix. + start_sample: Starting sample index in target buffer. + end_sample: Optional end sample index. If None, calculated from source length. + """ + if source.size == 0: + return + + if end_sample is None: + end_sample = start_sample + len(source) + + # Extend target buffer if needed + if end_sample > len(target): + new_length = end_sample + target = np.concatenate([ + target, + np.zeros(new_length - len(target), dtype="float32") + ]) + + # Perform the mix (additive) + target[start_sample:end_sample] += source + + +def normalize_audio( + audio: np.ndarray, + target_peak: float = 1.0, +) -> np.ndarray: + """Normalize audio buffer to prevent clipping. + + If the audio exceeds the target peak (default 1.0), it is scaled down + proportionally to prevent distortion. + + Args: + audio: Input audio buffer. + target_peak: Target maximum amplitude (default 1.0). + + Returns: + Normalized audio buffer (new array, original is not modified). + """ + if audio.size == 0: + return audio.copy() + + max_amplitude = float(np.abs(audio).max()) + + if max_amplitude <= target_peak: + return audio.copy() + + # Scale down to prevent clipping + scale_factor = target_peak / max_amplitude + return (audio * scale_factor).astype("float32") + + +def ensure_buffer_size( + buffer: np.ndarray, + min_samples: int, +) -> np.ndarray: + """Ensure audio buffer is at least min_samples long. + + If buffer is shorter, it is extended with zeros. + + Args: + buffer: Input audio buffer. + min_samples: Minimum required length in samples. + + Returns: + Buffer of at least min_samples length (new array if extended). + """ + if len(buffer) >= min_samples: + return buffer + + new_buffer = np.zeros(min_samples, dtype="float32") + new_buffer[:len(buffer)] = buffer + return new_buffer + + +def concatenate_audio(*buffers: np.ndarray) -> np.ndarray: + """Concatenate multiple audio buffers. + + Args: + *buffers: Audio buffers to concatenate. + + Returns: + Single concatenated audio buffer. + """ + non_empty = [b for b in buffers if b.size > 0] + if not non_empty: + return np.array([], dtype="float32") + return np.concatenate(non_empty) + + +def audio_duration(audio: np.ndarray, sample_rate: int = SAMPLE_RATE) -> float: + """Calculate duration of audio buffer in seconds. + + Args: + audio: Audio buffer. + sample_rate: Sample rate in Hz (default SAMPLE_RATE). + + Returns: + Duration in seconds. + """ + return len(audio) / sample_rate + + +def samples_for_duration(duration_seconds: float, sample_rate: int = SAMPLE_RATE) -> int: + """Calculate number of samples for a given duration. + + Args: + duration_seconds: Duration in seconds. + sample_rate: Sample rate in Hz (default SAMPLE_RATE). + + Returns: + Number of samples (rounded to nearest integer). + """ + return int(round(duration_seconds * sample_rate)) diff --git a/abogen/pyqt/conversion.py b/abogen/pyqt/conversion.py index 3ae5654..6ea017d 100644 --- a/abogen/pyqt/conversion.py +++ b/abogen/pyqt/conversion.py @@ -29,6 +29,12 @@ from abogen.domain.output_paths import ( sanitize_output_stem, ) from abogen.domain.audio_helpers import build_ffmpeg_command, to_float32 +from abogen.domain.audio_buffer import ( + create_silence, + mix_audio, + normalize_audio, + SAMPLE_RATE, +) import abogen.hf_tracker as hf_tracker import static_ffmpeg import threading # for efficient waiting @@ -1285,10 +1291,7 @@ class ConversionThread(QThread): # Add silence between chapters for merged output (except after the last chapter) if merge_chapters_at_end and chapter_idx < total_chapters: - silence_samples = int( - self.silence_duration * 24000 - ) # Silence duration at 24,000 Hz - silence_audio = np.zeros(silence_samples, dtype="float32") + silence_audio = create_silence(self.silence_duration) silence_bytes = silence_audio.tobytes() if merged_out_file: @@ -1596,9 +1599,8 @@ class ConversionThread(QThread): max_end_time = max( (end for _, end, _ in subtitles if end is not None), default=0 ) - audio_buffer = np.zeros( - int(max_end_time * rate) + rate, dtype="float32" - ) + buffer_samples = int(max_end_time * rate) + rate + audio_buffer = np.zeros(buffer_samples, dtype="float32") # Process each subtitle and mix into buffer self.etr_start_time = time.time() @@ -1799,33 +1801,14 @@ class ConversionThread(QThread): # Pad or trim to subtitle duration target_samples = int(subtitle_duration * rate) if len(full_audio) < target_samples: - full_audio = np.concatenate( - [ - full_audio, - np.zeros( - target_samples - len(full_audio), dtype="float32" - ), - ] - ) + padding_duration = (target_samples - len(full_audio)) / rate + full_audio = np.concatenate([full_audio, create_silence(padding_duration)]) elif len(full_audio) > target_samples: full_audio = full_audio[:target_samples] # Mix audio into buffer at the correct position (handles overlaps) start_sample = int(start_time * rate) - end_sample = start_sample + len(full_audio) - if end_sample > len(audio_buffer): - # Extend buffer if needed - audio_buffer = np.concatenate( - [ - audio_buffer, - np.zeros( - end_sample - len(audio_buffer), dtype="float32" - ), - ] - ) - - # Mix (add) the audio - this handles overlaps by combining them - audio_buffer[start_sample:end_sample] += full_audio + mix_audio(audio_buffer, full_audio, start_sample) # Write subtitle if subtitle_file: @@ -1860,12 +1843,11 @@ class ConversionThread(QThread): self.progress_updated.emit(percent, etr_str) # Normalize audio buffer to prevent clipping from mixed overlaps - max_amplitude = np.abs(audio_buffer).max() - if max_amplitude > 1.0: + if np.abs(audio_buffer).max() > 1.0: self.log_updated.emit( - f"\n -> Normalizing audio (peak: {max_amplitude:.2f})" + f"\n -> Normalizing audio (peak: {np.abs(audio_buffer).max():.2f})" ) - audio_buffer = audio_buffer / max_amplitude + audio_buffer = normalize_audio(audio_buffer) # Write the complete audio buffer self.log_updated.emit(("\nFinalizing audio. Please wait...", "grey")) diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 5b19954..42dbfc2 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -116,6 +116,11 @@ from abogen.domain.audio_helpers import ( to_float32 as _to_float32, apply_m4b_chapters_with_mutagen as _apply_m4b_chapters_with_mutagen, ) +from abogen.domain.audio_buffer import ( + create_silence as _create_silence, + normalize_audio as _normalize_audio, + SAMPLE_RATE, +) from .service import Job, JobStatus @@ -640,10 +645,9 @@ def run_conversion_job(job: Job) -> None: nonlocal current_time if duration_seconds <= 0: return - samples = int(round(duration_seconds * SAMPLE_RATE)) - if samples <= 0: + silence = _create_silence(duration_seconds) + if silence.size == 0: return - silence = np.zeros(samples, dtype="float32") if include_in_chapter and chapter_sink: chapter_sink.write(silence) if audio_sink: diff --git a/tests/test_audio_buffer.py b/tests/test_audio_buffer.py new file mode 100644 index 0000000..643f92f --- /dev/null +++ b/tests/test_audio_buffer.py @@ -0,0 +1,307 @@ +"""Tests for abogen.domain.audio_buffer module.""" + +import numpy as np +import pytest + +from abogen.domain.audio_buffer import ( + create_silence, + mix_audio, + normalize_audio, + ensure_buffer_size, + concatenate_audio, + audio_duration, + samples_for_duration, + SAMPLE_RATE, +) + + +class TestCreateSilence: + """Tests for create_silence function.""" + + def test_positive_duration(self): + """Test creating silence with positive duration.""" + duration = 1.0 # 1 second + silence = create_silence(duration) + + expected_samples = int(round(duration * SAMPLE_RATE)) + assert len(silence) == expected_samples + assert silence.dtype == np.float32 + assert np.all(silence == 0) + + def test_zero_duration(self): + """Test creating silence with zero duration returns empty array.""" + silence = create_silence(0) + assert len(silence) == 0 + assert silence.dtype == np.float32 + + def test_negative_duration(self): + """Test creating silence with negative duration returns empty array.""" + silence = create_silence(-1.0) + assert len(silence) == 0 + assert silence.dtype == np.float32 + + def test_very_small_duration(self): + """Test creating silence with very small duration.""" + duration = 0.001 # 1 millisecond + silence = create_silence(duration) + + # Should round to at least 1 sample or 0 + assert len(silence) >= 0 + assert silence.dtype == np.float32 + + def test_half_second(self): + """Test creating 0.5 second of silence.""" + silence = create_silence(0.5) + expected_samples = int(round(0.5 * SAMPLE_RATE)) + assert len(silence) == expected_samples + + +class TestMixAudio: + """Tests for mix_audio function.""" + + def test_basic_mix(self): + """Test basic audio mixing.""" + target = np.ones(100, dtype="float32") + source = np.ones(50, dtype="float32") * 2 + + mix_audio(target, source, start_sample=25) + + # First 25 samples should remain 1.0 + assert np.all(target[:25] == 1.0) + # Middle 50 samples should be 1.0 + 2.0 = 3.0 + assert np.all(target[25:75] == 3.0) + # Last 25 samples should remain 1.0 + assert np.all(target[75:] == 1.0) + + def test_empty_source(self): + """Test mixing empty source does nothing.""" + target = np.ones(100, dtype="float32") + original = target.copy() + + mix_audio(target, np.array([], dtype="float32"), start_sample=50) + + assert np.array_equal(target, original) + + def test_extend_target_buffer(self): + """Test that target buffer is extended when needed.""" + target = np.ones(100, dtype="float32") + source = np.ones(50, dtype="float32") * 2 + + # This should extend target to 150 samples + mix_audio(target, source, start_sample=120) + + assert len(target) >= 170 # 120 + 50 + # Check that source was mixed correctly + assert np.all(target[120:170] == 2.0) + + def test_start_at_zero(self): + """Test mixing starting at sample 0.""" + target = np.zeros(100, dtype="float32") + source = np.ones(50, dtype="float32") + + mix_audio(target, source, start_sample=0) + + assert np.all(target[:50] == 1.0) + assert np.all(target[50:] == 0.0) + + def test_explicit_end_sample(self): + """Test mixing with explicit end_sample.""" + target = np.zeros(100, dtype="float32") + source = np.ones(50, dtype="float32") + + mix_audio(target, source, start_sample=10, end_sample=60) + + # Only first 10 samples of source should be mixed (60-10=50, but source is only 50) + # Actually, end_sample overrides the length + assert target[10] == 1.0 + + +class TestNormalizeAudio: + """Tests for normalize_audio function.""" + + def test_no_normalization_needed(self): + """Test audio within range is not modified.""" + audio = np.ones(100, dtype="float32") * 0.5 + result = normalize_audio(audio) + + assert not np.share_memory(audio, result) # Should be a copy + assert np.array_equal(result, audio) + + def test_normalization_applied(self): + """Test audio above target peak is scaled down.""" + audio = np.ones(100, dtype="float32") * 2.0 + result = normalize_audio(audio) + + assert np.all(result <= 1.0) + assert np.isclose(result[0], 1.0) + + def test_empty_audio(self): + """Test normalizing empty audio returns empty copy.""" + audio = np.array([], dtype="float32") + result = normalize_audio(audio) + + assert len(result) == 0 + assert result.dtype == np.float32 + + def test_custom_target_peak(self): + """Test normalization with custom target peak.""" + audio = np.ones(100, dtype="float32") * 4.0 + result = normalize_audio(audio, target_peak=2.0) + + assert np.all(result <= 2.0) + assert np.isclose(result[0], 2.0) + + def test_negative_peak(self): + """Test normalization handles negative peaks.""" + audio = np.ones(100, dtype="float32") * -2.0 + result = normalize_audio(audio) + + assert np.all(result >= -1.0) + assert np.isclose(result[0], -1.0) + + def test_mixed_positive_negative(self): + """Test normalization with both positive and negative peaks.""" + audio = np.array([-3.0, 2.0, -1.0, 4.0], dtype="float32") + result = normalize_audio(audio) + + # Should scale by 1/4 (max absolute value is 4) + assert np.isclose(result[0], -0.75) + assert np.isclose(result[1], 0.5) + assert np.isclose(result[3], 1.0) + + +class TestEnsureBufferSize: + """Tests for ensure_buffer_size function.""" + + def test_buffer_already_large_enough(self): + """Test buffer that is already large enough is unchanged.""" + buffer = np.ones(100, dtype="float32") + result = ensure_buffer_size(buffer, 50) + + assert np.array_equal(result, buffer) + + def test_buffer_needs_extension(self): + """Test buffer is extended with zeros when too small.""" + buffer = np.ones(50, dtype="float32") + result = ensure_buffer_size(buffer, 100) + + assert len(result) == 100 + assert np.all(result[:50] == 1.0) + assert np.all(result[50:] == 0.0) + + def test_exact_size(self): + """Test buffer of exact size is unchanged.""" + buffer = np.ones(100, dtype="float32") + result = ensure_buffer_size(buffer, 100) + + assert len(result) == 100 + assert np.array_equal(result, buffer) + + +class TestConcatenateAudio: + """Tests for concatenate_audio function.""" + + def test_concatenate_two_buffers(self): + """Test concatenating two audio buffers.""" + a = np.ones(50, dtype="float32") + b = np.ones(50, dtype="float32") * 2 + result = concatenate_audio(a, b) + + assert len(result) == 100 + assert np.all(result[:50] == 1.0) + assert np.all(result[50:] == 2.0) + + def test_concatenate_multiple_buffers(self): + """Test concatenating multiple audio buffers.""" + a = np.ones(20, dtype="float32") + b = np.ones(30, dtype="float32") * 2 + c = np.ones(40, dtype="float32") * 3 + result = concatenate_audio(a, b, c) + + assert len(result) == 90 + assert np.all(result[:20] == 1.0) + assert np.all(result[20:50] == 2.0) + assert np.all(result[50:] == 3.0) + + def test_concatenate_empty_buffers(self): + """Test concatenating empty buffers returns empty array.""" + result = concatenate_audio( + np.array([], dtype="float32"), + np.array([], dtype="float32") + ) + assert len(result) == 0 + + def test_concatenate_with_empty(self): + """Test concatenating with some empty buffers.""" + a = np.ones(50, dtype="float32") + result = concatenate_audio(a, np.array([], dtype="float32")) + + assert len(result) == 50 + assert np.array_equal(result, a) + + +class TestAudioDuration: + """Tests for audio_duration function.""" + + def test_one_second_duration(self): + """Test duration calculation for 1 second of audio.""" + audio = np.zeros(SAMPLE_RATE, dtype="float32") + duration = audio_duration(audio) + + assert duration == 1.0 + + def test_half_second_duration(self): + """Test duration calculation for 0.5 second of audio.""" + audio = np.zeros(SAMPLE_RATE // 2, dtype="float32") + duration = audio_duration(audio) + + assert duration == 0.5 + + def test_empty_audio_duration(self): + """Test duration of empty audio is 0.""" + duration = audio_duration(np.array([], dtype="float32")) + assert duration == 0.0 + + def test_custom_sample_rate(self): + """Test duration with custom sample rate.""" + audio = np.zeros(48000, dtype="float32") # 48k samples + duration = audio_duration(audio, sample_rate=48000) + + assert duration == 1.0 + + +class TestSamplesForDuration: + """Tests for samples_for_duration function.""" + + def test_one_second(self): + """Test samples for 1 second at default rate.""" + samples = samples_for_duration(1.0) + assert samples == SAMPLE_RATE + + def test_half_second(self): + """Test samples for 0.5 second at default rate.""" + samples = samples_for_duration(0.5) + assert samples == SAMPLE_RATE // 2 + + def test_zero_duration(self): + """Test samples for 0 duration.""" + samples = samples_for_duration(0) + assert samples == 0 + + def test_negative_duration(self): + """Test samples for negative duration.""" + samples = samples_for_duration(-1.0) + assert samples == 0 + + def test_custom_sample_rate(self): + """Test samples with custom sample rate.""" + samples = samples_for_duration(1.0, sample_rate=44100) + assert samples == 44100 + + +class TestSampleRateConstant: + """Tests for SAMPLE_RATE constant.""" + + def test_sample_rate_value(self): + """Test SAMPLE_RATE is 24000.""" + assert SAMPLE_RATE == 24000