HPR4746: Monotone

この番組は、主催者によって「Clean」と分類されています。

Python とそのライブラリ

1. librosa — 音声ファイルの読み込みとリサンプリング

def load_audio(path):
   y, sr = librosa.load(path, sr=SR, mono=True)
   return y, sr

各エピソード、その元のサンプリングレートやチャンネル数に関わらず、16kHz（SR = 16000）の単一モノストリームに変換されます。これは、librosa、リサンプリング、およびステレオをモノに変換する1つの手順です。これにより、MP3エンコーダーやポッドキャストホストが最初に使用したものが何であっても、すべての後続のツール（VAD、ピッチ追跡、MFCC抽出）が、一貫性があり予測可能な入力で動作するようにしました。

2. パースメソッド（発声の追跡）— 音高と音質を追跡

def pitch_and_hnr(y, sr):
 snd = parselmouth.Sound(y, sampling_frequency=sr)
 pitch = snd.to_pitch(time_step=PITCH_TIME_STEP, pitch_floor=PITCH_FLOOR, pitch_ceiling=PITCH_CEILING)
 f0 = pitch.selected_array["frequency"]
 times = pitch.xs()
 try:
 harmonicity = snd.to_harmonicity_cc(time_step=PITCH_TIME_STEP)
 hnr = np.array([harmonicity.get_value(t) for t in times])
 hnr = np.nan_to_num(hnr, nan=-999.0)
 exceptException:
 hnr = np.full_like(f0, 999.0) # 失敗した場合
 return f0, hnr, times

parselmouthは、フォネティクス研究における標準的なツールであるPraatをラップするPythonライブラリです。これは、声の単調性測定の核心であり、`to_pitch()`は20msごとに基本周波数（F0、おおよそ「音程」）を返し、`to_harmonicity_cc()`も同じタイムスタンプでハーモニック・トゥ・ノイズ比（HNR）を返します。これは、後で、ノイズや圧縮されたオーディオから得られた音程推定を排除し、悪いオーディオを「平坦な」音程として見せかけるのを防ぐために使用される、信号品質のスコアです。

3. webrtcvad — どのフレームが実際に音声であるかを決定

def vad_speech_mask(y, sr, aggressiveness=2):
 vad = webrtcvad.Vad(aggressiveness)
 pcm16 = (np.clip(y, -1.0, 1.0) *32767).astype(np.int16)
 n_frames =len(pcm16) // VAD_FRAME_LEN
 mask = np.zeros(n_frames, dtype=bool)
 for i inrange(n_frames):
 chunk = pcm16[i * VAD_FRAME_LEN:(i +1) * VAD_FRAME_LEN]
 raw = struct.pack("%dh"%len(chunk), *chunk)
 try:
 mask[i] = vad.is_speech(raw, sr)
 exceptException:
 mask[i] =False
 return mask, times

webrtcvadは、GoogleのWebRTCプロジェクトの一部である音声活動検出器であり、フレームごとに分類を行う（「この30msの音声データは音声データか否か」）。そのため、浮動小数点数ではなく、生の16ビットPCMバイトが必要であるため、np.clipとstruct.packによる変換が必要。

4. librosa.feature — 音楽と会話を区別する

flatness = librosa.feature.spectral_flatness(y=y, hop_length=hop)[0]
chroma = librosa.feature.chroma_stft(y=y, sr=sr, hop_length=hop)
onset_env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=hop)
...
chroma_diff = np.mean(np.linalg.norm(np.diff(chroma_win, axis=1), axis=0))
chroma_stability = 1.0 / (1.0 + chroma_diff)
...
scores[w] = 0.4 * chroma_stability + 0.35 * periodicity + 0.25 * min(flat_mean * 3, 1.0)

ポッドキャストのイントロ/その他の音楽は、多くの場合音楽であり、音楽は音声とは異なる周波数特性を持つ。そのため、この特性を測定に含めることは、「単調」測定を歪める可能性がある。分類器を訓練する代わりに、3つのlibrosaの特徴と手動で調整された重みを組み合わせた。具体的には、スペクトル平坦性（音楽はより多くの音階構造を持ち、ノイズのようなスペクトルは少ない）、クロマの安定性（持続的な和音と、それ以外のもの）などが挙げられる。

(常に変化する音声の周波数帯、および発話の開始における周期性（規則的なリズム対、発話の不規則なリズム）)

