From f19fe3ce136ed3e95337ff33337f5898d1b80dff Mon Sep 17 00:00:00 2001 From: Rue <343175018+thatsrue@users.noreply.github.com> Date: Fri, 30 Oct 2026 21:42:55 -0400 Subject: [PATCH] fix: normalise every input, keep checkpoints after a cut write, rewrite a lost VTT from JSON Closes #77 --- tests/test_audio_utils.py | 3 - tests/test_batch.py | 93 +++++++++++++++++------ tests/test_pipeline_cache.py | 7 ++ transcript_combobulator/audio_utils.py | 13 ---- transcript_combobulator/pipeline.py | 21 ++--- transcript_combobulator/pipeline_cache.py | 17 ++++- 6 files changed, 101 insertions(+), 53 deletions(-) diff --git a/tests/test_audio_utils.py b/tests/test_audio_utils.py index e283aec..f008252 100644 --- a/tests/test_audio_utils.py +++ b/tests/test_audio_utils.py @@ -10,7 +10,6 @@ from transcript_combobulator.audio_utils import ( AudioValidationError, convert_to_wav, - needs_conversion, validate_audio_file, ) @@ -27,7 +26,6 @@ def test_ffmpeg_conversion_yields_16k_mono_peak_normalized(tmp_path: Path) -> No src = tmp_path / "in.wav" _stereo_44k(src) out = tmp_path / "out.wav" - assert needs_conversion(src) convert_to_wav(src, out) @@ -37,7 +35,6 @@ def test_ffmpeg_conversion_yields_16k_mono_peak_normalized(tmp_path: Path) -> No assert abs(info["duration"] - 2.0) < 0.002 data, _ = sf.read(str(out), dtype="float32") assert 0.98 <= float(np.abs(data).max()) <= 1.0 - assert not needs_conversion(out) assert not list(tmp_path.glob("*.unnormalized.wav")) diff --git a/tests/test_batch.py b/tests/test_batch.py index e89aaf4..11c1695 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -4,7 +4,9 @@ import json from unittest.mock import patch +import numpy as np import pytest +import soundfile as sf from rich.console import Console from transcript_combobulator.batch import ( @@ -285,7 +287,8 @@ def test_force_ignores_completed_pipeline(self, tmp_path): return_value=True, ), patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=False + "transcript_combobulator.pipeline.convert_to_wav", + side_effect=lambda source, destination: destination.touch(), ), patch( "transcript_combobulator.pipeline.process_audio", @@ -336,7 +339,8 @@ def fail_after_manifest_invalidation(*args, **kwargs): return_value=True, ), patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=False + "transcript_combobulator.pipeline.convert_to_wav", + side_effect=lambda source, destination: destination.touch(), ), patch( "transcript_combobulator.pipeline.process_audio", @@ -350,21 +354,23 @@ def fail_after_manifest_invalidation(*args, **kwargs): assert "vtt" not in load_manifest(manifest) - def test_cache_miss_replaces_existing_normalized_audio(self, tmp_path): - """A changed normalized source replaces the prior derived WAV.""" + def test_quiet_16k_mono_wav_is_peak_normalised(self, tmp_path): + """A 16 kHz mono WAV reaches VAD at the same level as any other input.""" from transcript_combobulator.pipeline import process_file as main - input_file = tmp_path / "input" / "speaker.wav" + input_file = tmp_path / "input" / "2-danica.wav" input_file.parent.mkdir() - input_file.write_bytes(b"new audio") + t = np.arange(16000) / 16000 + quiet = (0.1 * np.sin(2 * np.pi * 440 * t)).astype(np.float32) + sf.write(str(input_file), quiet, 16000, subtype="PCM_16") output_dir = tmp_path / "output" output_dir.mkdir() - output_file = output_dir / "speaker.wav" + output_file = output_dir / "2-danica.wav" output_file.write_bytes(b"old audio") transcription = { - "vtt_file": str(output_dir / "speaker.vtt"), - "json_file": str(output_dir / "speaker.json"), - "mapping_file": str(output_dir / "speaker_mapping.json"), + "vtt_file": str(output_dir / "2-danica.vtt"), + "json_file": str(output_dir / "2-danica.json"), + "mapping_file": str(output_dir / "2-danica_mapping.json"), } with ( @@ -376,9 +382,6 @@ def test_cache_miss_replaces_existing_normalized_audio(self, tmp_path): "transcript_combobulator.pipeline.is_pipeline_complete", return_value=False, ), - patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=False - ), patch( "transcript_combobulator.pipeline.process_audio", return_value=(output_dir, []), @@ -391,7 +394,9 @@ def test_cache_miss_replaces_existing_normalized_audio(self, tmp_path): ): main(str(input_file)) - assert output_file.read_bytes() == b"new audio" + data, sample_rate = sf.read(str(output_file), dtype="float32") + assert sample_rate == 16000 + assert 0.98 <= float(np.abs(data).max()) <= 1.0 def test_cache_miss_removes_existing_audio_before_conversion(self, tmp_path): """Conversion cannot silently reuse a valid but stale derived WAV.""" @@ -424,9 +429,6 @@ def convert_after_removal(source, destination): "transcript_combobulator.pipeline.is_pipeline_complete", return_value=False, ), - patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=True - ), patch( "transcript_combobulator.pipeline.convert_to_wav", side_effect=convert_after_removal, @@ -662,7 +664,8 @@ def _run(self, tmp_path, failed): return_value=False, ), patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=False + "transcript_combobulator.pipeline.convert_to_wav", + side_effect=lambda source, destination: destination.touch(), ), patch( "transcript_combobulator.pipeline.process_audio", @@ -713,7 +716,8 @@ def _run(self, tmp_path, input_file, session=None): return_value=False, ), patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=False + "transcript_combobulator.pipeline.convert_to_wav", + side_effect=lambda source, destination: destination.touch(), ), patch( "transcript_combobulator.pipeline.process_audio", @@ -801,9 +805,6 @@ def test_conversion_and_vad_are_skipped_when_recorded(self, tmp_path): return_value=output_dir, ), patch("transcript_combobulator.pipeline.convert_to_wav") as convert, - patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=True - ), patch("transcript_combobulator.pipeline.process_audio") as vad, patch( "transcript_combobulator.pipeline.transcribe_segments", @@ -855,3 +856,51 @@ def test_presentation_only_change_rewrites_vtt_without_inference(self, tmp_path) assert metrics["transcription"]["inference_cached"] is True assert "Hello" in vtt.read_text(encoding="utf-8") assert "vtt_rewrite_seconds" in metrics["stages"] + + def test_missing_vtt_is_rewritten_from_json_without_inference(self, tmp_path): + from transcript_combobulator.pipeline import process_file as main + + input_file, output_dir, manifest, fps = self._setup(tmp_path) + json_path = output_dir / "speaker_transcription.json" + vtt = output_dir / "speaker.vtt" + + def transcribe(*args, **kwargs): + json_path.write_text( + json.dumps( + { + "segments": [ + { + "start": 1.0, + "end": 2.0, + "text": "Hello", + "confidence": 90.0, + } + ] + } + ) + ) + vtt.write_text("WEBVTT\n", encoding="utf-8") + return { + "vtt_file": str(vtt), + "json_file": str(json_path), + "mapping_file": str(output_dir / "speaker_mapping.json"), + "metrics": {"chunk_count": 1, "failed_chunk_count": 0, "chunks": []}, + } + + with ( + patch( + "transcript_combobulator.pipeline.get_output_path_for_input", + return_value=output_dir, + ), + patch( + "transcript_combobulator.pipeline.transcribe_segments", + side_effect=transcribe, + ) as ts, + ): + main(str(input_file)) + vtt.unlink() + metrics = main(str(input_file)) + + ts.assert_called_once() + assert metrics["transcription"]["inference_cached"] is True + assert "Hello" in vtt.read_text(encoding="utf-8") diff --git a/tests/test_pipeline_cache.py b/tests/test_pipeline_cache.py index 19cb0d6..fed53a6 100644 --- a/tests/test_pipeline_cache.py +++ b/tests/test_pipeline_cache.py @@ -166,6 +166,13 @@ def test_chunk_checkpoint_roundtrip_and_truncation(tmp_path: Path) -> None: assert set(loaded) == {1, 2} assert loaded[1]["segments"][0]["text"] == "a" + # the next run's chunks are not lost behind the fragment + cp = ChunkCheckpoint(progress, key="k1") + cp.record(3, [], {"audio_seconds": 0.5}) + cp.record(4, [], {"audio_seconds": 0.5}) + cp.close(remove=False) + assert set(load_chunk_checkpoint(progress, "k1")) == {1, 2, 3, 4} + # a different key (settings changed) discards the file assert load_chunk_checkpoint(progress, "k2") == {} assert not progress.exists() diff --git a/transcript_combobulator/audio_utils.py b/transcript_combobulator/audio_utils.py index 49de5c0..a6a7655 100644 --- a/transcript_combobulator/audio_utils.py +++ b/transcript_combobulator/audio_utils.py @@ -46,19 +46,6 @@ def validate_audio_file(file_path: Path) -> dict[str, Any]: raise AudioValidationError(f"Failed to read audio file {file_path}: {e}") from e -def needs_conversion(file_path: Path, target_sample_rate: int = 16000) -> bool: - """True if the file isn't already 16kHz mono WAV.""" - try: - info = validate_audio_file(file_path) - except AudioValidationError: - return True - return bool( - info["format"] != ".wav" - or info["sample_rate"] != target_sample_rate - or info["channels"] != 1 - ) - - def convert_to_wav( input_path: Path, output_path: Path, target_sample_rate: int = 16000 ) -> None: diff --git a/transcript_combobulator/pipeline.py b/transcript_combobulator/pipeline.py index 957afd5..aaf0f3b 100644 --- a/transcript_combobulator/pipeline.py +++ b/transcript_combobulator/pipeline.py @@ -2,7 +2,6 @@ import json import logging -import shutil import time from contextlib import contextmanager from pathlib import Path @@ -11,7 +10,6 @@ from transcript_combobulator.audio_utils import ( AudioValidationError, convert_to_wav, - needs_conversion, validate_audio_file, ) from transcript_combobulator.config import ( @@ -140,17 +138,11 @@ def _progress_callback(phase: str, current: int, total: int) -> None: # A cache miss must not reuse audio derived from an older source. if output_file.resolve() != input_file: output_file.unlink(missing_ok=True) - conversion_required = needs_conversion(input_file) - file_metrics["conversion"] = { - "required": conversion_required, - "action": "converted" if conversion_required else "copied", - } - if conversion_required: - convert_to_wav(input_file, output_file) - else: - if not output_file.exists(): - shutil.copy(input_file, output_file) - logger.info("No conversion needed, copied to output directory") + # Even a 16 kHz mono WAV is converted: conversion is where the + # peak normalisation happens, so VAD sees the same level + # whatever the container. + file_metrics["conversion"] = {"required": True, "action": "converted"} + convert_to_wav(input_file, output_file) record_stage( manifest_path, "conversion", @@ -242,11 +234,12 @@ def _progress_callback(phase: str, current: int, total: int) -> None: ) if not stage_is_complete(stages, "inference", fingerprints["inference"]): + # Only the JSON: a missing VTT is rewritten from it, not re-inferred. record_stage( manifest_path, "inference", fingerprints["inference"], - [Path(transcription["json_file"]), Path(transcription["vtt_file"])], + [Path(transcription["json_file"])], ) artifacts = [ diff --git a/transcript_combobulator/pipeline_cache.py b/transcript_combobulator/pipeline_cache.py index ab35597..01ef9ef 100644 --- a/transcript_combobulator/pipeline_cache.py +++ b/transcript_combobulator/pipeline_cache.py @@ -207,6 +207,20 @@ def load_chunk_checkpoint(progress_path: Path, key: str) -> dict[int, dict[str, return completed +def _drop_partial_line(progress_path: Path) -> None: + """Cut an interrupted last write, so the next record starts its own line. + + Appending after the fragment would glue the next record onto it, and the + reader stops at the first line it cannot parse. + """ + if not progress_path.is_file(): + return + data = progress_path.read_bytes() + if data and not data.endswith(b"\n"): + with open(progress_path, "r+b") as f: + f.truncate(data.rfind(b"\n") + 1) + + class ChunkCheckpoint: """Append-only per-chunk progress. Each completed chunk is one JSON line.""" @@ -215,7 +229,8 @@ def __init__(self, progress_path: Path, key: str) -> None: self.key = key self.completed = load_chunk_checkpoint(progress_path, key) progress_path.parent.mkdir(parents=True, exist_ok=True) - new_file = not progress_path.is_file() + _drop_partial_line(progress_path) + new_file = not progress_path.is_file() or progress_path.stat().st_size == 0 self._handle = open(progress_path, "a", encoding="utf-8") if new_file: self._handle.write(json.dumps({"key": key}) + "\n")