Drop the unreachable guards, and split the STI chain

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)

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.