transcribe_qwen3_asr.py ダウンロード/コピー
transcribe_qwen3_asr.py をダウンロード
transcribe_qwen3_asr.py
transcribe_qwen3_asr.py
1"""
2Qwen3-ASRコマンドライン音声文字起こしツール。
3
4このスクリプトは、Qwen3-ASRモデルを使用して音声ファイルを文字起こしするための
5コマンドラインインターフェースを提供します。
6入力ファイルごとに以下の3つの出力ファイルを生成します。
7
8 1. ``<name>-time.txt`` : タイムスタンプ付きのテキスト（ForcedAlignerが必要な場合）
9 2. ``<name>.txt`` : プレーンな文字起こしテキスト
10 3. ``<name>-info.txt`` : 実行時間、モデル、オーディオ、および結果のメタデータ
11
12ASRエンジンは ``ASRBackend`` クラスの背後に隔離されており、
13CLI、出力ライター、環境レポートを変更することなく、
14faster-whisper、openai-whisper、または他のエンジンを追加できるよう設計されています。
15
16.. seealso::
17 :doc:`transcribe_qwen3_asr_usage` （利用ガイドへの架空のリンク）
18"""
19
20from __future__ import annotations
21
22import argparse
23import glob
24import importlib.metadata
25import json
26import os
27import platform
28import subprocess
29import sys
30import tempfile
31import time
32import traceback
33from abc import ABC, abstractmethod
34from dataclasses import asdict, dataclass, field
35from datetime import datetime, timezone
36from pathlib import Path
37from typing import Any, Callable, Optional
38
39
40MODEL_ALIASES = {
41 "0.6B": "Qwen/Qwen3-ASR-0.6B",
42 "1.7B": "Qwen/Qwen3-ASR-1.7B",
43 "Qwen/Qwen3-ASR-0.6B": "Qwen/Qwen3-ASR-0.6B",
44 "Qwen/Qwen3-ASR-1.7B": "Qwen/Qwen3-ASR-1.7B",
45}
46
47LANGUAGE_ALIASES = {
48 "ja": "Japanese",
49 "japanese": "Japanese",
50 "en": "English",
51 "english": "English",
52 "zh": "Chinese",
53 "chinese": "Chinese",
54 "ko": "Korean",
55 "korean": "Korean",
56}
57
58
59@dataclass
60class Segment:
61 """文字起こしされたテキストのセグメントを表すデータクラス。
62
63 :param start: セグメントの開始時間（秒）。
64 :type start: Optional[float]
65 :param end: セグメントの終了時間（秒）。
66 :type end: Optional[float]
67 :param text: セグメント内のテキスト。
68 :type text: str
69 """
70 start: Optional[float]
71 end: Optional[float]
72 text: str
73
74
75@dataclass
76class ASRResult:
77 """ASRエンジンの文字起こし結果を表すデータクラス。
78
79 :param text: 文字起こしされた全テキスト。
80 :type text: str
81 :param language: 検出された言語コードまたは名称。指定がない場合はNone。
82 :type language: Optional[str]
83 :param segments: タイムスタンプ付きのテキストセグメントのリスト。デフォルトは空リスト。
84 :type segments: list[Segment]
85 :param engine_metadata: エンジン固有の追加メタデータ。デフォルトは空の辞書。
86 :type engine_metadata: dict[str, Any]
87 """
88 text: str
89 language: Optional[str] = None
90 segments: list[Segment] = field(default_factory=list)
91 engine_metadata: dict[str, Any] = field(default_factory=dict)
92
93
94@dataclass
95class OutputPaths:
96 """出力ファイルのパスを保持するデータクラス。
97
98 :param timestamped: タイムスタンプ付きテキストの出力パス。
99 :type timestamped: Path
100 :param plain: プレーンテキストの出力パス。
101 :type plain: Path
102 :param info: 情報JSONファイルの出力パス。
103 :type info: Path
104 """
105 timestamped: Path
106 plain: Path
107 info: Path
108
109
110class ASRBackend(ABC):
111 """ローカルASRエンジン用の共通インターフェース。
112
113 この抽象基底クラスは、さまざまなASRバックエンド（例: Qwen3-ASR、Whisperなど）が
114 共通のインターフェースを通じて使用できるようにするための基本的なメソッドを定義します。
115 新しいASRエンジンを追加する際は、このクラスを継承し、抽象メソッドを実装する必要があります。
116 """
117
118 backend_name = "unknown"
119
120 @abstractmethod
121 def load(self) -> None:
122 """1つ以上のファイルを処理する前に、一度モデルをロードします。
123
124 このメソッドは、モデルの初期化やリソースの準備を行います。
125 """
126
127 @abstractmethod
128 def transcribe(self, audio_path: str, language: Optional[str]) -> ASRResult:
129 """1つの音声ファイルを文字起こしし、エンジンに依存しない結果を返します。
130
131 :param audio_path: 文字起こしする音声ファイルへのパス。
132 :type audio_path: str
133 :param language: 音声の言語。指定がない場合は自動検出されます。
134 :type language: Optional[str]
135 :returns: 文字起こし結果を含む :class:`ASRResult` オブジェクト。
136 :rtype: ASRResult
137 """
138
139 @abstractmethod
140 def describe(self) -> dict[str, Any]:
141 """このバックエンドで使用されている設定を辞書形式で返します。
142
143 :returns: バックエンドの設定を含む辞書。
144 :rtype: dict[str, Any]
145 """
146
147
148class Qwen3ASRBackend(ASRBackend):
149 """Qwen3-ASRモデルを使用するASRバックエンドの実装。
150
151 このクラスは、Qwen3-ASRライブラリをラップし、
152 ``ASRBackend`` インターフェースに準拠した文字起こし機能を提供します。
153
154 :param model_name: 使用するQwen3-ASRモデルの名前またはエイリアス。
155 :type model_name: str
156 :param device: モデルを実行するデバイス（例: "cpu", "cuda", "cuda:0"）。
157 :type device: str
158 :param dtype_name: モデルのデータ型（例: "float32", "float16", "bfloat16"）。
159 :type dtype_name: str
160 :param timestamps: タイムスタンプ付きセグメントを生成するかどうか（True/False）。
161 :type timestamps: bool
162 :param aligner_name: タイムスタンプ生成に使用するアライナーモデルの名前。
163 :type aligner_name: str
164 :param max_inference_batch_size: 推論時の最大バッチサイズ。
165 :type max_inference_batch_size: int
166 :param max_new_tokens: 生成される新しいトークンの最大数。
167 :type max_new_tokens: int
168 :param attention: Attentionの実装タイプ（例: "auto", "sdpa", "eager", "flash_attention_2"）。
169 :type attention: str
170 """
171 backend_name = "qwen3-asr"
172
173 def __init__(
174 self,
175 model_name: str,
176 device: str,
177 dtype_name: str,
178 timestamps: bool,
179 aligner_name: str,
180 max_inference_batch_size: int,
181 max_new_tokens: int,
182 attention: str,
183 ) -> None:
184 """Qwen3ASRBackendを初期化します。
185
186 :param model_name: 使用するQwen3-ASRモデルの名前またはエイリアス。
187 :type model_name: str
188 :param device: モデルを実行するデバイス（例: "cpu", "cuda", "cuda:0"）。
189 :type device: str
190 :param dtype_name: モデルのデータ型（例: "float32", "float16", "bfloat16"）。
191 :type dtype_name: str
192 :param timestamps: タイムスタンプ付きセグメントを生成するかどうか（True/False）。
193 :type timestamps: bool
194 :param aligner_name: タイムスタンプ生成に使用するアライナーモデルの名前。
195 :type aligner_name: str
196 :param max_inference_batch_size: 推論時の最大バッチサイズ。
197 :type max_inference_batch_size: int
198 :param max_new_tokens: 生成される新しいトークンの最大数。
199 :type max_new_tokens: int
200 :param attention: Attentionの実装タイプ（例: "auto", "sdpa", "eager", "flash_attention_2"）。
201 :type attention: str
202 """
203 self.model_name = MODEL_ALIASES.get(model_name, model_name)
204 self.requested_device = device
205 self.requested_dtype = dtype_name
206 self.timestamps = timestamps
207 self.aligner_name = aligner_name
208 self.max_inference_batch_size = max_inference_batch_size
209 self.max_new_tokens = max_new_tokens
210 self.attention = attention
211 self.model = None
212 self.torch = None
213 self.device_map = ""
214 self.dtype = None
215 self.dtype_label = ""
216
217 def _resolve_device_and_dtype(self) -> None:
218 """要求されたデバイスとデータ型をPyTorchの実際の値に解決します。
219
220 このメソッドは、指定された ``device`` と ``dtype_name`` の文字列値を
221 PyTorchが認識するデバイス文字列とデータ型オブジェクトに変換します。
222 GPUの利用可能性やbfloat16のサポート状況に応じて最適な値を自動選択します。
223
224 :raises RuntimeError: CUDAが要求されたにもかかわらず利用できない場合。
225 :raises ValueError: サポートされていないdtypeが指定された場合、またはCPU上でfloat16が指定された場合。
226 """
227 import torch
228
229 self.torch = torch
230 device = self.requested_device.strip().lower()
231 if not device or device == "auto":
232 device = "cuda:0" if torch.cuda.is_available() else "cpu"
233 elif device == "cuda":
234 device = "cuda:0"
235
236 if device.startswith("cuda") and not torch.cuda.is_available():
237 raise RuntimeError("CUDAが要求されましたが、torch.cuda.is_available()がFalseです")
238
239 dtype_name = self.requested_dtype.strip().lower()
240 if dtype_name == "auto":
241 if device.startswith("cuda"):
242 dtype_name = "bfloat16" if torch.cuda.is_bf16_supported() else "float16"
243 else:
244 dtype_name = "float32"
245
246 dtype_table = {
247 "bfloat16": torch.bfloat16,
248 "bf16": torch.bfloat16,
249 "float16": torch.float16,
250 "fp16": torch.float16,
251 "float32": torch.float32,
252 "fp32": torch.float32,
253 }
254 if dtype_name not in dtype_table:
255 raise ValueError(f"サポートされていないdtype: {self.requested_dtype}")
256 if device == "cpu" and dtype_name in {"float16", "fp16"}:
257 raise ValueError("CPU上でのfloat16は推奨されません。--dtype float32を使用してください")
258
259 self.device_map = device
260 self.dtype = dtype_table[dtype_name]
261 self.dtype_label = str(self.dtype).replace("torch.", "")
262
263 def load(self) -> None:
264 """Qwen3-ASRモデルと必要に応じてForcedAlignerをロードします。
265
266 デバイスとデータ型を解決した後、指定された設定に基づいて
267 Qwen3ASRModelをHugging Face Hubからロードします。
268 タイムスタンプが有効な場合は、ForcedAlignerも初期化されます。
269 """
270 self._resolve_device_and_dtype()
271 from qwen_asr import Qwen3ASRModel
272
273 kwargs: dict[str, Any] = {
274 "dtype": self.dtype,
275 "device_map": self.device_map,
276 "max_inference_batch_size": self.max_inference_batch_size,
277 "max_new_tokens": self.max_new_tokens,
278 }
279 if self.attention != "auto":
280 kwargs["attn_implementation"] = self.attention
281
282 if self.timestamps:
283 kwargs["forced_aligner"] = self.aligner_name
284 aligner_kwargs: dict[str, Any] = {
285 "dtype": self.dtype,
286 "device_map": self.device_map,
287 }
288 if self.attention != "auto":
289 aligner_kwargs["attn_implementation"] = self.attention
290 kwargs["forced_aligner_kwargs"] = aligner_kwargs
291
292 print(f"Loading ASR model : {self.model_name}", flush=True)
293 if self.timestamps:
294 print(f"Loading aligner : {self.aligner_name}", flush=True)
295 print(f"Device / dtype : {self.device_map} / {self.dtype_label}", flush=True)
296 self.model = Qwen3ASRModel.from_pretrained(self.model_name, **kwargs)
297
298 @staticmethod
299 def _timestamp_item_to_segment(item: Any) -> Segment:
300 """タイムスタンプ付きのアイテムを :class:`Segment` オブジェクトに変換します。
301
302 Qwen3-ASRモデルが返すタイムスタンプ形式から、
303 一貫した :class:`Segment` データクラスに変換します。
304
305 :param item: タイムスタンプ情報を含む辞書またはオブジェクト。
306 :type item: Any
307 :returns: 変換された :class:`Segment` オブジェクト。
308 :rtype: Segment
309 """
310 if isinstance(item, dict):
311 text = str(item.get("text", ""))
312 start = item.get("start_time", item.get("start"))
313 end = item.get("end_time", item.get("end"))
314 else:
315 text = str(getattr(item, "text", ""))
316 start = getattr(item, "start_time", getattr(item, "start", None))
317 end = getattr(item, "end_time", getattr(item, "end", None))
318 return Segment(
319 start=float(start) if start is not None else None,
320 end=float(end) if end is not None else None,
321 text=text,
322 )
323
324 @classmethod
325 def _convert_timestamps(cls, raw: Any) -> list[Segment]:
326 """Qwen3-ASRの生のタイムスタンプ出力をSegmentのリストに変換します。
327
328 モデルの出力が複数のリストレベルを持つ場合があるため、それを平坦化し、
329 各タイムスタンプアイテムを :class:`Segment` オブジェクトに変換します。
330
331 :param raw: Qwen3-ASRから返された生のタイムスタンプデータ。
332 :type raw: Any
333 :returns: :class:`Segment` オブジェクトのリスト。
334 :rtype: list[Segment]
335 """
336 if raw is None:
337 return []
338 # 一部のバージョンでは、単一の入力に対して余分なリストレベルを返す場合があります。
339 while isinstance(raw, (list, tuple)) and len(raw) == 1 and isinstance(raw[0], (list, tuple)):
340 raw = raw[0]
341 if not isinstance(raw, (list, tuple)):
342 raw = [raw]
343 return [cls._timestamp_item_to_segment(item) for item in raw]
344
345 def transcribe(self, audio_path: str, language: Optional[str]) -> ASRResult:
346 """Qwen3-ASRモデルを使用して音声ファイルを文字起こしします。
347
348 :param audio_path: 文字起こしする音声ファイルへのパス。
349 :type audio_path: str
350 :param language: 音声の言語。指定がない場合は自動検出されます。
351 :type language: Optional[str]
352 :returns: 文字起こし結果を含む :class:`ASRResult` オブジェクト。
353 :rtype: ASRResult
354 :raises RuntimeError: モデルがロードされていない場合、またはQwen3-ASRが結果を返さなかった場合。
355 """
356 if self.model is None:
357 raise RuntimeError("モデルがロードされていません")
358 results = self.model.transcribe(
359 audio=audio_path,
360 language=language,
361 return_time_stamps=self.timestamps,
362 )
363 if not results:
364 raise RuntimeError("Qwen3-ASRが結果を返しませんでした")
365 raw = results[0]
366 text = str(getattr(raw, "text", ""))
367 detected_language = getattr(raw, "language", None)
368 raw_timestamps = getattr(raw, "time_stamps", None)
369 segments = self._convert_timestamps(raw_timestamps)
370 return ASRResult(
371 text=text,
372 language=str(detected_language) if detected_language is not None else None,
373 segments=segments,
374 engine_metadata={"timestamp_items": len(segments)},
375 )
376
377 def describe(self) -> dict[str, Any]:
378 """このバックエンドで使用されている設定を辞書形式で返します。
379
380 :returns: バックエンドの設定を含む辞書。
381 :rtype: dict[str, Any]
382 """
383 return {
384 "backend": self.backend_name,
385 "model": self.model_name,
386 "device": self.device_map,
387 "dtype": self.dtype_label,
388 "timestamps": self.timestamps,
389 "aligner": self.aligner_name if self.timestamps else None,
390 "attention": self.attention,
391 "max_inference_batch_size": self.max_inference_batch_size,
392 "max_new_tokens": self.max_new_tokens,
393 }
394
395
396def normalize_language(value: str) -> Optional[str]:
397 """指定された言語の文字列を標準形式に正規化します。
398
399 "auto"や空文字列はNoneに変換され、一般的なエイリアスは正式な名称に変換されます。
400
401 :param value: 正規化する言語文字列。
402 :type value: str
403 :returns: 正規化された言語文字列、または自動検出を意味するNone。
404 :rtype: Optional[str]
405 """
406 value = value.strip()
407 if not value or value.lower() == "auto":
408 return None
409 return LANGUAGE_ALIASES.get(value.lower(), value)
410
411
412def package_version(name: str) -> Optional[str]:
413 """指定されたパッケージのバージョンを取得します。
414
415 :param name: バージョンを取得するパッケージの名前。
416 :type name: str
417 :returns: パッケージのバージョン文字列、またはパッケージが見つからない場合はNone。
418 :rtype: Optional[str]
419 """
420 try:
421 return importlib.metadata.version(name)
422 except importlib.metadata.PackageNotFoundError:
423 return None
424
425
426def command_output(command: list[str]) -> Optional[str]:
427 """指定されたコマンドを実行し、その標準出力を取得します。
428
429 コマンドの実行に失敗した場合や標準出力が空の場合はNoneを返します。
430
431 :param command: 実行するコマンドとその引数のリスト。
432 :type command: list[str]
433 :returns: コマンドの標準出力文字列、またはエラー/空出力の場合はNone。
434 :rtype: Optional[str]
435 """
436 try:
437 completed = subprocess.run(
438 command,
439 stdout=subprocess.PIPE,
440 stderr=subprocess.PIPE,
441 text=True,
442 check=False,
443 )
444 value = completed.stdout.strip()
445 return value or None
446 except (OSError, ValueError):
447 return None
448
449
450def get_audio_duration(path: str) -> Optional[float]:
451 """指定された音声ファイルの長さを秒単位で取得します。
452
453 ffprobeコマンドを使用して音声ファイルのduration情報を解析します。
454
455 :param path: 音声ファイルへのパス。
456 :type path: str
457 :returns: 音声ファイルの長さ（秒）、または取得できない場合はNone。
458 :rtype: Optional[float]
459 """
460 value = command_output([
461 "ffprobe", "-v", "error", "-show_entries", "format=duration",
462 "-of", "default=noprint_wrappers=1:nokey=1", path,
463 ])
464 try:
465 return float(value) if value is not None else None
466 except ValueError:
467 return None
468
469
470def format_hms(seconds: Optional[float]) -> str:
471 """秒数をHH:MM:SS.ss形式の文字列にフォーマットします。
472
473 :param seconds: フォーマットする秒数。Noneの場合は"??:??:??.??"を返します。
474 :type seconds: Optional[float]
475 :returns: フォーマットされた時間文字列。
476 :rtype: str
477 """
478 if seconds === None:
479 return "??:??:??.??"
480 seconds = max(0.0, float(seconds))
481 hours = int(seconds // 3600)
482 minutes = int((seconds % 3600) // 60)
483 secs = seconds % 60
484 return f"{hours:02d}:{minutes:02d}:{secs:05.2f}"
485
486
487def collect_environment(torch_module: Any) -> dict[str, Any]:
488 """現在の実行環境に関する情報を収集します。
489
490 OS、Python、インストールされている主要なパッケージのバージョン、
491 FFmpegのバージョン、CUDAの利用可能性とGPUの詳細などを取得します。
492
493 :param torch_module: PyTorchモジュールへの参照。
494 :type torch_module: Any
495 :returns: 環境情報を格納した辞書。
496 :rtype: dict[str, Any]
497 """
498 info: dict[str, Any] = {
499 "timestamp_utc": datetime.now(timezone.utc).isoformat(),
500 "platform": platform.platform(),
501 "os": platform.system(),
502 "os_release": platform.release(),
503 "machine": platform.machine(),
504 "processor": platform.processor(),
505 "python": platform.python_version(),
506 "python_executable": sys.executable,
507 "packages": {
508 name: package_version(name)
509 for name in (
510 "qwen-asr", "torch", "transformers", "accelerate",
511 "huggingface-hub", "soundfile", "librosa",
512 )
513 },
514 "ffmpeg": command_output(["ffmpeg", "-version"]),
515 }
516 if info["ffmpeg"]:
517 info["ffmpeg"] = str(info["ffmpeg"]).splitlines()[0]
518
519 cuda_available = bool(torch_module.cuda.is_available())
520 cuda: dict[str, Any] = {
521 "available": cuda_available,
522 "torch_cuda_version": torch_module.version.cuda,
523 "cudnn_version": torch_module.backends.cudnn.version() if cuda_available else None,
524 "device_count": torch_module.cuda.device_count() if cuda_available else 0,
525 "devices": [],
526 }
527 if cuda_available:
528 for index in range(torch_module.cuda.device_count()):
529 prop = torch_module.cuda.get_device_properties(index)
530 cuda["devices"].append({
531 "index": index,
532 "name": prop.name,
533 "total_memory_gib": round(prop.total_memory / 1024**3, 3),
534 "compute_capability": f"{prop.major}.{prop.minor}",
535 })
536 info["cuda"] = cuda
537 return info
538
539
540def print_environment(info: dict[str, Any]) -> None:
541 """収集した環境情報をコンソールに表示します。
542
543 :param info: :func:`collect_environment` によって収集された環境情報の辞書。
544 :type info: dict[str, Any]
545 """
546 print("\n=== ランタイム環境 ===")
547 print(f"OS : {info['platform']}")
548 print(f"Python : {info['python']}")
549 print(f"実行可能パス : {info['python_executable']}")
550 print(f"qwen-asr : {info['packages']['qwen-asr']}")
551 print(f"PyTorch : {info['packages']['torch']}")
552 print(f"Transformers : {info['packages']['transformers']}")
553 print(f"CUDA利用可能 : {info['cuda']['available']}")
554 print(f"Torch CUDA : {info['cuda']['torch_cuda_version']}")
555 for gpu in info["cuda"]["devices"]:
556 print(
557 f"GPU {gpu['index']} : {gpu['name']} "
558 f"({gpu['total_memory_gib']:.2f} GiB, CC {gpu['compute_capability']})"
559 )
560
561
562def make_output_paths(input_path: str, args: argparse.Namespace) -> OutputPaths:
563 """入力ファイルに基づいて出力ファイルのパスを生成します。
564
565 CLI引数で特定の出力ファイル名が指定されている場合はそれを使用し、
566 そうでない場合は入力ファイルのステム名に基づいてデフォルトのパスを作成します。
567
568 :param input_path: 入力音声ファイルへのパス。
569 :type input_path: str
570 :param args: コマンドライン引数をパースしたオブジェクト。
571 :type args: argparse.Namespace
572 :returns: 生成された出力パスを含む :class:`OutputPaths` オブジェクト。
573 :rtype: OutputPaths
574 """
575 stem = Path(input_path).stem
576 return OutputPaths(
577 timestamped=Path(args.outfile1) if args.outfile1 else Path(f"{stem}-time.txt"),
578 plain=Path(args.outfile2) if args.outfile2 else Path(f"{stem}.txt"),
579 info=Path(args.outfile3) if args.outfile3 else Path(f"{stem}-info.txt"),
580 )
581
582
583def write_transcript_outputs(
584 paths: OutputPaths,
585 result: ASRResult,
586) -> None:
587 """文字起こしされたテキストファイルを書き込みます。
588
589 この関数は、タイムスタンプ付きのテキストファイルとプレーンテキストファイルの両方を生成します。
590 長時間の処理が中断された場合でも、中間結果がディスクに残るように、
591 チャンク処理の各完了後にも呼び出されます。
592
593 :param paths: 出力ファイルのパスを含む :class:`OutputPaths` オブジェクト。
594 :type paths: OutputPaths
595 :param result: 文字起こし結果を含む :class:`ASRResult` オブジェクト。
596 :type result: ASRResult
597 """
598 with paths.timestamped.open("w", encoding="utf-8") as handle:
599 if result.segments:
600 for seg in result.segments:
601 if seg.start is None and seg.end is None:
602 handle.write(f"{seg.text}\n")
603 else:
604 handle.write(
605 f"[{format_hms(seg.start)} - {format_hms(seg.end)}] {seg.text}\n"
606 )
607 else:
608 handle.write("[タイムスタンプ利用不可]\n")
609 handle.write(result.text)
610 if result.text and not result.text.endswith("\n"):
611 handle.write("\n")
612
613 with paths.plain.open("w", encoding="utf-8") as handle:
614 handle.write(result.text)
615 if result.text and not result.text.endswith("\n"):
616 handle.write("\n")
617
618
619def write_outputs(
620 paths: OutputPaths,
621 result: ASRResult,
622 report: dict[str, Any],
623) -> None:
624 """すべての出力ファイル（文字起こしテキストと情報JSON）を書き込みます。
625
626 :param paths: 出力ファイルのパスを含む :class:`OutputPaths` オブジェクト。
627 :type paths: OutputPaths
628 :param result: 文字起こし結果を含む :class:`ASRResult` オブジェクト。
629 :type result: ASRResult
630 :param report: 実行レポートデータを含む辞書。
631 :type report: dict[str, Any]
632 """
633 write_transcript_outputs(paths, result)
634
635 with paths.info.open("w", encoding="utf-8") as handle:
636 json.dump(report, handle, ensure_ascii=False, indent=2, default=str)
637 handle.write("\n")
638
639
640def transcribe_with_optional_chunks(
641 backend: ASRBackend,
642 audio_path: str,
643 language: Optional[str],
644 duration: Optional[float],
645 chunk_seconds: float,
646 chunk_overlap: float,
647 use_chunks: bool,
648 progress_callback: Optional[Callable[[ASRResult, dict[str, Any]], None]] = None,
649) -> ASRResult:
650 """アライナーの入力長に制限がある場合に、長い音声を分割して文字起こしします。
651
652 この関数は、指定された ``chunk_seconds`` に基づいて音声を小さなチャンクに分割し、
653 各チャンクをASRバックエンドで文字起こしします。
654 チャンク処理はバックエンドの外で行われるため、さまざまなASRエンジンで再利用可能です。
655 タイムスタンプは元のファイルのタイムラインにシフトバックされます。
656 チャンクの境界には、後の処理（例: LLMによるテキスト結合）で役立つマーカーが挿入されます。
657
658 :param backend: 使用するASRバックエンドの実装。
659 :type backend: ASRBackend
660 :param audio_path: 文字起こしする音声ファイルへのパス。
661 :type audio_path: str
662 :param language: 音声の言語。指定がない場合は自動検出されます。
663 :type language: Optional[str]
664 :param duration: 音声ファイルの全体の長さ（秒）。
665 :type duration: Optional[float]
666 :param chunk_seconds: 音声を分割するチャンクの秒数。0以下の場合、分割は行われません。
667 :type chunk_seconds: float
668 :param chunk_overlap: 各チャンクの前後で重複させる秒数。重複により境界での情報損失を防ぎます。
669 :type chunk_overlap: float
670 :param use_chunks: チャンク分割を使用するかどうか。Falseの場合、全音声を一度に処理します。
671 :type use_chunks: bool
672 :param progress_callback: チャンクが完了するたびに呼び出されるコールバック関数。
673 部分的なASRResultと進捗情報を引数に取ります。
674 :type progress_callback: Optional[Callable[[ASRResult, dict[str, Any]], None]]
675 :returns: 全てのチャンクを結合した最終的な文字起こし結果。
676 :rtype: ASRResult
677 :raises RuntimeError: FFmpegがインストールされていない場合、またはFFmpegによるチャンク作成に失敗した場合。
678 :raises ValueError: ``--chunk-overlap`` が無効な値の場合。
679 """
680 if not use_chunks or chunk_seconds <= 0 or duration is None or duration <= chunk_seconds:
681 return backend.transcribe(audio_path, language)
682
683 if command_output(["ffmpeg", "-version"]) is None:
684 raise RuntimeError("タイムスタンプアライメントのために長い音声を分割するにはffmpegが必要です")
685
686 if chunk_overlap < 0:
687 raise ValueError("--chunk-overlapは0以上である必要があります")
688 if chunk_overlap * 2 >= chunk_seconds:
689 raise ValueError("--chunk-overlapは--chunk-secondsの半分未満である必要があります")
690
691 parts: list[str] = []
692 segments: list[Segment] = []
693 detected_languages: list[str] = []
694 core_start = 0.0
695 chunk_index = 0
696
697 with tempfile.TemporaryDirectory(prefix="qwen3_asr_") as temp_dir:
698 while core_start < duration:
699 chunk_index += 1
700 core_end = min(duration, core_start + chunk_seconds)
701 actual_start = max(0.0, core_start - (chunk_overlap if chunk_index > 1 else 0.0))
702 actual_end = min(
703 duration,
704 core_end + (chunk_overlap if core_end < duration else 0.0),
705 )
706 length = actual_end - actual_start
707 chunk_path = os.path.join(temp_dir, f"chunk_{chunk_index:04d}.wav")
708 command = [
709 "ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
710 "-ss", str(actual_start), "-t", str(length), "-i", audio_path,
711 "-vn", "-ac", "1", "-ar", "16000", chunk_path,
712 ]
713 completed = subprocess.run(command, check=False)
714 if completed.returncode != 0:
715 raise RuntimeError(f"チャンク {chunk_index} の作成中にffmpegが失敗しました")
716
717 print(
718 f"チャンク {chunk_index}: 実際の時間 {format_hms(actual_start)} - "
719 f"{format_hms(actual_end)}、コア時間 {format_hms(core_start)} - "
720 f"{format_hms(core_end)}",
721 flush=True,
722 )
723 chunk_started = time.perf_counter()
724 chunk_result = backend.transcribe(chunk_path, language)
725 chunk_elapsed = time.perf_counter() - chunk_started
726 marker = (
727 f"[[ASR_CHUNK {chunk_index:04d} "
728 f"ACTUAL={format_hms(actual_start)}..{format_hms(actual_end)} "
729 f"CORE={format_hms(core_start)}..{format_hms(core_end)} "
730 f"OVERLAP={chunk_overlap:.1f}s]]"
731 )
732 if chunk_index > 1:
733 boundary = (
734 f"[[ASR_BOUNDARY {chunk_index - 1:04d}|{chunk_index:04d} "
735 f"OVERLAP={chunk_overlap:.1f}s; "
736 "LLM: この境界をまたがる重複または不完全なテキストを調整してください]]"
737 )
738 parts.append(boundary)
739 segments.append(Segment(None, None, boundary))
740 parts.append(marker)
741 segments.append(Segment(None, None, marker))
742 if chunk_result.text.strip():
743 parts.append(chunk_result.text.strip())
744 parts.append(f"[[/ASR_CHUNK {chunk_index:04d}]]")
745 if chunk_result.language:
746 detected_languages.append(chunk_result.language)
747 for seg in chunk_result.segments:
748 segments.append(Segment(
749 start=(seg.start + actual_start) if seg.start is not None else None,
750 end=(seg.end + actual_start) if seg.end is not None else None,
751 text=seg.text,
752 ))
753 segments.append(Segment(None, None, f"[[/ASR_CHUNK {chunk_index:04d}]]"))
754 core_start = core_end
755
756 progress_percent = 100.0 * core_end / duration
757 print(
758 f"\n=== チャンク {chunk_index} 完了: "
759 f"{progress_percent:.1f}% ({format_hms(core_end)} / {format_hms(duration)}) ==="
760 )
761 print(f"チャンク処理時間: {chunk_elapsed:.2f} 秒")
762 print(chunk_result.text.strip() or "[テキストなし]")
763 print("=== チャンク結果の終わり ===\n", flush=True)
764
765 if progress_callback is not None:
766 partial_result = ASRResult(
767 text="\n".join(parts),
768 language=detected_languages[0] if detected_languages else None,
769 segments=list(segments),
770 engine_metadata={
771 "timestamp_items": len(segments),
772 "audio_chunks_completed": chunk_index,
773 "processed_seconds": core_end,
774 "progress_percent": progress_percent,
775 "chunk_seconds": chunk_seconds,
776 "chunk_overlap_seconds": chunk_overlap,
777 "boundary_markers": True,
778 },
779 )
780 progress_callback(
781 partial_result,
782 {
783 "chunk_index": chunk_index,
784 "processed_seconds": core_end,
785 "duration_seconds": duration,
786 "progress_percent": progress_percent,
787 "chunk_elapsed_seconds": chunk_elapsed,
788 },
789 )
790
791 detected_language = detected_languages[0] if detected_languages else None
792 return ASRResult(
793 text="\n".join(parts),
794 language=detected_language,
795 segments=segments,
796 engine_metadata={
797 "timestamp_items": len(segments),
798 "audio_chunks": chunk_index,
799 "chunk_seconds": chunk_seconds,
800 "chunk_overlap_seconds": chunk_overlap,
801 "boundary_markers": True,
802 },
803 )
804
805
806def create_backend(args: argparse.Namespace) -> ASRBackend:
807 """コマンドライン引数に基づいてASRバックエンドインスタンスを作成します。
808
809 :param args: コマンドライン引数をパースしたオブジェクト。
810 :type args: argparse.Namespace
811 :returns: 初期化されたASRバックエンドのインスタンス。
812 :rtype: ASRBackend
813 """
814 factories = {
815 "qwen3-asr": lambda: Qwen3ASRBackend(
816 model_name=args.model,
817 device=args.device,
818 dtype_name=args.dtype,
819 timestamps=bool(args.timestamps),
820 aligner_name=args.aligner,
821 max_inference_batch_size=args.max_inference_batch_size,
822 max_new_tokens=args.max_new_tokens,
823 attention=args.attention,
824 )
825 }
826 return factories[args.backend]()
827
828
829def build_parser() -> argparse.ArgumentParser:
830 """コマンドライン引数をパースするためのArgumentParserを作成します。
831
832 利用可能なASRバックエンド、モデル、デバイス、データ型、チャンク設定など、
833 様々なオプションを定義します。
834
835 :returns: 設定済みの :class:`argparse.ArgumentParser` オブジェクト。
836 :rtype: argparse.ArgumentParser
837 """
838 parser = argparse.ArgumentParser(
839 description="Qwen3-ASR音声文字起こしツール（ASRバックエンド拡張対応）"
840 )
841 parser.add_argument("infile", help="入力音声ファイル名（glob可）")
842 parser.add_argument(
843 "--backend", choices=["qwen3-asr"], default="qwen3-asr",
844 help="ASRバックエンド (default: qwen3-asr)",
845 )
846 parser.add_argument(
847 "-m", "--model", default="0.6B",
848 choices=list(MODEL_ALIASES),
849 help="Qwen3-ASRモデル (default: 0.6B)",
850 )
851 parser.add_argument(
852 "-l", "--lang", default="ja",
853 help="言語コード/名称。autoまたは空文字で自動判定 (default: ja)",
854 )
855 parser.add_argument(
856 "-d", "--device", default="auto",
857 help="auto, cpu, cuda, cuda:0など (default: auto)",
858 )
859 parser.add_argument(
860 "--dtype", choices=["auto", "bfloat16", "float16", "float32"],
861 default="auto", help="推論精度 (default: auto)",
862 )
863 parser.add_argument(
864 "--timestamps", type=int, choices=[0, 1], default=1,
865 help="ForcedAlignerで時刻を付けるか (default: 1)",
866 )
867 parser.add_argument(
868 "--aligner", default="Qwen/Qwen3-ForcedAligner-0.6B",
869 help="時刻推定モデル",
870 )
871 parser.add_argument(
872 "--attention", choices=["auto", "sdpa", "eager", "flash_attention_2"],
873 default="auto", help="Attention実装 (default: auto)",
874 )
875 parser.add_argument(
876 "--max-inference-batch-size", type=int, default=1,
877 help="推論バッチ上限。小さい値はVRAMを節約 (default: 1)",
878 )
879 parser.add_argument(
880 "--max-new-tokens", type=int, default=4096,
881 help="生成トークン上限。長時間音声では大きくする (default: 4096)",
882 )
883 parser.add_argument(
884 "--chunk-seconds", type=float, default=240.0,
885 help="時刻付け時の長時間音声分割秒数。0で分割しない (default: 240)",
886 )
887 parser.add_argument(
888 "--chunk-overlap", type=float, default=10.0,
889 help="チャンク前後の重複秒数。境界は出力に明示 (default: 10)",
890 )
891 parser.add_argument("--outfile1", default="", help="時刻付き出力ファイル")
892 parser.add_argument("--outfile2", default="", help="本文出力ファイル")
893 parser.add_argument("--outfile3", default="", help="環境・設定情報JSONファイル")
894 parser.add_argument(
895 "--pause", type=int, choices=[0, 1], default=0,
896 help="終了時にENTERを待つか (default: 0)",
897 )
898 return parser
899
900
901def main() -> int:
902 """Qwen3-ASR音声文字起こしツールのメインエントリポイント。
903
904 コマンドライン引数をパースし、指定された音声ファイルを文字起こしします。
905 環境情報を収集・表示し、ASRモデルをロードして文字起こしを実行し、
906 結果を複数のファイルに出力します。
907 複数の入力ファイルが指定された場合、または長い音声ファイルがチャンク分割される場合、
908 それぞれ個別に処理されます。
909
910 :returns: 終了コード。成功時は0、エラー時は1、PyTorch未インストールの場合は2、
911 ユーザーによる中断時は130を返します。
912 :rtype: int
913 """
914 parser = build_parser()
915 args = parser.parse_args()
916 files = sorted(glob.glob(args.infile))
917 if not files:
918 parser.error(f"ファイルが見つかりません: {args.infile}")
919 if len(files) > 1 and (args.outfile1 or args.outfile2 or args.outfile3):
920 parser.error("複数の入力ファイルで明示的な出力ファイル名は使用できません")
921
922 try:
923 import torch
924 except ImportError:
925 print("エラー: PyTorchがこのPython環境にインストールされていません。", file=sys.stderr)
926 return 2
927
928 environment = collect_environment(torch)
929 print_environment(environment)
930 language = normalize_language(args.lang)
931 backend = create_backend(args)
932
933 load_started = time.perf_counter()
934 backend.load()
935 load_seconds = time.perf_counter() - load_started
936 print(f"モデルロード時間: {load_seconds:.2f} 秒", flush=True)
937
938 for input_path in files:
939 paths = make_output_paths(input_path, args)
940 duration = get_audio_duration(input_path)
941 file_size = os.path.getsize(input_path)
942
943 print("\n=== 文字起こし ===")
944 print(f"入力ファイル : {input_path}")
945 print(f"音声長さ : {format_hms(duration)}")
946 print(f"言語 : {language or '自動検出'}")
947 print(f"出力1 (時刻付き) : {paths.timestamped}")
948 print(f"出力2 (テキスト) : {paths.plain}")
949 print(f"出力3 (情報) : {paths.info}")
950
951 def save_partial_result(
952 partial_result: ASRResult,
953 progress: dict[str, Any],
954 ) -> None:
955 """チャンク処理中に中間結果を保存するためのコールバック関数。
956
957 :param partial_result: 部分的な文字起こし結果。
958 :type partial_result: ASRResult
959 :param progress: 進捗情報を含む辞書。
960 :type progress: dict[str, Any]
961 """
962 write_transcript_outputs(paths, partial_result)
963 print(
964 f"部分結果を保存しました: チャンク {progress['chunk_index']} "
965 f"({progress['progress_percent']:.1f}%) -> "
966 f"{paths.timestamped}, {paths.plain}",
967 flush=True,
968 )
969
970 started = time.perf_counter()
971 result = transcribe_with_optional_chunks(
972 backend=backend,
973 audio_path=input_path,
974 language=language,
975 duration=duration,
976 chunk_seconds=args.chunk_seconds,
977 chunk_overlap=args.chunk_overlap,
978 use_chunks=bool(args.timestamps),
979 progress_callback=save_partial_result,
980 )
981 elapsed = time.perf_counter() - started
982 realtime_factor = elapsed / duration if duration and duration > 0 else None
983
984 report = {
985 "environment": environment,
986 "backend": backend.describe(),
987 "input": {
988 "path": os.path.abspath(input_path),
989 "size_bytes": file_size,
990 "duration_seconds": duration,
991 "requested_language": language,
992 },
993 "result": {
994 "detected_language": result.language,
995 "characters": len(result.text),
996 "segments": len(result.segments),
997 "elapsed_seconds": elapsed,
998 "realtime_factor": realtime_factor,
999 **result.engine_metadata,
1000 },
1001 "outputs": {key: str(value.resolve()) for key, value in asdict(paths).items()},
1002 "model_load_seconds": load_seconds,
1003 }
1004 write_outputs(paths, result, report)
1005
1006 print(f"検出言語 : {result.language}")
1007 print(f"経過時間 : {elapsed:.2f} 秒")
1008 if realtime_factor is not None:
1009 print(f"リアルタイム係数 : {realtime_factor:.4f}x")
1010 print("\n=== 文字起こしテキスト ===")
1011 print(result.text, flush=True)
1012
1013 return 0
1014
1015
1016if __name__ == "__main__":
1017 exit_code = 1
1018 try:
1019 exit_code = main()
1020 except KeyboardInterrupt:
1021 print("\n中断されました。", file=sys.stderr)
1022 exit_code = 130
1023 except Exception:
1024 traceback.print_exc()
1025 exit_code = 1
1026 finally:
1027 if "--pause" in sys.argv:
1028 try:
1029 # --pause 1 が指定されている場合のみ待機
1030 index = sys.argv.index("--pause")
1031 if index + 1 < len(sys.argv) and sys.argv[index + 1] == "1":
1032 input("\nENTERキーを押して終了します。\n")
1033 except (ValueError, EOFError):
1034 pass
1035 raise SystemExit(exit_code)