diff --git a/tests/test_audio_utils.py b/tests/test_audio_utils.py index 929c351..d3a2781 100644 --- a/tests/test_audio_utils.py +++ b/tests/test_audio_utils.py @@ -11,7 +11,6 @@ from transcript_combobulator.audio_utils import ( AudioValidationError, convert_to_wav, - needs_conversion, validate_audio_file, ) @@ -28,7 +27,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) @@ -38,7 +36,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 6952c8b..fab6ba4 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -5,7 +5,9 @@ import subprocess 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 ( @@ -347,7 +349,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", @@ -398,7 +401,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", @@ -412,21 +416,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 ( @@ -438,9 +444,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, []), @@ -453,7 +456,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.""" @@ -486,9 +491,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, @@ -724,7 +726,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", @@ -775,7 +778,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", @@ -827,9 +831,7 @@ def fake_transcribe(audio_path, output_dir, **kwargs): } with ( - patch( - "transcript_combobulator.pipeline.needs_conversion", return_value=False - ), + patch("transcript_combobulator.pipeline.convert_to_wav"), patch( "transcript_combobulator.pipeline.process_audio", return_value=(tmp_path, []), @@ -905,9 +907,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", @@ -959,3 +958,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 7d6ac17..a886a0f 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 86fb7cf..f12bea9 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 @@ -12,7 +11,6 @@ AudioValidationError, convert_to_wav, input_duration, - needs_conversion, ) from transcript_combobulator.config import ( get_output_path_for_input, @@ -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")