From 8de05a54687f54fbd6480aaca09148a345d442c8 Mon Sep 17 00:00:00 2001 From: Con-Benksl Date: Wed, 9 Sep 2026 12:51:11 -0700 Subject: [PATCH] Fix time-domain convolution kernel orientation --- speechbrain/processing/signal_processing.py | 3 +- tests/unittests/test_convolve1d.py | 86 +++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) create mode 100644 tests/unittests/test_convolve1d.py diff --git a/speechbrain/processing/signal_processing.py b/speechbrain/processing/signal_processing.py index ccccd9c381..149a84d6e9 100644 --- a/speechbrain/processing/signal_processing.py +++ b/speechbrain/processing/signal_processing.py @@ -295,9 +295,10 @@ def convolve1d( # Use the implementation given by torch, which should be efficient on GPU else: + # conv1d performs cross-correlation, so reverse the kernel for convolution. convolved = torch.nn.functional.conv1d( input=waveform, - weight=kernel, + weight=kernel.flip(-1), stride=stride, groups=groups, padding=padding if not isinstance(padding, tuple) else 0, diff --git a/tests/unittests/test_convolve1d.py b/tests/unittests/test_convolve1d.py new file mode 100644 index 0000000000..0a143896ec --- /dev/null +++ b/tests/unittests/test_convolve1d.py @@ -0,0 +1,86 @@ +import pytest +import torch + + +@pytest.mark.parametrize("padding", [0, 1, (2, 2), (2, 0), (0, 2)]) +@pytest.mark.parametrize("stride", [1, 2]) +@pytest.mark.parametrize("dtype", [torch.float32, torch.float64]) +def test_convolve1d_time_domain(device, padding, stride, dtype): + import numpy as np + + from speechbrain.processing.signal_processing import convolve1d + + waveform = torch.tensor([1, 2, 3, 4], device=device, dtype=dtype) + kernel = torch.tensor([1, 2, 4], device=device, dtype=dtype) + pad_width = padding if isinstance(padding, tuple) else (padding, padding) + expected = np.convolve( + np.pad(waveform.cpu().numpy(), pad_width), + kernel.cpu().numpy(), + mode="valid", + )[::stride] + + result = convolve1d( + waveform.view(1, -1, 1), + kernel.view(1, -1, 1), + padding=padding, + stride=stride, + ) + + assert result.dtype == dtype + np.testing.assert_allclose(result[0, :, 0].cpu().numpy(), expected) + + +@pytest.mark.parametrize("dtype", [torch.float32, torch.float64]) +def test_convolve1d_time_domain_matches_fft(device, dtype): + from speechbrain.processing.signal_processing import convolve1d + + waveform = torch.tensor([1, 2, 3, 4], device=device, dtype=dtype).view( + 1, -1, 1 + ) + kernel = torch.tensor([1, 2, 4], device=device, dtype=dtype).view(1, -1, 1) + time_result = convolve1d(waveform, kernel, padding=(2, 2)) + fft_result = convolve1d(waveform, kernel, padding=(0, 2), use_fft=True) + expected = torch.tensor( + [1, 4, 11, 18, 20, 16], device=device, dtype=dtype + ).view(1, -1, 1) + + torch.testing.assert_close(time_result, expected) + torch.testing.assert_close(time_result, fft_result) + + +@pytest.mark.parametrize("stride", [1, 2]) +def test_convolve1d_grouped(device, stride): + import numpy as np + + from speechbrain.processing.signal_processing import convolve1d + + waveform = torch.arange(24, device=device, dtype=torch.float64).view( + 2, 6, 2 + ) + kernel = torch.arange(1, 13, device=device, dtype=torch.float64).view( + 4, 3, 1 + ) + result = convolve1d(waveform, kernel, groups=2, stride=stride) + + for batch in range(2): + for channel in range(4): + expected = np.convolve( + waveform[batch, :, channel // 2].cpu().numpy(), + kernel[channel, :, 0].cpu().numpy(), + mode="valid", + )[::stride] + np.testing.assert_allclose( + result[batch, :, channel].cpu().numpy(), expected + ) + + +def test_convolve1d_gradients(device): + from speechbrain.processing.signal_processing import convolve1d + + waveform = torch.randn( + 1, 4, 1, device=device, dtype=torch.float64, requires_grad=True + ) + kernel = torch.randn( + 1, 3, 1, device=device, dtype=torch.float64, requires_grad=True + ) + assert torch.autograd.gradcheck(convolve1d, (waveform, kernel))