wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3205 from gui217/feat/rnnoise

Browse Source 
Feat/rnnoise
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-17 18:22:22 -08:00
committed by GitHub
parent 9623575b78 1c0e25a90d
commit 6c0e4be4ac
6 changed files with 573 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
tests/test_rnnoise_cancellation.py Normal file
View File

@@ -0,0 +1,169 @@
import asyncio
import unittest
import numpy as np
import pytest
try:
import pyrnnoise
except ImportError:
pyrnnoise = None
from pipecat.audio.filters.rnnoise_filter import RNNoiseFilter
from pipecat.frames.frames import FilterEnableFrame
class TestRNNoiseCancellation(unittest.IsolatedAsyncioTestCase):
async def test_rnnoise_cancellation_functionality(self):
print("\nStarting Noise Cancellation Test")
# 1. Check for pyrnnoise
if pyrnnoise is None:
self.skipTest("pyrnnoise not installed. Cannot verify actual noise cancellation.")
# 2. Generate clean speech-like audio (Harmonic series)
sample_rate = 48000
duration = 2.0
t = np.linspace(0, duration, int(sample_rate * duration), endpoint=False)
# Fundamental 200Hz + harmonics
clean_signal = np.sin(2 * np.pi * 200 * t) * 0.5
clean_signal += np.sin(2 * np.pi * 400 * t) * 0.3
clean_signal += np.sin(2 * np.pi * 600 * t) * 0.2
# Apply envelope to simulate speech (bursts)
# sin(2*pi*2*t) has period 0.5s.
envelope = np.sin(2 * np.pi * 2 * t)
envelope = np.clip(envelope, 0, 1)
clean_signal *= envelope
# 3. Add Noise (White Noise)
noise_level = 0.1 # Reduced noise level slightly to make speech clearer for alignment
noise = np.random.normal(0, noise_level, len(t))
noisy_signal = clean_signal + noise
# Normalize to int16 range
noisy_signal = np.clip(noisy_signal, -1, 1)
noisy_int16 = (noisy_signal * 32767).astype(np.int16)
noisy_bytes = noisy_int16.tobytes()
clean_int16 = (clean_signal * 32767).astype(np.int16)
print(f"Generated 2s of noisy audio at {sample_rate}Hz")
# 4. Initialize RNNoiseFilter
rnnoise_filter = RNNoiseFilter()
await rnnoise_filter.start(sample_rate)
await rnnoise_filter.process_frame(FilterEnableFrame(enable=True))
# 5. Process
# Feed in chunks
chunk_size = 960 # 20ms
processed_audio = b""
for i in range(0, len(noisy_bytes), chunk_size):
chunk = noisy_bytes[i : i + chunk_size]
result = await rnnoise_filter.filter(chunk)
processed_audio += result
await rnnoise_filter.stop()
print(f"Output audio size: {len(processed_audio)}")
# 6. Verify Noise Reduction
output_int16 = np.frombuffer(processed_audio, dtype=np.int16)
# Truncate to min length
min_len = min(len(clean_int16), len(output_int16))
clean_trunc = clean_int16[:min_len]
output_trunc = output_int16[:min_len]
noisy_trunc = noisy_int16[:min_len]
# 7. Compensate for Delay
# Use cross-correlation on a segment to find delay
# We expect output to be delayed relative to clean (lag is positive)
# search window +/- 2000 samples (~40ms)
search_range = 2400 # 50ms
# Use the middle of the signal to avoid edge effects and have strong signal
mid_point = min_len // 2
window_len = 4800 # 100ms
ref_sig = clean_trunc[mid_point : mid_point + window_len].astype(float)
target_sig = output_trunc[
mid_point - search_range : mid_point + window_len + search_range
].astype(float)
correlation = np.correlate(target_sig, ref_sig, mode="valid")
best_idx = np.argmax(correlation)
# The 'valid' mode correlation result corresponds to shifts.
# index 0 matches alignment where ref starts at target start.
# target start is (mid_point - search_range).
# ref start is mid_point.
# So index 0 means target is shifted left by search_range (or delay = -search_range).
# delay = best_idx - search_range
delay = best_idx - search_range
print(f"Detected delay: {delay} samples ({delay / sample_rate * 1000:.2f} ms)")
# Shift output to align
if delay > 0:
# Output is delayed, so we need to look at output[delay:] to match clean[0:]
aligned_output = output_trunc[delay:]
aligned_clean = clean_trunc[: len(aligned_output)]
aligned_noisy = noisy_trunc[: len(aligned_output)]
elif delay < 0:
# Output is ahead (unlikely for causal filter), but handling it
aligned_output = output_trunc[:delay]
aligned_clean = clean_trunc[-delay:]
aligned_noisy = noisy_trunc[-delay:]
else:
aligned_output = output_trunc
aligned_clean = clean_trunc
aligned_noisy = noisy_trunc
# Recalculate MSE on aligned signals
mse_input = np.mean((aligned_noisy.astype(float) - aligned_clean.astype(float)) ** 2)
mse_output = np.mean((aligned_output.astype(float) - aligned_clean.astype(float)) ** 2)
print(f"MSE (Input vs Clean): {mse_input:.2f}")
print(f"MSE (Output vs Clean): {mse_output:.2f}")
# Also check noise reduction in silent regions
# Clean signal envelope is 0 at t=0, 0.25, 0.5...
# Let's find indices where aligned_clean is very small
threshold = 100 # amplitude threshold (out of 32767)
silent_mask = np.abs(aligned_clean) < threshold
if np.sum(silent_mask) > 1000:
noise_power_input = np.mean(aligned_noisy[silent_mask].astype(float) ** 2)
noise_power_output = np.mean(aligned_output[silent_mask].astype(float) ** 2)
print(f"Noise Power in Silence (Input): {noise_power_input:.2f}")
print(f"Noise Power in Silence (Output): {noise_power_output:.2f}")
self.assertLess(
noise_power_output, noise_power_input, "Noise power in silence not reduced"
)
else:
print("Warning: Not enough silent samples found for noise floor check.")
# Main assertion: MSE should improve
# Relax assertion slightly because RNNoise introduces distortion even on clean speech
# But for noisy speech, it should generally be better or at least remove noise.
# If MSE doesn't improve (due to speech distortion), at least Noise Power in Silence should drop.
if mse_output >= mse_input:
print(
"Warning: Overall MSE did not improve (speech distortion?). Relying on Noise Power check."
)
# If we passed the noise power check above, we are good.
self.assertTrue(
np.sum(silent_mask) > 1000
and np.mean(aligned_output[silent_mask].astype(float) ** 2)
< np.mean(aligned_noisy[silent_mask].astype(float) ** 2)
)
else:
self.assertLess(mse_output, mse_input, "MSE did not improve")
print("Test Passed: Noise cancellation verified.")

