Initial Commit

app.py

@@ -0,0 +1,286 @@
+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"),
+    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="音声ファイルをアップロードすると、各話者の名前がアサインされた文字起こしが作成されます。"
+)
+
+app.launch(debug=True)

requirements.txt

@@ -0,0 +1,2 @@
+openai==0.27.2
+pyannote.audio