kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 20:07:04 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

_transcribe_diarize: cover 0% -> 100% before splitting

Browse Source 
The largest untested Python file left in the repo, at 356 lines against the
cap. Tests land first so the split that follows is verified against tests
that predate the seam.

63 tests: duration probing through both ffmpeg paths, the transcode fallback
and temp-file cleanup, audio loading's three outcomes, linear resampling,
cosine k-means, the ECAPA classifier loader, per-segment embedding
extraction, and diarize_segments end to end.

numpy is real here, not a double. It is a declared dependency, and the core
of this file is clustering maths - assertions against a mocked numpy would
prove nothing about whether two speakers actually get two labels. The
_try_import seam passes numpy through while stubbing torch and speechbrain,
so the tests still run in CI with neither installed. The clustering tests
assert real properties: two opposed groups separate, magnitude is ignored
because the metric is cosine, and a fixed seed reproduces.

The last three coverage gaps were _load_audio's ffmpeg fallback in both
outcomes, the minimum-slice widening, and the zero-iteration k-means branch.

Split across six test modules after two intermediate arrangements went over
the cap, and one move silently broke a test by separating it from its
doubles - caught by the test count holding at 63 through every rearrangement.
338 tests pass in the directory, up from 275 by exactly the 63 added.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011w2Pj4Ek3ZNMQVYcFHgXDY
This commit is contained in:
Krzysztof Rudnicki 2026-08-17 13:54:04 +02:00
parent 47ee01a3b5
commit fb51e991fe
8 changed files with 1151 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
docs/superpowers/contracts/transcribe-diarize-tests-2026-08-17.json Normal file
View File

@ -0,0 +1,18 @@
{
"title": "Cover _transcribe_diarize.py before splitting it",
"objective": "_transcribe_diarize.py is 356 lines against a 250-line cap and was completely untested - the largest untested Python file left in the repo. This commit raises it to 100% line and branch coverage first, so the following split can be verified against tests that predate the seam. numpy is real here rather than mocked, because the file's core is clustering maths and a mocked numpy would make those assertions vacuous.",
"acceptance_criteria": [
"_transcribe_diarize.py reaches 100% line and branch coverage, from 0%",
"_transcribe_diarize.py is byte-identical to HEAD - this commit adds tests only",
"The tests run without torch, speechbrain, soundfile or ffmpeg present",
"The clustering tests assert real separation properties, not that a mock was called",
"No suppression of any kind",
"Every new test file is under the 250-line cap"
],
"out_of_scope": [
"The split of _transcribe_diarize.py itself, which is the next commit",
"Any behaviour change",
"Moving these files under python_pkg/"
],
"verifier": "pytest linux_configuration/tests/test_transcribe_diarize_*.py --cov=_transcribe_diarize --cov-branch --cov-report=term-missing for the coverage claim and the test count; pre-commit run --files for lint; bash ~/utils/scripts/check_file_length.sh --all to confirm no new over-cap file."
}

54
docs/superpowers/evidence/transcribe-diarize-tests-2026-08-17.json Normal file
View File

@ -0,0 +1,54 @@
{
"intent": "Cover _transcribe_diarize.py from 0% to 100% line and branch, so the following split of the 356-line file is verified against tests written before the seam existed.",
"scope": [
"New: six test modules under linux_configuration/tests/ named test_transcribe_diarize_*",
"Non-goal: any change to _transcribe_diarize.py, which is byte-identical to HEAD",
"Non-goal: the split itself, which is the next commit"
],
"changes": [
"Added 63 tests covering duration probing through both ffmpeg paths, the ffmpeg transcode fallback and temp-file cleanup, audio loading with its three outcomes, linear resampling, cosine k-means, the ECAPA classifier loader, per-segment embedding extraction, and diarize_segments end to end.",
"numpy is a declared dependency, so the resampling and clustering tests run against the real library rather than a double. The _try_import seam is patched to pass numpy through while stubbing torch and speechbrain: clustering that 'worked' against a mocked numpy would prove nothing about whether two speakers actually get two labels.",
"The clustering tests assert real properties - two opposed groups produce two labels, magnitude is ignored because the metric is cosine, and a fixed seed is reproducible.",
"Split across six test modules, each under the cap, after two intermediate arrangements went over it."
],
"verification": [
{
"command": "git diff --stat -- .../tools/_transcribe_diarize.py",
"result": "pass",
"evidence": "No change: this commit adds tests only."
},
{
"command": "pytest linux_configuration/tests/test_transcribe_diarize_*.py --cov=_transcribe_diarize --cov-branch --cov-report=term-missing",
"result": "pass",
"evidence": "63 passed. _transcribe_diarize.py: 180 statements, 0 missing, 52 branches, 0 partial, 100%. The final three gaps were _load_audio's ffmpeg fallback in both outcomes, the minimum-slice widening at line 312, and the zero-iteration k-means branch."
},
{
"command": "pytest linux_configuration/tests/test_transcribe_diarize_*.py (count across each test-file rearrangement)",
"result": "pass",
"evidence": "63 passed before and after every move. Two intermediate arrangements were rejected for going over the cap, and one move silently broke a test by separating it from its doubles - caught by the count, which is the discriminating check for a test-file split."
},
{
"command": "pytest linux_configuration/tests/ -q",
"result": "pass",
"evidence": "338 passed, up from 275 by exactly the 63 tests added; no existing test was altered."
},
{
"command": "pre-commit run --files <changed files>",
"result": "pass",
"evidence": "All hooks pass. Fixed rather than suppressed: five ANN401 warnings by typing the doubles with npt.NDArray[np.float32] instead of Any, a mypy exit-return error on a context-manager double, two func-returns-value errors from append-inside-lambda, three S108 hardcoded /tmp paths replaced with tmp_path, and a codespell hit on 'unparseable'."
},
{
"command": "bash ~/utils/scripts/check_file_length.sh --all",
"result": "pass",
"evidence": "33 files over cap, unchanged by this commit. The six test modules are 237/205/184/184/177/103 lines."
}
],
"risks": [
"torch and speechbrain are doubled, so the embedding extraction is proved to slice, pad and batch correctly, but not that the real ECAPA model produces separable embeddings. That is inherent to testing without the models present in CI.",
"The clustering tests use fixed seeds. A change to numpy's default_rng stream would change the labels; the tests assert cluster structure (which points share a label) rather than specific label values, which is stable under that."
],
"rollback": [
"git revert the commit; the six test modules disappear together.",
"After rollback, confirm pytest linux_configuration/tests/ reports 275 passed and _transcribe_diarize.py is unchanged."
]
}