102
tests/test_rnnoise_filter.py Normal file
View File

@@ -0,0 +1,102 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import unittest
import numpy as np
try:
import pyrnnoise
except ImportError:
pyrnnoise = None
from pipecat.audio.filters.rnnoise_filter import RNNoiseFilter
from pipecat.frames.frames import FilterEnableFrame
class TestRNNoiseFilter(unittest.IsolatedAsyncioTestCase):
def setUp(self):
if pyrnnoise is None:
self.skipTest("pyrnnoise not installed")
async def test_rnnoise_filter_reduces_noise(self):
"""Test that RNNoise filter reduces noise in audio."""
filter = RNNoiseFilter()
# Initialize with 48kHz sample rate (RNNoise requirement)
await filter.start(sample_rate=48000)
# Create noisy audio: clean signal + noise
# Generate a simple sine wave as clean signal
# Need at least 480 samples (one frame) for processing
duration = 0.02 # 20ms = 960 samples at 48kHz (2 frames)
sample_rate = 48000
t = np.linspace(0, duration, int(sample_rate * duration), False)
frequency = 440.0 # A4 note
clean_signal = np.sin(2 * np.pi * frequency * t)
# Add white noise
noise = np.random.normal(0, 0.3, clean_signal.shape)
noisy_signal = clean_signal + noise
# Convert to int16 format
noisy_audio_int16 = (noisy_signal * 32767).astype(np.int16)
noisy_audio_bytes = noisy_audio_int16.tobytes()
# Process through filter
filtered_audio_bytes = await filter.filter(noisy_audio_bytes)
# Convert back to numpy array for comparison
filtered_audio = np.frombuffer(filtered_audio_bytes, dtype=np.int16)
# Verify output is not empty (should have at least one processed frame)
self.assertGreater(len(filtered_audio), 0)
# Verify the filtered audio is different from input (noise reduction occurred)
# The filtered audio should have less variance/noise
self.assertIsNotNone(filtered_audio_bytes)
await filter.stop()
async def test_rnnoise_filter_passthrough_when_disabled(self):
"""Test that RNNoise filter passes through audio when disabled."""
filter = RNNoiseFilter()
await filter.start(sample_rate=48000)
# Disable filtering
await filter.process_frame(FilterEnableFrame(enable=False))
# Create test audio
test_audio = np.random.randint(-32768, 32767, 480, dtype=np.int16).tobytes()
# Process through filter
filtered_audio = await filter.filter(test_audio)
# Should pass through unchanged when disabled
self.assertEqual(filtered_audio, test_audio)
await filter.stop()
async def test_rnnoise_filter_buffering(self):
"""Test that RNNoise filter properly buffers incomplete frames."""
filter = RNNoiseFilter()
await filter.start(sample_rate=48000)
# Send a small chunk that's less than a full frame (480 samples)
small_chunk = np.random.randint(-32768, 32767, 100, dtype=np.int16).tobytes()
# First call should return empty (buffering, not enough for a frame)
result1 = await filter.filter(small_chunk)
self.assertEqual(result1, b"")
# Send more data to complete a frame (100 + 500 = 600 samples > 480)
more_data = np.random.randint(-32768, 32767, 500, dtype=np.int16).tobytes()
result2 = await filter.filter(more_data)
# Should return processed audio for at least one complete frame
self.assertGreater(len(result2), 0)
await filter.stop()

