diff --git a/scripts/run_mvp_meeting.py b/scripts/run_mvp_meeting.py new file mode 100644 index 0000000..baf6741 --- /dev/null +++ b/scripts/run_mvp_meeting.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +"""Run the minimal Meeting Lab MVP: audio -> Whisper -> direct protocol.""" + +from __future__ import annotations + +import argparse +import json +import re +import shutil +import sys +import time +from datetime import datetime +from pathlib import Path +from typing import Any, Callable + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from src.meeting_lab.llm.ollama import DEFAULT_ENDPOINT # noqa: E402 +from src.meeting_lab.models.meeting_context import load_meeting_context # noqa: E402 +from src.meeting_lab.protocol.generate_direct_protocol import ( # noqa: E402 + DEFAULT_MODEL, + DirectProtocolResult, + generate_direct_protocol, + load_compact_transcript, +) +from src.meeting_lab.transcription.whisper import transcribe_audio # noqa: E402 + + +DEFAULT_OUTPUT_ROOT = Path("meeting_data/runs") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Transcribe one meeting and generate one direct protocol." + ) + parser.add_argument("audio_file", type=Path) + parser.add_argument("--whisper-model", type=Path, required=True) + parser.add_argument("--whisper-executable", default="whisper-cli") + parser.add_argument("--context", type=Path) + parser.add_argument("--output-root", type=Path, default=DEFAULT_OUTPUT_ROOT) + parser.add_argument("--language", default="de") + parser.add_argument("--threads", default="auto", help="Thread count or 'auto' for physical CPU cores (default: auto).") + parser.add_argument("--model", default=DEFAULT_MODEL) + parser.add_argument("--ollama-endpoint", default=DEFAULT_ENDPOINT) + return parser.parse_args(argv) + + +def create_unique_run_dir( + output_root: Path, + meeting_name: str, + now: Callable[[], datetime] = datetime.now, +) -> Path: + safe_name = re.sub(r"[^A-Za-z0-9_.-]+", "_", meeting_name).strip("._-") or "meeting" + base = output_root / f"{safe_name}_{now().strftime('%Y%m%d_%H%M%S')}" + candidate = base + suffix = 1 + while candidate.exists(): + candidate = output_root / f"{base.name}_{suffix:02d}" + suffix += 1 + candidate.mkdir(parents=True) + return candidate + + +def write_json(path: Path, value: Any) -> None: + path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +def validate_inputs(args: argparse.Namespace) -> None: + if not args.audio_file.is_file(): + raise FileNotFoundError(f"Audio file does not exist: {args.audio_file}") + if not args.whisper_model.is_file(): + raise FileNotFoundError(f"Whisper model does not exist: {args.whisper_model}") + if args.context is not None: + if not args.context.is_file(): + raise FileNotFoundError(f"Meeting Context file does not exist: {args.context}") + load_meeting_context(args.context) + + +def persist_protocol(run_dir: Path, result: DirectProtocolResult) -> Path: + protocol_dir = run_dir / "protocol" + protocol_dir.mkdir(exist_ok=True) + (protocol_dir / "exact_prompt.txt").write_text(result.exact_prompt, encoding="utf-8") + write_json(protocol_dir / "raw_response.json", result.raw_response) + write_json(protocol_dir / "runtime_metadata.json", result.runtime_metadata) + protocol_path = run_dir / "protocol.md" + protocol_path.write_text(result.protocol_text, encoding="utf-8") + return protocol_path + + +def run(args: argparse.Namespace) -> tuple[int, Path | None, Path | None]: + overall_started = time.perf_counter() + validation_started = time.perf_counter() + try: + validate_inputs(args) + except Exception as exc: + print(f"Error: {type(exc).__name__}: {exc}", file=sys.stderr) + return 2, None, None + + validation_runtime = time.perf_counter() - validation_started + run_dir = create_unique_run_dir(args.output_root, args.audio_file.stem) + timestamp = datetime.now().astimezone().isoformat(timespec="seconds") + transcript_path = run_dir / "transcript" / "transcript.json" + protocol_path = run_dir / "protocol.md" + stage_runtimes: dict[str, float | None] = { + "validation": round(validation_runtime, 3), + "setup": None, + "whisper": None, + "transcript_validation": None, + "protocol": None, + } + metadata: dict[str, Any] = { + "run_id": run_dir.name, + "timestamp": timestamp, + "input_audio": str(args.audio_file.resolve()), + "transcript_output": str(transcript_path.resolve()), + "protocol_output": str(protocol_path.resolve()), + "whisper_model": str(args.whisper_model.resolve()), + "model": args.model, + "ollama_endpoint": args.ollama_endpoint, + "status": "running", + "stage_runtimes_seconds": stage_runtimes, + "total_runtime_seconds": None, + "failure": None, + } + current_stage = "setup" + stage_started = time.perf_counter() + + try: + audio_dir = run_dir / "audio" + transcript_dir = run_dir / "transcript" + context_dir = run_dir / "context" + protocol_dir = run_dir / "protocol" + audio_dir.mkdir() + transcript_dir.mkdir() + context_dir.mkdir() + protocol_dir.mkdir() + write_json( + audio_dir / "input_manifest.json", + { + "source_file": str(args.audio_file.resolve()), + "filename": args.audio_file.name, + "size_bytes": args.audio_file.stat().st_size, + }, + ) + + preserved_context: Path | None = None + if args.context is not None: + preserved_context = context_dir / "meeting_context.yaml" + shutil.copy2(args.context, preserved_context) + stage_runtimes["setup"] = round(time.perf_counter() - stage_started, 3) + + current_stage = "whisper" + stage_started = time.perf_counter() + transcription = transcribe_audio( + args.audio_file, + args.whisper_model, + transcript_dir, + args.language, + executable=args.whisper_executable, + threads=args.threads, + ) + stage_runtimes["whisper"] = round(time.perf_counter() - stage_started, 3) + + current_stage = "transcript_validation" + stage_started = time.perf_counter() + load_compact_transcript(transcription.transcript_json) + stage_runtimes["transcript_validation"] = round( + time.perf_counter() - stage_started, 3 + ) + + current_stage = "protocol" + stage_started = time.perf_counter() + result = generate_direct_protocol( + transcription.transcript_json, + preserved_context, + model=args.model, + endpoint=args.ollama_endpoint, + ) + stage_runtimes["protocol"] = round(time.perf_counter() - stage_started, 3) + protocol_path = persist_protocol(run_dir, result) + metadata["status"] = "completed" + except Exception as exc: + if current_stage in stage_runtimes and stage_runtimes[current_stage] is None: + stage_runtimes[current_stage] = round(time.perf_counter() - stage_started, 3) + metadata["status"] = "failed" + metadata["failure"] = { + "stage": current_stage, + "type": type(exc).__name__, + "message": str(exc), + } + protocol_path = None + print(f"Error: {type(exc).__name__}: {exc}", file=sys.stderr) + finally: + metadata["total_runtime_seconds"] = round(time.perf_counter() - overall_started, 3) + write_json(run_dir / "run_metadata.json", metadata) + + return (0 if metadata["status"] == "completed" else 2), run_dir, protocol_path + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + code, _run_dir, protocol_path = run(args) + if protocol_path is not None: + print(protocol_path) + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/transcribe_meeting.py b/scripts/transcribe_meeting.py new file mode 100644 index 0000000..e096cdc --- /dev/null +++ b/scripts/transcribe_meeting.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python3 +"""Transcribe one audio file with whisper.cpp; do not generate a protocol.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from src.meeting_lab.transcription.whisper import ( # noqa: E402 + TranscriptionError, + transcribe_audio, +) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Create a compact Meeting Lab transcript with whisper.cpp.") + parser.add_argument("audio_file", type=Path) + parser.add_argument("--model", type=Path, required=True, help="Path to a whisper.cpp GGML model.") + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--language", default="auto", help="Language code or 'auto' (default: auto).") + parser.add_argument("--threads", default="auto", help="Thread count or 'auto' for physical CPU cores (default: auto).") + parser.add_argument("--whisper-executable", default="whisper-cli", help="whisper.cpp CLI executable (default: whisper-cli).") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + try: + result = transcribe_audio( + args.audio_file, + args.model, + args.output_dir, + args.language, + executable=args.whisper_executable, + threads=args.threads, + ) + except TranscriptionError as exc: + print(f"Error: {exc}") + return 1 + print(f"Transcript: {result.transcript_json}") + print(f"Runtime: {result.runtime_seconds:.3f} seconds") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/meeting_lab/transcription/whisper.py b/src/meeting_lab/transcription/whisper.py new file mode 100644 index 0000000..5e795c2 --- /dev/null +++ b/src/meeting_lab/transcription/whisper.py @@ -0,0 +1,290 @@ +"""Isolated whisper.cpp wrapper producing Meeting Lab compact transcripts.""" + +from __future__ import annotations + +import json +import os +import platform +import re +import shutil +import subprocess +import tempfile +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Sequence + + +BACKEND = "whisper.cpp" +RAW_FILENAME = "whisper_raw.json" +TRANSCRIPT_FILENAME = "transcript.json" +TEXT_FILENAME = "transcript.txt" +METADATA_FILENAME = "runtime_metadata.json" +DEFAULT_THREADS = "auto" + + +class TranscriptionError(RuntimeError): + """Raised when parameters, Whisper execution, or output are invalid.""" + + +@dataclass(frozen=True) +class TranscriptionResult: + output_dir: Path + raw_output: Path + transcript_json: Path + transcript_text: Path + runtime_metadata: Path + runtime_seconds: float + + +def _logical_cpu_count() -> int: + """Return the available logical CPU count as a last-resort fallback.""" + if hasattr(os, "sched_getaffinity"): + try: + count = len(os.sched_getaffinity(0)) + if count > 0: + return count + except OSError: + pass + return os.cpu_count() or 1 + + +def _linux_physical_core_count() -> int | None: + affinity = None + if hasattr(os, "sched_getaffinity"): + try: + affinity = os.sched_getaffinity(0) + except OSError: + pass + cores: set[tuple[str, str]] = set() + for cpu_dir in Path("/sys/devices/system/cpu").glob("cpu[0-9]*"): + try: + cpu_number = int(cpu_dir.name[3:]) + if affinity is not None and cpu_number not in affinity: + continue + topology = cpu_dir / "topology" + package = (topology / "physical_package_id").read_text().strip() + core = (topology / "core_id").read_text().strip() + cores.add((package, core)) + except (OSError, ValueError): + continue + return len(cores) or None + + +def _darwin_physical_core_count() -> int | None: + try: + completed = subprocess.run( + ("sysctl", "-n", "hw.physicalcpu"), + check=False, + capture_output=True, + text=True, + ) + count = int(completed.stdout.strip()) + return count if completed.returncode == 0 and count > 0 else None + except (OSError, ValueError): + return None + + +def physical_core_count() -> int: + """Detect physical cores where supported, falling back to available threads.""" + system = platform.system() + detected = _linux_physical_core_count() if system == "Linux" else None + if system == "Darwin": + detected = _darwin_physical_core_count() + return detected or _logical_cpu_count() + + +def resolve_threads( + threads: str | int, + detector: Callable[[], int] = physical_core_count, +) -> int: + if isinstance(threads, bool): + raise TranscriptionError("Threads must be 'auto' or a positive integer.") + if threads == "auto": + count = detector() + else: + try: + count = int(threads) + except (TypeError, ValueError) as exc: + raise TranscriptionError("Threads must be 'auto' or a positive integer.") from exc + if count <= 0: + raise TranscriptionError("Threads must be 'auto' or a positive integer.") + return count + + +def _vulkan_support(raw: dict[str, Any]) -> bool | None: + system_info = raw.get("systeminfo") + if not isinstance(system_info, str) or "VULKAN" not in system_info.upper(): + return None + return re.search(r"VULKAN\s*=\s*1", system_info, re.IGNORECASE) is not None + + +def _json_object(path: Path) -> dict[str, Any]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise TranscriptionError(f"Cannot read Whisper JSON output {path}: {exc}") from exc + if not isinstance(data, dict): + raise TranscriptionError("Whisper JSON output must contain a top-level object.") + return data + + +def compact_transcript(raw: dict[str, Any]) -> dict[str, Any]: + """Convert whisper.cpp JSON without linguistic cleanup or reordering.""" + entries = raw.get("transcription") + if not isinstance(entries, list): + raise TranscriptionError("Whisper JSON output must contain a 'transcription' list.") + + segments: list[dict[str, Any]] = [] + for index, entry in enumerate(entries): + if not isinstance(entry, dict): + raise TranscriptionError(f"transcription[{index}] must be an object.") + offsets = entry.get("offsets") + if not isinstance(offsets, dict): + raise TranscriptionError(f"transcription[{index}].offsets must be an object.") + start_ms = offsets.get("from") + end_ms = offsets.get("to") + if not isinstance(start_ms, (int, float)) or isinstance(start_ms, bool): + raise TranscriptionError(f"transcription[{index}].offsets.from must be a number.") + if not isinstance(end_ms, (int, float)) or isinstance(end_ms, bool): + raise TranscriptionError(f"transcription[{index}].offsets.to must be a number.") + if end_ms < start_ms: + raise TranscriptionError( + f"transcription[{index}].offsets.to must be greater than or equal to offsets.from." + ) + text_value = entry.get("text", "") + if not isinstance(text_value, str): + raise TranscriptionError(f"transcription[{index}].text must be a string.") + text = text_value.strip() + if text: + segments.append( + { + "id": len(segments), + "start": float(start_ms) / 1000.0, + "end": float(end_ms) / 1000.0, + "text": text, + } + ) + return {"text": " ".join(item["text"] for item in segments), "segments": segments} + + +def _timestamp(seconds: float) -> str: + milliseconds = int(round(seconds * 1000)) + hours, remainder = divmod(milliseconds, 3_600_000) + minutes, remainder = divmod(remainder, 60_000) + secs, millis = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}" + + +def transcript_text(transcript: dict[str, Any]) -> str: + lines = [ + f"[{_timestamp(item['start'])} - {_timestamp(item['end'])}] {item['text']}" + for item in transcript["segments"] + ] + return "\n".join(lines) + ("\n" if lines else "") + + +def _write_json(path: Path, value: dict[str, Any]) -> None: + path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +def transcribe_audio( + audio_path: Path, + model_path: Path, + output_dir: Path, + language: str = "auto", + *, + executable: str = "whisper-cli", + threads: str | int = DEFAULT_THREADS, + thread_detector: Callable[[], int] = physical_core_count, + runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run, + monotonic: Callable[[], float] = time.monotonic, + now: Callable[[], datetime] = lambda: datetime.now(timezone.utc), +) -> TranscriptionResult: + """Run one whisper.cpp call and write raw, compact, text, and metadata outputs.""" + audio_path = Path(audio_path) + model_path = Path(model_path) + output_dir = Path(output_dir) + if not audio_path.is_file(): + raise TranscriptionError(f"Audio file does not exist: {audio_path}") + if not model_path.is_file(): + raise TranscriptionError(f"Whisper model does not exist: {model_path}") + if not isinstance(language, str) or not language.strip(): + raise TranscriptionError("Language must be a non-empty string.") + if not executable.strip(): + raise TranscriptionError("Whisper executable must be a non-empty string.") + if output_dir.exists() and not output_dir.is_dir(): + raise TranscriptionError(f"Output directory path is not a directory: {output_dir}") + thread_count = resolve_threads(threads, thread_detector) + + output_dir.mkdir(parents=True, exist_ok=True) + raw_output = output_dir / RAW_FILENAME + started_at = now().astimezone(timezone.utc) + started = monotonic() + + with tempfile.TemporaryDirectory(prefix=".whisper-", dir=output_dir) as temp_name: + temporary_prefix = Path(temp_name) / "whisper_raw" + command: Sequence[str] = ( + executable, + "-m", str(model_path), + "-f", str(audio_path), + "-l", language.strip(), + "-t", str(thread_count), + "-fa", + "-oj", + "-of", str(temporary_prefix), + ) + try: + completed = runner(command, check=False, capture_output=True, text=True) + except OSError as exc: + raise TranscriptionError(f"Could not start {BACKEND}: {exc}") from exc + runtime_seconds = monotonic() - started + temporary_raw = temporary_prefix.with_suffix(".json") + if temporary_raw.is_file(): + shutil.copyfile(temporary_raw, raw_output) + if completed.returncode != 0: + detail = completed.stderr.strip() or completed.stdout.strip() or "no diagnostic output" + raise TranscriptionError( + f"{BACKEND} failed with exit code {completed.returncode}: {detail}" + ) + if not raw_output.is_file(): + raise TranscriptionError(f"{BACKEND} completed without producing JSON output.") + + raw_data = _json_object(raw_output) + transcript = compact_transcript(raw_data) + transcript_json_path = output_dir / TRANSCRIPT_FILENAME + transcript_text_path = output_dir / TEXT_FILENAME + metadata_path = output_dir / METADATA_FILENAME + _write_json(transcript_json_path, transcript) + transcript_text_path.write_text(transcript_text(transcript), encoding="utf-8") + duration = max((item["end"] for item in transcript["segments"]), default=None) + metadata = { + "input_file": str(audio_path.resolve()), + "model": str(model_path.resolve()), + "backend": BACKEND, + "whisper_executable": executable, + "language": language.strip(), + "threads": thread_count, + "threads_option": str(threads), + "flash_attention": True, + "vulkan_support_detected": _vulkan_support(raw_data), + "duration_seconds": duration, + "runtime_seconds": runtime_seconds, + "timestamp": started_at.isoformat(), + "output_files": { + "whisper_raw": RAW_FILENAME, + "transcript_json": TRANSCRIPT_FILENAME, + "transcript_text": TEXT_FILENAME, + "runtime_metadata": METADATA_FILENAME, + }, + } + _write_json(metadata_path, metadata) + return TranscriptionResult( + output_dir=output_dir, + raw_output=raw_output, + transcript_json=transcript_json_path, + transcript_text=transcript_text_path, + runtime_metadata=metadata_path, + runtime_seconds=runtime_seconds, + ) diff --git a/tests/test_mvp_orchestrator.py b/tests/test_mvp_orchestrator.py new file mode 100644 index 0000000..4113b05 --- /dev/null +++ b/tests/test_mvp_orchestrator.py @@ -0,0 +1,248 @@ +import json +import tempfile +import unittest +from datetime import datetime +from pathlib import Path +from unittest.mock import Mock, patch + +from scripts import run_mvp_meeting +from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult +from src.meeting_lab.transcription.whisper import TranscriptionError, TranscriptionResult + + +VALID_CONTEXT = """schema_version: "1" +meeting: + meeting_id: "mvp-test" + title: "MVP Test" + language: "de" +participants: [] +mentioned_people: [] +organization: + departments: [] +known_entities: {} +""" + + +def protocol_result(model: str = "chosen:model") -> DirectProtocolResult: + text = "# Protokoll\n\nUnverändert. \n" + return DirectProtocolResult( + protocol_text=text, + exact_prompt="exact prompt\n", + model_metadata={"model": model}, + runtime_metadata={"model": model, "request_count": 1, "client_wall_time_seconds": 0.5}, + raw_response={"response": text, "done": True}, + ) + + +def fake_transcribe( + audio_path: Path, + model_path: Path, + output_dir: Path, + language: str, + *, + executable: str, + threads: str | int, +) -> TranscriptionResult: + output_dir.mkdir(parents=True, exist_ok=True) + raw = output_dir / "whisper_raw.json" + transcript = output_dir / "transcript.json" + text = output_dir / "transcript.txt" + metadata = output_dir / "runtime_metadata.json" + raw.write_text('{"transcription": []}\n', encoding="utf-8") + transcript.write_text( + json.dumps({"text": "Besprechungstext.", "segments": []}) + "\n", + encoding="utf-8", + ) + text.write_text("Besprechungstext.\n", encoding="utf-8") + metadata.write_text('{"runtime_seconds": 1.25}\n', encoding="utf-8") + return TranscriptionResult(output_dir, raw, transcript, text, metadata, 1.25) + + +class MvpOrchestratorTests(unittest.TestCase): + def create_inputs(self, root: Path) -> tuple[Path, Path, Path]: + audio = root / "team meeting.wav" + whisper_model = root / "ggml-model.bin" + context = root / "source-context.yaml" + audio.write_bytes(b"audio") + whisper_model.write_bytes(b"model") + context.write_text(VALID_CONTEXT, encoding="utf-8") + return audio, whisper_model, context + + def args(self, root: Path, extra: list[str] | None = None): + audio, whisper_model, context = self.create_inputs(root) + values = [ + str(audio), + "--whisper-model", str(whisper_model), + "--context", str(context), + "--output-root", str(root / "runs"), + "--language", "de", + "--model", "chosen:model", + "--ollama-endpoint", "http://ollama.test:11434", + ] + if extra: + values.extend(extra) + return run_mvp_meeting.parse_args(values) + + def test_successful_full_orchestration_and_artifact_layout(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.args(root) + with ( + patch.object(run_mvp_meeting, "transcribe_audio", side_effect=fake_transcribe) as whisper, + patch.object( + run_mvp_meeting, + "generate_direct_protocol", + return_value=protocol_result(), + ) as protocol, + ): + code, run_dir, protocol_path = run_mvp_meeting.run(args) + + self.assertEqual(code, 0) + self.assertIsNotNone(run_dir) + self.assertEqual(protocol_path, run_dir / "protocol.md") + expected = { + "run_metadata.json", + "audio/input_manifest.json", + "transcript/whisper_raw.json", + "transcript/transcript.json", + "transcript/transcript.txt", + "transcript/runtime_metadata.json", + "context/meeting_context.yaml", + "protocol/exact_prompt.txt", + "protocol/raw_response.json", + "protocol/runtime_metadata.json", + "protocol.md", + } + self.assertTrue(all((run_dir / item).is_file() for item in expected)) + metadata = json.loads((run_dir / "run_metadata.json").read_text()) + self.assertEqual(metadata["status"], "completed") + self.assertEqual(metadata["model"], "chosen:model") + self.assertIsNone(metadata["failure"]) + self.assertEqual(whisper.call_count, 1) + self.assertEqual(protocol.call_count, 1) + + def test_context_model_endpoint_and_whisper_options_are_forwarded(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.args( + root, + ["--whisper-executable", "/tools/whisper-cli", "--threads", "4"], + ) + with ( + patch.object(run_mvp_meeting, "transcribe_audio", side_effect=fake_transcribe) as whisper, + patch.object( + run_mvp_meeting, + "generate_direct_protocol", + return_value=protocol_result(), + ) as protocol, + ): + code, run_dir, _ = run_mvp_meeting.run(args) + + self.assertEqual(code, 0) + self.assertEqual(whisper.call_args.args[3], "de") + self.assertEqual(whisper.call_args.kwargs["executable"], "/tools/whisper-cli") + self.assertEqual(whisper.call_args.kwargs["threads"], "4") + self.assertEqual(protocol.call_args.args[1], run_dir / "context/meeting_context.yaml") + self.assertEqual(protocol.call_args.kwargs["model"], "chosen:model") + self.assertEqual( + protocol.call_args.kwargs["endpoint"], "http://ollama.test:11434" + ) + + def test_whisper_failure_is_recorded_and_protocol_is_not_called(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.args(root) + with ( + patch.object( + run_mvp_meeting, + "transcribe_audio", + side_effect=TranscriptionError("whisper stopped"), + ), + patch.object(run_mvp_meeting, "generate_direct_protocol") as protocol, + ): + code, run_dir, protocol_path = run_mvp_meeting.run(args) + + metadata = json.loads((run_dir / "run_metadata.json").read_text()) + self.assertEqual(code, 2) + self.assertIsNone(protocol_path) + self.assertEqual(metadata["status"], "failed") + self.assertEqual(metadata["failure"]["stage"], "whisper") + self.assertIn("whisper stopped", metadata["failure"]["message"]) + protocol.assert_not_called() + self.assertTrue((run_dir / "audio/input_manifest.json").is_file()) + + def test_protocol_failure_preserves_transcript_and_failure_metadata(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.args(root) + with ( + patch.object(run_mvp_meeting, "transcribe_audio", side_effect=fake_transcribe), + patch.object( + run_mvp_meeting, + "generate_direct_protocol", + side_effect=ValueError("generation stopped"), + ), + ): + code, run_dir, protocol_path = run_mvp_meeting.run(args) + + metadata = json.loads((run_dir / "run_metadata.json").read_text()) + self.assertEqual(code, 2) + self.assertIsNone(protocol_path) + self.assertEqual(metadata["failure"]["stage"], "protocol") + self.assertTrue((run_dir / "transcript/transcript.json").is_file()) + self.assertFalse((run_dir / "protocol.md").exists()) + + def test_unique_run_directories_do_not_overwrite(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + fixed = datetime(2026, 8, 21, 9, 15, 30) + first = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed) + marker = first / "keep.txt" + marker.write_text("keep", encoding="utf-8") + second = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed) + self.assertEqual(first.name, "team_meeting_20260821_091530") + self.assertEqual(second.name, "team_meeting_20260821_091530_01") + self.assertEqual(marker.read_text(), "keep") + + def test_semantic_pipeline_functions_are_never_invoked(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.args(root) + with ( + patch.object(run_mvp_meeting, "transcribe_audio", side_effect=fake_transcribe), + patch.object( + run_mvp_meeting, + "generate_direct_protocol", + return_value=protocol_result(), + ), + patch("src.meeting_lab.chunking.chunk_transcript.build_chunks") as chunking, + patch("src.meeting_lab.extraction.extract_chunks.extract_input") as extraction, + patch("src.meeting_lab.consolidation.consolidate_facts.call_ollama") as consolidation, + ): + code, _, _ = run_mvp_meeting.run(args) + + self.assertEqual(code, 0) + chunking.assert_not_called() + extraction.assert_not_called() + consolidation.assert_not_called() + + def test_main_returns_nonzero_for_whisper_failure(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + args = self.args(root) + argv = [ + str(args.audio_file), + "--whisper-model", str(args.whisper_model), + "--context", str(args.context), + "--output-root", str(args.output_root), + ] + with patch.object( + run_mvp_meeting, + "transcribe_audio", + side_effect=TranscriptionError("failed"), + ): + self.assertEqual(run_mvp_meeting.main(argv), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_transcription.py b/tests/test_transcription.py new file mode 100644 index 0000000..3765255 --- /dev/null +++ b/tests/test_transcription.py @@ -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()