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: split 356 -> 164 + 143 + 105

Browse Source 
The last Python file over the cap:

  _transcribe_media.py    143  _try_import, both duration probes,
                               get_media_duration, ffmpeg transcode, cleanup
  _transcribe_cluster.py  105  _resample_linear, _kmeans_cosine and its
                               iteration loop
  _transcribe_diarize.py  164  the pipeline: audio loading, the classifier,
                               embedding extraction, diarize_segments

The graph is a tree: _transcribe_media imports nothing local, _cluster takes
only _try_import from it, _diarize imports from both. All 13 top-level
definitions verified AST-identical to HEAD, and an AST scan confirms the
three module constants travelled to the modules that use them.

transcribe_fw.py now imports get_media_duration from _transcribe_media
rather than through _transcribe_diarize. Ruff removed that re-export as
unused and broke the import; running the program is what caught it, and
naming the owning module is the correct fix under the no-barrels rule.

Verified: transcribe_fw.py /nonexistent.wav still exits 2 and --help exits
0; 63 tests pass at 100% line and branch on all three modules; 338 pass in
the directory. Over-cap files 33 -> 32, and no Python file is over the cap
any more - the remaining 32 are 29 shell, 2 kotlin, 1 dart.

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:59:39 +02:00
parent fb51e991fe
commit dbf2f29b00
9 changed files with 350 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -0,0 +1,18 @@
{
"title": "Split _transcribe_diarize.py 356 -> 164 + 143 + 105",
"objective": "_transcribe_diarize.py was the last Python file over the 250-line cap. Now that it sits at 100% line and branch coverage, split it three ways along its natural seams: media probing, clustering maths, and the diarization pipeline itself. Code moves verbatim; the tests predate the seam and are re-pointed, not rewritten.",
"acceptance_criteria": [
"All three resulting modules are under 250 lines; the over-cap total falls 33 -> 32 and no Python file remains over the cap",
"All 13 top-level definitions are AST-identical to the pre-split file",
"The same 63 tests pass, with all three modules at 100% line and branch coverage",
"python3 transcribe_fw.py /nonexistent.wav still exits 2, the code transcribe.sh branches on",
"Import sites name the owning module - no re-export barrels",
"No suppression of any kind"
],
"out_of_scope": [
"Any behaviour change",
"The 29 over-cap shell scripts, which are the remaining work",
"Moving anything under python_pkg/"
],
"verifier": "A throwaway ast.dump identity check against HEAD; pytest with --cov of all three modules --cov-branch; running transcribe_fw.py and reading $? directly; pre-commit run --files; bash ~/utils/scripts/check_file_length.sh --all."
}

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