librosa.feature.mfcc + sklearn.cluster.KMeans + sklearn.metrics.silhouette_score — 独立した話者数推定

mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13, hop_length=hop)

MFCC（メル周波数スペクトル係数）は、短いオーディオウィンドウのスペクトルの形状を要約する—これは、広く音声タスクで使用されている「音声指紋」です。アイデア：各エピソードを2秒のチャンクに分割し、各チャンクを指紋ベクトルに変換し、特徴を標準化（Zスコア）し、KMeansを使用して、チャンクが自然に2つ以上のクラスタ（異なる話者）に分割されるかどうかを確認するか、1つの話者全体に分割されるかどうかを確認します。silhouette_scoreは、クラスタがどれだけ分離されているかを評価し、最適なkを選択するか、品質基準を満たさない場合は「1人の話者」に戻ります。この調査の一部は機能しませんでした。ノイズを別の話者として検出しました。

scipy.stats.mannwhitneyu + 手書きのCliffのデルタ—正規分布を仮定せずに2つのグループを比較

def cliffs_delta(a, b):

scipy.stats.mannwhitneyu は統計的検定です。これは、数値の2つのグループ（HPR対比較ポッドキャスト、特定の指標ごとに）を比較しますが、それらが正規分布に従っていることを前提としません。これは、ピッチ範囲の測定など、歪んだ現実世界のデータに適しています。scipyは、この検定に効果サイズを伴せて提供しないため、Cliffのデルタ、「可能な(HPRエピソード、比較エピソード)のすべてのペアの中で、HPRの値がより小さい値である頻度はどれくらいか？」というペア（有意性検定と効果サイズの数値）が、この検定全体で使用されていました。

7. pandas — 結果のグループ化、階層化、クロス集計

print(df.groupby("corpus")["retained_fraction"].describe())
 ...
 print(pd.crosstab(df["corpus"], df["register"]))

一度、data/results.jsonl (分析されたエピソードごとに1つのJSONオブジェクト) を pandas.DataFrame にロードすると、以下の2つのワンライナーが、それまでカスタムコードが必要だった2つの異なる質問に答えることができます: .groupby(...).describe() は、1回の呼び出しで、コーパスごとのオーディオ品質の分布サマリー (カウント、平均、標準偏差、四分位) を提供し、pd.crosstab() は、男性的な/女性的な話し方の対比における HPR (High-Prestige Register) をクロス集計します。これにより、2つのポッドキャストのコレクションが類似した性別的な話し方を持っていたかどうかを迅速に確認できます。

8. matplotlib — 箱ひげ図による主要な比較

fig, ax = plt.subplots(figsize=(6, 4))
data = [df[(df.corpus == c) & (df[metric].notna())][metric].values for c in ["hpr", "comparison"]]
ax.boxplot(data, tick_labels=["HPR", "Comparison"], showmeans=True)

ピッチ範囲メトリックの箱ひげ図、HPR、比較ポッドキャストとの比較、平均値、中央値

9. spacy の単語ベクトル — トピックの漂動を測定するだけでなく、単語の選択を測定する

nlp = get_nlp() # spacy.load("en_core_web_md")
 ...
 chunk_docs = [nlp(c) for c in chunks]
 chunk_vecs = []
 for d in chunk_docs:
 v = d.vector
 norm = np.linalg.norm(v)
 if norm > 1e-6:
 chunk_vecs.append(v / norm)
 ...
 adj_sims = [float(np.dot(V[i], V[i + 1])) for i in range(n - 1)]

各約130単語のトランスクリプトのチャンクは、spaCyのミドル英語モデル（en_core_web_md、内蔵された単語ベクトルを持つ）を使用して単一のベクトルとしてエンコードされ、その後L2正規化され、2つのベクト間の内積を計算するis彼らのコサイン類似度。連続するチャンクの類似度を比較することで、「トピックが1つのチャンクから次のチャンクへほとんど移動しているか」というスコアが得られます。

10. 語彙の多様性 - 転写の長さの制限を回避しながら

lr = LexicalRichness(text)
try:
 mtld = lr.mtld(threshold=0.72)
except Exception:
 mtld = None

「単純な語彙多様性スコア（ユニークな単語 ÷ 総単語数）は、テキストの長さが長くなるにつれて悪化する。これは、この指標が定義されている方法によるものであり、話者が実際に繰り返し話しているわけではない。」

