refactor(cli): одиночный файл и батч переведены на module выполнения

CLI строит один ExecutionRequest, создаёт Transcriber после prescan
и получает результат с ExecutionInfo: разрешение auto, device-aware
умолчания, проверка пословного контракта, строка устройства для шапки
и fallback ушли за seam. --verbose печатает движок, потоки и runtime_info().
Удалены load_model, _transcribe_file, TranscribeFileResult,
apply_device_defaults и тесты переноса полей; сценарные тесты CLI идут
через CliRunner с fake adapter на get_backend.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dmitry DementievandClaude Opus 5 committed 2026-09-22 00:05:27 +03:00
1 parent e22b2f1bae
commit 0f8ecb5bd3
13 files changed
+402 -1606

No files matched your search

@@ -167,7 +167,6 @@ class FasterWhisperBackend:
except Exception as exc: # noqa: BLE001 — диагностика не должна ронять запуск except Exception as exc: # noqa: BLE001 — диагностика не должна ронять запуск
cuda_devices = f"недоступно ({exc})" cuda_devices = f"недоступно ({exc})"
return { return {
"engine": "faster-whisper",
"faster_whisper": package_version("faster-whisper"), "faster_whisper": package_version("faster-whisper"),
"ctranslate2": package_version("ctranslate2"), "ctranslate2": package_version("ctranslate2"),
"cuda_devices": cuda_devices, "cuda_devices": cuda_devices,
@@ -229,7 +229,6 @@ class OnnxAsrBackend:
configured = ", ".join(self._providers) configured = ", ".join(self._providers)
return { return {
"engine": "onnx-asr",
"onnxruntime": package_version("onnxruntime"), "onnxruntime": package_version("onnxruntime"),
"onnx_asr": package_version("onnx-asr"), "onnx_asr": package_version("onnx-asr"),
"available_providers": ", ".join(onnxruntime.get_available_providers()), "available_providers": ", ".join(onnxruntime.get_available_providers()),
@@ -209,7 +209,6 @@ class OpenVINOBackend:
from openvino import Core from openvino import Core
return { return {
"engine": "openvino",
"openvino": package_version("openvino"), "openvino": package_version("openvino"),
"available_devices": ", ".join(Core().available_devices), "available_devices": ", ".join(Core().available_devices),
"device": self.actual_ov_device or "", "device": self.actual_ov_device or "",
+90 -179
View File
@@ -13,8 +13,6 @@ from rich.status import Status
from .config import ( from .config import (
CliValues, CliValues,
ResolvedConfig,
apply_device_defaults,
load_config, load_config,
resolve_defaults, resolve_defaults,
) )
@@ -39,11 +37,12 @@ from .quality import (
) )
from .speaker_diarizer import SpeakerDiarizer, load_speaker_diarizer from .speaker_diarizer import SpeakerDiarizer, load_speaker_diarizer
from .transcriber import ( from .transcriber import (
ExecutionInfo,
ExecutionRequest,
Segment, Segment,
Transcriber,
TranscribeResult, TranscribeResult,
_transcribe_file,
cuda_error_hint, cuda_error_hint,
load_model,
) )
from .types import ( from .types import (
UNKNOWN_LANGUAGE, UNKNOWN_LANGUAGE,
@@ -53,10 +52,7 @@ from .types import (
) )
from .utils import ( from .utils import (
build_output_path, build_output_path,
detect_device,
expand_globs, expand_globs,
get_gpu_name,
get_intel_gpu_name,
has_existing_transcript, has_existing_transcript,
validate_input_file, validate_input_file,
) )
@@ -95,21 +91,6 @@ def _with_cli_warning_renderer(
return wrapped return wrapped
def _format_device_info(device_used: str) -> str:
"""Формирует строку устройства для шапки транскрипта."""
if device_used == "cuda":
gpu_name = get_gpu_name()
return f"CUDA ({gpu_name or 'Unknown GPU'})"
if device_used == "openvino-gpu":
gpu_name = get_intel_gpu_name()
return f"OpenVINO ({gpu_name or 'Intel GPU'})"
if device_used in ("openvino", "openvino-cpu"):
return "OpenVINO (CPU)"
if device_used == "onnx":
return "ONNX (CPU)"
return "CPU"
def _format_language_mode(requested_language: str, result: TranscribeResult) -> str: def _format_language_mode(requested_language: str, result: TranscribeResult) -> str:
"""Описывает источник языка, не выдавая профиль модели за детектор.""" """Описывает источник языка, не выдавая профиль модели за детектор."""
if requested_language != "auto": if requested_language != "auto":
@@ -348,7 +329,7 @@ def main(
) )
raise SystemExit(2) raise SystemExit(2)
resolved_device: str | None = None requested_device: str | None = None
try: try:
config = load_config() config = load_config()
cli_values: CliValues = { cli_values: CliValues = {
@@ -360,11 +341,19 @@ def main(
} }
defaults = resolve_defaults(cli_values, config) defaults = resolve_defaults(cli_values, config)
diarize_enabled = defaults["diarize"] is True or speakers is not None diarize_enabled = defaults["diarize"] is True or speakers is not None
requested_device = defaults["device"]
resolved_device = detect_device(defaults["device"]) # Явные model/compute_type; умолчания исполнения разрешает module.
defaults = apply_device_defaults(defaults, resolved_device, cli_values, config) request = ExecutionRequest(
device=requested_device,
ct_explicit = compute_type is not None or "compute_type" in config model=model if model is not None else config.get("model"),
compute_type=(
compute_type if compute_type is not None else config.get("compute_type")
),
language=defaults["language"],
cpu_threads=threads,
require_word_timestamps=diarize_enabled,
)
expanded = expand_globs(files) expanded = expand_globs(files)
if not expanded: if not expanded:
@@ -381,22 +370,18 @@ def main(
if is_batch: if is_batch:
_run_batch( _run_batch(
expanded, expanded,
defaults, request,
verbose, verbose,
force, force,
ct_explicit,
cpu_threads=threads,
diarize=diarize_enabled, diarize=diarize_enabled,
speakers=speakers, speakers=speakers,
) )
else: else:
_run_single( _run_single(
expanded[0], expanded[0],
defaults, request,
output, output,
verbose, verbose,
ct_explicit,
cpu_threads=threads,
diarize=diarize_enabled, diarize=diarize_enabled,
speakers=speakers, speakers=speakers,
) )
@@ -406,15 +391,15 @@ def main(
except SystemExit: except SystemExit:
raise raise
except ValueError as exc: except ValueError as exc:
_print_cuda_hint(exc, resolved_device) _print_cuda_hint(exc, requested_device)
console.print(f"Ошибка: {exc}", style="red bold") console.print(f"Ошибка: {exc}", style="red bold")
raise SystemExit(1) raise SystemExit(1)
except (FileNotFoundError,) as exc: except (FileNotFoundError,) as exc:
_print_cuda_hint(exc, resolved_device) _print_cuda_hint(exc, requested_device)
console.print(f"Ошибка: {exc}", style="red bold") console.print(f"Ошибка: {exc}", style="red bold")
raise SystemExit(1) raise SystemExit(1)
except Exception as exc: except Exception as exc:
_print_cuda_hint(exc, resolved_device) _print_cuda_hint(exc, requested_device)
if verbose: if verbose:
console.print_exception() console.print_exception()
else: else:
@@ -423,13 +408,48 @@ def main(
raise SystemExit(1) raise SystemExit(1)
def _print_execution_header(info: ExecutionInfo, verbose: bool) -> None:
"""Печатает выбранное исполнение; в --verbose — версии runtime для диагностики."""
console.print(
f"Модель: [bold]{info.model}[/bold] "
f"Устройство: [bold]{info.device}[/bold] "
f"Compute: [bold]{info.compute_type}[/bold]"
)
if info.device == "openvino-gpu" and info.model != "large-v3":
console.print(
"Совет: --model large-v3 даёт лучшее качество на GPU (~2x дольше)",
style="dim",
)
if info.device != info.resolved_device:
console.print(
f"Запрошено {info.requested_device}, используется {info.device}",
style="yellow",
)
if verbose:
threads = info.cpu_threads or "по умолчанию библиотеки"
console.print(f"Движок: {info.engine} Потоки: {threads}", style="dim")
for key, value in info.runtime.items():
console.print(f" {key}: {value}", style="dim", markup=False)
def _load_diarizer(
diarize: bool, speakers: int | None, cpu_threads: int
) -> SpeakerDiarizer | None:
"""Готовит диаризатор до первого ASR; пословный контракт уже проверен module."""
if not diarize:
return None
return load_speaker_diarizer(
speakers=speakers,
threads=cpu_threads,
on_status=lambda message: console.print(message),
)
def _run_single( def _run_single(
file: Path, file: Path,
defaults: ResolvedConfig, request: ExecutionRequest,
output: Path | None, output: Path | None,
verbose: bool, verbose: bool,
compute_type_explicit: bool = False,
cpu_threads: int = 0,
diarize: bool = False, diarize: bool = False,
speakers: int | None = None, speakers: int | None = None,
) -> None: ) -> None:
@@ -437,9 +457,6 @@ def _run_single(
start = time.monotonic() start = time.monotonic()
validated_file = validate_input_file(file) validated_file = validate_input_file(file)
requested_device = defaults["device"]
resolved_device = detect_device(requested_device)
strict = requested_device != "auto"
output_path = build_output_path(validated_file, output) output_path = build_output_path(validated_file, output)
console.print(f"Файл: [bold]{validated_file.name}[/bold]") console.print(f"Файл: [bold]{validated_file.name}[/bold]")
@@ -447,59 +464,18 @@ def _run_single(
def on_segment(seg: Segment) -> None: def on_segment(seg: Segment) -> None:
console.print(f" [{seg.start:.2f}s] {seg.text.strip()}") console.print(f" [{seg.start:.2f}s] {seg.text.strip()}")
model_obj, actual_device, backend, model_path = load_model( transcriber = Transcriber(request)
defaults["model"], info = transcriber.prepare(on_status=lambda msg: console.print(msg))
resolved_device, _print_execution_header(info, verbose)
defaults["compute_type"], speaker_diarizer = _load_diarizer(diarize, speakers, request.cpu_threads)
on_status=lambda msg: console.print(msg),
strict_device=strict,
compute_type_explicit=compute_type_explicit,
cpu_threads=cpu_threads,
)
actual_ct = (
getattr(backend, "actual_compute_type", defaults["compute_type"])
or defaults["compute_type"]
)
console.print(
f"Модель: [bold]{defaults['model']}[/bold] "
f"Устройство: [bold]{actual_device}[/bold] "
f"Compute: [bold]{actual_ct}[/bold]"
)
if actual_device == "openvino-gpu" and defaults["model"] != "large-v3":
console.print(
"Совет: --model large-v3 даёт лучшее качество на GPU (~2x дольше)",
style="dim",
)
speaker_diarizer = None
if diarize:
if not backend.word_timestamps_available:
raise ValueError(
"Выбранный движок или модель не поддерживает пословные таймкоды"
)
speaker_diarizer = load_speaker_diarizer(
speakers=speakers,
threads=cpu_threads,
on_status=lambda message: console.print(message),
)
with Status("Подготавливаю запуск...", console=console) as status: with Status("Подготавливаю запуск...", console=console) as status:
tfr = _transcribe_file( result = transcriber.transcribe(
model=model_obj, validated_file,
actual_device=actual_device,
backend=backend,
model_path=model_path,
file_path=validated_file,
model_name=defaults["model"],
compute_type=defaults["compute_type"],
language=defaults["language"] if defaults["language"] != "auto" else None,
on_segment=on_segment if verbose else None, on_segment=on_segment if verbose else None,
on_status=status.update, on_status=status.update,
strict_device=strict,
cpu_threads=cpu_threads,
) )
result = tfr.result
speaker_transcript = None speaker_transcript = None
diarization_warning = None diarization_warning = None
diarization_degraded = False diarization_degraded = False
@@ -526,18 +502,12 @@ def _run_single(
if diarization_warning is not None: if diarization_warning is not None:
console.print(f"Внимание: {diarization_warning}", style="yellow") console.print(f"Внимание: {diarization_warning}", style="yellow")
if tfr.actual_device != resolved_device: execution = transcriber.execution
if requested_device == "auto": if execution.device != info.device:
console.print( console.print(
f"Определено устройство {resolved_device}, " f"Запрошено {execution.requested_device}, использовано {execution.device}",
f"но использовано {tfr.actual_device} (fallback)", style="yellow",
style="yellow", )
)
else:
console.print(
f"Запрошено {requested_device}, использовано {tfr.actual_device}",
style="yellow",
)
if len(result.segments) == 0: if len(result.segments) == 0:
message = f"Речь не обнаружена в файле {validated_file.name}" message = f"Речь не обнаружена в файле {validated_file.name}"
@@ -545,14 +515,13 @@ def _run_single(
message += "; диаризация не запускалась" message += "; диаризация не запускалась"
console.print(message, style="yellow") console.print(message, style="yellow")
device_info = _format_device_info(result.device_used) language_mode = _format_language_mode(request.language or "auto", result)
language_mode = _format_language_mode(defaults["language"], result)
content = format_transcript( content = format_transcript(
result=result, result=result,
source_filename=validated_file.name, source_filename=validated_file.name,
model_name=defaults["model"], model_name=execution.model,
device_info=device_info, device_info=execution.description,
language_mode=language_mode, language_mode=language_mode,
speaker_transcript=speaker_transcript, speaker_transcript=speaker_transcript,
diarization_warning=diarization_warning, diarization_warning=diarization_warning,
@@ -569,11 +538,9 @@ def _run_single(
def _run_batch( def _run_batch(
files: list[Path], files: list[Path],
defaults: ResolvedConfig, request: ExecutionRequest,
verbose: bool, verbose: bool,
force: bool, force: bool,
compute_type_explicit: bool = False,
cpu_threads: int = 0,
diarize: bool = False, diarize: bool = False,
speakers: int | None = None, speakers: int | None = None,
) -> None: ) -> None:
@@ -603,50 +570,11 @@ def _run_batch(
raise SystemExit(1) raise SystemExit(1)
return return
# Phase 2: Load model (ensure + create в одном вызове) # Phase 2: module выполнения создаётся, когда есть работа
requested_device = defaults["device"] transcriber = Transcriber(request)
resolved_device = detect_device(requested_device) info = transcriber.prepare(on_status=lambda msg: console.print(msg))
strict = requested_device != "auto" _print_execution_header(info, verbose)
model_obj, actual_device, backend, model_path = load_model( speaker_diarizer = _load_diarizer(diarize, speakers, request.cpu_threads)
defaults["model"],
resolved_device,
defaults["compute_type"],
on_status=lambda msg: console.print(msg),
strict_device=strict,
compute_type_explicit=compute_type_explicit,
cpu_threads=cpu_threads,
)
if actual_device == "openvino-gpu" and defaults["model"] != "large-v3":
console.print(
"Совет: --model large-v3 даёт лучшее качество на GPU (~2x дольше)",
style="dim",
)
if actual_device != resolved_device:
if requested_device == "auto":
console.print(
f"Определено устройство {resolved_device}, "
f"но используется {actual_device} (fallback)",
style="yellow",
)
else:
console.print(
f"Запрошено {requested_device}, используется {actual_device}",
style="yellow",
)
speaker_diarizer = None
if diarize:
if not backend.word_timestamps_available:
raise ValueError(
"Выбранный движок или модель не поддерживает пословные таймкоды"
)
speaker_diarizer = load_speaker_diarizer(
speakers=speakers,
threads=cpu_threads,
on_status=lambda message: console.print(message),
)
# Phase 3: Transcribe # Phase 3: Transcribe
processed = 0 processed = 0
@@ -663,39 +591,24 @@ def _run_batch(
def on_segment(seg: Segment) -> None: def on_segment(seg: Segment) -> None:
console.print(f" [{seg.start:.2f}s] {seg.text.strip()}") console.print(f" [{seg.start:.2f}s] {seg.text.strip()}")
device_before = transcriber.execution.device
with Status(f"{prefix}...", console=console) as status: with Status(f"{prefix}...", console=console) as status:
tfr = _transcribe_file( result = transcriber.transcribe(
model=model_obj, file,
actual_device=actual_device,
backend=backend,
model_path=model_path,
file_path=file,
model_name=defaults["model"],
compute_type=defaults["compute_type"],
language=defaults["language"]
if defaults["language"] != "auto"
else None,
on_segment=on_segment if verbose else None, on_segment=on_segment if verbose else None,
on_status=status.update on_status=status.update
if not verbose if not verbose
else lambda msg: console.print(msg), else lambda msg: console.print(msg),
strict_device=strict,
cpu_threads=cpu_threads,
) )
if tfr.actual_device != actual_device: execution = transcriber.execution
if execution.device != device_before:
console.print( console.print(
f" {file.name}: fallback на {tfr.actual_device} при транскрипции", f" {file.name}: fallback на {execution.device} при транскрипции",
style="yellow", style="yellow",
) )
# Обновляем после возможного mid-stream fallback
model_obj = tfr.model
actual_device = tfr.actual_device
backend = tfr.backend
model_path = tfr.model_path
result = tfr.result language_mode = _format_language_mode(request.language or "auto", result)
language_mode = _format_language_mode(defaults["language"], result)
speaker_transcript = None speaker_transcript = None
diarization_warning = None diarization_warning = None
file_degraded = False file_degraded = False
@@ -735,13 +648,11 @@ def _run_batch(
message += "; диаризация не запускалась" message += "; диаризация не запускалась"
console.print(message, style="yellow") console.print(message, style="yellow")
device_info = _format_device_info(result.device_used)
content = format_transcript( content = format_transcript(
result=result, result=result,
source_filename=file.name, source_filename=file.name,
model_name=defaults["model"], model_name=execution.model,
device_info=device_info, device_info=execution.description,
language_mode=language_mode, language_mode=language_mode,
speaker_transcript=speaker_transcript, speaker_transcript=speaker_transcript,
diarization_warning=diarization_warning, diarization_warning=diarization_warning,
@@ -760,7 +671,7 @@ def _run_batch(
except KeyboardInterrupt: except KeyboardInterrupt:
raise raise
except Exception as exc: except Exception as exc:
_print_cuda_hint(exc, actual_device) _print_cuda_hint(exc, request.device)
if verbose: if verbose:
console.print_exception() console.print_exception()
else: else:
-20
View File
@@ -171,23 +171,3 @@ def _resolve_value(cli_value: _T | None, config_value: _T | None, default: _T) -
if config_value is not None: if config_value is not None:
return config_value return config_value
return default return default
def apply_device_defaults(
defaults: ResolvedConfig,
resolved_device: str,
cli_values: CliValues,
config: ConfigValues,
) -> ResolvedConfig:
"""Применяет device-aware дефолты для model и compute_type,
если они не были явно заданы через CLI или конфиг."""
device_defs = DEVICE_DEFAULTS.get(resolved_device, {})
if not device_defs:
return defaults
result = dict(defaults)
for key in ("model", "compute_type"):
if cli_values.get(key) is None and key not in config:
if key in device_defs:
result[key] = device_defs[key]
return cast(ResolvedConfig, result)
+1 -118
View File
@@ -10,9 +10,8 @@ from local_transcriber.backends import get_backend
from local_transcriber.config import DEVICE_DEFAULTS, HARDCODED_DEFAULTS from local_transcriber.config import DEVICE_DEFAULTS, HARDCODED_DEFAULTS
# Re-export из types.py для обратной совместимости # Re-export из types.py для обратной совместимости
from local_transcriber.types import ( from local_transcriber.types import ( # noqa: F401
Segment, Segment,
TranscribeFileResult,
TranscribeResult, TranscribeResult,
WordTimestampsUnavailableError, WordTimestampsUnavailableError,
) )
@@ -228,120 +227,6 @@ def _describe_device(device: str) -> str:
return "CPU" return "CPU"
def load_model(
model_name: str,
device: str,
compute_type: str,
on_status: Callable[[str], None] | None = None,
strict_device: bool = False,
compute_type_explicit: bool = False,
cpu_threads: int = 0,
) -> tuple[Any, str, Any, str]:
"""Загружает модель: ensure + create с fallback.
Возвращает (model, actual_device, backend, model_path).
compute_type_explicit: True если пользователь явно указал --compute-type.
cpu_threads: число потоков для CPU inference (0 = дефолт библиотеки).
"""
backend = get_backend(device, compute_type_explicit=compute_type_explicit)
actual_device = device
model_path = backend.ensure_model_available(model_name, compute_type, on_status)
try:
_notify_status(on_status, f"Инициализирую модель на {device}...")
model = backend.create_model(
model_path, device, compute_type, cpu_threads=cpu_threads
)
# Резолвим actual_device по реальному OpenVINO device
ov_dev = getattr(backend, "actual_ov_device", None)
if ov_dev == "GPU" and actual_device != "openvino-gpu":
actual_device = "openvino-gpu"
elif (
ov_dev == "CPU"
and actual_device.startswith("openvino")
and actual_device != "openvino-cpu"
):
actual_device = "openvino-cpu"
except (RuntimeError, ValueError) as exc:
if device != "cpu" and _is_backend_error(exc, device):
if strict_device:
raise
warnings.warn(
f"Не удалось загрузить модель на {device}: {exc}. Переключение на CPU.",
stacklevel=2,
)
actual_device = "cpu"
backend = get_backend("cpu")
model_path = backend.ensure_model_available(
model_name, compute_type, on_status
)
_notify_status(on_status, "Инициализирую модель на cpu...")
model = backend.create_model(
model_path, "cpu", compute_type, cpu_threads=cpu_threads
)
else:
raise
return model, actual_device, backend, model_path
def _transcribe_file(
model: Any,
actual_device: str,
backend: Any,
model_path: str,
file_path: Path,
model_name: str,
compute_type: str,
language: str | None = None,
on_segment: Callable[[Segment], None] | None = None,
on_status: Callable[[str], None] | None = None,
strict_device: bool = False,
cpu_threads: int = 0,
) -> TranscribeFileResult:
"""Транскрибирует один файл. При mid-stream fallback перезагружает модель."""
lang_arg = language if language and language != "auto" else None
try:
_notify_status(on_status, "Транскрибирую...")
result = backend.transcribe(model, file_path, lang_arg, on_segment, on_status)
result.device_used = actual_device
except (RuntimeError, ValueError) as exc:
if actual_device != "cpu" and _is_backend_error(exc, actual_device):
if strict_device:
raise
warnings.warn(
f"Ошибка при транскрипции на {actual_device}: {exc}. "
"Переключение на CPU и повтор.",
stacklevel=2,
)
actual_device = "cpu"
backend = get_backend("cpu")
model_path = backend.ensure_model_available(
model_name, compute_type, on_status
)
_notify_status(on_status, "Инициализирую модель на cpu...")
model = backend.create_model(
model_path, "cpu", compute_type, cpu_threads=cpu_threads
)
_notify_status(on_status, "Транскрибирую...")
result = backend.transcribe(
model, file_path, lang_arg, on_segment, on_status
)
result.device_used = actual_device
else:
raise
return TranscribeFileResult(
result=result,
model=model,
actual_device=actual_device,
backend=backend,
model_path=model_path,
)
def transcribe( def transcribe(
file_path: Path, file_path: Path,
model_name: str | None = None, model_name: str | None = None,
@@ -377,8 +262,6 @@ def ensure_model_available(
on_status: Callable[[str], None] | None = None, on_status: Callable[[str], None] | None = None,
) -> str: ) -> str:
"""Публичный helper: гарантирует наличие модели для указанного бэкенда.""" """Публичный helper: гарантирует наличие модели для указанного бэкенда."""
from local_transcriber.config import DEVICE_DEFAULTS, HARDCODED_DEFAULTS
if compute_type is None: if compute_type is None:
device_defs = DEVICE_DEFAULTS.get(device, {}) device_defs = DEVICE_DEFAULTS.get(device, {})
compute_type = device_defs.get( compute_type = device_defs.get(
-10
View File
@@ -2,7 +2,6 @@
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any
# Единый признак «язык неизвестен» для всех бэкендов # Единый признак «язык неизвестен» для всех бэкендов
UNKNOWN_LANGUAGE = "unknown" UNKNOWN_LANGUAGE = "unknown"
@@ -83,14 +82,5 @@ class TranscribeResult:
words: list[Word] = field(default_factory=list) words: list[Word] = field(default_factory=list)
@dataclass
class TranscribeFileResult:
result: TranscribeResult
model: Any # backend-specific model handle
actual_device: str
backend: Any = None # backend instance (для переиспользования в батче)
model_path: str = "" # путь к модели (меняется при cross-backend fallback)
StatusCallback = Callable[[str], None] | None StatusCallback = Callable[[str], None] | None
SegmentCallback = Callable[[Segment], None] | None SegmentCallback = Callable[[Segment], None] | None
-1
View File
@@ -93,7 +93,6 @@ def test_runtime_info_reports_versions_and_cuda_devices(monkeypatch):
info = backend.runtime_info() info = backend.runtime_info()
assert info["engine"] == "faster-whisper"
assert info["ctranslate2"] assert info["ctranslate2"]
assert info["faster_whisper"] assert info["faster_whisper"]
assert info["cuda_devices"] == "0" assert info["cuda_devices"] == "0"
-1
View File
@@ -476,7 +476,6 @@ def test_runtime_info_reports_available_and_actual_devices(monkeypatch):
info = backend.runtime_info() info = backend.runtime_info()
assert info["engine"] == "openvino"
assert info["openvino"] assert info["openvino"]
assert info["available_devices"] == "CPU, GPU" assert info["available_devices"] == "CPU, GPU"
assert info["device"] == "GPU" assert info["device"] == "GPU"
+229 -906
View File
File diff suppressed because it is too large. Load diff
-73
View File
@@ -3,7 +3,6 @@ from unittest.mock import patch
import pytest import pytest
from local_transcriber.config import ( from local_transcriber.config import (
apply_device_defaults,
find_config_file, find_config_file,
load_config, load_config,
resolve_defaults, resolve_defaults,
@@ -136,54 +135,6 @@ def test_resolve_defaults_diarize_priority(cli_value, config_value, expected):
assert resolve_defaults(cli, config)["diarize"] is expected assert resolve_defaults(cli, config)["diarize"] is expected
def test_apply_device_defaults_cuda():
defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = apply_device_defaults(defaults, "cuda", cli, {})
assert result["model"] == "medium"
assert result["compute_type"] == "float16"
def test_apply_device_defaults_cpu():
defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = apply_device_defaults(defaults, "cpu", cli, {})
assert result["model"] == "medium"
assert result["compute_type"] == "float32"
def test_apply_device_defaults_onnx_uses_readable_model():
defaults = {
"model": "medium",
"language": "ru",
"device": "auto",
"compute_type": "float32",
}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = apply_device_defaults(defaults, "onnx", cli, {})
assert result["model"] == "gigaam-v3-e2e-rnnt"
assert result["compute_type"] == "int8"
def test_apply_device_defaults_cli_overrides():
defaults = {"model": "large-v3", "language": "ru", "device": "auto", "compute_type": "int8"}
cli = {"model": "large-v3", "language": None, "device": None, "compute_type": "int8"}
result = apply_device_defaults(defaults, "cuda", cli, {})
assert result["model"] == "large-v3"
assert result["compute_type"] == "int8"
def test_apply_device_defaults_config_overrides():
defaults = {"model": "small", "language": "ru", "device": "auto", "compute_type": "int8"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
config = {"model": "small", "compute_type": "int8"}
result = apply_device_defaults(defaults, "cuda", cli, config)
assert result["model"] == "small"
assert result["compute_type"] == "int8"
def test_load_config_openvino_device(tmp_path): def test_load_config_openvino_device(tmp_path):
config = tmp_path / "config.toml" config = tmp_path / "config.toml"
config.write_text('device = "openvino"\n') config.write_text('device = "openvino"\n')
@@ -203,27 +154,3 @@ def test_load_config_openvino_cpu_device(tmp_path):
config.write_text('device = "openvino-cpu"\n') config.write_text('device = "openvino-cpu"\n')
result = load_config(config) result = load_config(config)
assert result == {"device": "openvino-cpu"} assert result == {"device": "openvino-cpu"}
def test_apply_device_defaults_openvino():
defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = apply_device_defaults(defaults, "openvino", cli, {})
assert result["model"] == "medium"
assert result["compute_type"] == "int8"
def test_apply_device_defaults_openvino_gpu():
defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = apply_device_defaults(defaults, "openvino-gpu", cli, {})
assert result["model"] == "medium"
assert result["compute_type"] == "int8"
def test_apply_device_defaults_openvino_cpu():
defaults = {"model": "medium", "language": "ru", "device": "auto", "compute_type": "float32"}
cli = {"model": None, "language": None, "device": None, "compute_type": None}
result = apply_device_defaults(defaults, "openvino-cpu", cli, {})
assert result["model"] == "medium"
assert result["compute_type"] == "int8"
-1
View File
@@ -706,7 +706,6 @@ class TestRuntimeInfo:
info = backend.runtime_info() info = backend.runtime_info()
assert info["engine"] == "onnx-asr"
assert info["onnxruntime"] assert info["onnxruntime"]
assert info["onnx_asr"] assert info["onnx_asr"]
assert info["available_providers"] == "CoreMLExecutionProvider, CPUExecutionProvider" assert info["available_providers"] == "CoreMLExecutionProvider, CPUExecutionProvider"
+82 -294
View File
@@ -8,10 +8,8 @@ from local_transcriber.transcriber import (
Segment, Segment,
Transcriber, Transcriber,
TranscribeResult, TranscribeResult,
_transcribe_file,
cuda_error_hint, cuda_error_hint,
ensure_model_available, ensure_model_available,
load_model,
transcribe, transcribe,
) )
from local_transcriber.types import WordTimestampsUnavailableError from local_transcriber.types import WordTimestampsUnavailableError
@@ -297,101 +295,6 @@ def test_transcribe_strict_cuda_error_during_transcription(mock_get_backend):
) )
# === load_model() tests ===
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_cuda_fallback(mock_get_backend):
cuda_backend = _make_backend(create_model_error=RuntimeError("CUDA out of memory"))
cpu_model = MagicMock()
cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model")
def backend_for_device(device, **kwargs):
return cuda_backend if device == "cuda" else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
model, actual_device, backend, model_path = load_model("tiny", "cuda", "int8")
assert actual_device == "cpu"
assert model is cpu_model
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_strict_raises(mock_get_backend):
backend = _make_backend(create_model_error=RuntimeError("CUDA out of memory"))
mock_get_backend.return_value = backend
with pytest.raises(RuntimeError, match="CUDA out of memory"):
load_model("tiny", "cuda", "int8", strict_device=True)
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_returns_backend_and_path(mock_get_backend):
backend = _make_backend(model_path="/mock/model/path")
mock_get_backend.return_value = backend
model, actual_device, returned_backend, model_path = load_model(
"tiny", "cpu", "int8"
)
assert returned_backend is backend
assert model_path == "/mock/model/path"
assert actual_device == "cpu"
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_passes_cpu_threads_to_backend(mock_get_backend):
backend = _make_backend(model_path="/mock/model/path")
mock_get_backend.return_value = backend
load_model("tiny", "cpu", "int8", cpu_threads=8)
assert backend.create_model.call_args.kwargs["cpu_threads"] == 8
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_fallback_preserves_cpu_threads(mock_get_backend):
cuda_backend = _make_backend(create_model_error=RuntimeError("CUDA out of memory"))
cpu_model = MagicMock()
cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model")
def backend_for_device(device, **kwargs):
return cuda_backend if device == "cuda" else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
load_model("tiny", "cuda", "int8", cpu_threads=6)
assert cuda_backend.create_model.call_args.kwargs["cpu_threads"] == 6
assert cpu_backend.create_model.call_args.kwargs["cpu_threads"] == 6
# === _transcribe_file() tests ===
def test__transcribe_file_basic():
result_data = _make_result(count=2)
backend = _make_backend(transcribe_result=result_data)
tfr = _transcribe_file(
model=MagicMock(),
actual_device="cpu",
backend=backend,
model_path="/mock/model",
file_path=Path("test.mp3"),
model_name="tiny",
compute_type="int8",
)
assert len(tfr.result.segments) == 2
assert tfr.actual_device == "cpu"
assert tfr.backend is backend
assert tfr.model_path == "/mock/model"
# === ensure_model_available() tests (через FasterWhisperBackend) === # === ensure_model_available() tests (через FasterWhisperBackend) ===
@@ -494,202 +397,6 @@ def test_ensure_model_available_rejects_incomplete_local_directory(tmp_path):
ensure_model_available(str(model_dir)) ensure_model_available(str(model_dir))
# === Cross-backend fallback (openvino → cpu) ===
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_openvino_gpu_fallback_to_cpu(mock_get_backend):
"""OpenVINO GPU ошибка при init → fallback на CPU (FasterWhisper)."""
ov_backend = _make_backend(
create_model_error=RuntimeError("OpenVINO model load failed"),
)
cpu_model = MagicMock()
cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model")
def backend_for_device(device, **kwargs):
return ov_backend if device == "openvino-gpu" else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
model, actual_device, backend, model_path = load_model(
"medium",
"openvino-gpu",
"fp16",
)
assert actual_device == "cpu"
assert model is cpu_model
assert backend is cpu_backend
assert model_path == "/mock/cpu/model"
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_openvino_cpu_fallback_to_cpu(mock_get_backend):
"""OpenVINO CPU ошибка при init → fallback на CPU (FasterWhisper)."""
ov_backend = _make_backend(
create_model_error=RuntimeError("OpenVINO model load failed"),
)
cpu_model = MagicMock()
cpu_backend = _make_backend(model=cpu_model, model_path="/mock/cpu/model")
def backend_for_device(device, **kwargs):
return ov_backend if device == "openvino-cpu" else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
model, actual_device, backend, model_path = load_model(
"medium",
"openvino-cpu",
"int8",
)
assert actual_device == "cpu"
assert model is cpu_model
assert backend is cpu_backend
assert model_path == "/mock/cpu/model"
@patch("local_transcriber.transcriber.get_backend")
def test_transcribe_file_openvino_gpu_midstream_fallback(mock_get_backend):
"""OpenVINO GPU ошибка при транскрипции → fallback на CPU."""
ov_backend = _make_backend(
transcribe_error=RuntimeError("OpenVINO inference error"),
)
cpu_backend = _make_backend(
transcribe_result=_make_result(count=2, device_used="cpu"),
model_path="/mock/cpu/model",
)
def backend_for_device(device, **kwargs):
return ov_backend if device.startswith("openvino") else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
tfr = _transcribe_file(
model=MagicMock(),
actual_device="openvino-gpu",
backend=ov_backend,
model_path="/mock/ov/model",
file_path=Path("test.mp3"),
model_name="medium",
compute_type="fp16",
)
assert tfr.actual_device == "cpu"
assert tfr.backend is cpu_backend
assert tfr.model_path == "/mock/cpu/model"
@patch("local_transcriber.transcriber.get_backend")
def test_transcribe_file_does_not_fallback_for_missing_word_timestamps(
mock_get_backend,
):
ov_backend = _make_backend(
transcribe_error=WordTimestampsUnavailableError("нет таймкодов"),
)
with pytest.raises(WordTimestampsUnavailableError, match="нет таймкодов"):
_transcribe_file(
model=MagicMock(),
actual_device="openvino-gpu",
backend=ov_backend,
model_path="/mock/ov/model",
file_path=Path("test.mp3"),
model_name="medium",
compute_type="fp16",
)
mock_get_backend.assert_not_called()
@patch("local_transcriber.transcriber.get_backend")
def test_transcribe_file_midstream_fallback_preserves_cpu_threads(mock_get_backend):
ov_backend = _make_backend(
transcribe_error=RuntimeError("OpenVINO inference error"),
)
cpu_backend = _make_backend(
transcribe_result=_make_result(count=2, device_used="cpu"),
model_path="/mock/cpu/model",
)
def backend_for_device(device, **kwargs):
return ov_backend if device.startswith("openvino") else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
_transcribe_file(
model=MagicMock(),
actual_device="openvino-gpu",
backend=ov_backend,
model_path="/mock/ov/model",
file_path=Path("test.mp3"),
model_name="medium",
compute_type="fp16",
cpu_threads=6,
)
assert cpu_backend.create_model.call_args.kwargs["cpu_threads"] == 6
@patch("local_transcriber.transcriber.get_backend")
def test_openvino_gpu_strict_device_no_fallback(mock_get_backend):
"""strict_device=True + OpenVINO GPU ошибка → raise."""
backend = _make_backend(
create_model_error=RuntimeError("OpenVINO model load failed"),
)
mock_get_backend.return_value = backend
with pytest.raises(RuntimeError, match="OpenVINO"):
load_model("medium", "openvino-gpu", "fp16", strict_device=True)
@patch("local_transcriber.transcriber.get_backend")
def test_openvino_runtime_error_triggers_fallback(mock_get_backend):
"""Любой RuntimeError от OpenVINO бэкенда → fallback."""
ov_backend = _make_backend(
create_model_error=RuntimeError("Exception from src/inference/..."),
)
cpu_backend = _make_backend(model_path="/mock/cpu/model")
def backend_for_device(device, **kwargs):
return ov_backend if device.startswith("openvino") else cpu_backend
mock_get_backend.side_effect = backend_for_device
with pytest.warns(UserWarning, match="Переключение на CPU"):
_, actual_device, _, _ = load_model("medium", "openvino-cpu", "int8")
assert actual_device == "cpu"
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_resolves_openvino_gpu_device(mock_get_backend):
"""load_model обновляет actual_device по backend.actual_ov_device."""
backend = _make_backend()
backend.actual_ov_device = "GPU"
mock_get_backend.return_value = backend
_, actual_device, _, _ = load_model("medium", "openvino-cpu", "int8")
assert actual_device == "openvino-gpu"
@patch("local_transcriber.transcriber.get_backend")
def test_load_model_resolves_openvino_cpu_device(mock_get_backend):
"""load_model обновляет actual_device по backend.actual_ov_device."""
backend = _make_backend()
backend.actual_ov_device = "CPU"
mock_get_backend.return_value = backend
_, actual_device, _, _ = load_model("medium", "openvino-gpu", "fp16")
assert actual_device == "openvino-cpu"
def test_ensure_model_available_openvino_default_compute_type(): def test_ensure_model_available_openvino_default_compute_type():
"""ensure_model_available(device='openvino-cpu') без compute_type не падает.""" """ensure_model_available(device='openvino-cpu') без compute_type не падает."""
from local_transcriber.backends.openvino import OpenVINOBackend from local_transcriber.backends.openvino import OpenVINOBackend
@@ -775,7 +482,11 @@ def test_run_falls_back_to_cpu_at_load_and_keeps_state_for_next_file(
run = Transcriber( run = Transcriber(
ExecutionRequest( ExecutionRequest(
device="cuda", model="tiny", compute_type="int8", strict_device=False device="cuda",
model="tiny",
compute_type="int8",
strict_device=False,
cpu_threads=3,
) )
) )
with pytest.warns(UserWarning, match="Переключение на CPU"): with pytest.warns(UserWarning, match="Переключение на CPU"):
@@ -785,6 +496,7 @@ def test_run_falls_back_to_cpu_at_load_and_keeps_state_for_next_file(
assert first.device_used == "cpu" assert first.device_used == "cpu"
assert second.device_used == "cpu" assert second.device_used == "cpu"
assert cpu_backend.create_model.call_count == 1 assert cpu_backend.create_model.call_count == 1
assert cpu_backend.create_model.call_args.kwargs["cpu_threads"] == 3
assert run.execution.resolved_device == "cuda" assert run.execution.resolved_device == "cuda"
assert run.execution.device == "cpu" assert run.execution.device == "cpu"
assert run.execution.engine == "faster-whisper" assert run.execution.engine == "faster-whisper"
@@ -961,3 +673,79 @@ def test_public_transcribe_uses_the_same_execution_module(mock_get_backend):
) )
assert result.device_used == "onnx" assert result.device_used == "onnx"
assert len(result.segments) == 1 assert len(result.segments) == 1
@pytest.mark.parametrize(
("device", "ov_device", "gpu_name", "intel_name", "expected"),
[
("cpu", None, None, None, "CPU"),
("onnx", None, "RTX 3060", None, "ONNX (CPU)"),
("cuda", None, "RTX 3060", None, "CUDA (RTX 3060)"),
("cuda", None, None, None, "CUDA (Unknown GPU)"),
("openvino-gpu", "GPU", None, "Intel Arc 140T", "OpenVINO (Intel Arc 140T)"),
("openvino-gpu", "GPU", None, None, "OpenVINO (Intel GPU)"),
("openvino-cpu", "CPU", None, None, "OpenVINO (CPU)"),
],
)
@patch("local_transcriber.transcriber.get_backend")
def test_run_description_names_hardware_only_from_driver_data(
mock_get_backend, device, ov_device, gpu_name, intel_name, expected
):
"""Строка исполнения для шапки: ONNX не выдаётся за CPU faster-whisper, GPU — по драйверу."""
backend = _make_run_backend()
backend.actual_ov_device = ov_device
mock_get_backend.return_value = backend
with (
patch("local_transcriber.transcriber.get_gpu_name", return_value=gpu_name),
patch("local_transcriber.transcriber.get_intel_gpu_name", return_value=intel_name),
):
info = Transcriber(ExecutionRequest(device=device, model="m")).prepare()
assert info.description == expected
@patch("local_transcriber.transcriber.get_backend")
def test_run_openvino_runtime_error_falls_back_when_not_strict(mock_get_backend):
"""Любой RuntimeError OpenVINO считается ошибкой движка: не-strict запуск уходит на CPU."""
ov_backend = _make_run_backend(
create_model_error=RuntimeError("Exception from src/inference/src/core.cpp")
)
cpu_backend = _make_run_backend()
mock_get_backend.side_effect = lambda device, **_: (
ov_backend if device.startswith("openvino") else cpu_backend
)
run = Transcriber(
ExecutionRequest(device="openvino-gpu", model="medium", strict_device=False)
)
with pytest.warns(UserWarning, match="Переключение на CPU"):
result = run.transcribe(Path("a.mp3"))
assert result.device_used == "cpu"
assert run.execution.engine == "faster-whisper"
@patch("local_transcriber.transcriber.get_backend")
def test_run_openvino_strict_raises_without_fallback(mock_get_backend):
mock_get_backend.return_value = _make_run_backend(
create_model_error=RuntimeError("GPU plugin failed")
)
with pytest.raises(RuntimeError, match="GPU plugin"):
Transcriber(ExecutionRequest(device="openvino-gpu", model="medium")).prepare()
assert mock_get_backend.call_count == 1
@patch("local_transcriber.transcriber.get_backend")
def test_run_reports_openvino_gpu_chosen_for_cpu_request(mock_get_backend):
"""Фактическое устройство OpenVINO важнее запрошенного и в обратную сторону."""
backend = _make_run_backend()
backend.actual_ov_device = "GPU"
mock_get_backend.return_value = backend
info = Transcriber(ExecutionRequest(device="openvino-cpu", model="medium")).prepare()
assert info.resolved_device == "openvino-cpu"
assert info.device == "openvino-gpu"