222 lines
9.2 KiB
Python
222 lines
9.2 KiB
Python
import json
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from scripts import transcribe_meeting
|
|
from src.meeting_lab.transcription.whisper import (
|
|
TranscriptionError,
|
|
compact_transcript,
|
|
transcribe_audio,
|
|
)
|
|
|
|
|
|
RAW_RESULT = {
|
|
"params": {"language": "de"},
|
|
"transcription": [
|
|
{
|
|
"timestamps": {"from": "00:00:00,000", "to": "00:00:01,200"},
|
|
"offsets": {"from": 0, "to": 1200},
|
|
"text": " Hallo ",
|
|
},
|
|
{
|
|
"timestamps": {"from": "00:00:01,200", "to": "00:00:02,500"},
|
|
"offsets": {"from": 1200, "to": 2500},
|
|
"text": "Welt",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self, *, returncode: int = 0, write_raw: bool = True) -> None:
|
|
self.returncode = returncode
|
|
self.write_raw = write_raw
|
|
self.commands: list[list[str]] = []
|
|
|
|
def __call__(self, command, **kwargs):
|
|
command = list(command)
|
|
self.commands.append(command)
|
|
if self.write_raw:
|
|
prefix = Path(command[command.index("-of") + 1])
|
|
prefix.with_suffix(".json").write_text(
|
|
json.dumps(RAW_RESULT, ensure_ascii=False), encoding="utf-8"
|
|
)
|
|
return subprocess.CompletedProcess(
|
|
command, self.returncode, stdout="", stderr="backend failure"
|
|
)
|
|
|
|
|
|
class TranscriptionTests(unittest.TestCase):
|
|
def create_inputs(self, root: Path) -> tuple[Path, Path]:
|
|
audio = root / "meeting.wav"
|
|
model = root / "ggml-model.bin"
|
|
audio.write_bytes(b"audio")
|
|
model.write_bytes(b"model")
|
|
return audio, model
|
|
|
|
def test_missing_audio_file_is_rejected_before_execution(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
model = root / "model.bin"
|
|
model.write_bytes(b"model")
|
|
runner = FakeRunner()
|
|
with self.assertRaisesRegex(TranscriptionError, "Audio file does not exist"):
|
|
transcribe_audio(root / "missing.wav", model, root / "out", runner=runner)
|
|
self.assertEqual(runner.commands, [])
|
|
|
|
def test_invalid_parameters_are_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
with self.assertRaisesRegex(TranscriptionError, "model does not exist"):
|
|
transcribe_audio(audio, root / "missing.bin", root / "out")
|
|
with self.assertRaisesRegex(TranscriptionError, "Language must"):
|
|
transcribe_audio(audio, model, root / "out", " ")
|
|
with self.assertRaisesRegex(TranscriptionError, "executable must"):
|
|
transcribe_audio(audio, model, root / "out", executable="")
|
|
with self.assertRaisesRegex(TranscriptionError, "Threads must"):
|
|
transcribe_audio(audio, model, root / "out", threads="zero")
|
|
|
|
def test_auto_threads_use_detected_physical_core_count_and_flash_attention(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
runner = FakeRunner()
|
|
result = transcribe_audio(
|
|
audio,
|
|
model,
|
|
root / "output",
|
|
runner=runner,
|
|
thread_detector=lambda: 6,
|
|
)
|
|
|
|
command = runner.commands[0]
|
|
self.assertEqual(command[command.index("-t") + 1], "6")
|
|
self.assertIn("-fa", command)
|
|
metadata = json.loads(result.runtime_metadata.read_text())
|
|
self.assertEqual(metadata["threads"], 6)
|
|
self.assertEqual(metadata["threads_option"], "auto")
|
|
self.assertTrue(metadata["flash_attention"])
|
|
self.assertEqual(metadata["whisper_executable"], "whisper-cli")
|
|
|
|
def test_explicit_thread_count_is_forwarded_unchanged(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
runner = FakeRunner()
|
|
result = transcribe_audio(
|
|
audio, model, root / "output", threads=12, runner=runner
|
|
)
|
|
|
|
command = runner.commands[0]
|
|
self.assertEqual(command[command.index("-t") + 1], "12")
|
|
metadata = json.loads(result.runtime_metadata.read_text())
|
|
self.assertEqual(metadata["threads"], 12)
|
|
self.assertEqual(metadata["threads_option"], "12")
|
|
|
|
def test_cli_forwards_explicit_thread_count_to_wrapper(self) -> None:
|
|
fake_result = Mock(
|
|
transcript_json=Path("output/transcript.json"), runtime_seconds=1.0
|
|
)
|
|
with patch.object(
|
|
transcribe_meeting, "transcribe_audio", return_value=fake_result
|
|
) as transcribe:
|
|
code = transcribe_meeting.main(
|
|
[
|
|
"meeting.wav",
|
|
"--model", "model.bin",
|
|
"--output-dir", "output",
|
|
"--threads", "6",
|
|
]
|
|
)
|
|
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(transcribe.call_args.kwargs["threads"], "6")
|
|
|
|
def test_compact_transcript_has_established_schema(self) -> None:
|
|
self.assertEqual(
|
|
compact_transcript(RAW_RESULT),
|
|
{
|
|
"text": "Hallo Welt",
|
|
"segments": [
|
|
{"id": 0, "start": 0.0, "end": 1.2, "text": "Hallo"},
|
|
{"id": 1, "start": 1.2, "end": 2.5, "text": "Welt"},
|
|
],
|
|
},
|
|
)
|
|
|
|
def test_success_preserves_raw_output_and_creates_metadata(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
runner = FakeRunner()
|
|
times = iter([10.0, 12.25])
|
|
result = transcribe_audio(
|
|
audio,
|
|
model,
|
|
root / "output",
|
|
"de",
|
|
runner=runner,
|
|
monotonic=lambda: next(times),
|
|
now=lambda: datetime(2026, 8, 21, 8, 30, tzinfo=timezone.utc),
|
|
)
|
|
|
|
self.assertEqual(json.loads(result.raw_output.read_text()), RAW_RESULT)
|
|
self.assertEqual(json.loads(result.transcript_json.read_text())["text"], "Hallo Welt")
|
|
self.assertEqual(
|
|
result.transcript_text.read_text(),
|
|
"[00:00:00.000 - 00:00:01.200] Hallo\n"
|
|
"[00:00:01.200 - 00:00:02.500] Welt\n",
|
|
)
|
|
metadata = json.loads(result.runtime_metadata.read_text())
|
|
self.assertEqual(metadata["backend"], "whisper.cpp")
|
|
self.assertEqual(metadata["language"], "de")
|
|
self.assertEqual(metadata["duration_seconds"], 2.5)
|
|
self.assertEqual(metadata["runtime_seconds"], 2.25)
|
|
self.assertEqual(metadata["timestamp"], "2026-08-21T08:30:00+00:00")
|
|
self.assertEqual(metadata["input_file"], str(audio.resolve()))
|
|
self.assertEqual(metadata["model"], str(model.resolve()))
|
|
|
|
def test_output_paths_are_deterministic(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
output = root / "chosen-output"
|
|
result = transcribe_audio(audio, model, output, runner=FakeRunner())
|
|
self.assertEqual(result.raw_output, output / "whisper_raw.json")
|
|
self.assertEqual(result.transcript_json, output / "transcript.json")
|
|
self.assertEqual(result.transcript_text, output / "transcript.txt")
|
|
self.assertEqual(result.runtime_metadata, output / "runtime_metadata.json")
|
|
self.assertEqual(
|
|
set(json.loads(result.runtime_metadata.read_text())["output_files"].values()),
|
|
{"whisper_raw.json", "transcript.json", "transcript.txt", "runtime_metadata.json"},
|
|
)
|
|
|
|
def test_backend_failure_raises_and_preserves_available_raw_output(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
output = root / "output"
|
|
with self.assertRaisesRegex(TranscriptionError, "exit code 7.*backend failure"):
|
|
transcribe_audio(audio, model, output, runner=FakeRunner(returncode=7))
|
|
self.assertEqual(json.loads((output / "whisper_raw.json").read_text()), RAW_RESULT)
|
|
self.assertFalse((output / "transcript.json").exists())
|
|
self.assertFalse((output / "runtime_metadata.json").exists())
|
|
|
|
def test_missing_backend_output_is_reported(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
audio, model = self.create_inputs(root)
|
|
with self.assertRaisesRegex(TranscriptionError, "without producing JSON"):
|
|
transcribe_audio(
|
|
audio, model, root / "output", runner=FakeRunner(write_raw=False)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|