MTLD (Measure of Textual Lexical Diversity) は、テキスト全体をスキャンし、多様性が特定の閾値を下回るまでに必要な単語数をカウントし、その平均を計算することで、この問題を回避します。これは、10分間のエピソードと40分間のエピソード間の比較を維持するように特別に設計された指標です。

11. vaderSentiment — 辞書ベースの感情スコア

vader = get_vader()  # SentimentIntensityAnalyzer()
 sentiments = [vader.polarity_scores(c)["compound"] for c in chunks]
 ...
 "sentiment_std": float(np.std(sentiments)) if len(sentiments) > 1 else None

VADERは、手動で構築された辞書とルールセットを使用して、各チャンクの感情スコアを-1から+1の範囲で評価します（ニューラルネットワークではありません）。この方法が選択されたのは、速度、決定性、および数百のエピソードごとに1つのチャンクごとに実行できるコストが低いという点です。使用される値は平均感情ではなく、標準偏差エピソードのセグメントごとの感情の分析：感情的な

声が常に変化するか、それとも常に一定の高さで維持されるか？

12. zlib (標準ライブラリ) — 圧縮比をエントロピーの指標として

def compression_ratio(text):
 raw = text.encode("utf-8")
 if len(raw) < 200:
 return None
 compressed = zlib.compress(raw, level=9)
 return len(compressed) / len(raw)

第三のNLPライブラリは一切使用されていません—標準ライブラリのzlibモジュールのみを使用。 gzipのような圧縮機能は、繰り返し現れるパターンを特定し、利用します。もし、トランスクリプトを大幅に圧縮できるのであれば、そのトランスクリプトは比較的繰り返し、予測可能な内容であり、ほとんど圧縮できない場合は、その文言はより多様であると推測できます。これは、おおよそ「口語の単調さ」を示しています。

標準ライブラリ（re）—「HPRは単独ホストを優先するか」を検証可能なヒューリスティックに変換

MULTI_PRESENTER_PATTERNS = [

音響スピーカークラスタリングの試み（5番、信頼性が低いことが判明）ではなく、より信頼できる情報源は、平文テキストから得られたものでした。HPR自身のエピソードメタデータ（ローカルの知識ベースから取得）には、ホストが書いた概要が含まれており、「ゲスト」「インタビュー」「私と話す」などの正規表現パターンと、リンクされたコレスポンデントプロファイルのカウントだけで、複数のプレゼンターが含まれるエピソードを特定することができました。

multiprocessing.Pool — 多数のファイルに対して同じ分析を実行し、何百もの待ち時間をなくす

work = [(r, local_paths[r["id"]]) for r in batch]

すべてのバッチパイプライン（音響ピッチ分析、話者数カウント）は、同じ構造を使用していました：Pool(workers).imap_unordered(...) を使用して、複数のCPUコアで同時に各エピソードの関数を実行し、完了するごとに結果をJSON-linesファイルに出力していました。これにより、バッチ全体が完了するのを待つのではなく、個々の結果を即座に出力することが可能になりました。imap_unordered は、単一の遅いエピソードが完了するのを待つのではなく、すでに完了した結果の書き込みを妨げるのを避けるために、map よりも選択されました。

15. サブプロセス + 繰り返し処理の最適化 — ディスク容量の制限内で動作

def load_done_ids(results_path):
  if not os.path.exists(results_path):
    return set()
  done = set()
  with open(results_path) as f:
    for line in f:
      ...
      done.add(json.loads(line)["id"])
  return done
  ...
  rows = [r for r in rows if r["id"] not in done]
  ...
  for batch_start in range(0, len(rows), args.batch_size):
    batch = rows[batch_start:batch_start + args.batch_size]
    local_paths = fetch_batch(batch, args.scratch) # scpからリモートホストから
  ...
  for p in local_paths.values():
    os.remove(p) # 次のバッチの前に削除

すべてのポッドキャスト (約220GB) は、ローカルディスク (約54GB空き容量) に収まらなかった。解決策はループ処理: scp (サブプロセス経由) でオーディオの1つのバッチを読み込み、分析し、削除し、次のバッチに移動する。load_done_ids() は、出力ファイルにすでに存在するものを読み戻すことで、パイプライン全体を再開できるようにする。これにより、クラッシュや途中で意図的に停止した場合でも、完了した作業が失われない。

リンク

提供フィードバックこのエピソードについて.

HPR4746: モノトーン