129
tests/test_rnnoise_resampling.py Normal file
View File

@@ -0,0 +1,129 @@
import asyncio
import sys
import unittest
from unittest.mock import MagicMock, patch
import numpy as np
# We don't need to mock sys.modules here if we use patch on the imported module member
# But we need to ensure RNNoiseFilter is imported so we can patch its member
try:
from pipecat.audio.filters.rnnoise_filter import RNNoiseFilter
from pipecat.frames.frames import FilterEnableFrame
except ImportError as e:
# If dependencies are missing (like numpy?), we can't test
print(f"Failed to import RNNoiseFilter: {e}")
sys.exit(1)
class TestRNNoiseResampling(unittest.IsolatedAsyncioTestCase):
@patch("pipecat.audio.filters.rnnoise_filter.RNNoise")
async def test_rnnoise_resampling_16k_to_48k_and_back(self, mock_rnnoise_class):
print("\nStarting Resampling Test: 16kHz -> 48kHz -> 16kHz")
# Configure Mock with buffering behavior
processed_chunks_count = 0
buffer = np.array([], dtype=np.int16)
def side_effect_process_chunk(audio_samples, partial=False):
nonlocal buffer, processed_chunks_count
# Append new samples to buffer
if len(audio_samples) > 0:
buffer = np.concatenate((buffer, audio_samples))
# Yield 480-sample chunks
while len(buffer) >= 480:
chunk = buffer[:480]
buffer = buffer[480:]
processed_chunks_count += 1
# Simulate processing (pass through)
# Convert int16 -> float32 [-1, 1]
normalized = chunk.astype(np.float32) / 32768.0
yield 0.99, normalized
mock_rnnoise_instance = MagicMock()
mock_rnnoise_instance.denoise_chunk.side_effect = side_effect_process_chunk
mock_rnnoise_class.return_value = mock_rnnoise_instance
# 1. Generate 1 second of 16kHz audio (sine wave 440Hz)
sample_rate = 16000
duration = 1.0
t = np.linspace(0, duration, int(sample_rate * duration), endpoint=False)
audio_data = (np.sin(2 * np.pi * 440 * t) * 32767).astype(np.int16)
audio_bytes = audio_data.tobytes()
print(
f"Input audio: {len(audio_bytes)} bytes, {len(audio_data)} samples at {sample_rate}Hz"
)
# 2. Initialize RNNoiseFilter
# This will use the patched RNNoise
rnnoise_filter = RNNoiseFilter()
await rnnoise_filter.start(sample_rate)
# Enable filtering
await rnnoise_filter.process_frame(FilterEnableFrame(enable=True))
# 3. Process audio in chunks
chunk_size = 320 # 160 samples (10ms at 16k) * 2 bytes
processed_audio = b""
for i in range(0, len(audio_bytes), chunk_size):
chunk = audio_bytes[i : i + chunk_size]
result = await rnnoise_filter.filter(chunk)
processed_audio += result
await rnnoise_filter.stop()
print(f"Output audio: {len(processed_audio)} bytes")
print(f"Processed chunks (internal 480 samples): {processed_chunks_count}")
# 4. Verify output length
# Expect roughly same length
expected_chunks = (len(audio_data) * 48000 // sample_rate) // 480
print(f"Expected chunks: ~{expected_chunks}")
# Check that we actually processed something
self.assertGreaterEqual(
processed_chunks_count, expected_chunks - 5, "Too few chunks processed"
)
# Check output length
self.assertGreater(len(processed_audio), 0, "Output should not be empty")
# Check length matches input (with some tolerance for buffering latency)
# 100ms tolerance?
byte_tolerance = int(0.2 * sample_rate * 2)
self.assertGreaterEqual(
len(processed_audio),
len(audio_bytes) - byte_tolerance,
f"Output too short: {len(processed_audio)} vs {len(audio_bytes)}",
)
self.assertLessEqual(
len(processed_audio),
len(audio_bytes) + byte_tolerance,
f"Output too long: {len(processed_audio)} vs {len(audio_bytes)}",
)
# 5. Check sample rate / pitch preservation
output_data = np.frombuffer(processed_audio, dtype=np.int16)
if len(output_data) > 2000:
# Use a window in the middle
start_idx = len(output_data) // 4
end_idx = 3 * len(output_data) // 4
segment = output_data[start_idx:end_idx]
fft = np.fft.rfft(segment)
freqs = np.fft.rfftfreq(len(segment), d=1 / sample_rate)
peak_idx = np.argmax(np.abs(fft))
peak_freq = freqs[peak_idx]
print(f"Peak frequency: {peak_freq:.2f} Hz")
self.assertLess(
abs(peak_freq - 440), 50, f"Frequency shifted significantly: {peak_freq} vs 440"
)
print("Test Passed: Resampling logic verified (with mocked RNNoise).")