Spaces:
Runtime error
Runtime error
| import os | |
| import wave | |
| import numpy as np | |
| import contextlib | |
| from pydub import AudioSegment | |
| from pyannote.core import Segment | |
| from pyannote.audio import Audio | |
| from pyannote.audio.pipelines.speaker_verification import PretrainedSpeakerEmbedding | |
| import torch | |
| from typing import Dict, List, Tuple | |
| def convert_to_wav(input_file: str, output_file: str = "output_file.wav") -> str: | |
| """ | |
| 音声ファイルをWAV形式に変換します。 | |
| Parameters | |
| ---------- | |
| input_file: str | |
| 変換する音声ファイルのパス | |
| output_file: str | |
| 変換後のWAVファイルの出力先パス（デフォルトは"output_file.wav"） | |
| Returns | |
| ------- | |
| str | |
| 変換後のWAVファイルのパス | |
| """ | |
| file_format = os.path.splitext(input_file)[1][1:] | |
| audio = AudioSegment.from_file(input_file, format=file_format) | |
| audio.export(output_file, format="wav") | |
| return output_file | |
| def segment_embedding( | |
| file_name: str, | |
| duration: float, | |
| segment, | |
| embedding_model: PretrainedSpeakerEmbedding | |
| ) -> np.ndarray: | |
| """ | |
| 音声ファイルから指定されたセグメントの埋め込みを計算します。 | |
| Parameters | |
| ---------- | |
| file_name: str | |
| 音声ファイルのパス | |
| duration: float | |
| 音声ファイルの継続時間 | |
| segment: whisperのtranscribeのsegment | |
| embedding_model: PretrainedSpeakerEmbedding | |
| 埋め込みモデル | |
| Returns | |
| ------- | |
| np.ndarray | |
| 計算された埋め込みベクトル | |
| """ | |
| audio = Audio() | |
| start = segment["start"] | |
| end = min(duration, segment["end"]) | |
| clip = Segment(start, end) | |
| waveform, sample_rate = audio.crop(file_name, clip) | |
| return embedding_model(waveform[None]) | |
| def reference_audio_embedding( | |
| file_name: str | |
| ) -> np.ndarray: | |
| """ | |
| 参考音声の埋め込みを出力します。 | |
| Parameters | |
| ---------- | |
| file_name: str | |
| 音声ファイルのパス | |
| Returns | |
| ------- | |
| np.ndarray | |
| 計算された埋め込みベクトル | |
| """ | |
| audio = Audio() | |
| waveform, sample_rate = audio(file_name) | |
| embedding_model = embedding_model = PretrainedSpeakerEmbedding("speechbrain/spkrec-ecapa-voxceleb", device='cpu') | |
| return embedding_model(waveform[None])[0] | |
| def generate_speaker_embeddings( | |
| meeting_file_path: str, | |
| transcript | |
| ) -> np.ndarray: | |
| """ | |
| 音声ファイルから話者の埋め込みを計算します。 | |
| Parameters | |
| ---------- | |
| meeting_file_path: str | |
| 音声ファイルのパス | |
| transcript: Whisper API の transcribe メソッドの出力結果 | |
| Returns | |
| ------- | |
| np.ndarray | |
| 計算された話者の埋め込み群 | |
| """ | |
| output_file = convert_to_wav(meeting_file_path) | |
| segments = transcript['segments'] | |
| embedding_model = PretrainedSpeakerEmbedding("speechbrain/spkrec-ecapa-voxceleb", device='cpu') | |
| embeddings = np.zeros(shape=(len(segments), 192)) | |
| with contextlib.closing(wave.open(output_file, 'r')) as f: | |
| frames = f.getnframes() | |
| rate = f.getframerate() | |
| duration = frames / float(rate) | |
| for i, segment in enumerate(segments): | |
| embeddings[i] = segment_embedding(output_file, duration, segment, embedding_model) | |
| embeddings = np.nan_to_num(embeddings) | |
| return embeddings | |
| import numpy as np | |
| from sklearn.cluster import AgglomerativeClustering | |
| from typing import List, Tuple | |
| def clustering_embeddings(speaker_count: int, embeddings: np.ndarray) -> AgglomerativeClustering: | |
| """ | |
| 埋め込みデータをクラスタリングして、クラスタリングオブジェクトを返します。 | |
| Parameters | |
| ---------- | |
| embeddings: np.ndarray | |
| 分散表現（埋め込み）のリスト。 | |
| Returns | |
| ------- | |
| AgglomerativeClustering | |
| クラスタリングオブジェクト。 | |
| """ | |
| clustering = AgglomerativeClustering(speaker_count).fit(embeddings) | |
| return clustering | |
| def format_speaker_output_by_segment(clustering: AgglomerativeClustering, transcript: dict) -> str: | |
| """ | |
| クラスタリングの結果をもとに、各発話者ごとにセグメントを整形して出力します | |
| Parameters | |
| ---------- | |
| clustering: AgglomerativeClustering | |
| クラスタリングオブジェクト。 | |
| transcript: dict | |
| Whisper API の transcribe メソッドの出力結果 | |
| Returns | |
| ------- | |
| str | |
| 発話者ごとに整形されたセグメントの文字列 | |
| """ | |
| labeled_segments = [] | |
| for label, segment in zip(clustering.labels_, transcript["segments"]): | |
| labeled_segments.append((label, segment["start"], segment["text"])) | |
| output = "" | |
| for speaker, _, text in labeled_segments: | |
| output += f"話者{speaker + 1}: 「{text}」\n" | |
| return output | |
| from sklearn.cluster import KMeans | |
| from sklearn.metrics.pairwise import pairwise_distances | |
| def clustering_embeddings2(speaker_count: int, embeddings: np.ndarray) -> KMeans: | |
| """ | |
| 埋め込みデータをクラスタリングして、クラスタリングオブジェクトを返します。 | |
| Parameters | |
| ---------- | |
| embeddings: np.ndarray | |
| 分散表現（埋め込み）のリスト。 | |
| Returns | |
| ------- | |
| KMeans | |
| クラスタリングオブジェクト。 | |
| """ | |
| # コサイン類似度行列を計算 | |
| cosine_distances = pairwise_distances(embeddings, metric='cosine') | |
| clustering = KMeans(n_clusters=speaker_count).fit(cosine_distances) | |
| return clustering | |
| from scipy.spatial.distance import cosine | |
| def closest_reference_speaker(embedding: np.ndarray, references: List[Tuple[str, np.ndarray]]) -> str: | |
| """ | |
| 与えられた埋め込みに最も近い参照話者を返します。 | |
| Parameters | |
| ---------- | |
| embedding: np.ndarray | |
| 話者の埋め込み | |
| references: List[Tuple[str, np.ndarray]] | |
| 参照話者の名前と埋め込みのリスト | |
| Returns | |
| ------- | |
| str | |
| 最も近い参照話者の名前 | |
| """ | |
| min_distance = float('inf') | |
| closest_speaker = None | |
| for name, reference_embedding in references: | |
| distance = cosine(embedding, reference_embedding) | |
| if distance < min_distance: | |
| min_distance = distance | |
| closest_speaker = name | |
| return closest_speaker | |
| def format_speaker_output_by_segment2(embeddings: np.ndarray, transcript: dict, reference_embeddings: List[Tuple[str, np.ndarray]]) -> str: | |
| """ | |
| 各発話者の埋め込みに基づいて、セグメントを整形して出力します。 | |
| Parameters | |
| ---------- | |
| embeddings: np.ndarray | |
| 話者の埋め込みのリスト | |
| transcript: dict | |
| Whisper API の transcribe メソッドの出力結果 | |
| reference_embeddings: List[Tuple[str, np.ndarray]] | |
| 参照話者の名前と埋め込みのリスト | |
| Returns | |
| ------- | |
| str | |
| 発話者ごとに整形されたセグメントの文字列。 | |
| """ | |
| labeled_segments = [] | |
| for embedding, segment in zip(embeddings, transcript["segments"]): | |
| speaker_name = closest_reference_speaker(embedding, reference_embeddings) | |
| labeled_segments.append((speaker_name, segment["start"], segment["text"])) | |
| output = "" | |
| for speaker, _, text in labeled_segments: | |
| output += f"{speaker}: 「{text}」\n" | |
| return output | |
| import gradio as gr | |
| import openai | |
| def create_transcription_with_speaker(openai_key, main_audio, reference_audio_1, reference1_name, | |
| reference_audio_2, reference2_name, speaker_count = 2): | |
| openai.api_key = openai_key | |
| # 文字起こし | |
| transcript = openai.Audio.transcribe("whisper-1", open(main_audio, "rb"), response_format="verbose_json") | |
| # 各発話をembeddingsに変換 | |
| embeddings = generate_speaker_embeddings(main_audio, transcript) | |
| # 各発話のembeddingsをクラスタリング | |
| clustering = clustering_embeddings(speaker_count, embeddings) | |
| # クラスタリングで作られた仮のラベルで各セグメントに名前付け | |
| output_by_segment1 = format_speaker_output_by_segment(clustering, transcript) | |
| reference1 = reference_audio_embedding(reference_audio_1) | |
| reference2 = reference_audio_embedding(reference_audio_2) | |
| reference_embeddings = [(reference1_name, reference1), (reference2_name, reference2)] | |
| output_by_segment2 = format_speaker_output_by_segment2(embeddings, transcript, reference_embeddings) | |
| return output_by_segment1, output_by_segment2 | |
| inputs = [ | |
| gr.Textbox(lines=1, label="openai_key", type="password"), | |
| gr.Audio(type="filepath", label="メイン音声ファイル"), | |
| gr.Audio(type="filepath", label="話者 (1) 参考音声ファイル"), | |
| gr.Textbox(lines=1, label="話者 (1) の名前"), | |
| gr.Audio(type="filepath", label="話者 (2) 参考音声ファイル"), | |
| gr.Textbox(lines=1, label="話者 (2) の名前") | |
| ] | |
| outputs = [ | |
| gr.Textbox(label="話者クラスタリング文字起こし"), | |
| gr.Textbox(label="話者アサイン文字起こし"), | |
| ] | |
| app = gr.Interface( | |
| fn=create_transcription_with_speaker, | |
| inputs=inputs, | |
| outputs=outputs, | |
| title="話者アサイン機能付き書き起こしアプリ", | |
| description="音声ファイルをアップロードすると、各話者の名前がアサインされた文字起こしが作成されます。参考: https://huggingface.co/spaces/vumichien/Whisper_speaker_diarization" | |
| ) | |
| app.launch(debug=True) |