怒った通話をTTSで合成する — Amazon Connect/KVS互換のMKVを作り、架電せずに検証する

def _track_entry(number: int, name: str) -> bytes:
    return _elem(
        b"\xae",
        _elem(b"\xd7", bytes([number]))                    # TrackNumber
        + _elem(b"\x53\x6e", name.encode())                # Name
        # CodecIDは実機と同じく "A_AAC" を名乗らせる(中身は生L16のまま)。
        # 詐称ごと再現しておかないと、実データで起きる問題が試験で出ない
        + _elem(b"\x86", b"A_AAC"),
    )

def build_mkv(customer: np.ndarray, agent: np.ndarray) -> bytes:
    """話者別PCMを、KVSが出すのと同じ構造のMKVに詰める。"""
    n = max(customer.size, agent.size)
    cust = np.pad(customer, (0, n - customer.size))
    agnt = np.pad(agent, (0, n - agent.size))

    tracks = _elem(
        b"\x16\x54\xae\x6b",
        _track_entry(1, "AUDIO_TO_CUSTOMER") + _track_entry(2, "AUDIO_FROM_CUSTOMER"),
    )

    step = RATE * 20 // 1000  # 20msずつ交互に積む(実機と同じ粒度)
    blocks = []
    for i in range(0, n, step):
        blocks.append(_simple_block(1, agnt[i:i + step].tobytes()))
        blocks.append(_simple_block(2, cust[i:i + step].tobytes()))
    cluster = _elem(b"\x1f\x43\xb6\x75", _elem(b"\xe7", b"\x00") + b"".join(blocks))

    # EBMLヘッダは中身を見ていないので最小限。Segmentの下にTracksとClusterを置く
    header = _elem(b"\x1a\x45\xdf\xa3", _elem(b"\x42\x82", b"matroska"))
    return header + _elem(b"\x18\x53\x80\x67", tracks + cluster)

def _vint(value: int, length: int) -> bytes:
    """EBMLの可変長整数(サイズ用。先頭にマーカービットを立てる)。"""
    return (value | (1 << (7 * length))).to_bytes(length, "big")


def _elem(elem_id: bytes, payload: bytes) -> bytes:
    return elem_id + _vint(len(payload), 8) + payload

def to_telephone_band(x: np.ndarray) -> np.ndarray:
    """24kHz → 8kHz。折り返しを防ぐため先に低域だけ残す。

    電話帯域に落とすこと自体が検証の一部。TTSの綺麗な音のままでは、
    実際の通話で起きる聞き取り困難を再現できない。
    """
    taps = np.hamming(61) * np.sinc(np.arange(-30, 31) * 2 * 3400 / TTS_RATE)
    taps /= taps.sum()
    filtered = np.convolve(x.astype(np.float64), taps, mode="same")
    return np.rint(filtered[::3]).clip(-32768, 32767).astype("<i2")

SCRIPT: list[tuple[str, str, str]] = [
    # (話者, 台詞, TTSへの口調指示)
    ("agent", "解析を開始しました。ご用件をお話しください。", "落ち着いた案内の口調で"),
    ("customer", "もしもし、先週から言ってる件、どうなってますか。", "普通の問い合わせの口調で"),
    ("agent", "確認いたしますので少々お待ちください。", "丁寧な事務的口調で"),
    ("customer", "確認って、さっきもそう言いましたよね。いつまで待たせるんですか。",
     "少し苛立った、語気の強まった口調で"),
    ("agent", "申し訳ございません。", "恐縮した口調で"),
    ("customer", "謝れば済むと思ってんのか。あんたじゃ話にならん、責任者を出せよ今すぐ!",
     "強い怒りをぶつける、大きな声で"),
]

for speaker, text, tone in SCRIPT:
    voice = args.customer_voice if speaker == "customer" else args.agent_voice
    pcm = to_telephone_band(synth(text, tone, voice, key))
    other = "agent" if speaker == "customer" else "customer"
    # 喋る側にはこの位置から音を置き、相手側には同じ長さの無音を置く
    tracks[speaker].append((cursor, pcm))
    tracks[other].append((cursor, np.zeros(pcm.size, dtype="<i2")))
    cursor += pcm.size + gap

[ 10] customer もしもし
      customer 先週から言ってる件、どうなってますか?
[ 35] customer 確認って、さっきもそう言いましたよね?
      customer いつまで待たせるんですか。
[ 45] customer 謝れば          ← ここで判定が止まっている
      customer 済むと思ってんのか。      ← 無判定
      customer あんたじゃ話にならん。    ← 無判定
      customer 責任者を出せよ、今すぐ。  ← 無判定

def should_judge(self, msg: dict) -> bool:
    """相手の確定発話だけを対象にし、間隔を空ける。"""
    if msg.get("speaker") != "customer" or not msg.get("final") or not msg.get("text"):
        return False
    if self._busy:
        return False
    return (time.monotonic() - self._last_at) >= ANGER_MIN_INTERVAL_SEC

async def run(self, messages: list[dict], target: dict) -> None:
    """判定して、対象の発話に結果を書き込みブラウザへ流す。

    間隔を空けるが**発話は捨てない**。判定中に来たものは最新の1件に畳み込み、
    直前の判定が終わり次第それを判定する。捨てる実装にすると、発話が立て込む
    場面——つまり怒りが高まっている場面——ほど判定が抜ける。
    """
    self._pending = (messages, target)
    if self._busy:
        return  # 走っているループが拾う
    self._busy = True
    try:
        while self._pending is not None:
            wait = ANGER_MIN_INTERVAL_SEC - (time.monotonic() - self._last_at)
            if wait > 0:
                await asyncio.sleep(wait)
            msgs, tgt = self._pending
            self._pending = None
            self._last_at = time.monotonic()

            result = await judge(build_window(msgs))
            if result is None:
                continue

            # 発話そのものにスコアを載せる(記録にもそのまま残る)
            tgt["anger_score"] = result["score"]
            tgt["anger_reason"] = result["reason"]
            self.max_score = max(self.max_score, result["score"])

            await self._emit({
                "type": "emotion",
                "speaker": "customer",
                "item_id": tgt.get("item_id"),
                **result,
            })
    finally:
        self._busy = False

[ 10] customer もしもし
[ 25] customer 先週から言ってる件、どうなってますか?
[ 35] customer 確認って、さっきもそう言いましたよね。
[ 45] customer いつまで待たせるんですか。
[ 45] customer 謝れば
[ 75] ALERT customer 済むと思ってんのか。
[ 75] ALERT customer あんたじゃ話にならん。
[ 85] ALERT customer 責任者を出せよ、今すぐ。

 0  もしもし。
20  ちょっとシステムに欠陥があるんだけどどうなってるの?
45  すげーバグってんだけど。            ← 「罵倒や人格否定はない」と切り分け
75  このバグは異常だよマジで責任者出してくんね。  ALERT
65  どうやって直すんこれ。              ← 下がった
75  責任者出せ、責任者。
75  だから責任者出せって言ってんでしょ。