@ -0,0 +1,61 @@
{
"intent": "Split the 356-line _transcribe_diarize.py into media probing, clustering maths, and the diarization pipeline, all under the 250-line cap, moving every definition verbatim.",
"scope": [
"_transcribe_diarize.py 356 -> 164 (audio loading, classifier, embeddings, diarize_segments)",
"New: _transcribe_media.py 143 (duration probing, ffmpeg transcode, temp cleanup)",
"New: _transcribe_cluster.py 105 (linear resampling, cosine k-means)",
"transcribe_fw.py: one import line re-pointed",
"Non-goal: any behaviour change"
],
"changes": [
"Extracted _try_import, both duration probes, get_media_duration, the ffmpeg transcode and _cleanup_temp into _transcribe_media.py; _resample_linear, _kmeans_cosine and _run_kmeans_iterations into _transcribe_cluster.py. What remains in _transcribe_diarize.py is the pipeline: loading audio, loading the classifier, extracting embeddings, and diarize_segments.",
"The dependency graph is a tree: _transcribe_media imports nothing local, _transcribe_cluster imports only _try_import from it, and _transcribe_diarize imports from both.",
"transcribe_fw.py now imports get_media_duration from _transcribe_media rather than through _transcribe_diarize. This was forced by ruff removing the now-unused re-export, and is the correct outcome under the no-barrels rule - import sites should name the owning module.",
"Test modules were re-pointed to the module that owns each function. The audio tests span two modules: _load_audio stayed put and still patches its own namespace, while the direct transcode and cleanup calls now name _transcribe_media."
],
"verification": [
{
"command": "python3 verify_split.py HEAD .../_transcribe_diarize.py <three modules>",
"result": "pass",
"evidence": "IDENTICAL: all 13 top-level definitions moved verbatim. Re-run after ruff's autofixes and the import re-point, and still identical."
},
{
"command": "AST scan for names used but not bound in each of the three modules",
"result": "pass",
"evidence": "All three report no unbound names, confirming the three module-level constants (_NDIM_2D, _SAMPLE_RATE_16K, _MIN_SAMPLES_DIAR) travelled to the modules that use them."
},
{
"command": "pytest linux_configuration/tests/test_transcribe_diarize_*.py --cov of all three modules --cov-branch",
"result": "pass",
"evidence": "63 passed - the same count as before the split. _transcribe_cluster.py 51 statements / 18 branches, _transcribe_diarize.py 81 / 22, _transcribe_media.py 59 / 12, all 100%."
},
{
"command": "python3 transcribe_fw.py /nonexistent.wav; echo $?",
"result": "pass",
"evidence": "Exit 2, the code transcribe.sh branches on, and --help exits 0. This is what caught a real breakage: ruff had removed the get_media_duration re-export as unused, and transcribe_fw.py imported it from there. Only running it (and the full suite) surfaced that."
},
{
"command": "pytest linux_configuration/tests/ -q",
"result": "pass",
"evidence": "338 passed, unchanged from before the split."
},
{
"command": "pre-commit run --files <changed files>",
"result": "pass",
"evidence": "All hooks pass. The shebang-executable hook reads the git index rather than the filesystem, so the new modules' mode change had to be staged with git add --chmod=+x before it would pass."
},
{
"command": "bash ~/utils/scripts/check_file_length.sh --all",
"result": "pass",
"evidence": "32 files over cap, down from 33. The three modules are 164, 143 and 105 lines. No Python file in the repo is now over the cap."
}
],
"risks": [
"A three-way split adds two import edges between sibling modules resolved by bare name on sys.path[0]. That is the existing convention in this directory and the real run confirms it resolves.",
"The get_media_duration re-export is gone, so any out-of-tree caller importing it from _transcribe_diarize would break. Only transcribe_fw.py imported it, and it was updated in the same commit."
],
"rollback": [
"git revert the commit; _transcribe_diarize.py returns to its 356-line form and the two new modules disappear.",
"After rollback, confirm python3 transcribe_fw.py /nonexistent.wav still returns 2 and pytest linux_configuration/tests/ reports 338 passed."
]
}

104
linux_configuration/scripts/single_use/misc/testsAndMisc-bash/tools/_transcribe_cluster.py Executable file
View File

