Skip to content

Commit 3b2fded

Browse files
authored
fix: overloads, WeightingFilter multichannel zi, version sync & SonarCloud CI (#45)
* fix: complete overloads, WeightingFilter multichannel zi, version sync - Add missing calculate_level=True overloads so mypy resolves explicit calculate_level=True calls correctly (4 clean overloads covering all sigbands × calculate_level combinations) - Fix WeightingFilter multichannel zi bug: lazy allocation deferred to filter() so channel dimension matches actual input shape (same pattern as OctaveFilterBank) - Sync __version__ from '1.1.2' to '1.1.4' to match pyproject.toml - Update WeightingFilter tests for lazy init + add multichannel test * fix(ci): pin sonarcloud action to v7 and use standard GITHUB_TOKEN - Change SonarSource/sonarqube-scan-action@master to @v7 (stable) - Replace secrets.TOKEN_GH with secrets.GITHUB_TOKEN (auto-available) * fix: WeightingFilter multichannel→1D zi transition & SonarCloud coverage paths - Fix needs_init to handle all ndim transitions (1D↔2D) in WeightingFilter - Add [tool.coverage.run] relative_files=true so coverage.xml uses relative source paths matching sonar.sources=src - Add test_weighting_filter_multichannel_to_mono_transition test * fix: reduce WeightingFilter.filter() cognitive complexity and fix CI coverage - Extract _init_filter_state() and _needs_zi_reinit() helpers from filter() to reduce Cognitive Complexity from 26 to ~5 (SonarCloud max: 15) - Change pip install to editable mode (-e) so coverage can trace src/ code (fixes 'No data was collected' warning and 0% new_coverage in SonarCloud)
1 parent 0a3d317 commit 3b2fded

6 files changed

Lines changed: 116 additions & 38 deletions

File tree

.github/workflows/python-app.yml

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ jobs:
5656
python -m pip install --upgrade pip
5757
pip install -r requirements.txt
5858
pip install -r requirements-dev.txt
59-
pip install .
59+
pip install -e .
6060
- name: Run tests
6161
env:
6262
NUMBA_DISABLE_JIT: 1
@@ -84,9 +84,9 @@ jobs:
8484
with:
8585
name: test-results-ubuntu-latest-3.13
8686
- name: SonarCloud Scan
87-
uses: SonarSource/sonarqube-scan-action@master
87+
uses: SonarSource/sonarqube-scan-action@v7
8888
env:
89-
GITHUB_TOKEN: ${{ secrets.TOKEN_GH }}
89+
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
9090
SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
9191

9292
pr-comment:

pyproject.toml

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ ignore_missing_imports = true
3838
[tool.bandit]
3939
exclude_dirs = ["tests", ".venv"]
4040

41+
[tool.coverage.run]
42+
relative_files = true
43+
4144
[tool.pytest.ini_options]
4245
filterwarnings = [
4346
"ignore:Low sampling rate:UserWarning",

src/pyoctaveband/__init__.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
# Use non-interactive backend for plots
2020
matplotlib.use("Agg")
2121

22-
__version__ = "1.1.2"
22+
__version__ = "1.1.4"
2323

2424
# Public methods
2525
__all__ = [

src/pyoctaveband/core.py

Lines changed: 25 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -131,52 +131,51 @@ def __repr__(self) -> str:
131131

132132
@overload
133133
def filter(
134-
self,
135-
x: List[float] | np.ndarray,
134+
self,
135+
x: List[float] | np.ndarray,
136136
sigbands: Literal[False] = False,
137137
mode: str = "rms",
138-
detrend: bool = True
138+
detrend: bool = True,
139+
calculate_level: Literal[True] = True,
139140
) -> Tuple[np.ndarray, List[float]]: ...
140141

141142
@overload
142143
def filter(
143-
self,
144-
x: List[float] | np.ndarray,
144+
self,
145+
x: List[float] | np.ndarray,
145146
sigbands: Literal[True],
146147
mode: str = "rms",
147-
detrend: bool = True
148+
detrend: bool = True,
149+
calculate_level: Literal[True] = True,
148150
) -> Tuple[np.ndarray, List[float], List[np.ndarray]]: ...
149151

150-
# New overloads with calculate_level
151152
@overload
152153
def filter(
153-
self,
154-
x: List[float] | np.ndarray,
155-
sigbands: Literal[False] = False,
156-
mode: str = "rms",
157-
detrend: bool = True,
158-
calculate_level: Literal[False] = False
159-
) -> Tuple[None, List[float]]:
160-
...
154+
self,
155+
x: List[float] | np.ndarray,
156+
sigbands: Literal[False] = False,
157+
mode: str = "rms",
158+
detrend: bool = True,
159+
calculate_level: Literal[False] = False,
160+
) -> Tuple[None, List[float]]: ...
161161

162162
@overload
163163
def filter(
164-
self,
165-
x: List[float] | np.ndarray,
166-
sigbands: Literal[True],
167-
mode: str = "rms",
168-
detrend: bool = True,
169-
calculate_level: Literal[False] = False
170-
) -> Tuple[None, List[float], List[np.ndarray]]:
171-
...
164+
self,
165+
x: List[float] | np.ndarray,
166+
sigbands: Literal[True],
167+
mode: str = "rms",
168+
detrend: bool = True,
169+
calculate_level: Literal[False] = False,
170+
) -> Tuple[None, List[float], List[np.ndarray]]: ...
172171

173172
def filter(
174-
self,
175-
x: List[float] | np.ndarray,
173+
self,
174+
x: List[float] | np.ndarray,
176175
sigbands: bool = False,
177176
mode: str = "rms",
178177
detrend: bool = True,
179-
calculate_level: bool =True
178+
calculate_level: bool = True,
180179
) -> Tuple[np.ndarray | None, List[float]] | Tuple[np.ndarray | None, List[float], List[np.ndarray]]:
181180
"""
182181
Apply the pre-designed filter bank to a signal.

src/pyoctaveband/parametric_filters.py

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -84,25 +84,51 @@ def __init__(self, fs: int, curve: str = "A",
8484
zd, pd, kd = signal.bilinear_zpk(z, p, k, fs)
8585
self.sos = signal.zpk2sos(zd, pd, kd)
8686

87-
# Calculate initial conditions for filter state
87+
# Initialize filter state for stateful block-wise processing.
88+
# Uses lazy allocation: zi is sized on first filter() call so that
89+
# the channel dimension matches the actual input shape.
8890
if self.stateful:
89-
if not steady_ic:
90-
self.zi = np.zeros((self.sos.shape[0], 2))
91-
else:
91+
self.zi = np.array([])
92+
self._steady_ic = steady_ic
93+
94+
def _init_filter_state(self, x_proc: np.ndarray) -> None:
95+
"""Allocate or reallocate ``zi`` to match the input shape."""
96+
n_sections = self.sos.shape[0]
97+
if x_proc.ndim == 1:
98+
if self._steady_ic:
9299
self.zi = signal.sosfilt_zi(self.sos)
100+
else:
101+
self.zi = np.zeros((n_sections, 2))
102+
else:
103+
n_channels = x_proc.shape[0]
104+
if self._steady_ic:
105+
zi_base = signal.sosfilt_zi(self.sos)
106+
self.zi = np.tile(zi_base[:, np.newaxis, :], (1, n_channels, 1))
107+
else:
108+
self.zi = np.zeros((n_sections, n_channels, 2))
109+
110+
def _needs_zi_reinit(self, x_proc: np.ndarray) -> bool:
111+
"""Check whether ``zi`` must be (re)allocated for *x_proc*."""
112+
if self.zi.size == 0:
113+
return True
114+
if x_proc.ndim == 1:
115+
return self.zi.ndim != 2
116+
return self.zi.ndim != 3 or self.zi.shape[1] != x_proc.shape[0]
93117

94118
def filter(self, x: List[float] | np.ndarray) -> np.ndarray:
95119
"""
96120
Apply the weighting filter to a signal.
97121
98-
:param x: Input signal.
122+
:param x: Input signal (1D or 2D [channels, samples]).
99123
:return: Weighted signal.
100124
"""
101125
x_proc = _typesignal(x)
102126
if self.curve == "Z":
103127
return x_proc
104128

105129
if self.stateful:
130+
if self._needs_zi_reinit(x_proc):
131+
self._init_filter_state(x_proc)
106132
y, self.zi = signal.sosfilt(self.sos, x_proc, axis=-1, zi=self.zi)
107133
else:
108134
y = signal.sosfilt(self.sos, x_proc, axis=-1)

tests/test_stateful_weighting_filter.py

Lines changed: 53 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,58 @@ def test_weighting_filter_steady_ic_initialization():
4444
# Create a stateful weighting filter with steady_ic=True
4545
wf = WeightingFilter(fs=48000, stateful=True, steady_ic=True)
4646

47-
# Check that zi has the expected shape
47+
# Lazy init: zi is empty until first filter() call
48+
assert wf.zi.size == 0
49+
50+
# Trigger lazy init with 1D signal
51+
x = np.zeros(100)
52+
wf.filter(x)
53+
4854
n_sections = wf.sos.shape[0]
49-
assert wf.zi.shape[0] == n_sections
50-
assert wf.zi.shape[1] == 2
55+
assert wf.zi.shape == (n_sections, 2)
56+
57+
58+
def test_weighting_filter_multichannel_to_mono_transition():
59+
from pyoctaveband import WeightingFilter
60+
"""zi must reinit when input switches from multichannel to 1D."""
61+
rng = np.random.default_rng(77)
62+
fs = 48000
63+
wf = WeightingFilter(fs, "A", stateful=True)
64+
65+
# First call: multichannel (4 channels)
66+
x_multi = rng.standard_normal((4, 1200))
67+
wf.filter(x_multi)
68+
assert wf.zi.ndim == 3
69+
70+
# Second call: 1D — must not crash
71+
x_mono = rng.standard_normal(1200)
72+
y = wf.filter(x_mono)
73+
assert wf.zi.ndim == 2
74+
assert y.shape == x_mono.shape
75+
76+
77+
def test_weighting_filter_multichannel():
78+
from pyoctaveband import WeightingFilter
79+
"""Stateful block-wise multichannel weighting must match full-signal processing."""
80+
rng = np.random.default_rng(99)
81+
fs = 48000
82+
n_channels = 4
83+
n_samples = 4800
84+
block_size = 1200
85+
86+
x = rng.standard_normal((n_channels, n_samples))
87+
88+
# Full-signal reference (stateless)
89+
ref_wf = WeightingFilter(fs, "A")
90+
ref_out = ref_wf.filter(x)
91+
92+
# Block-wise stateful
93+
stateful_wf = WeightingFilter(fs, "A", stateful=True)
94+
blocks = []
95+
for start in range(0, n_samples, block_size):
96+
block = x[:, start:start + block_size]
97+
blocks.append(stateful_wf.filter(block))
98+
99+
stateful_out = np.concatenate(blocks, axis=-1)
100+
np.testing.assert_allclose(stateful_out, ref_out, rtol=1e-10, atol=1e-12)
51101

0 commit comments

Comments
 (0)