204
linux_configuration/tests/test_transcribe_diarize_audio.py Normal file
View File

@ -0,0 +1,204 @@
"""Tests for the ffmpeg transcode fallback, temp files, and audio loading.
``_load_audio`` reads with soundfile and, when that fails, transcodes through
ffmpeg to a temporary 16 kHz mono WAV. Both binaries are patched, so these
pass whether or not ffmpeg is installed on the machine running them.
"""
from __future__ import annotations
import logging
import shutil
import subprocess
from typing import TYPE_CHECKING
import _transcribe_diarize as diar
import pytest
if TYPE_CHECKING:
from pathlib import Path
def _importer(available: dict[str, object]) -> object:
"""Return a _try_import double resolving only the named modules."""
def _try(name: str) -> object | None:
return available.get(name)
return _try
def _only(name: str, module: object) -> object:
"""Return a _try_import double resolving exactly one module name."""
def _try(requested: str) -> object | None:
return module if requested == name else None
return _try
# --------------------------------------------------------------------------- #
# _ffmpeg_transcode_to_wav16_mono / _cleanup_temp
# --------------------------------------------------------------------------- #
def test_transcode_returns_none_without_ffmpeg(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""No ffmpeg binary means no fallback transcode."""
monkeypatch.setattr(shutil, "which", lambda _b: None)
assert diar._ffmpeg_transcode_to_wav16_mono("clip.mp4") is None
def test_transcode_returns_the_temp_path_on_success(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A successful transcode hands back the temporary WAV path."""
monkeypatch.setattr(shutil, "which", lambda _b: "/usr/bin/ffmpeg")
commands: list[list[str]] = []
def _record(cmd: list[str], **_kwargs: object) -> None:
commands.append(cmd)
monkeypatch.setattr(subprocess, "run", _record)
result = diar._ffmpeg_transcode_to_wav16_mono("clip.mp4")
assert result is not None
assert result.endswith(".wav")
assert "16000" in commands[0]
assert "-ac" in commands[0]
diar._cleanup_temp(result)
@pytest.mark.parametrize(
"error", [OSError("boom"), subprocess.CalledProcessError(1, "ffmpeg")]
)
def test_transcode_cleans_up_after_a_failure(
monkeypatch: pytest.MonkeyPatch, error: Exception
) -> None:
"""A failed transcode removes the temp file it had already created."""
monkeypatch.setattr(shutil, "which", lambda _b: "/usr/bin/ffmpeg")
removed: list[str] = []
monkeypatch.setattr(diar.Path, "unlink", lambda self: removed.append(str(self)))
def _boom(*_a: object, **_k: object) -> None:
raise error
monkeypatch.setattr(subprocess, "run", _boom)
assert diar._ffmpeg_transcode_to_wav16_mono("clip.mp4") is None
assert len(removed) == 1
def test_cleanup_temp_removes_an_existing_file(tmp_path: Path) -> None:
"""The temp WAV is deleted once diarization is done with it."""
victim = tmp_path / "scratch.wav"
victim.write_bytes(b"RIFF")
diar._cleanup_temp(str(victim))
assert not victim.exists()
def test_cleanup_temp_ignores_none() -> None:
"""No temp file was created, so there is nothing to remove."""
diar._cleanup_temp(None)
def test_cleanup_temp_survives_an_already_deleted_file(tmp_path: Path) -> None:
"""A file removed by something else is not an error."""
diar._cleanup_temp(str(tmp_path / "never-existed.wav"))
def test_load_audio_returns_none_without_soundfile(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Diarization needs soundfile; without it there is no audio to read."""
monkeypatch.setattr(diar, "_try_import", lambda _n: None)
assert diar._load_audio("clip.wav") is None
def test_load_audio_warns_when_the_ffmpeg_fallback_is_unavailable(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""An unreadable file with no transcode path warns and gives up."""
class _Sf:
def read(self, *_a: object, **_k: object) -> None:
msg = "unsupported format"
raise OSError(msg)
monkeypatch.setattr(diar, "_try_import", _only("soundfile", _Sf()))
monkeypatch.setattr(diar, "_ffmpeg_transcode_to_wav16_mono", lambda _p: None)
with caplog.at_level(logging.WARNING):
assert diar._load_audio("clip.mp4") is None
assert "no ffmpeg fallback" in caplog.text
class _SfFailingThenOk:
"""A soundfile double that fails on the original and reads the transcode."""
def __init__(self, *, second_fails: bool = False) -> None:
self.calls: list[str] = []
self._second_fails = second_fails
def read(self, path: str, **_kwargs: object) -> tuple[str, int]:
"""Fail for the source file; succeed (or not) for the transcode."""
self.calls.append(path)
if len(self.calls) == 1:
msg = "unsupported format"
raise OSError(msg)
if self._second_fails:
msg = "still unreadable"
raise OSError(msg)
return "wav-data", 16000
def test_load_audio_falls_back_to_a_transcoded_copy(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""An unreadable container is transcoded, then read from the temp WAV."""
scratch = str(tmp_path / "scratch.wav")
sf = _SfFailingThenOk()
monkeypatch.setattr(diar, "_try_import", _only("soundfile", sf))
monkeypatch.setattr(diar, "_ffmpeg_transcode_to_wav16_mono", lambda _p: scratch)
result = diar._load_audio("clip.mp4")
assert result == ("wav-data", 16000, scratch)
assert sf.calls == ["clip.mp4", scratch]
def test_load_audio_cleans_up_when_the_transcode_is_also_unreadable(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""If even the transcode cannot be read, the temp file is removed."""
temp = tmp_path / "scratch.wav"
temp.write_bytes(b"RIFF")
monkeypatch.setattr(
diar, "_try_import", _only("soundfile", _SfFailingThenOk(second_fails=True))
)
monkeypatch.setattr(diar, "_ffmpeg_transcode_to_wav16_mono", lambda _p: str(temp))
with caplog.at_level(logging.WARNING):
assert diar._load_audio("clip.mp4") is None
assert "Could not read transcoded audio" in caplog.text
assert not temp.exists()
def test_load_audio_returns_the_source_when_it_reads_cleanly(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A readable file needs no transcode and reports no temp path."""
class _Sf:
def read(self, _path: str, **_kwargs: object) -> tuple[str, int]:
return "wav-data", 44100
monkeypatch.setattr(diar, "_try_import", _only("soundfile", _Sf()))
assert diar._load_audio("clip.wav") == ("wav-data", 44100, None)

88
linux_configuration/tests/test_transcribe_diarize_classifier.py Normal file
View File

@ -0,0 +1,88 @@
"""Tests for loading the ECAPA speaker-embedding classifier.
speechbrain is never installed in CI and is reached only through
``_try_import``, so the encoder is doubled here. A model that fails to load
must clean up any temporary audio the caller had already transcoded.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
import _transcribe_diarize as diar
import pytest
if TYPE_CHECKING:
from pathlib import Path
def _importer(available: dict[str, object]) -> object:
"""Return a _try_import double resolving only the named modules."""
def _try(name: str) -> object | None:
return available.get(name)
return _try
# --------------------------------------------------------------------------- #
# _load_speaker_classifier
# --------------------------------------------------------------------------- #
def test_load_speaker_classifier_returns_none_without_speechbrain(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Without speechbrain there is no embedding model to load."""
monkeypatch.setattr(diar, "_try_import", lambda _n: None)
assert diar._load_speaker_classifier(None) is None
def test_load_speaker_classifier_loads_the_ecapa_model(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The ECAPA model is loaded on CPU into a cache directory."""
recorded: dict[str, object] = {}
class _Encoder:
@staticmethod
def from_hparams(**kwargs: object) -> str:
recorded.update(kwargs)
return "classifier"
module = type("M", (), {"EncoderClassifier": _Encoder})()
monkeypatch.setattr(
diar, "_try_import", _importer({"speechbrain.inference": module})
)
assert diar._load_speaker_classifier(None) == "classifier"
assert recorded["source"] == "speechbrain/spkrec-ecapa-voxceleb"
assert recorded["run_opts"] == {"device": "cpu"}
@pytest.mark.parametrize("error", [OSError("no net"), RuntimeError("bad weights")])
def test_load_speaker_classifier_cleans_up_after_a_failure(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
tmp_path: Path,
error: Exception,
) -> None:
"""A model that will not load warns and removes the temp audio."""
temp = tmp_path / "scratch.wav"
temp.write_bytes(b"RIFF")
class _Encoder:
@staticmethod
def from_hparams(**_kwargs: object) -> None:
raise error
module = type("M", (), {"EncoderClassifier": _Encoder})()
monkeypatch.setattr(
diar, "_try_import", _importer({"speechbrain.inference": module})
)
with caplog.at_level(logging.WARNING):
assert diar._load_speaker_classifier(str(temp)) is None
assert "Could not load speaker embedding model" in caplog.text
assert not temp.exists()

184
linux_configuration/tests/test_transcribe_diarize_cluster.py Normal file
View File

@ -0,0 +1,184 @@
"""Tests for the resampling and cosine k-means used to label speakers.
numpy is a declared dependency (meta/requirements.txt), so these run against
the real library rather than a double: clustering that "works" against a
mocked numpy would prove nothing about whether two speakers get two labels.
"""
from __future__ import annotations
import _transcribe_diarize as diar
import numpy as np
import pytest
_SR_16K = 16000
_TWO_SPEAKERS = 2
# --------------------------------------------------------------------------- #
# _resample_linear
# --------------------------------------------------------------------------- #
def test_resample_linear_returns_the_input_unchanged_at_the_same_rate() -> None:
"""Resampling to the rate it already has is a no-op."""
audio = np.linspace(0.0, 1.0, num=100, dtype=np.float32)
result = diar._resample_linear(audio, _SR_16K, _SR_16K)
assert result is audio
def test_resample_linear_downsamples_to_the_expected_length() -> None:
"""Halving the rate halves the sample count."""
audio = np.linspace(0.0, 1.0, num=1000, dtype=np.float32)
result = diar._resample_linear(audio, 32000, _SR_16K)
assert result.shape[0] == 500
assert result.dtype == np.float32
def test_resample_linear_upsamples_to_the_expected_length() -> None:
"""Doubling the rate doubles the sample count."""
audio = np.linspace(0.0, 1.0, num=100, dtype=np.float32)
assert diar._resample_linear(audio, 8000, _SR_16K).shape[0] == 200
def test_resample_linear_preserves_a_constant_signal() -> None:
"""Linear interpolation of a flat signal stays flat."""
audio = np.full(100, 0.5, dtype=np.float32)
result = diar._resample_linear(audio, 8000, _SR_16K)
assert np.allclose(result, 0.5)
def test_resample_linear_never_returns_an_empty_array() -> None:
"""An extreme downsample still yields at least one sample."""
audio = np.linspace(0.0, 1.0, num=10, dtype=np.float32)
assert diar._resample_linear(audio, 48000, 1).shape[0] >= 1
def test_resample_linear_requires_numpy(monkeypatch: pytest.MonkeyPatch) -> None:
"""Without numpy the failure is explicit rather than an AttributeError."""
monkeypatch.setattr(diar, "_try_import", lambda _n: None)
with pytest.raises(RuntimeError, match="numpy is required"):
diar._resample_linear(np.zeros(10, dtype=np.float32), 8000, _SR_16K)
# --------------------------------------------------------------------------- #
# _kmeans_cosine
# --------------------------------------------------------------------------- #
def test_kmeans_cosine_separates_two_opposed_clusters() -> None:
"""Embeddings pointing in two directions get two distinct labels."""
rng = np.random.default_rng(0)
group_a = np.tile([1.0, 0.0], (10, 1)) + rng.normal(0, 0.01, (10, 2))
group_b = np.tile([0.0, 1.0], (10, 1)) + rng.normal(0, 0.01, (10, 2))
embeddings = list(np.vstack([group_a, group_b]).astype(np.float32))
labels = diar._kmeans_cosine(embeddings, k=_TWO_SPEAKERS)
assert len(set(labels.tolist())) == _TWO_SPEAKERS
assert len(set(labels[:10].tolist())) == 1
assert len(set(labels[10:].tolist())) == 1
assert labels[0] != labels[10]
def test_kmeans_cosine_ignores_magnitude() -> None:
"""Cosine clustering keys on direction, so scale must not matter."""
embeddings = [
np.array([1.0, 0.0], dtype=np.float32),
np.array([100.0, 0.0], dtype=np.float32),
np.array([0.0, 1.0], dtype=np.float32),
np.array([0.0, 50.0], dtype=np.float32),
]
labels = diar._kmeans_cosine(embeddings, k=_TWO_SPEAKERS).tolist()
assert labels[0] == labels[1]
assert labels[2] == labels[3]
assert labels[0] != labels[2]
def test_kmeans_cosine_is_deterministic_for_a_fixed_seed() -> None:
"""The same input twice gives the same labels, so runs are reproducible."""
rng = np.random.default_rng(1)
embeddings = list(rng.normal(size=(20, 8)).astype(np.float32))
first = diar._kmeans_cosine(embeddings, k=3, seed=7).tolist()
second = diar._kmeans_cosine(embeddings, k=3, seed=7).tolist()
assert first == second
def test_kmeans_cosine_returns_empty_for_no_embeddings() -> None:
"""A file with no segments produces no labels."""
assert diar._kmeans_cosine([], k=_TWO_SPEAKERS).shape == (0,)
def test_kmeans_cosine_returns_empty_for_the_wrong_dimensionality() -> None:
"""A 1-D input is not a stack of embeddings and is rejected."""
assert diar._kmeans_cosine([1.0, 2.0, 3.0], k=_TWO_SPEAKERS).shape == (0,)
def test_kmeans_cosine_pads_when_asked_for_more_speakers_than_segments() -> None:
"""Requesting 5 speakers from 2 segments still returns one label each."""
embeddings = [
np.array([1.0, 0.0], dtype=np.float32),
np.array([0.0, 1.0], dtype=np.float32),
]
labels = diar._kmeans_cosine(embeddings, k=5)
assert labels.shape == (2,)
def test_kmeans_cosine_labels_every_segment_with_one_speaker() -> None:
"""k=1 assigns every segment the same label."""
rng = np.random.default_rng(2)
embeddings = list(rng.normal(size=(6, 4)).astype(np.float32))
assert set(diar._kmeans_cosine(embeddings, k=1).tolist()) == {0}
def test_kmeans_cosine_requires_numpy(monkeypatch: pytest.MonkeyPatch) -> None:
"""Without numpy the failure is explicit."""
monkeypatch.setattr(diar, "_try_import", lambda _n: None)
with pytest.raises(RuntimeError, match="numpy is required"):
diar._kmeans_cosine([np.zeros(4, dtype=np.float32)], k=1)
# --------------------------------------------------------------------------- #
# _run_kmeans_iterations
# --------------------------------------------------------------------------- #
def test_run_kmeans_iterations_converges_and_stops_early() -> None:
"""Already-separated data converges on the first pass and breaks out."""
features = np.array(
[[1.0, 0.0], [1.0, 0.0], [0.0, 1.0], [0.0, 1.0]], dtype=np.float32
)
centroids = np.array([[1.0, 0.0], [0.0, 1.0]], dtype=np.float32)
labels = diar._run_kmeans_iterations(np, features, centroids, 2, 50)
assert labels.tolist() == [0, 0, 1, 1]
def test_run_kmeans_iterations_keeps_an_empty_cluster_centroid() -> None:
"""A centroid that attracts no points is carried forward, not zeroed."""
features = np.array([[1.0, 0.0], [1.0, 0.0]], dtype=np.float32)
centroids = np.array([[1.0, 0.0], [0.0, 1.0]], dtype=np.float32)
labels = diar._run_kmeans_iterations(np, features, centroids, 2, 5)
assert labels.tolist() == [0, 0]
def test_run_kmeans_iterations_returns_none_when_asked_for_no_iterations() -> None:
"""A zero-iteration run never assigns labels; the loop body is skipped."""
features = np.array([[1.0, 0.0]], dtype=np.float32)
centroids = np.array([[1.0, 0.0]], dtype=np.float32)
assert diar._run_kmeans_iterations(np, features, centroids, 1, 0) is None

235
linux_configuration/tests/test_transcribe_diarize_pipeline.py Normal file
View File

@ -0,0 +1,235 @@
"""Tests for the diarize_segments pipeline end to end.
torch and speechbrain are never installed in CI and are reached only through
``_try_import``, so both are doubled here. The embedding double returns a
direction per segment, which lets the clustering downstream be real numpy.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
import _transcribe_diarize as diar
import numpy as np
if TYPE_CHECKING:
from contextlib import AbstractContextManager
from pathlib import Path
import numpy.typing as npt
import pytest
_SR_16K = 16000
_TWO_SPEAKERS = 2
class _Segment:
"""A transcription segment, of which only start and end are read."""
def __init__(self, start: float | None, end: float | None) -> None:
self.start = start
self.end = end
class _Tensor:
"""A torch tensor double carrying the slice it was built from."""
def __init__(self, data: npt.NDArray[np.float32]) -> None:
self.data = data
def unsqueeze(self, _dim: int) -> _Tensor:
"""torch tensors are reshaped before batching; the data is unchanged."""
return self
class _Embedding:
"""The chain classifier.encode_batch(...).squeeze().squeeze().cpu().numpy()."""
def __init__(self, vector: npt.NDArray[np.float32]) -> None:
self._vector = vector
def squeeze(self, _dim: int) -> _Embedding:
return self
def cpu(self) -> _Embedding:
return self
def numpy(self) -> npt.NDArray[np.float32]:
"""Return the embedding as the array the caller will cast."""
return self._vector
class _Torch:
"""A torch module double providing tensor() and no_grad()."""
def tensor(self, data: npt.NDArray[np.float32]) -> _Tensor:
"""Wrap a slice of audio the way torch.tensor would."""
return _Tensor(data)
def no_grad(self) -> AbstractContextManager[None]:
"""Return a context manager, as torch.no_grad() does."""
class _Ctx:
def __enter__(self) -> None:
return None
def __exit__(self, *_a: object) -> None:
return None
return _Ctx()
class _Classifier:
"""A speaker classifier double: one direction per call, recorded."""
def __init__(self, vectors: list[npt.NDArray[np.float32]] | None = None) -> None:
self.seen: list[int] = []
self._vectors = vectors
self._calls = 0
def encode_batch(self, tensor: _Tensor) -> _Embedding:
"""Record the sample count and return the next canned embedding."""
self.seen.append(len(tensor.data))
if self._vectors is not None:
vector = self._vectors[self._calls % len(self._vectors)]
else:
vector = np.array([1.0, 0.0], dtype=np.float32)
self._calls += 1
return _Embedding(vector)
def _importer(available: dict[str, object]) -> object:
"""Return a _try_import double resolving the named modules, plus numpy.
numpy is a real declared dependency and the resampling and clustering
code fetches it through the same seam, so it is always passed through -
stubbing it would make the clustering assertions meaningless.
"""
def _try(name: str) -> object | None:
if name == "numpy":
return np
return available.get(name)
return _try
# --------------------------------------------------------------------------- #
# diarize_segments
# --------------------------------------------------------------------------- #
def test_diarize_segments_labels_two_speakers(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The whole pipeline turns four segments into two speaker labels."""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: (wav, _SR_16K, None))
monkeypatch.setattr(
diar,
"_load_speaker_classifier",
lambda _t: _Classifier(
[
np.array([1.0, 0.0], dtype=np.float32),
np.array([0.0, 1.0], dtype=np.float32),
]
),
)
segments = [_Segment(float(i), float(i) + 1.0) for i in range(4)]
labels = diar.diarize_segments("clip.wav", segments, num_speakers=_TWO_SPEAKERS)
assert labels is not None
assert len(labels) == len(segments)
assert len(set(labels)) == _TWO_SPEAKERS
def test_diarize_segments_warns_without_torch(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""No torch means speaker labels are skipped, not an error."""
monkeypatch.setattr(diar, "_try_import", lambda _n: None)
with caplog.at_level(logging.WARNING):
assert diar.diarize_segments("clip.wav", [_Segment(0.0, 1.0)]) is None
assert "skipping speaker labels" in caplog.text
def test_diarize_segments_gives_up_when_the_audio_cannot_be_read(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Unreadable audio ends diarization before any model is loaded."""
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: None)
assert diar.diarize_segments("clip.wav", [_Segment(0.0, 1.0)]) is None
def test_diarize_segments_gives_up_without_a_classifier(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A missing embedding model ends diarization."""
wav = np.zeros(_SR_16K, dtype=np.float32)
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: (wav, _SR_16K, None))
monkeypatch.setattr(diar, "_load_speaker_classifier", lambda _t: None)
assert diar.diarize_segments("clip.wav", [_Segment(0.0, 1.0)]) is None
def test_diarize_segments_returns_none_for_no_segments(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""No segments means no embeddings and so no labels."""
wav = np.zeros(_SR_16K, dtype=np.float32)
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: (wav, _SR_16K, None))
monkeypatch.setattr(diar, "_load_speaker_classifier", lambda _t: _Classifier())
assert diar.diarize_segments("clip.wav", []) is None
def test_diarize_segments_downmixes_stereo(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A 2-D stereo array is averaged to mono before resampling."""
stereo = np.zeros((_SR_16K * 2, 2), dtype=np.float32)
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: (stereo, _SR_16K, None))
monkeypatch.setattr(diar, "_load_speaker_classifier", lambda _t: _Classifier())
labels = diar.diarize_segments("clip.wav", [_Segment(0.0, 1.0)], num_speakers=1)
assert labels == [0]
def test_diarize_segments_removes_the_temp_file(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A transcoded temp WAV is deleted once labels are computed."""
temp = tmp_path / "scratch.wav"
temp.write_bytes(b"RIFF")
wav = np.zeros(_SR_16K * 2, dtype=np.float32)
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: (wav, _SR_16K, str(temp)))
monkeypatch.setattr(diar, "_load_speaker_classifier", lambda _t: _Classifier())
diar.diarize_segments("clip.wav", [_Segment(0.0, 1.0)], num_speakers=1)
assert not temp.exists()
def test_diarize_segments_treats_zero_speakers_as_one(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""num_speakers=0 would break k-means, so it is floored at one."""
wav = np.zeros(_SR_16K * 2, dtype=np.float32)
monkeypatch.setattr(diar, "_try_import", _importer({"torch": _Torch()}))
monkeypatch.setattr(diar, "_load_audio", lambda _p: (wav, _SR_16K, None))
monkeypatch.setattr(diar, "_load_speaker_classifier", lambda _t: _Classifier())
assert diar.diarize_segments("clip.wav", [_Segment(0.0, 1.0)], num_speakers=0) == [
0
]

172
linux_configuration/tests/test_transcribe_diarize_probe.py Normal file
View File

@ -0,0 +1,172 @@
"""Tests for media-duration probing and temp-file handling.
Duration is probed twice: through the ffmpeg-python binding if it is
importable, then by shelling out to ffprobe. Both are patched here, so the
tests pass whether or not ffmpeg is installed on the machine running them.
"""
from __future__ import annotations
import shutil
import subprocess
import _transcribe_diarize as diar
import pytest
class _FfmpegModule:
"""An ffmpeg-python double whose probe() returns a fixed payload."""
def __init__(self, payload: object) -> None:
self._payload = payload
def probe(self, _path: str) -> object:
"""Return the canned probe result, or raise it if it is an error."""
if isinstance(self._payload, Exception):
raise self._payload
return self._payload
def _only(name: str, module: object) -> object:
"""Return a _try_import double resolving exactly one module name."""
def _try(requested: str) -> object | None:
return module if requested == name else None
return _try
# --------------------------------------------------------------------------- #
# The optional-dependency import seam
# --------------------------------------------------------------------------- #
def test_try_import_returns_the_module_when_present() -> None:
"""The seam the other tests patch works against a real module."""
assert diar._try_import("json") is not None
def test_try_import_returns_none_for_a_missing_module() -> None:
"""A missing optional dependency is None, not an ImportError."""
assert diar._try_import("no_such_module_anywhere_12345") is None
# --------------------------------------------------------------------------- #
# _probe_with_ffmpeg_python
# --------------------------------------------------------------------------- #
def test_probe_with_ffmpeg_python_reads_the_format_duration(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The binding reports duration under probe()['format']['duration']."""
module = _FfmpegModule({"format": {"duration": "123.5"}})
monkeypatch.setattr(diar, "_try_import", _only("ffmpeg", module))
assert diar._probe_with_ffmpeg_python("clip.wav") == pytest.approx(123.5)
def test_probe_with_ffmpeg_python_returns_none_without_the_binding(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Without ffmpeg-python installed this probe declines."""
monkeypatch.setattr(diar, "_try_import", lambda _n: None)
assert diar._probe_with_ffmpeg_python("clip.wav") is None
def test_probe_with_ffmpeg_python_returns_none_without_a_duration_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A probe result with no duration falls through to the next method."""
module = _FfmpegModule({"format": {}})
monkeypatch.setattr(diar, "_try_import", _only("ffmpeg", module))
assert diar._probe_with_ffmpeg_python("clip.wav") is None
@pytest.mark.parametrize("error", [OSError("boom"), RuntimeError("boom")])
def test_probe_with_ffmpeg_python_survives_a_probe_error(
monkeypatch: pytest.MonkeyPatch, error: Exception
) -> None:
"""A failing probe returns None rather than propagating."""
monkeypatch.setattr(diar, "_try_import", _only("ffmpeg", _FfmpegModule(error)))
assert diar._probe_with_ffmpeg_python("clip.wav") is None
# --------------------------------------------------------------------------- #
# _probe_with_ffprobe
# --------------------------------------------------------------------------- #
def test_probe_with_ffprobe_parses_the_duration(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""ffprobe prints a bare float, which is parsed to seconds."""
monkeypatch.setattr(shutil, "which", lambda _b: "/usr/bin/ffprobe")
monkeypatch.setattr(subprocess, "check_output", lambda *_a, **_k: b" 42.25 \n")
assert diar._probe_with_ffprobe("clip.wav") == pytest.approx(42.25)
def test_probe_with_ffprobe_returns_none_when_not_installed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""No ffprobe on PATH means no duration."""
monkeypatch.setattr(shutil, "which", lambda _b: None)
assert diar._probe_with_ffprobe("clip.wav") is None
@pytest.mark.parametrize(
"error",
[
OSError("boom"),
subprocess.CalledProcessError(1, "ffprobe"),
ValueError("not a float"),
],
)
def test_probe_with_ffprobe_survives_every_failure_mode(
monkeypatch: pytest.MonkeyPatch, error: Exception
) -> None:
"""A crash, a non-zero exit, or unparsable output all yield None."""
monkeypatch.setattr(shutil, "which", lambda _b: "/usr/bin/ffprobe")
def _boom(*_a: object, **_k: object) -> None:
raise error
monkeypatch.setattr(subprocess, "check_output", _boom)
assert diar._probe_with_ffprobe("clip.wav") is None
# --------------------------------------------------------------------------- #
# get_media_duration
# --------------------------------------------------------------------------- #
def test_get_media_duration_prefers_the_python_binding(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""When the binding answers, ffprobe is never invoked."""
called: list[str] = []
def _record_ffprobe(_path: str) -> None:
called.append("ffprobe")
monkeypatch.setattr(diar, "_probe_with_ffmpeg_python", lambda _p: 10.0)
monkeypatch.setattr(diar, "_probe_with_ffprobe", _record_ffprobe)
assert diar.get_media_duration("clip.wav") == pytest.approx(10.0)
assert called == []
def test_get_media_duration_falls_back_to_ffprobe(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Without the binding, the CLI is tried next."""
monkeypatch.setattr(diar, "_probe_with_ffmpeg_python", lambda _p: None)
monkeypatch.setattr(diar, "_probe_with_ffprobe", lambda _p: 20.0)
assert diar.get_media_duration("clip.wav") == pytest.approx(20.0)
def test_get_media_duration_is_none_when_both_fail(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""An unprobeable file is not an error - the caller drops the percentage."""
monkeypatch.setattr(diar, "_probe_with_ffmpeg_python", lambda _p: None)
monkeypatch.setattr(diar, "_probe_with_ffprobe", lambda _p: None)

196
linux_configuration/tests/test_transcribe_diarize_segments.py Normal file
View File

@ -0,0 +1,196 @@
"""Tests for speaker-embedding extraction and the diarize_segments pipeline.
torch and speechbrain are never installed in CI and are reached only through
``_try_import``, so both are doubled here. The embedding double returns a
direction per segment, which lets the clustering downstream be real numpy.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import _transcribe_diarize as diar
import numpy as np
if TYPE_CHECKING:
from contextlib import AbstractContextManager
import numpy.typing as npt
_SR_16K = 16000
_TWO_SPEAKERS = 2
class _Segment:
"""A transcription segment, of which only start and end are read."""
def __init__(self, start: float | None, end: float | None) -> None:
self.start = start
self.end = end
class _Tensor:
"""A torch tensor double carrying the slice it was built from."""
def __init__(self, data: npt.NDArray[np.float32]) -> None:
self.data = data
def unsqueeze(self, _dim: int) -> _Tensor:
"""torch tensors are reshaped before batching; the data is unchanged."""
return self
class _Embedding:
"""The chain classifier.encode_batch(...).squeeze().squeeze().cpu().numpy()."""
def __init__(self, vector: npt.NDArray[np.float32]) -> None:
self._vector = vector
def squeeze(self, _dim: int) -> _Embedding:
return self
def cpu(self) -> _Embedding:
return self
def numpy(self) -> npt.NDArray[np.float32]:
"""Return the embedding as the array the caller will cast."""
return self._vector
class _Torch:
"""A torch module double providing tensor() and no_grad()."""
def tensor(self, data: npt.NDArray[np.float32]) -> _Tensor:
"""Wrap a slice of audio the way torch.tensor would."""
return _Tensor(data)
def no_grad(self) -> AbstractContextManager[None]:
"""Return a context manager, as torch.no_grad() does."""
class _Ctx:
def __enter__(self) -> None:
return None
def __exit__(self, *_a: object) -> None:
return None
return _Ctx()
class _Classifier:
"""A speaker classifier double: one direction per call, recorded."""
def __init__(self, vectors: list[npt.NDArray[np.float32]] | None = None) -> None:
self.seen: list[int] = []
self._vectors = vectors
self._calls = 0
def encode_batch(self, tensor: _Tensor) -> _Embedding:
"""Record the sample count and return the next canned embedding."""
self.seen.append(len(tensor.data))
if self._vectors is not None:
vector = self._vectors[self._calls % len(self._vectors)]
else:
vector = np.array([1.0, 0.0], dtype=np.float32)
self._calls += 1
return _Embedding(vector)
def _importer(available: dict[str, object]) -> object:
"""Return a _try_import double resolving the named modules, plus numpy.
numpy is a real declared dependency and the resampling and clustering
code fetches it through the same seam, so it is always passed through -
stubbing it would make the clustering assertions meaningless.
"""
def _try(name: str) -> object | None:
if name == "numpy":
return np
return available.get(name)
return _try
# --------------------------------------------------------------------------- #
# _extract_embeddings
# --------------------------------------------------------------------------- #
def test_extract_embeddings_returns_one_vector_per_segment() -> None:
"""Every segment yields exactly one embedding, in order."""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
segments = [_Segment(0.0, 1.0), _Segment(1.0, 2.0), _Segment(2.0, 3.0)]
classifier = _Classifier()
embeddings = diar._extract_embeddings(segments, wav, classifier, _Torch())
assert len(embeddings) == 3
assert all(e.dtype == np.float32 for e in embeddings)
def test_extract_embeddings_pads_around_each_segment() -> None:
"""A 50 ms pad each side widens the slice beyond the segment itself."""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
classifier = _Classifier()
diar._extract_embeddings([_Segment(1.0, 2.0)], wav, classifier, _Torch())
# 1 s of audio plus 0.05 s of padding at each end.
assert classifier.seen == [int(_SR_16K * 1.1)]
def test_extract_embeddings_enforces_a_minimum_slice() -> None:
"""A very short segment is widened so the encoder has enough samples."""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
classifier = _Classifier()
diar._extract_embeddings([_Segment(1.0, 1.001)], wav, classifier, _Torch())
assert classifier.seen[0] >= diar._MIN_SAMPLES_DIAR
def test_extract_embeddings_handles_a_segment_with_no_duration() -> None:
"""An end at or before the start is given a default 0.2 s span."""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
classifier = _Classifier()
diar._extract_embeddings([_Segment(5.0, 5.0)], wav, classifier, _Torch())
assert classifier.seen[0] >= diar._MIN_SAMPLES_DIAR
def test_extract_embeddings_treats_missing_times_as_zero() -> None:
"""A segment with None timings starts at the beginning rather than failing."""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
classifier = _Classifier()
embeddings = diar._extract_embeddings(
[_Segment(None, None)], wav, classifier, _Torch()
)
assert len(embeddings) == 1
def test_extract_embeddings_clamps_to_the_end_of_the_audio() -> None:
"""A segment running past the recording is clipped to what exists."""
wav = np.zeros(_SR_16K, dtype=np.float32)
classifier = _Classifier()
diar._extract_embeddings([_Segment(0.5, 100.0)], wav, classifier, _Torch())
assert classifier.seen[0] <= len(wav)
def test_extract_embeddings_widens_a_slice_below_the_minimum() -> None:
"""A segment shorter than the padding still reaches the minimum width.
The padded slice of a 1 ms segment is 1600 samples exactly at the start of
the file, so this uses a segment at offset 0 where only one side is padded
and the widening branch is the one that has to fire.
"""
wav = np.zeros(_SR_16K * 10, dtype=np.float32)
classifier = _Classifier()
diar._extract_embeddings([_Segment(0.0, 0.001)], wav, classifier, _Torch())
assert classifier.seen[0] == diar._MIN_SAMPLES_DIAR