@ -0,0 +1,104 @@
#!/usr/bin/env python3
"""Resampling and cosine k-means, the maths behind speaker labelling.
numpy is fetched through ``_try_import`` rather than imported at module
scope, so this file remains importable on a machine without it - which is
what lets the transcribe tools be unit-tested in CI.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from _transcribe_media import _try_import
if TYPE_CHECKING:
import types
import numpy as np
import numpy.typing as npt
_NDIM_2D = 2
def _resample_linear(
x: npt.NDArray[np.float32],
src_sr: int,
tgt_sr: int,
) -> npt.NDArray[np.float32]:
"""Linearly resample 1-D audio array."""
np_mod = _try_import("numpy")
if np_mod is None:
msg = "numpy is required for resampling"
raise RuntimeError(msg)
if src_sr == tgt_sr:
return x
ratio = float(tgt_sr) / float(src_sr)
n_out = max(1, round(x.shape[-1] * ratio))
xp = np_mod.linspace(0.0, 1.0, num=x.shape[-1], endpoint=False)
xq = np_mod.linspace(0.0, 1.0, num=n_out, endpoint=False)
y = np_mod.interp(xq, xp, x.astype(np_mod.float32))
return y.astype(np_mod.float32)
def _kmeans_cosine(
embs: list[Any],
k: int,
iters: int = 50,
seed: int = 0,
) -> npt.NDArray[np.int64]:
"""Cluster embeddings with cosine-similarity k-means."""
np_mod = _try_import("numpy")
if np_mod is None:
msg = "numpy is required for clustering"
raise RuntimeError(msg)
rng = np_mod.random.default_rng(seed)
features = np_mod.asarray(embs, dtype=np_mod.float32)
if features.ndim != _NDIM_2D or features.shape[0] == 0:
return np_mod.zeros((0,), dtype=np_mod.int64)
features = features / (np_mod.linalg.norm(features, axis=1, keepdims=True) + 1e-8)
idxs = rng.choice(
features.shape[0],
size=min(k, features.shape[0]),
replace=False,
)
centroids = features[idxs]
if centroids.shape[0] < k:
pad = rng.standard_normal(
size=(
k - centroids.shape[0],
features.shape[1],
)
).astype(np_mod.float32)
pad /= np_mod.linalg.norm(pad, axis=1, keepdims=True) + 1e-8
centroids = np_mod.concatenate([centroids, pad], axis=0)
return _run_kmeans_iterations(np_mod, features, centroids, k, iters)
def _run_kmeans_iterations(
np_mod: types.ModuleType,
features: npt.NDArray[np.float32],
centroids: npt.NDArray[np.float32],
k: int,
iters: int,
) -> npt.NDArray[np.intp]:
"""Run k-means iteration loop and return labels."""
labels: Any = None
for _ in range(iters):
sims = features @ centroids.T
labels = sims.argmax(axis=1)
new_c = np_mod.zeros_like(centroids)
for j in range(k):
sel = features[labels == j]
if sel.shape[0] == 0:
new_c[j] = centroids[j]
else:
v = sel.mean(axis=0)
v /= np_mod.linalg.norm(v) + 1e-8
new_c[j] = v
if np_mod.allclose(new_c, centroids, atol=1e-4):
break
centroids = new_c
return labels

220
linux_configuration/scripts/single_use/misc/testsAndMisc-bash/tools/_transcribe_diarize.py Normal file → Executable file
View File

@ -1,16 +1,24 @@
"""Speaker diarization and audio processing utilities."""
#!/usr/bin/env python3
"""Speaker diarization: embed each segment, then cluster the embeddings.
Every heavy dependency (torch, speechbrain, soundfile) is reached through
``_try_import``, so a machine without them degrades to no speaker labels
rather than failing the transcription.
"""
from __future__ import annotations
import contextlib
import importlib
import logging
from pathlib import Path
import shutil
import subprocess
import tempfile
from typing import TYPE_CHECKING, Any
from _transcribe_cluster import _kmeans_cosine, _resample_linear
from _transcribe_media import (
_cleanup_temp,
_ffmpeg_transcode_to_wav16_mono,
_try_import,
)
if TYPE_CHECKING:
import types
@ -24,206 +32,6 @@ _SAMPLE_RATE_16K = 16000
_MIN_SAMPLES_DIAR = 1600
def _try_import(name: str) -> types.ModuleType | None:
"""Attempt to import a module, returning None on failure."""
try:
return importlib.import_module(name)
except ImportError:
return None
def _probe_with_ffmpeg_python(
path: str,
) -> float | None:
"""Try ffmpeg-python to get duration."""
ffmpeg_mod = _try_import("ffmpeg")
if ffmpeg_mod is None:
return None
try:
probe = ffmpeg_mod.probe(path)
fmt = probe.get("format", {})
if "duration" in fmt:
return float(fmt["duration"])
except (OSError, RuntimeError):
pass
return None
def _probe_with_ffprobe(path: str) -> float | None:
"""Try ffprobe CLI to get duration."""
ffprobe_bin = shutil.which("ffprobe")
if ffprobe_bin is None:
return None
try:
out = subprocess.check_output(
[
ffprobe_bin,
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
path,
],
stderr=subprocess.DEVNULL,
)
return float(out.decode().strip())
except (
OSError,
subprocess.CalledProcessError,
ValueError,
):
return None
def get_media_duration(path: str) -> float | None:
"""Try to get media duration in seconds.
Returns None if unavailable.
"""
result = _probe_with_ffmpeg_python(path)
if result is not None:
return result
return _probe_with_ffprobe(path)
def _resample_linear(
x: npt.NDArray[np.float32],
src_sr: int,
tgt_sr: int,
) -> npt.NDArray[np.float32]:
"""Linearly resample 1-D audio array."""
np_mod = _try_import("numpy")
if np_mod is None:
msg = "numpy is required for resampling"
raise RuntimeError(msg)
if src_sr == tgt_sr:
return x
ratio = float(tgt_sr) / float(src_sr)
n_out = max(1, round(x.shape[-1] * ratio))
xp = np_mod.linspace(0.0, 1.0, num=x.shape[-1], endpoint=False)
xq = np_mod.linspace(0.0, 1.0, num=n_out, endpoint=False)
y = np_mod.interp(xq, xp, x.astype(np_mod.float32))
return y.astype(np_mod.float32)
def _kmeans_cosine(
embs: list[Any],
k: int,
iters: int = 50,
seed: int = 0,
) -> npt.NDArray[np.int64]:
"""Cluster embeddings with cosine-similarity k-means."""
np_mod = _try_import("numpy")
if np_mod is None:
msg = "numpy is required for clustering"
raise RuntimeError(msg)
rng = np_mod.random.default_rng(seed)
features = np_mod.asarray(embs, dtype=np_mod.float32)
if features.ndim != _NDIM_2D or features.shape[0] == 0:
return np_mod.zeros((0,), dtype=np_mod.int64)
features = features / (np_mod.linalg.norm(features, axis=1, keepdims=True) + 1e-8)
idxs = rng.choice(
features.shape[0],
size=min(k, features.shape[0]),
replace=False,
)
centroids = features[idxs]
if centroids.shape[0] < k:
pad = rng.standard_normal(
size=(
k - centroids.shape[0],
features.shape[1],
)
).astype(np_mod.float32)
pad /= np_mod.linalg.norm(pad, axis=1, keepdims=True) + 1e-8
centroids = np_mod.concatenate([centroids, pad], axis=0)
return _run_kmeans_iterations(np_mod, features, centroids, k, iters)
def _run_kmeans_iterations(
np_mod: types.ModuleType,
features: npt.NDArray[np.float32],
centroids: npt.NDArray[np.float32],
k: int,
iters: int,
) -> npt.NDArray[np.intp]:
"""Run k-means iteration loop and return labels."""
labels: Any = None
for _ in range(iters):
sims = features @ centroids.T
labels = sims.argmax(axis=1)
new_c = np_mod.zeros_like(centroids)
for j in range(k):
sel = features[labels == j]
if sel.shape[0] == 0:
new_c[j] = centroids[j]
else:
v = sel.mean(axis=0)
v /= np_mod.linalg.norm(v) + 1e-8
new_c[j] = v
if np_mod.allclose(new_c, centroids, atol=1e-4):
break
centroids = new_c
return labels
def _ffmpeg_transcode_to_wav16_mono(
src_path: str,
) -> str | None:
"""Transcode input to a temporary 16k mono WAV.
Returns its path, or None if ffmpeg is unavailable.
"""
ffmpeg_bin = shutil.which("ffmpeg")
if ffmpeg_bin is None:
return None
with tempfile.NamedTemporaryFile(
prefix="fw_diar_",
suffix=".wav",
delete=False,
) as tmp:
tmp_path = tmp.name
cmd = [
ffmpeg_bin,
"-y",
"-v",
"error",
"-i",
src_path,
"-ac",
"1",
"-ar",
"16000",
"-f",
"wav",
tmp_path,
]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except (OSError, subprocess.CalledProcessError):
with contextlib.suppress(OSError):
Path(tmp_path).unlink()
return None
return tmp_path
def _cleanup_temp(path: str | None) -> None:
"""Remove a temporary file if it exists."""
if path is not None:
with contextlib.suppress(OSError):
Path(path).unlink()
def _load_audio(
audio_path: str,
) -> tuple[Any, int, str | None] | None:

140
linux_configuration/scripts/single_use/misc/testsAndMisc-bash/tools/_transcribe_media.py Executable file
View File

@ -0,0 +1,140 @@
#!/usr/bin/env python3
"""Media duration probing and the ffmpeg transcode fallback.
Duration is probed through the ffmpeg-python binding first, then by shelling
out to ffprobe; both are optional, and an unprobeable file simply reports no
duration rather than failing the run.
"""
from __future__ import annotations
import contextlib
import importlib
import logging
from pathlib import Path
import shutil
import subprocess
import tempfile
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import types
logger = logging.getLogger(__name__)
def _try_import(name: str) -> types.ModuleType | None:
"""Attempt to import a module, returning None on failure."""
try:
return importlib.import_module(name)
except ImportError:
return None
def _probe_with_ffmpeg_python(
path: str,
) -> float | None:
"""Try ffmpeg-python to get duration."""
ffmpeg_mod = _try_import("ffmpeg")
if ffmpeg_mod is None:
return None
try:
probe = ffmpeg_mod.probe(path)
fmt = probe.get("format", {})
if "duration" in fmt:
return float(fmt["duration"])
except (OSError, RuntimeError):
pass
return None
def _probe_with_ffprobe(path: str) -> float | None:
"""Try ffprobe CLI to get duration."""
ffprobe_bin = shutil.which("ffprobe")
if ffprobe_bin is None:
return None
try:
out = subprocess.check_output(
[
ffprobe_bin,
"-v",
"error",
"-show_entries",
"format=duration",
"-of",
"default=noprint_wrappers=1:nokey=1",
path,
],
stderr=subprocess.DEVNULL,
)
return float(out.decode().strip())
except (
OSError,
subprocess.CalledProcessError,
ValueError,
):
return None
def get_media_duration(path: str) -> float | None:
"""Try to get media duration in seconds.
Returns None if unavailable.
"""
result = _probe_with_ffmpeg_python(path)
if result is not None:
return result
return _probe_with_ffprobe(path)
def _ffmpeg_transcode_to_wav16_mono(
src_path: str,
) -> str | None:
"""Transcode input to a temporary 16k mono WAV.
Returns its path, or None if ffmpeg is unavailable.
"""
ffmpeg_bin = shutil.which("ffmpeg")
if ffmpeg_bin is None:
return None
with tempfile.NamedTemporaryFile(
prefix="fw_diar_",
suffix=".wav",
delete=False,
) as tmp:
tmp_path = tmp.name
cmd = [
ffmpeg_bin,
"-y",
"-v",
"error",
"-i",
src_path,
"-ac",
"1",
"-ar",
"16000",
"-f",
"wav",
tmp_path,
]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except (OSError, subprocess.CalledProcessError):
with contextlib.suppress(OSError):
Path(tmp_path).unlink()
return None
return tmp_path
def _cleanup_temp(path: str | None) -> None:
"""Remove a temporary file if it exists."""
if path is not None:
with contextlib.suppress(OSError):
Path(path).unlink()

3
linux_configuration/scripts/single_use/misc/testsAndMisc-bash/tools/transcribe_fw.py
View File

@ -11,7 +11,8 @@ from pathlib import Path
import sys
from typing import TYPE_CHECKING, Any
from _transcribe_diarize import diarize_segments, get_media_duration
from _transcribe_diarize import diarize_segments
from _transcribe_media import get_media_duration
from _transcribe_model import download_model_with_progress
from _transcribe_output import (
hhmmss,

17
linux_configuration/tests/test_transcribe_diarize_audio.py
View File

@ -13,6 +13,7 @@ import subprocess
from typing import TYPE_CHECKING
import _transcribe_diarize as diar
import _transcribe_media as media
import pytest
if TYPE_CHECKING:
@ -46,7 +47,7 @@ def test_transcode_returns_none_without_ffmpeg(
"""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
assert media._ffmpeg_transcode_to_wav16_mono("clip.mp4") is None
def test_transcode_returns_the_temp_path_on_success(
@ -61,13 +62,13 @@ def test_transcode_returns_the_temp_path_on_success(
monkeypatch.setattr(subprocess, "run", _record)
result = diar._ffmpeg_transcode_to_wav16_mono("clip.mp4")
result = media._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)
media._cleanup_temp(result)
@pytest.mark.parametrize(
@ -79,14 +80,14 @@ def test_transcode_cleans_up_after_a_failure(
"""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)))
monkeypatch.setattr(media.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 media._ffmpeg_transcode_to_wav16_mono("clip.mp4") is None
assert len(removed) == 1
@ -95,19 +96,19 @@ def test_cleanup_temp_removes_an_existing_file(tmp_path: Path) -> None:
victim = tmp_path / "scratch.wav"
victim.write_bytes(b"RIFF")
diar._cleanup_temp(str(victim))
media._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)
media._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"))
media._cleanup_temp(str(tmp_path / "never-existed.wav"))
def test_load_audio_returns_none_without_soundfile(

2
linux_configuration/tests/test_transcribe_diarize_cluster.py
View File

@ -7,7 +7,7 @@ mocked numpy would prove nothing about whether two speakers get two labels.
from __future__ import annotations
import _transcribe_diarize as diar
import _transcribe_cluster as diar
import numpy as np
import pytest

2
linux_configuration/tests/test_transcribe_diarize_probe.py
View File

@ -10,7 +10,7 @@ from __future__ import annotations
import shutil
import subprocess
import _transcribe_diarize as diar
import _transcribe_media as diar
import pytest