From e528a190f83d6623b1cd87646e202ce63788d74b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20M=2E=20Requena=20Plens?= Date: Thu, 6 Aug 2026 05:36:36 +0200 Subject: [PATCH] Drop the unreachable guards, and split the STI chain Three of the four issues SonarCloud still reports on main, plus the dead code the previous branch uncovered but left alone. `_validated_task_levels` re-checked what `Task.__post_init__` already rejects. `Task` is a frozen dataclass that refuses an empty sample set and a non-positive duration at construction, so no invalid task can reach the function and neither guard could ever fire. They are gone and the function is `_task_levels` now, since it no longer validates anything. This is what the `pytest.raises` work found: the test that appeared to cover those guards was in fact exercising the constructor, one frame earlier. `_with_advisory` had a single caller and existed only to name one call to `dataclasses.replace`. Inlining it removes the declared return type that a static analyser cannot reconcile with `replace`'s own signature, and nothing about the behaviour changes. `_sti_from_mtf` was at cognitive complexity 23 against a limit of 15. Three helpers come out of it, each one a step the standard names: `_truncated_mtf` (the A.5.3 NOTE 1 validity check and the truncation to 1,0), `_snr_vector` (the per-band signal-to-noise ratios) and `_corrected_mtf` (the A.5.3 noise, masking and reception-threshold corrections). Validation order is unchanged, so which error fires first is unchanged, and the truncation warning gained a stack level to keep pointing at the caller rather than at library code. The fourth issue is deliberately left open. `OctaveFilterBank.filter` returns two or three values depending on `sigbands`, which is what the rule objects to, but that arity is the documented public contract: four `@overload` stubs type it, 18 call sites unpack two values, 16 unpack three, and 20 documentation pages show both. Making the length uniform would break every one of them to satisfy a rule that is describing the API correctly. --- .../hearing/occupational_exposure.py | 30 ++-- src/phonometry/speech/sti.py | 143 +++++++++++------- 2 files changed, 100 insertions(+), 73 deletions(-) diff --git a/src/phonometry/hearing/occupational_exposure.py b/src/phonometry/hearing/occupational_exposure.py index 890b113c6..51da0e06e 100644 --- a/src/phonometry/hearing/occupational_exposure.py +++ b/src/phonometry/hearing/occupational_exposure.py @@ -378,17 +378,17 @@ class _BudgetOptions: warn: bool -def _validated_task_levels(tasks: Sequence[Task]) -> tuple[list[float], list[float]]: - """Per-task energy-average levels (Eq 7) and mean durations ``T_m``, validated.""" - levels: list[float] = [] - durations: list[float] = [] - for task in tasks: - if len(task.samples) == 0: - raise ValueError("Each task needs at least one sample.") - if task.duration_hours <= 0: - raise ValueError("Task 'duration_hours' must be positive.") - levels.append(energy_mean(task.samples)) # Eq 7 - durations.append(task.duration_hours) +def _task_levels(tasks: Sequence[Task]) -> tuple[list[float], list[float]]: + """Per-task energy-average levels (Eq 7) and mean durations ``T_m``. + + No validation here: :class:`Task` is a frozen dataclass that rejects an + empty sample set and a non-positive duration in ``__post_init__``, so an + invalid task cannot reach this function. The guards that used to repeat + those two checks were unreachable, and the test that appeared to cover them + was in fact exercising the constructor. + """ + levels = [energy_mean(task.samples) for task in tasks] # Eq 7 + durations = [task.duration_hours for task in tasks] return levels, durations @@ -494,7 +494,7 @@ def task_based_exposure( raise ValueError("'u3' must be non-negative.") # First pass: task levels and durations (Eq 7, 8). - levels, durations = _validated_task_levels(tasks) + levels, durations = _task_levels(tasks) # Daily level: energy sum of contributions (Eq 9 == Eq 10). energy = sum((t_m / _T0) * 10.0 ** (0.1 * lp) for lp, t_m in zip(levels, durations)) @@ -624,7 +624,7 @@ def job_based_exposure( OccupationalExposureWarning, stacklevel=2, ) - result = _with_advisory(result) + result = replace(result, sampling_advisory=True) return result @@ -665,7 +665,3 @@ def full_day_exposure( samples, effective_duration_hours, instrument, u3, "full_day", warn, spread_advisory=spread ) - -def _with_advisory(result: ExposureResult) -> ExposureResult: - """Return a copy of ``result`` with the sampling advisory flag set.""" - return replace(result, sampling_advisory=True) diff --git a/src/phonometry/speech/sti.py b/src/phonometry/speech/sti.py index 32873425c..8d778f17a 100644 --- a/src/phonometry/speech/sti.py +++ b/src/phonometry/speech/sti.py @@ -229,28 +229,11 @@ def _validate_band_vector(values: Sequence[float] | np.ndarray, name: str) -> np return arr -def _sti_from_mtf( - mtf: np.ndarray, - snr: float | Sequence[float] | np.ndarray | None = None, - level: Sequence[float] | np.ndarray | None = None, - ambient: Sequence[float] | np.ndarray | None = None, -) -> STIResult: - r"""STI computation chain from a matrix of modulation transfer values. - - Applies, in order: m > 1 truncation (Ed.4 A.5.3 NOTE 1), the optional - signal-to-noise and level-dependent corrections (Ed.4 A.5.3 = Ed.5), - the effective SNR with +/-15 dB limits (A.5.4), the transmission - indices :math:`TI = (SNR_{\mathrm{eff}} + 15)/30` (A.5.5), the band MTIs - and the final - weighted STI truncated to 1.0 (A.5.6, male factors of Ed.5 Table A.1). +def _truncated_mtf(mtf: np.ndarray) -> np.ndarray: + """Validate the modulation transfer matrix and truncate it to 1,0. - Noise handling: ``snr`` alone multiplies ``m`` by - :math:`1/(1 + 10^{-SNR/10})`, - which is exactly the :math:`I_k/(I_k + I_{n,k})` factor of the standard. - When ``level`` is provided the full intensity-domain correction - :math:`m' = m \cdot I_k / (I_k + I_{am,k} + I_{rt,k} + I_{n,k})` is used - instead, with ``ambient`` (or ``level - snr``) defining :math:`I_{n,k}`, - so the noise degradation is never applied twice. + Ed.4 A.5.3 NOTE 1 (= Ed.5): a value above 1,3 means the measurement is + invalid, and every value is truncated to 1,0 before the chain runs. """ m = np.array(mtf, dtype=np.float64) if m.ndim != 2 or m.shape[0] != _NUM_BANDS: @@ -261,30 +244,45 @@ def _sti_from_mtf( if np.any(m < 0.0) or not np.all(np.isfinite(m)): raise ValueError("Modulation transfer values must be finite and >= 0.") if np.any(m > 1.3): - # Ed.4 A.5.3 NOTE 1 (= Ed.5): m > 1,3 indicates an invalid measurement. warnings.warn( "Modulation transfer values above 1.3 detected: the measurement " "is likely invalid (IEC 60268-16 A.5.3). Values truncated to 1.0.", STIWarning, - stacklevel=3, + stacklevel=4, + ) + return np.minimum(m, 1.0) + + +def _snr_vector( + snr: float | Sequence[float] | np.ndarray | None, +) -> np.ndarray | None: + """The per-band signal-to-noise ratios, broadcast from a scalar if given.""" + if snr is None: + return None + snr_arr = np.asarray(snr, dtype=np.float64) + if snr_arr.ndim == 0: + return np.full(_NUM_BANDS, float(snr_arr)) + if snr_arr.shape != (_NUM_BANDS,): + raise ValueError( + f"'snr' must be a scalar or a vector of {_NUM_BANDS} " + f"octave-band values, got shape {snr_arr.shape}." ) - m = np.minimum(m, 1.0) + return snr_arr - if snr is not None and ambient is not None: - raise ValueError("Provide either 'snr' or 'ambient' noise levels, not both.") - snr_arr: np.ndarray | None = None - if snr is not None: - snr_arr = np.asarray(snr, dtype=np.float64) - if snr_arr.ndim == 0: - snr_arr = np.full(_NUM_BANDS, float(snr_arr)) - elif snr_arr.shape != (_NUM_BANDS,): - raise ValueError( - f"'snr' must be a scalar or a vector of {_NUM_BANDS} " - f"octave-band values, got shape {snr_arr.shape}." - ) +def _corrected_mtf( + m: np.ndarray, + snr_arr: np.ndarray | None, + level: Sequence[float] | np.ndarray | None, + ambient: Sequence[float] | np.ndarray | None, +) -> tuple[np.ndarray, np.ndarray | None]: + """Apply the noise and level-dependent corrections of A.5.3. - band_levels: np.ndarray | None = None + Without absolute levels only the signal-to-noise correction applies; with + them the auditory masking and absolute reception threshold join it. Returns + the corrected matrix and the validated band levels, which the result + carries. + """ if level is None: if ambient is not None: raise ValueError( @@ -296,26 +294,59 @@ def _sti_from_mtf( m = m / (1.0 + 10.0 ** (-snr_arr[:, np.newaxis] / 10.0)) # No absolute level information: the auditory masking and absolute # reception threshold corrections are skipped. + return m, None + + band_levels = _validate_band_vector(level, "level") + i_signal = 10.0 ** (band_levels / 10.0) + if ambient is not None: + ambient_arr = _validate_band_vector(ambient, "ambient") + elif snr_arr is not None: + ambient_arr = band_levels - snr_arr else: - band_levels = _validate_band_vector(level, "level") - i_signal = 10.0 ** (band_levels / 10.0) - if ambient is not None: - ambient_arr = _validate_band_vector(ambient, "ambient") - elif snr_arr is not None: - ambient_arr = band_levels - snr_arr - else: - ambient_arr = None - i_noise = ( - 10.0 ** (ambient_arr / 10.0) if ambient_arr is not None else np.zeros(_NUM_BANDS) - ) - i_total = i_signal + i_noise - level_total = 10.0 * np.log10(i_total) - # Masking only acts on the next higher band; 125 Hz is unmasked. - i_masking = np.zeros(_NUM_BANDS) - i_masking[1:] = i_total[:-1] * 10.0 ** (_masking_amdb(level_total[:-1]) / 10.0) - i_threshold = 10.0 ** (_ART_DB / 10.0) - factor = i_signal / (i_signal + i_masking + i_threshold + i_noise) - m = m * factor[:, np.newaxis] + ambient_arr = None + i_noise = ( + 10.0 ** (ambient_arr / 10.0) if ambient_arr is not None else np.zeros(_NUM_BANDS) + ) + i_total = i_signal + i_noise + level_total = 10.0 * np.log10(i_total) + # Masking only acts on the next higher band; 125 Hz is unmasked. + i_masking = np.zeros(_NUM_BANDS) + i_masking[1:] = i_total[:-1] * 10.0 ** (_masking_amdb(level_total[:-1]) / 10.0) + i_threshold = 10.0 ** (_ART_DB / 10.0) + factor = i_signal / (i_signal + i_masking + i_threshold + i_noise) + return m * factor[:, np.newaxis], band_levels + + +def _sti_from_mtf( + mtf: np.ndarray, + snr: float | Sequence[float] | np.ndarray | None = None, + level: Sequence[float] | np.ndarray | None = None, + ambient: Sequence[float] | np.ndarray | None = None, +) -> STIResult: + r"""STI computation chain from a matrix of modulation transfer values. + + Applies, in order: m > 1 truncation (Ed.4 A.5.3 NOTE 1), the optional + signal-to-noise and level-dependent corrections (Ed.4 A.5.3 = Ed.5), + the effective SNR with +/-15 dB limits (A.5.4), the transmission + indices :math:`TI = (SNR_{\mathrm{eff}} + 15)/30` (A.5.5), the band MTIs + and the final + weighted STI truncated to 1.0 (A.5.6, male factors of Ed.5 Table A.1). + + Noise handling: ``snr`` alone multiplies ``m`` by + :math:`1/(1 + 10^{-SNR/10})`, + which is exactly the :math:`I_k/(I_k + I_{n,k})` factor of the standard. + When ``level`` is provided the full intensity-domain correction + :math:`m' = m \cdot I_k / (I_k + I_{am,k} + I_{rt,k} + I_{n,k})` is used + instead, with ``ambient`` (or ``level - snr``) defining :math:`I_{n,k}`, + so the noise degradation is never applied twice. + """ + m = _truncated_mtf(mtf) + + if snr is not None and ambient is not None: + raise ValueError("Provide either 'snr' or 'ambient' noise levels, not both.") + + snr_arr = _snr_vector(snr) + m, band_levels = _corrected_mtf(m, snr_arr, level, ambient) # Effective SNR clipped to +/-15 dB (A.5.4); m = 0 and m = 1 map to the # clip limits through the log divergences.