Add stateful block-wise processing to OctaveFilterBank and WeightingFilter #42
Merged
jmrplens merged 9 commits intojmrplens:mainfrom Mar 8, 2026
Merged
Conversation
added 6 commits
February 26, 2026 13:57
Reviewer's GuideImplements stateful, block-wise processing support for OctaveFilterBank and WeightingFilter, including persistent SOS filter state handling, optional steady-state initialization, an option to skip SPL computation, documentation updates, and tests validating equivalence to full-signal processing and enforcing unsupported combinations. Sequence diagram for stateful block-wise processing pipelinesequenceDiagram
actor User
participant SF as SoundFile
participant WF as WeightingFilter
participant OFB as OctaveFilterBank
User->>WF: WeightingFilter(fs, curve="A", stateful=True, steady_ic=False)
User->>OFB: OctaveFilterBank(fs, fraction=1, stateful=True, resample=False)
User->>SF: blocks("measurement.wav", blocksize=256, overlap=0)
loop for each block
SF-->>User: block
User->>WF: filter(block)
alt stateful weighting filter
WF->>WF: sosfilt(sos, block, zi)
WF->>WF: update zi
end
WF-->>User: weighted
User->>OFB: filter(weighted, sigbands=True, detrend=False, calculate_level=True)
OFB->>OFB: _process_bands(x_proc, num_channels, sigbands, mode, calculate_level)
loop for each band idx
OFB->>OFB: _filter_and_resample(x_proc, idx)
alt stateful octave filter bank
OFB->>OFB: sosfilt(sos[idx], sd, zi[idx])
OFB->>OFB: update zi[idx]
else stateless
OFB->>OFB: sosfilt(sos[idx], sd)
end
OFB->>OFB: _calculate_level(filtered_signal, mode)
end
OFB-->>User: block_spl, freq, block_output
end
Class diagram for updated OctaveFilterBank and WeightingFilterclassDiagram
class OctaveFilterBank {
+int fs
+int fraction
+int order
+str filter_type
+str limits
+bool stateful
+bool resample
+int num_bands
+List[int] factor
+List[np.ndarray] sos
+List[np.ndarray] zi
+OctaveFilterBank(int fs, int fraction, int order, str filter_type, str limits, bool stateful, bool steady_ic, bool resample, bool show, str plot_file, float calibration_factor, bool dbfs)
+filter(List[float] x, bool sigbands, str mode, bool detrend, bool calculate_level)
+_process_bands(np.ndarray x_proc, int num_channels, bool sigbands, str mode, bool calculate_level)
+_filter_and_resample(np.ndarray x, int idx)
+_calculate_level(np.ndarray y, str mode) float
}
class WeightingFilter {
+int fs
+str curve
+bool stateful
+np.ndarray sos
+np.ndarray zi
+WeightingFilter(int fs, str curve, bool stateful, bool steady_ic)
+filter(List[float] x) np.ndarray
}
OctaveFilterBank "1" o-- "many" WeightingFilter : used_with_in_pipeline
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This PR introduces stateful block-wise processing to OctaveFilterBank and WeightingFilter, a valuable enhancement for streaming audio analysis. The implementation is well-structured and includes thorough tests to ensure correctness. My review focuses on improving the warning mechanism to prevent unintended program termination and enhancing the clarity of the documentation for the new feature. With these minor adjustments, this will be an excellent addition to the library.
src/pyoctaveband/core.py
Outdated
Comment on lines
+192
to
+193
| if self.stateful: | ||
| raise Warning("You should not detrend when doing block processing!") |
There was a problem hiding this comment.
Using raise Warning(...) will raise an exception and terminate the program flow, which is likely more disruptive than intended for a warning. The standard practice for issuing a warning without halting execution is to use the warnings module. This allows the user to be notified of the potential issue with detrending during block processing (which can cause discontinuities) without stopping their workflow.
Suggested change
| if self.stateful: | |
| raise Warning("You should not detrend when doing block processing!") | |
| if self.stateful: | |
| import warnings | |
| warnings.warn( | |
| "Detrending is not recommended for stateful block processing as it can " | |
| "introduce discontinuities between blocks.", | |
| UserWarning, | |
| ) |
README.md
Outdated
| Create a stateful filter bank with `stateful=True`. The internal state is zero-initialized by default | ||
| but may be initialized for step-response steady-state (like `scipy.signal.sosfilt_zi`) with `steady_ic=True`. | ||
| Notes when using a stateful `OctaveFilterBank`: | ||
| - You should not remove the DC-content when block processing (`detrend=False`). |
There was a problem hiding this comment.
The current phrasing is slightly ambiguous. It could be interpreted as "you should not detrend, which is done by setting detrend=False". Clarifying that detrend=False is the recommended setting would improve readability for users of this new feature.
Suggested change
| - You should not remove the DC-content when block processing (`detrend=False`). | |
| - Detrending should be disabled when block processing (`detrend=False`), as it can introduce discontinuities between blocks. |
Comment on lines
+87
to
+88
| with pytest.raises(Warning): | ||
| bank.filter(signal, detrend=True) |
There was a problem hiding this comment.
To align with the suggested change of using warnings.warn instead of raise Warning for a non-halting warning, this test should be updated to use pytest.warns. This correctly tests for the presence of a warning without treating it as a program-halting exception.
Suggested change
| with pytest.raises(Warning): | |
| bank.filter(signal, detrend=True) | |
| with pytest.warns(UserWarning, match="Detrending is not recommended"): | |
| bank.filter(signal, detrend=True) |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 8 issues, and left some high level feedback:
- The new
calculate_levelflag can causesplto beNonewhilefilter()and its overloads still advertisenp.ndarrayin the return type; consider tightening the typing/overloads or hiding this flag in a private helper so the public API contract remains consistent. - Using
raise Warninginfilter()for the stateful+detrend case is unusual for runtime warnings; consider switching towarnings.warn(..., RuntimeWarning)(or a custom exception) so callers can manage it via the standard warnings or exception mechanisms. - The stateful
OctaveFilterBankinitial conditions are sized with a single channel dimension ((n_sections, 1, 2)), so multiple channels will share the same filter state; if per-channel independence is desired, you may want to allocateziper channel or document that only mono input is supported for stateful mode.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new `calculate_level` flag can cause `spl` to be `None` while `filter()` and its overloads still advertise `np.ndarray` in the return type; consider tightening the typing/overloads or hiding this flag in a private helper so the public API contract remains consistent.
- Using `raise Warning` in `filter()` for the stateful+detrend case is unusual for runtime warnings; consider switching to `warnings.warn(..., RuntimeWarning)` (or a custom exception) so callers can manage it via the standard warnings or exception mechanisms.
- The stateful `OctaveFilterBank` initial conditions are sized with a single channel dimension (`(n_sections, 1, 2)`), so multiple channels will share the same filter state; if per-channel independence is desired, you may want to allocate `zi` per channel or document that only mono input is supported for stateful mode.
## Individual Comments
### Comment 1
<location path="src/pyoctaveband/core.py" line_range="110-119" />
<code_context>
)
+ # Calculate initial conditions for filter state
+ if self.stateful:
+ self.zi = [None for _ in range(self.num_bands)]
+ for idx in range(self.num_bands):
</code_context>
<issue_to_address>
**issue (bug_risk):** Raising `Warning` as an exception here is likely unintended and makes the default `detrend=True` unusable for stateful filters.
Using `raise Warning(...)` turns this into an exception, so `stateful=True` with the default `detrend=True` will always fail at runtime. If the intent is to discourage detrending with block processing, consider instead:
- changing the default to `detrend=False` when `stateful=True`,
- using `warnings.warn(...)`, or
- raising a more specific exception (e.g. `ValueError`) and documenting that `detrend` must be `False` for stateful processing.
The current behavior is a runtime trap and `Warning` is not typically used as an exception type.
</issue_to_address>
### Comment 2
<location path="src/pyoctaveband/core.py" line_range="174" />
<code_context>
mode: str = "rms",
- detrend: bool = True
+ detrend: bool = True,
+ calculate_level: bool =True
) -> Tuple[np.ndarray, List[float]] | Tuple[np.ndarray, List[float], List[np.ndarray]]:
"""
</code_context>
<issue_to_address>
**issue (bug_risk):** When `calculate_level=False`, `spl` becomes `None` but the public API and overloads still advertise a `np.ndarray` return type.
`_process_bands` sets `spl` to `None` when `calculate_level=False`, and that value is returned from `filter` as part of `(spl, self.freq, xb)`. Callers will expect `spl` to be an array based on the type hints and docstring, and your new overloads for `calculate_level: Literal[False]` still advertise `Tuple[np.ndarray, List[float]]` / `Tuple[np.ndarray, List[float], List[np.ndarray]]`. This contract/type mismatch can cause runtime errors and confuse static type checkers.
Please either ensure `spl` is always an array (e.g. zeros/NaNs when `calculate_level=False`), or update the `filter` signature and overloads so that `spl` can be `None` or omitted when `calculate_level=False` (potentially with a distinct overload for the bands-only case).
</issue_to_address>
### Comment 3
<location path="src/pyoctaveband/core.py" line_range="112-117" />
<code_context>
+ # Calculate initial conditions for filter state
+ if self.stateful:
+ self.zi = [None for _ in range(self.num_bands)]
+ for idx in range(self.num_bands):
+ if not steady_ic:
+ self.zi[idx] = np.zeros((self.sos[idx].shape[0], 1, 2))
+ else:
+ zi = signal.sosfilt_zi(self.sos[idx])
+ self.zi[idx] = zi[:, np.newaxis, :] # add a dimension since we are filtering along an axis in a 2D-array
+
+
</code_context>
<issue_to_address>
**issue (bug_risk):** The shape of `zi` for stateful filtering likely does not match `sosfilt`’s expectations for multi-channel input.
For stateful runs, `self.zi[idx]` is created as `(n_sections, 1, 2)` or `(n_sections, channels, 2)` and then passed to `signal.sosfilt(..., axis=-1)`. However, `sosfilt` expects `zi` to be shaped `(n_sections, ...)` where `...` matches the input with the filter axis removed (e.g., `(n_sections, num_channels)` for a `(num_channels, num_samples)` input). The extra trailing `2` dimension is closer to the `lfilter` API and may not broadcast correctly. Please confirm the SciPy contract for `zi` with your `x_proc` shape and adjust the initialization (e.g. drop the last dim) to avoid shape errors or incorrect cross-channel state handling.
</issue_to_address>
### Comment 4
<location path="src/pyoctaveband/parametric_filters.py" line_range="43-47" />
<code_context>
)
+ # Calculate initial conditions for filter state
+ if self.stateful:
+ self.zi = [None for _ in range(self.num_bands)]
+ for idx in range(self.num_bands):
</code_context>
<issue_to_address>
**suggestion:** Stateful `WeightingFilter` uses a `zi` shape tailored for 1D signals; multi-channel inputs may not be handled correctly.
`self.zi` is currently sized `(n_sections, 2)` (or zeros of that shape), which fits 1D signals with `axis=-1` but not multi-channel inputs (e.g. `(channels, samples)`), where `sosfilt` would expect `(n_sections, channels)`. Consider either enforcing 1D-only input in `_typesignal` or making `zi` initialization depend on the input shape to avoid incorrect state alignment if/when multi-channel data is passed in.
</issue_to_address>
### Comment 5
<location path="tests/test_stateful_octave_filter_bank.py" line_range="4-13" />
<code_context>
+@pytest.mark.parametrize("block_size", [8, 256, 1024])
</code_context>
<issue_to_address>
**suggestion (testing):** Extend block-wise vs full-signal test to cover SPL output and a multichannel case
This test currently only compares `block_output_signal` vs `full_output_signal` for 1D input and doesn’t exercise SPL or multichannel behavior.
To strengthen it:
1. Also compare SPL outputs by calling `filter(..., sigbands=True, calculate_level=True)` for both full and block-wise processing and asserting that the concatenated per-band SPL values match.
2. Add a multichannel case (e.g. parameterize `n_channels` for 1 and 2, or add a separate `(n_channels, n_samples)` test) so the stateful logic and per-band `zi` handling are exercised for multi-channel input as well.
Suggested implementation:
```python
import numpy as np
import pytest
@pytest.mark.parametrize("block_size", [8, 256, 1024])
@pytest.mark.parametrize("n_channels", [1, 2])
def test_block_processing_matches_full_signal(block_size: int, n_channels: int):
from pyoctaveband import OctaveFilterBank
"""
Ensure that block-wise processing with preserved filter state
produces the same result as filtering the full signal at once,
for both single-channel and multichannel input.
Also verify that SPL outputs produced via full-signal and
block-wise processing are identical when concatenated.
"""
rng = np.random.default_rng(42)
fs = 48000
```
To fully implement the requested behavior, you should update the body of `test_block_processing_matches_full_signal` (after `fs = 48000`) roughly as follows:
1. **Create multi-channel input**
- Define a number of samples, e.g. `n_samples = 48000`.
- Generate a 2D signal when `n_channels > 1`:
```python
if n_channels == 1:
x = rng.standard_normal(n_samples)
else:
x = rng.standard_normal((n_channels, n_samples))
```
Adjust shape/order if the rest of your tests expect `(n_samples, n_channels)` instead.
2. **Instantiate the filter bank**
- Reuse whatever center frequencies / bands you use in the existing version of this test, e.g.:
```python
ofb = OctaveFilterBank(fs=fs, fraction=1, order=4)
```
3. **Full-signal processing (time-domain output + SPL)**
- Call the filter for the full signal, once for band signals, once for SPL:
```python
full_output_signal = ofb.filter(x)
full_output_signal_spl, full_spl = ofb.filter(
x, sigbands=True, calculate_level=True
)
```
- If `.filter` already returns both band signals and SPL when `calculate_level=True`, adapt accordingly (e.g. `full_output_signal_spl, full_spl = ofb.filter(..., calculate_level=True)`).
4. **Block-wise processing (time-domain output + SPL)**
- Reset/init the filter bank state before block-wise processing if required by your API.
- Iterate over the signal in steps of `block_size`, preserving filter state:
```python
block_outputs = []
block_outputs_spl = []
block_spl_values = []
for start in range(0, n_samples, block_size):
stop = min(start + block_size, n_samples)
x_block = x[..., start:stop] # works for 1D and multi-channel
y_block = ofb.filter(x_block)
block_outputs.append(y_block)
y_block_spl, spl_block = ofb.filter(
x_block, sigbands=True, calculate_level=True
)
block_outputs_spl.append(y_block_spl)
block_spl_values.append(spl_block)
```
5. **Concatenate block-wise outputs**
- Concatenate the lists along the time axis:
```python
block_output_signal = np.concatenate(block_outputs, axis=-1)
block_output_signal_spl = np.concatenate(block_outputs_spl, axis=-1)
block_spl = np.concatenate(block_spl_values, axis=-1)
```
6. **Assertions for time-domain equality (existing behavior)**
- Keep or adapt the existing assertion comparing `block_output_signal` vs `full_output_signal`, using `np.allclose` with appropriate tolerances and axes to handle `n_channels`.
7. **Assertions for SPL equality (new behavior)**
- Add assertions to verify SPL outputs match for full vs block-wise processing:
```python
assert block_output_signal_spl.shape == full_output_signal_spl.shape
np.testing.assert_allclose(block_output_signal_spl, full_output_signal_spl, rtol=1e-6, atol=1e-9)
assert block_spl.shape == full_spl.shape
np.testing.assert_allclose(block_spl, full_spl, rtol=1e-6, atol=1e-9)
```
Adjust argument ordering (`sigbands`, `calculate_level`), return values, and array shapes to match the actual `OctaveFilterBank.filter` API and the conventions used elsewhere in your tests.
</issue_to_address>
### Comment 6
<location path="tests/test_stateful_weighting_filter.py" line_range="4-6" />
<code_context>
+import numpy as np
+import pytest
+
+@pytest.mark.parametrize("block_size", [8, 256, 1024])
+def test_block_processing_matches_full_signal(block_size: int):
+ from pyoctaveband import OctaveFilterBank
</code_context>
<issue_to_address>
**suggestion (testing):** Add explicit coverage for stateful weighting filter with steady-state initial conditions
The current parametrized test only covers the default `steady_ic=False` path. Please also add coverage for `WeightingFilter(..., stateful=True, steady_ic=True)`, either via a separate test or by extending the parametrization, to ensure the steady-state initial-conditions path preserves block vs full-signal equivalence.
</issue_to_address>
### Comment 7
<location path="tests/test_stateful_octave_filter_bank.py" line_range="53-62" />
<code_context>
+ bank = OctaveFilterBank(48000, resample=True, stateful=True)
+
+
+def test_stateful_steady_ic_initialization():
+ from pyoctaveband.core import OctaveFilterBank
+ # Create a stateful filter bank with steady_ic=True
+ bank = OctaveFilterBank(
+ fs=48000,
+ stateful=True,
+ steady_ic=True,
+ order=2, # small order is enough for coverage
+ fraction=1, # 1-octave
+ resample=False # avoid the resampling branch
+ )
+
+ # Check that zi is a list of numpy arrays with the expected shape
+ for idx, zi in enumerate(bank.zi):
+ # zi should have 3 dimensions: (n_sections, 1, 2)
+ assert isinstance(zi, np.ndarray)
+ assert zi.ndim == 3
+ n_sections = bank.sos[idx].shape[0]
+ assert zi.shape[0] == n_sections
+ assert zi.shape[1] == 1
+ assert zi.shape[2] == 2
+
+def test_detrend_stateful_warning():
</code_context>
<issue_to_address>
**suggestion (testing):** Consider asserting that stateful OctaveFilterBank with steady_ic=True also behaves correctly on actual data
Right now this only validates the structure of `zi`. To also exercise behavior, consider extending the test to process a small known signal with both:
- a stateless `OctaveFilterBank`, and
- a stateful `OctaveFilterBank` with `steady_ic=True` (and `calculate_level=False` if you only care about band outputs),
then assert that concatenated block-wise outputs match the full-signal outputs, similar to `test_block_processing_matches_full_signal`.
Suggested implementation:
```python
def test_stateful_steady_ic_initialization():
from pyoctaveband.core import OctaveFilterBank
# Create a stateful filter bank with steady_ic=True
bank = OctaveFilterBank(
fs=48000,
stateful=True,
steady_ic=True,
order=2, # small order is enough for coverage
fraction=1, # 1-octave
resample=False # avoid the resampling branch
)
# Check that zi is a list of numpy arrays with the expected shape
for idx, zi in enumerate(bank.zi):
# zi should have 3 dimensions: (n_sections, 1, 2)
assert isinstance(zi, np.ndarray)
assert zi.ndim == 3
n_sections = bank.sos[idx].shape[0]
assert zi.shape[0] == n_sections
assert zi.shape[1] == 1
assert zi.shape[2] == 2
# Additionally verify that a stateful bank with steady_ic=True behaves
# consistently with a stateless bank when processing actual data.
rng = np.random.default_rng(0)
fs = 48000
duration = 0.25
n_samples = int(fs * duration)
# Random test signal
signal = rng.standard_normal(n_samples)
# Stateless reference (process full signal at once)
stateless_bank = OctaveFilterBank(
fs=fs,
stateful=False,
steady_ic=False,
order=2,
fraction=1,
resample=False,
calculate_level=False, # only care about band outputs
)
full_output = stateless_bank.filter(signal)
# Stateful bank with steady_ic=True (process in blocks)
stateful_bank = OctaveFilterBank(
fs=fs,
stateful=True,
steady_ic=True,
order=2,
fraction=1,
resample=False,
calculate_level=False, # only care about band outputs
)
block_size = 64
block_outputs = []
for start in range(0, n_samples, block_size):
block = signal[start:start + block_size]
y_block = stateful_bank.filter(block)
block_outputs.append(y_block)
# Concatenate block-wise outputs and compare to full-signal stateless output
concat_output = np.concatenate(block_outputs, axis=-1)
assert concat_output.shape == full_output.shape
np.testing.assert_allclose(concat_output, full_output, rtol=1e-6, atol=1e-8)
def test_detrend_stateful_warning():
```
The above implementation assumes:
1. `OctaveFilterBank` exposes a `filter(signal: np.ndarray)` method that returns band outputs with a time axis that can be concatenated along `axis=-1`. If your API instead uses a different method name (e.g. `process`, `__call__`, or `filter_block`) or a different output shape, update the calls to `stateless_bank.filter(...)`, `stateful_bank.filter(...)`, and the `axis` argument in `np.concatenate` accordingly.
2. The constructor of `OctaveFilterBank` accepts `calculate_level`. If this argument does not exist or is named differently, remove or rename it to match the actual signature used elsewhere in the tests (e.g. copy the usage from `test_block_processing_matches_full_signal`).
3. If your tests use a helper to perform the "block vs full" comparison (e.g. a shared fixture or function in this file), you can refactor the block-processing part to use that helper for consistency.
</issue_to_address>
### Comment 8
<location path="tests/test_coverage_fix.py" line_range="152-158" />
<code_context>
with pytest.raises(ValueError):
bank._calculate_level(np.array([1.0]), "invalid_mode")
+
+def test_dont_calculate_level():
+ from pyoctaveband.core import OctaveFilterBank
+ bank = OctaveFilterBank(48000)
+ x = np.zeros((bank.num_bands, 100))
+ spl, y = bank._process_bands(x, num_channels=bank.num_bands, calculate_level=False, sigbands=True)
+ assert spl is None
+ assert y is not None
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for the public calculate_level=False API and strengthen assertions on band outputs
This currently only exercises the private `_process_bands` helper. Since `calculate_level` is a public `filter` argument, please also add tests that call:
- `OctaveFilterBank.filter(x, sigbands=False, calculate_level=False)` and assert the SPL return matches the intended contract (e.g. `None`), and
- `OctaveFilterBank.filter(x, sigbands=True, calculate_level=False)` and assert both outputs are as expected.
It would also help to strengthen the assertions on `y` by checking the number of bands and shapes, e.g. `len(y) == bank.num_bands` and each element has shape `(num_channels, n_samples)` after resampling. This will better validate the new option and reduce reliance on private methods in tests.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+43
to
47
| if self.stateful: | ||
| self.zi = np.array([]) | ||
| return | ||
|
|
||
| if self.curve not in ["A", "C"]: |
There was a problem hiding this comment.
suggestion: Stateful WeightingFilter uses a zi shape tailored for 1D signals; multi-channel inputs may not be handled correctly.
self.zi is currently sized (n_sections, 2) (or zeros of that shape), which fits 1D signals with axis=-1 but not multi-channel inputs (e.g. (channels, samples)), where sosfilt would expect (n_sections, channels). Consider either enforcing 1D-only input in _typesignal or making zi initialization depend on the input shape to avoid incorrect state alignment if/when multi-channel data is passed in.
Comment on lines
+4
to
+13
| @pytest.mark.parametrize("block_size", [8, 256, 1024]) | ||
| def test_block_processing_matches_full_signal(block_size: int): | ||
| from pyoctaveband import OctaveFilterBank | ||
| """ | ||
| Ensure that block-wise processing with preserved filter state | ||
| produces the same result as filtering the full signal at once. | ||
| """ | ||
|
|
||
| rng = np.random.default_rng(42) | ||
|
|
There was a problem hiding this comment.
suggestion (testing): Extend block-wise vs full-signal test to cover SPL output and a multichannel case
This test currently only compares block_output_signal vs full_output_signal for 1D input and doesn’t exercise SPL or multichannel behavior.
To strengthen it:
- Also compare SPL outputs by calling
filter(..., sigbands=True, calculate_level=True)for both full and block-wise processing and asserting that the concatenated per-band SPL values match. - Add a multichannel case (e.g. parameterize
n_channelsfor 1 and 2, or add a separate(n_channels, n_samples)test) so the stateful logic and per-bandzihandling are exercised for multi-channel input as well.
Suggested implementation:
import numpy as np
import pytest
@pytest.mark.parametrize("block_size", [8, 256, 1024])
@pytest.mark.parametrize("n_channels", [1, 2])
def test_block_processing_matches_full_signal(block_size: int, n_channels: int):
from pyoctaveband import OctaveFilterBank
"""
Ensure that block-wise processing with preserved filter state
produces the same result as filtering the full signal at once,
for both single-channel and multichannel input.
Also verify that SPL outputs produced via full-signal and
block-wise processing are identical when concatenated.
"""
rng = np.random.default_rng(42)
fs = 48000To fully implement the requested behavior, you should update the body of test_block_processing_matches_full_signal (after fs = 48000) roughly as follows:
-
Create multi-channel input
- Define a number of samples, e.g.
n_samples = 48000. - Generate a 2D signal when
n_channels > 1:Adjust shape/order if the rest of your tests expectif n_channels == 1: x = rng.standard_normal(n_samples) else: x = rng.standard_normal((n_channels, n_samples))
(n_samples, n_channels)instead.
- Define a number of samples, e.g.
-
Instantiate the filter bank
- Reuse whatever center frequencies / bands you use in the existing version of this test, e.g.:
ofb = OctaveFilterBank(fs=fs, fraction=1, order=4)
- Reuse whatever center frequencies / bands you use in the existing version of this test, e.g.:
-
Full-signal processing (time-domain output + SPL)
- Call the filter for the full signal, once for band signals, once for SPL:
full_output_signal = ofb.filter(x) full_output_signal_spl, full_spl = ofb.filter( x, sigbands=True, calculate_level=True )
- If
.filteralready returns both band signals and SPL whencalculate_level=True, adapt accordingly (e.g.full_output_signal_spl, full_spl = ofb.filter(..., calculate_level=True)).
- Call the filter for the full signal, once for band signals, once for SPL:
-
Block-wise processing (time-domain output + SPL)
- Reset/init the filter bank state before block-wise processing if required by your API.
- Iterate over the signal in steps of
block_size, preserving filter state:block_outputs = [] block_outputs_spl = [] block_spl_values = [] for start in range(0, n_samples, block_size): stop = min(start + block_size, n_samples) x_block = x[..., start:stop] # works for 1D and multi-channel y_block = ofb.filter(x_block) block_outputs.append(y_block) y_block_spl, spl_block = ofb.filter( x_block, sigbands=True, calculate_level=True ) block_outputs_spl.append(y_block_spl) block_spl_values.append(spl_block)
-
Concatenate block-wise outputs
- Concatenate the lists along the time axis:
block_output_signal = np.concatenate(block_outputs, axis=-1) block_output_signal_spl = np.concatenate(block_outputs_spl, axis=-1) block_spl = np.concatenate(block_spl_values, axis=-1)
- Concatenate the lists along the time axis:
-
Assertions for time-domain equality (existing behavior)
- Keep or adapt the existing assertion comparing
block_output_signalvsfull_output_signal, usingnp.allclosewith appropriate tolerances and axes to handlen_channels.
- Keep or adapt the existing assertion comparing
-
Assertions for SPL equality (new behavior)
- Add assertions to verify SPL outputs match for full vs block-wise processing:
assert block_output_signal_spl.shape == full_output_signal_spl.shape np.testing.assert_allclose(block_output_signal_spl, full_output_signal_spl, rtol=1e-6, atol=1e-9) assert block_spl.shape == full_spl.shape np.testing.assert_allclose(block_spl, full_spl, rtol=1e-6, atol=1e-9)
- Add assertions to verify SPL outputs match for full vs block-wise processing:
Adjust argument ordering (sigbands, calculate_level), return values, and array shapes to match the actual OctaveFilterBank.filter API and the conventions used elsewhere in your tests.
Comment on lines
+4
to
+6
| @pytest.mark.parametrize("block_size", [8, 256, 1024]) | ||
| @pytest.mark.parametrize("filter_type", ["A", "C", "Z"]) | ||
| def test_weighting_filter_block_processing_matches_full_signal(block_size: int, filter_type: str): |
There was a problem hiding this comment.
suggestion (testing): Add explicit coverage for stateful weighting filter with steady-state initial conditions
The current parametrized test only covers the default steady_ic=False path. Please also add coverage for WeightingFilter(..., stateful=True, steady_ic=True), either via a separate test or by extending the parametrization, to ensure the steady-state initial-conditions path preserves block vs full-signal equivalence.
Comment on lines
+53
to
+62
| def test_stateful_steady_ic_initialization(): | ||
| from pyoctaveband.core import OctaveFilterBank | ||
| # Create a stateful filter bank with steady_ic=True | ||
| bank = OctaveFilterBank( | ||
| fs=48000, | ||
| stateful=True, | ||
| steady_ic=True, | ||
| order=2, # small order is enough for coverage | ||
| fraction=1, # 1-octave | ||
| resample=False # avoid the resampling branch |
There was a problem hiding this comment.
suggestion (testing): Consider asserting that stateful OctaveFilterBank with steady_ic=True also behaves correctly on actual data
Right now this only validates the structure of zi. To also exercise behavior, consider extending the test to process a small known signal with both:
- a stateless
OctaveFilterBank, and - a stateful
OctaveFilterBankwithsteady_ic=True(andcalculate_level=Falseif you only care about band outputs),
then assert that concatenated block-wise outputs match the full-signal outputs, similar totest_block_processing_matches_full_signal.
Suggested implementation:
def test_stateful_steady_ic_initialization():
from pyoctaveband.core import OctaveFilterBank
# Create a stateful filter bank with steady_ic=True
bank = OctaveFilterBank(
fs=48000,
stateful=True,
steady_ic=True,
order=2, # small order is enough for coverage
fraction=1, # 1-octave
resample=False # avoid the resampling branch
)
# Check that zi is a list of numpy arrays with the expected shape
for idx, zi in enumerate(bank.zi):
# zi should have 3 dimensions: (n_sections, 1, 2)
assert isinstance(zi, np.ndarray)
assert zi.ndim == 3
n_sections = bank.sos[idx].shape[0]
assert zi.shape[0] == n_sections
assert zi.shape[1] == 1
assert zi.shape[2] == 2
# Additionally verify that a stateful bank with steady_ic=True behaves
# consistently with a stateless bank when processing actual data.
rng = np.random.default_rng(0)
fs = 48000
duration = 0.25
n_samples = int(fs * duration)
# Random test signal
signal = rng.standard_normal(n_samples)
# Stateless reference (process full signal at once)
stateless_bank = OctaveFilterBank(
fs=fs,
stateful=False,
steady_ic=False,
order=2,
fraction=1,
resample=False,
calculate_level=False, # only care about band outputs
)
full_output = stateless_bank.filter(signal)
# Stateful bank with steady_ic=True (process in blocks)
stateful_bank = OctaveFilterBank(
fs=fs,
stateful=True,
steady_ic=True,
order=2,
fraction=1,
resample=False,
calculate_level=False, # only care about band outputs
)
block_size = 64
block_outputs = []
for start in range(0, n_samples, block_size):
block = signal[start:start + block_size]
y_block = stateful_bank.filter(block)
block_outputs.append(y_block)
# Concatenate block-wise outputs and compare to full-signal stateless output
concat_output = np.concatenate(block_outputs, axis=-1)
assert concat_output.shape == full_output.shape
np.testing.assert_allclose(concat_output, full_output, rtol=1e-6, atol=1e-8)
def test_detrend_stateful_warning():The above implementation assumes:
OctaveFilterBankexposes afilter(signal: np.ndarray)method that returns band outputs with a time axis that can be concatenated alongaxis=-1. If your API instead uses a different method name (e.g.process,__call__, orfilter_block) or a different output shape, update the calls tostateless_bank.filter(...),stateful_bank.filter(...), and theaxisargument innp.concatenateaccordingly.- The constructor of
OctaveFilterBankacceptscalculate_level. If this argument does not exist or is named differently, remove or rename it to match the actual signature used elsewhere in the tests (e.g. copy the usage fromtest_block_processing_matches_full_signal). - If your tests use a helper to perform the "block vs full" comparison (e.g. a shared fixture or function in this file), you can refactor the block-processing part to use that helper for consistency.
Comment on lines
+152
to
+158
| def test_dont_calculate_level(): | ||
| from pyoctaveband.core import OctaveFilterBank | ||
| bank = OctaveFilterBank(48000) | ||
| x = np.zeros((bank.num_bands, 100)) | ||
| spl, y = bank._process_bands(x, num_channels=bank.num_bands, calculate_level=False, sigbands=True) | ||
| assert spl is None | ||
| assert y is not None |
There was a problem hiding this comment.
suggestion (testing): Add tests for the public calculate_level=False API and strengthen assertions on band outputs
This currently only exercises the private _process_bands helper. Since calculate_level is a public filter argument, please also add tests that call:
OctaveFilterBank.filter(x, sigbands=False, calculate_level=False)and assert the SPL return matches the intended contract (e.g.None), andOctaveFilterBank.filter(x, sigbands=True, calculate_level=False)and assert both outputs are as expected.
It would also help to strengthen the assertions on y by checking the number of bands and shapes, e.g. len(y) == bank.num_bands and each element has shape (num_channels, n_samples) after resampling. This will better validate the new option and reduce reliance on private methods in tests.
|
Warning Rate limit exceeded
⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughAdds stateful block-processing to OctaveFilterBank and WeightingFilter (new ctor options Changes
Sequence Diagram(s)sequenceDiagram
participant User as User Code
participant FB as OctaveFilterBank<br/>(stateful=True)
participant SOS as SOS Filter<br/>(sosfilt)
participant State as State Storage (zi)
User->>FB: __init__(stateful=True, steady_ic=False)
FB->>State: initialize per-band zi (zeros or steady IC)
User->>FB: filter(block_1, calculate_level=True)
FB->>SOS: sosfilt(block_1, zi_old)
SOS-->>FB: filtered_block_1, zi_new
FB->>State: store zi_new
FB-->>User: filtered_block_1 (+ level if requested)
User->>FB: filter(block_2)
FB->>SOS: sosfilt(block_2, zi_updated)
SOS-->>FB: filtered_block_2, zi_new
FB->>State: store zi_new
FB-->>User: filtered_block_2
sequenceDiagram
participant User as User Code
participant FB as OctaveFilterBank
participant Validator as Init Validation
User->>FB: __init__(stateful=True, resample=True)
FB->>Validator: check config
Validator-->>FB: incompatible
FB-->>User: ValueError (resample + stateful not allowed)
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@README.md`:
- Line 143: The fenced code block opened with ``` in README.md lacks a language
specifier (triggers MD040); edit the opening fence to include the language token
(e.g., change ``` to ```python) so the code block is syntax-highlighted and the
linter warning is resolved—look for the triple-backtick code block start in the
README and add the appropriate language label.
In `@src/pyoctaveband/core.py`:
- Around line 145-166: The bug is that when calculate_level is False,
_process_bands returns spl = None but filter() unconditionally does spl = spl[0]
for 1D inputs; update filter() to only index spl when spl is not None (e.g.,
replace spl = spl[0] with spl = spl[0] if spl is not None else None) and
likewise guard the other occurrences where spl is indexed (the similar blocks
around the other overload-handling code at the noted spots); ensure branches
that build the return tuple handle spl being None so single-channel calls with
calculate_level=False return the expected tuple shapes without raising.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
README.mdsrc/pyoctaveband/core.pysrc/pyoctaveband/parametric_filters.pytests/test_coverage_fix.pytests/test_stateful_octave_filter_bank.pytests/test_stateful_weighting_filter.py
jmrplens
reviewed
Mar 8, 2026
Owner
jmrplens
left a comment
jmrplens
left a comment
There was a problem hiding this comment.
🔍 Code Review — PR #42: Stateful block-wise processing
Great work on this feature, @ninoblumer! The overall design of stateful filtering is well thought out, and the block-vs-full equivalence tests are exactly what's needed to validate correctness.
I found 2 functional bugs, 1 CI failure, and 4 improvements that should be addressed before merge. Details in the inline comments below 👇
Summary:
| # | Severity | File | Issue |
|---|---|---|---|
| 1 | 🔴 Bug | core.py:193 |
raise Warning(...) raises exception instead of warning |
| 2 | 🔴 Bug | core.py:209 |
spl[0] crashes when calculate_level=False |
| 3 | core.py:154 |
calculate_level=False overloads promise np.ndarray but return None |
|
| 4 | 🔧 CI | test:50 |
Unused bank variable (ruff F841) — breaks CI lint |
| 5 | 🔧 Test | test:87 |
pytest.raises(Warning) coupled to bug #1 |
| 6 | 📝 Docs | README:143 |
Code fence missing python (MD040) |
| 7 | 📝 Docs | README:139 |
Ambiguous detrend wording |
We'll prepare a commit with all the fixes. 🚀
src/pyoctaveband/core.py
Outdated
| # Handle DC offset removal | ||
| if detrend: | ||
| if self.stateful: | ||
| raise Warning("You should not detrend when doing block processing!") |
Owner
There was a problem hiding this comment.
🐛 Bug: raise Warning(...) raises an exception instead of emitting a warning
This is a subtle but critical bug. raise Warning("...") doesn't emit a warning — it raises a Warning exception (which inherits from Exception). In practice, calling filter() with stateful=True and detrend=True (the default!) always crashes with an uncaught exception.
A user simply doing bank.filter(signal) on a stateful bank gets a crash instead of a friendly heads-up.
Fix: Use warnings.warn(...) from the stdlib warnings module:
import warnings
warnings.warn(
"Detrending is not recommended during block processing "
"as it can introduce discontinuities between blocks.",
UserWarning,
stacklevel=2
)The stacklevel=2 makes the warning point to the caller's code, not this internal line.
|
|
||
| # Process signal across all bands and channels | ||
| spl, xb = self._process_bands(x_proc, num_channels, sigbands, mode=mode) | ||
| spl, xb = self._process_bands(x_proc, num_channels, sigbands, mode=mode, calculate_level=calculate_level) |
Owner
There was a problem hiding this comment.
🐛 Bug: spl[0] crash when calculate_level=False on mono signals
When calculate_level=False, _process_bands() returns spl = None. But a few lines below (line 209), spl = spl[0] runs unconditionally for 1D inputs → TypeError: 'NoneType' is not subscriptable.
This affects any user calling with calculate_level=False on a single-channel signal (the most common case).
Fix: Guard with a None check on line 209:
if not is_multichannel:
if spl is not None:
spl = spl[0]
if sigbands and xb is not None:
xb = [band[0] for band in xb]
src/pyoctaveband/core.py
Outdated
| mode: str = "rms", | ||
| detrend: bool = True, | ||
| calculate_level: Literal[False] = False | ||
| ) -> Tuple[np.ndarray, List[float]]: |
Owner
There was a problem hiding this comment.
calculate_level=False overloads promise np.ndarray but return None
The overloads with calculate_level: Literal[False] declare a return type of Tuple[np.ndarray, List[float]], but when calculate_level=False the SPL is actually None, not an np.ndarray.
This is a broken type contract — type checkers like mypy/pyright will assume spl is an array and won't warn users about potential None access.
Fix: Change the return types to Tuple[None, List[float]] and Tuple[None, List[float], List[np.ndarray]] for the sigbands=True variant.
| def test_resample_and_stateful(): | ||
| from pyoctaveband.core import OctaveFilterBank | ||
| with pytest.raises(ValueError): | ||
| bank = OctaveFilterBank(48000, resample=True, stateful=True) |
Owner
There was a problem hiding this comment.
🔧 CI failure cause: unused variable bank (ruff F841)
This is the root cause of the quality job failure in CI. ruff flags bank as assigned but never used — makes sense since the test only verifies the constructor raises ValueError.
Fix: Drop the assignment:
with pytest.raises(ValueError):
OctaveFilterBank(48000, resample=True, stateful=True)| signal = rng.standard_normal(n_samples) | ||
|
|
||
| bank = OctaveFilterBank(fs, stateful=True, resample=False) | ||
| with pytest.raises(Warning): |
Owner
There was a problem hiding this comment.
🔧 Test coupled to the raise Warning bug
This test works right now because raise Warning(...) raises an exception, and pytest.raises(Warning) catches it. But once we fix the raise → warnings.warn bug (see comment on core.py:193), this test will break since there won't be an exception to catch anymore.
Fix: Switch to pytest.warns, which is the correct mechanism for testing warnings:
with pytest.warns(UserWarning, match="block processing"):
bank.filter(signal, detrend=True)
README.md
Outdated
| - Resampling is not supported for block processing, so you need to set `resample=False`. | ||
|
|
||
| Example | ||
| ``` |
Owner
There was a problem hiding this comment.
📝 Missing language specifier on fenced code block (MD040)
The code fence is missing python after the triple backticks. This prevents syntax highlighting on GitHub/IDEs and triggers a markdownlint MD040 warning.
Fix: ``` → ```python
README.md
Outdated
| Create a stateful filter bank with `stateful=True`. The internal state is zero-initialized by default | ||
| but may be initialized for step-response steady-state (like `scipy.signal.sosfilt_zi`) with `steady_ic=True`. | ||
| Notes when using a stateful `OctaveFilterBank`: | ||
| - You should not remove the DC-content when block processing (`detrend=False`). |
Owner
There was a problem hiding this comment.
📝 Ambiguous wording about detrending
"You should not remove the DC-content when block processing" could be read as "don't use detrend" or "don't remove DC by other means". For new users of the library, it's better to be explicit about which parameter to set.
Suggestion:
Detrending should be disabled during block processing (
detrend=False), as it can introduce discontinuities between blocks.
- Replace 'raise Warning(...)' with 'warnings.warn(...)' to emit a proper UserWarning instead of throwing an exception (core.py) - Guard 'spl[0]' with 'if spl is not None' to prevent TypeError when calculate_level=False (core.py) - Fix overload return types and implementation signature to use Optional[np.ndarray] when calculate_level=False (core.py) - Remove unused variable 'bank' to fix ruff F841 lint error (test_stateful_octave_filter_bank.py) - Change pytest.raises(Warning) to pytest.warns(UserWarning) to match the corrected warning behavior (test_stateful_octave_filter_bank.py) - Add 'python' language specifier to fenced code block (README.md) - Clarify detrend guidance wording (README.md)
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/pyoctaveband/core.py (2)
79-81: Unreachable comment afterraise.The comment on line 81 appears after the
raisestatement, making it unreachable code. Consider moving it above theifblock or to the docstring for better discoverability.♻️ Suggested improvement
+ # Note: a stateful resampling algorithm would be required to support this combination if resample and stateful: raise ValueError("Resampling and stateful behaviour (block processing) are not supported.") - # a stateful resampling algorithm would be required...🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/pyoctaveband/core.py` around lines 79 - 81, The inline comment after the raise in the if block checking "if resample and stateful" is unreachable; move the explanatory comment ("a stateful resampling algorithm would be required...") to a reachable location such as directly above the if statement or into the function/class docstring that contains the check so it is discoverable; ensure you update the comment near the symbols resample, stateful, and the ValueError to clearly explain why the combination is unsupported.
128-167: Consider adding explicitcalculate_level: Literal[True]overloads for complete type coverage.The existing overloads at lines 128-144 don't include
calculate_levelin their signatures. While they handle the default case (calculate_level defaults to True), explicit calls likefilter(x, calculate_level=True)may not resolve correctly with some type checkers.♻️ Suggested addition for complete overload coverage
`@overload` def filter( self, x: List[float] | np.ndarray, sigbands: Literal[False] = False, mode: str = "rms", detrend: bool = True, calculate_level: Literal[True] = True ) -> Tuple[np.ndarray, List[float]]: ... `@overload` def filter( self, x: List[float] | np.ndarray, sigbands: Literal[True], mode: str = "rms", detrend: bool = True, calculate_level: Literal[True] = True ) -> Tuple[np.ndarray, List[float], List[np.ndarray]]: ...🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/pyoctaveband/core.py` around lines 128 - 167, The overload set for the filter method is missing explicit variants for calculate_level: Literal[True], which can confuse type checkers; add two overloads mirroring the existing sigbands=False/True signatures but with calculate_level: Literal[True] = True and the correct return types (first element np.ndarray rather than None) so that filter(self, ..., calculate_level=True) is properly typed—update the overloads surrounding the existing filter definitions (the overloads and the filter method) to include these two calculate_level=True variants.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/pyoctaveband/core.py`:
- Around line 110-119: The zi initialization currently hardcodes a
single-channel shape (n_sections, 1, 2) causing a mismatch for multichannel
inputs; modify lazy initialization so self.zi is sized per-band based on actual
channel count when _filter_and_resample() first receives data: keep the existing
placeholder list in __init__ or where stateful is set (self.zi = [None ...]) but
defer creating per-band arrays until _filter_and_resample() examines the
incoming array shape (channels = data.shape[0] if 2D else 1) and then allocate
zi[idx] = np.zeros((self.sos[idx].shape[0], channels, 2)) for steady_ic False or
use signal.sosfilt_zi(self.sos[idx]) and reshape to (n_sections, channels, 2)
for steady_ic True; ensure you update any code paths that assume a 1-channel zi
and add a small branch in _filter_and_resample() to initialize per-band zi when
self.zi[idx] is None or has mismatched second-dimension size.
---
Nitpick comments:
In `@src/pyoctaveband/core.py`:
- Around line 79-81: The inline comment after the raise in the if block checking
"if resample and stateful" is unreachable; move the explanatory comment ("a
stateful resampling algorithm would be required...") to a reachable location
such as directly above the if statement or into the function/class docstring
that contains the check so it is discoverable; ensure you update the comment
near the symbols resample, stateful, and the ValueError to clearly explain why
the combination is unsupported.
- Around line 128-167: The overload set for the filter method is missing
explicit variants for calculate_level: Literal[True], which can confuse type
checkers; add two overloads mirroring the existing sigbands=False/True
signatures but with calculate_level: Literal[True] = True and the correct return
types (first element np.ndarray rather than None) so that filter(self, ...,
calculate_level=True) is properly typed—update the overloads surrounding the
existing filter definitions (the overloads and the filter method) to include
these two calculate_level=True variants.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7e733152-4058-4ffa-8d35-d06e5ea7ab04
📒 Files selected for processing (3)
README.mdsrc/pyoctaveband/core.pytests/test_stateful_octave_filter_bank.py
🚧 Files skipped from review as they are similar to previous changes (2)
- README.md
- tests/test_stateful_octave_filter_bank.py
- Fix mypy line 103: use np.ones() instead of list comprehension for self.factor - Fix mypy line 115: type self.zi as List[np.ndarray] with proper initialization - Fix mypy line 255: add spl is not None guard for type narrowing - Reduce cyclomatic complexity: extract _init_filter_state() from __init__
zi arrays were hardcoded to shape (n_sections, 1, 2), assuming single-channel input. When multichannel input (e.g. stereo) was passed, scipy.sosfilt raised ValueError because zi shape didn't match the channel dimension. Changed to lazy initialization: zi is allocated on first filter() call with the correct shape (n_sections, n_channels, 2) based on actual input channels. This also handles channel count changes between calls. Added test_stateful_multichannel to verify 4-channel stateful processing. Updated test_stateful_steady_ic_initialization for lazy init behavior. Addresses CodeRabbit review: r2901335966
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/pyoctaveband/core.py (1)
133-180:⚠️ Potential issue | 🟡 MinorThe overload set is incomplete for
calculate_level=True.Typed calls using
filter(..., calculate_level=True)will fail type checking despite the implementation accepting it. The first two overloads (withoutcalculate_levellisted) cannot match explicit keyword arguments, and the next two requireLiteral[False]. Add overloads forcalculate_level: Literal[True]returning the filtered arrays.🧩 Minimal typing fix
`@overload` def filter( self, x: List[float] | np.ndarray, sigbands: Literal[False] = False, mode: str = "rms", - detrend: bool = True + detrend: bool = True, + calculate_level: Literal[True] = True, ) -> Tuple[np.ndarray, List[float]]: ... `@overload` def filter( self, x: List[float] | np.ndarray, sigbands: Literal[True], mode: str = "rms", - detrend: bool = True + detrend: bool = True, + calculate_level: Literal[True] = True, ) -> Tuple[np.ndarray, List[float], List[np.ndarray]]: ...🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/pyoctaveband/core.py` around lines 133 - 180, The current overloads for the method filter are missing signatures for calculate_level=True, so type checkers can't match calls like filter(..., calculate_level=True); add two overloads mirroring the existing ones but with calculate_level: Literal[True] (one for sigbands: Literal[False] returning Tuple[np.ndarray, List[float]] and one for sigbands: Literal[True] returning Tuple[np.ndarray, List[float], List[np.ndarray]]), keeping the same mode and detrend defaults and ensuring the runtime implementation signature (def filter(..., calculate_level: bool = True)) matches these overloads.
♻️ Duplicate comments (1)
src/pyoctaveband/core.py (1)
115-123:⚠️ Potential issue | 🔴 CriticalStateful
ziis still hard-coded to a single channel.
filter()still accepts 2D[channels, samples]input, but these initial states are always created as(n_sections, 1, 2). In stateful mode that reaches Line 280 unchanged, so any multichannel block will handsosfilt()aziwhose channel dimension does not matchsd.shape[0]. This breaks stateful multichannel processing at runtime.🔧 Suggested fix
def _init_filter_state(self, steady_ic: bool) -> None: """Initialize filter state (zi) for stateful block-wise processing.""" - self.zi: List[np.ndarray] = [np.array([]) for _ in range(self.num_bands)] - for idx in range(self.num_bands): - if not steady_ic: - self.zi[idx] = np.zeros((self.sos[idx].shape[0], 1, 2)) - else: - zi = signal.sosfilt_zi(self.sos[idx]) - self.zi[idx] = zi[:, np.newaxis, :] # add a dimension since we are filtering along an axis in a 2D-array + self._steady_ic = steady_ic + self.zi: List[np.ndarray | None] = [None for _ in range(self.num_bands)] @@ if self.stateful: + channels = sd.shape[0] if sd.ndim > 1 else 1 + if self.zi[idx] is None or self.zi[idx].shape[1] != channels: + if not self._steady_ic: + self.zi[idx] = np.zeros((self.sos[idx].shape[0], channels, 2)) + else: + zi = signal.sosfilt_zi(self.sos[idx]) + self.zi[idx] = np.repeat(zi[:, np.newaxis, :], channels, axis=1) y, self.zi[idx] = signal.sosfilt(self.sos[idx], sd, axis=-1, zi=self.zi[idx])🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/pyoctaveband/core.py` around lines 115 - 123, _init_filter_state currently creates zi with a hard-coded single-channel shape (n_sections, 1, 2), which breaks multichannel stateful filtering in filter(); update _init_filter_state to allocate the channel dimension to match the actual number of channels used during filtering (use self.num_channels if available, or add a num_channels parameter to _init_filter_state and pass it from filter()), and ensure both the steady_ic branch (use zeros of shape (n_sections, num_channels, 2)) and the steady-state branch (broadcast the result of signal.sosfilt_zi(self.sos[idx]) to shape (n_sections, num_channels, 2), e.g. by inserting a newaxis and repeating along the channel axis) so that self.zi has the correct per-band, per-channel shape before calling signal.sosfilt in filter().
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@src/pyoctaveband/core.py`:
- Around line 133-180: The current overloads for the method filter are missing
signatures for calculate_level=True, so type checkers can't match calls like
filter(..., calculate_level=True); add two overloads mirroring the existing ones
but with calculate_level: Literal[True] (one for sigbands: Literal[False]
returning Tuple[np.ndarray, List[float]] and one for sigbands: Literal[True]
returning Tuple[np.ndarray, List[float], List[np.ndarray]]), keeping the same
mode and detrend defaults and ensuring the runtime implementation signature (def
filter(..., calculate_level: bool = True)) matches these overloads.
---
Duplicate comments:
In `@src/pyoctaveband/core.py`:
- Around line 115-123: _init_filter_state currently creates zi with a hard-coded
single-channel shape (n_sections, 1, 2), which breaks multichannel stateful
filtering in filter(); update _init_filter_state to allocate the channel
dimension to match the actual number of channels used during filtering (use
self.num_channels if available, or add a num_channels parameter to
_init_filter_state and pass it from filter()), and ensure both the steady_ic
branch (use zeros of shape (n_sections, num_channels, 2)) and the steady-state
branch (broadcast the result of signal.sosfilt_zi(self.sos[idx]) to shape
(n_sections, num_channels, 2), e.g. by inserting a newaxis and repeating along
the channel axis) so that self.zi has the correct per-band, per-channel shape
before calling signal.sosfilt in filter().
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f59add5a-6079-4700-937d-37e106a0a58e
📒 Files selected for processing (1)
src/pyoctaveband/core.py
This was referenced Mar 8, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR introduces stateful block-wise processing to
OctaveFilterBankandWeightingFilter, allowing filters to maintain their internal states between successive calls. This enables efficient streaming or block-based processing without recalculating the entire signal each time.Changes:
OctaveFilterBank.__init__stateful=True→ carry filter states across callssteady_ic=True→ initialize states using sosfilt_zi for steady-state conditions, not really necessary, could also be removed._filter_and_resample()to propagateziwhenstateful=TrueExample usage:
Tests included:
Notes:
resample=Trueandstateful=Trueare currently not supported together. Raising an error if attempted. Reason is, that the resampler itself is not stateful.Future work:
time_weighting()Summary by Sourcery
Introduce stateful, block-wise processing support for octave filter banks and weighting filters, including optional steady-state initialization and the ability to skip SPL calculation when only band signals are needed.
New Features:
Enhancements:
Documentation:
Tests:
Summary by CodeRabbit