martin/meeting-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use physical CPU cores for Whisper runtime defaults

Browse Source
This commit is contained in:
admin
2026-08-21 11:11:24 +02:00
parent a9dab7c81a
commit f2d21c1faf
5 changed files with 1021 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_transcription.py
+221
View File
@@ -0,0 +1,221 @@
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()