diff --git a/tests/test_emotion.py b/tests/test_emotion.py new file mode 100644 index 0000000..1130244 --- /dev/null +++ b/tests/test_emotion.py @@ -0,0 +1,76 @@ +"""Emotion-tag parsing for expressive TTS. A ``[감정]`` tag must steer the +pitch/speed of the text that follows without being spoken; a bracket that is NOT +a known emotion word must be kept as ordinary spoken content.""" + +from wsai.backends.emotion import ( + EMOTION_PARAMS, + match_emotion, + parse_segments, +) + +BASE = 1.3 + + +def test_emotion_tag_is_not_spoken_and_sets_delivery(): + segs = parse_segments("[기쁨] 오늘 날씨 좋다", BASE) + assert len(segs) == 1 + assert "기쁨" not in segs[0].text # the tag word is dropped + assert segs[0].text == "오늘 날씨 좋다" + mult, semis = EMOTION_PARAMS["happy"] + assert segs[0].speed == BASE * mult + assert segs[0].pitch == semis + + +def test_midreply_emotion_change_splits_segments(): + segs = parse_segments("[속상함] 정말 힘들었겠다. [힘차게] 하지만 넌 할 수 있어!", BASE) + assert len(segs) == 2 + assert segs[0].text == "정말 힘들었겠다." + assert segs[1].text == "하지만 넌 할 수 있어!" + assert segs[0].pitch == EMOTION_PARAMS["sad"][1] + assert segs[1].pitch == EMOTION_PARAMS["hopeful"][1] + + +def test_non_emotion_bracket_is_spoken_without_brackets(): + segs = parse_segments("[기쁨] 첫째는 [1번] 항목이야", BASE) + assert len(segs) == 1 + # "1번" is not an emotion -> read it; brackets themselves are gone. + assert "1번" in segs[0].text + assert "[" not in segs[0].text and "]" not in segs[0].text + assert segs[0].pitch == EMOTION_PARAMS["happy"][1] + + +def test_text_before_first_tag_is_neutral(): + segs = parse_segments("잠깐만. [신남] 찾았다!", BASE) + assert segs[0].text == "잠깐만." + assert segs[0].speed == BASE and segs[0].pitch == 0.0 + assert segs[1].pitch == EMOTION_PARAMS["excited"][1] + + +def test_plain_text_is_one_neutral_segment(): + segs = parse_segments("그냥 평범한 문장이야", BASE) + assert len(segs) == 1 + assert segs[0].speed == BASE and segs[0].pitch == 0.0 + + +def test_empty_input_yields_no_segments(): + assert parse_segments("", BASE) == [] + assert parse_segments(" ", BASE) == [] + + +def test_only_emotion_tags_yield_no_segments(): + # A reply that is nothing but tags has nothing to say. + assert parse_segments("[기쁨][신남]", BASE) == [] + + +def test_match_emotion_is_synonym_and_space_tolerant(): + assert match_emotion("속상함") == "sad" + assert match_emotion(" 힘 차게 ") == "hopeful" # squeezed + trimmed + assert match_emotion("행복하게") == "happy" + assert match_emotion("메모") is None # not an emotion + + +def test_persona_examples_are_all_recognised(): + # Every emotion the brain persona advertises must resolve, or it would be + # read aloud instead of shaping the voice. + for word in ["힘차게", "궁금", "반가움", "차분하게", "웃으며", "속상함"]: + assert match_emotion(word) is not None, word diff --git a/wsai/backends/claude.py b/wsai/backends/claude.py index 363ef62..acbf4c4 100644 --- a/wsai/backends/claude.py +++ b/wsai/backends/claude.py @@ -127,8 +127,12 @@ class ClaudeBrain: "화면 설명을 참고해 자연스러운 반말/존댓말은 사용자에 맞추고, 짧고 대화하듯 답해. " "화면을 못 봤으면 솔직히 말해. " "네 답변은 음성으로 읽히니 마크다운·코드블록·백틱 같은 서식 없이 평범한 말로만 답해. " - "답변은 반드시 대괄호 감정 표시 한 개로 시작해. 예: [힘차게], [궁금], [반가움], [차분하게], [웃으며]. " - "감정 태그는 답변 맨 앞에 딱 한 번만 붙이고, 그 뒤에 실제 답을 이어써. " + "감정은 대괄호 태그로 표현해. 태그 자체는 소리로 읽히지 않고, 그 뒤 문장의 목소리 톤(피치·속도)을 바꿔줘. " + "답변 맨 앞에 감정 태그 하나로 시작하고, 답변 도중 감정이 바뀌면 그 지점에 새 태그를 넣어. " + "예: [속상함] 정말 힘들었겠다. [힘차게] 하지만 넌 할 수 있어! " + "쓸 수 있는 감정: 기쁨, 신남, 희망(힘차게), 슬픔(속상함), 화남, 두려움, 놀람, 차분, 다정, 진지, 실망, 피곤, " + "사랑스럽게, 장난스럽게(웃으며), 속삭임, 외침, 단호, 안도, 궁금, 반가움. " + "감정 태그가 아닌 진짜 대괄호 내용(예: [1번], [메모])은 그대로 읽히니 필요하면 그렇게 써도 돼. " "답변은 최대한 짧고 간결하게, 한두 문장 이내로 해." ) diff --git a/wsai/backends/emotion.py b/wsai/backends/emotion.py new file mode 100644 index 0000000..98a080a --- /dev/null +++ b/wsai/backends/emotion.py @@ -0,0 +1,139 @@ +"""Emotion tags for expressive TTS. + +Claude can sprinkle ``[감정]`` tags through a reply to colour the delivery, e.g. + + "[속상함] 정말 힘들었겠다. [힘차게] 하지만 넌 할 수 있어!" + +An emotion tag is NOT spoken — instead it shifts the *following* text's pitch and +speed until the next tag. A bracketed word that is NOT a known emotion is left as +ordinary spoken content (the brackets are dropped, the words are read). + +The emotion vocabulary is grounded in the de-facto industry set used by Azure +Neural TTS speaking styles (cheerful, sad, angry, excited, friendly, hopeful, +terrified, shouting, whispering) together with Ekman's six basic emotions +(happiness, sadness, anger, fear, surprise, disgust). Each canonical emotion maps +to a ``(speed_multiplier, pitch_semitones)`` pair; the multiplier scales the base +synthesis speed and the semitone offset is applied as a pitch shift on the wav. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +# Canonical emotion -> (speed multiplier relative to base, pitch shift in semitones). +# Kept deliberately modest so delivery stays natural, not cartoonish. +EMOTION_PARAMS: dict[str, tuple[float, float]] = { + "happy": (1.08, 2.0), # 기쁨 / cheerful + "excited": (1.15, 3.0), # 신남 / excited + "hopeful": (1.10, 1.5), # 희망 / 힘차게 + "sad": (0.90, -2.5), # 슬픔 / sad + "angry": (1.12, 1.0), # 화남 / angry + "fearful": (1.12, 2.0), # 두려움 / terrified + "surprised": (1.05, 3.0), # 놀람 / surprise + "disgust": (0.96, -1.0), # 혐오 / disgust + "calm": (0.95, -1.0), # 차분 / calm + "friendly": (1.00, 1.0), # 다정 / friendly + "serious": (0.97, -1.0), # 진지 / serious + "disappointed":(0.92, -2.0), # 실망 / disappointed + "tired": (0.90, -2.0), # 피곤 / 지침 + "affectionate":(0.98, 1.0), # 사랑스럽게 / affectionate + "playful": (1.08, 2.0), # 장난스럽게 / playful + "whisper": (0.92, -1.5), # 속삭임 / whispering + "shout": (1.05, 2.5), # 외침 / shouting + "determined": (1.05, 0.5), # 단호 / determined + "relieved": (0.95, 0.5), # 안도 / relieved + "curious": (1.03, 1.5), # 궁금 / curious +} + +# Every spelling Claude might realistically emit, mapped to a canonical emotion. +# False negatives (reading an emotion word aloud) are harmless; false positives +# (silently dropping real content) are not — so match spellings exactly rather +# than fuzzily. +_SYNONYMS: dict[str, str] = {} + + +def _register(canonical: str, *words: str) -> None: + for w in words: + _SYNONYMS[_norm(w)] = canonical + + +def _norm(word: str) -> str: + # Compare on a squeezed, lower-cased form so "힘 차게" == "힘차게". + return re.sub(r"\s+", "", word).lower() + + +_register("happy", "기쁨", "기쁘게", "기뻐", "기뻐하며", "행복", "행복하게", "행복하게도", "즐겁게", "즐거움", "밝게", "반가움", "반갑게", "반가워", "cheerful", "happy", "joyful") +_register("excited", "신남", "신나게", "신나서", "흥분", "들뜬", "들떠서", "설렘", "설레며", "excited", "thrilled") +_register("hopeful", "희망", "희망차게", "힘차게", "힘내", "힘내서", "응원", "응원하며", "격려", "hopeful", "encouraging") +_register("sad", "슬픔", "슬프게", "슬퍼", "슬퍼하며", "속상함", "속상하게", "속상해", "우울", "우울하게", "안타깝게", "울먹이며", "sad", "sorrowful") +_register("angry", "화남", "화나게", "화나서", "화가남", "분노", "분노하며", "짜증", "짜증내며", "angry", "furious") +_register("fearful", "두려움", "두렵게", "무섭게", "무서워하며", "불안", "불안하게", "겁먹은", "겁먹고", "떨리는", "fearful", "terrified", "anxious") +_register("surprised", "놀람", "놀라며", "놀랍게", "놀라서", "깜짝", "경악", "surprised", "shocked") +_register("disgust", "혐오", "역겹게", "질색", "disgust", "disgusted") +_register("calm", "차분", "차분하게", "침착", "침착하게", "담담하게", "잔잔하게", "calm", "gentle") +_register("friendly", "다정", "다정하게", "친근", "친근하게", "부드럽게", "따뜻하게", "friendly", "warm") +_register("serious", "진지", "진지하게", "무겁게", "엄숙하게", "serious", "solemn") +_register("disappointed", "실망", "실망스럽게", "실망하며", "낙담", "disappointed") +_register("tired", "피곤", "피곤하게", "지침", "지쳐서", "지친", "힘없이", "tired", "weary", "exhausted") +_register("affectionate", "사랑스럽게", "애정", "애정어린", "다정스럽게", "affectionate", "loving") +_register("playful", "장난스럽게", "장난치며", "유쾌하게", "익살스럽게", "웃으며", "웃으면서", "playful", "teasing") +_register("curious", "궁금", "궁금하게", "궁금해하며", "궁금해서", "호기심", "curious", "inquisitive") +_register("whisper", "속삭임", "속삭이며", "조용히", "나지막이", "whisper", "whispering") +_register("shout", "외침", "외치며", "큰소리로", "소리치며", "우렁차게", "shout", "shouting") +_register("determined", "단호", "단호하게", "결연하게", "당당하게", "determined", "confident") +_register("relieved", "안도", "안도하며", "안심", "안심하며", "relieved") + + +def match_emotion(inner: str) -> str | None: + """Return the canonical emotion for a bracket's inner text, or None if the + text is not a recognised emotion word (and should therefore be spoken).""" + return _SYNONYMS.get(_norm(inner)) + + +@dataclass +class Segment: + text: str + speed: float + pitch: float # semitones; 0.0 == no shift + + +_TAG_RE = re.compile(r"\[([^\[\]]*)\]") + + +def parse_segments(text: str, base_speed: float) -> list[Segment]: + """Split ``text`` into consecutive spoken segments, each carrying the speed + and pitch implied by the most recent emotion tag. + + * An emotion tag switches the active emotion for everything after it and is + not spoken. + * A non-emotion bracket keeps its inner words as spoken text (brackets gone). + * Text before any tag is spoken with neutral delivery (base speed, no shift). + """ + segments: list[Segment] = [] + cur_speed, cur_pitch = base_speed, 0.0 + buf: list[str] = [] + + def flush() -> None: + joined = "".join(buf).strip() + if joined: + segments.append(Segment(joined, cur_speed, cur_pitch)) + buf.clear() + + pos = 0 + for m in _TAG_RE.finditer(text): + emotion = match_emotion(m.group(1)) + buf.append(text[pos:m.start()]) + pos = m.end() + if emotion is None: + # Not an emotion — read the bracket's contents, drop the brackets. + buf.append(m.group(1)) + else: + # Emotion tag — everything so far belongs to the previous emotion; + # flush it, then switch delivery for what follows. + flush() + mult, semis = EMOTION_PARAMS[emotion] + cur_speed, cur_pitch = base_speed * mult, semis + buf.append(text[pos:]) + flush() + return segments # empty when there is nothing speakable (blank or all-tags) diff --git a/wsai/backends/melo.py b/wsai/backends/melo.py index 776bac7..2d39c9a 100644 --- a/wsai/backends/melo.py +++ b/wsai/backends/melo.py @@ -28,6 +28,7 @@ from pathlib import Path from typing import Awaitable, Callable from ..interfaces import Reply +from .emotion import parse_segments log = logging.getLogger("wsai.tts.melo") @@ -173,9 +174,22 @@ class MeloTTS: callers that want the wav directly (e.g. the Discord voice bridge).""" await self._ensure() text = normalize_for_speech(text) + # Split on [감정] tags: each tag steers pitch/speed for the text that + # follows (and is itself not spoken); non-emotion brackets stay as words. + segments = parse_segments(text, self.speed) self._n += 1 out = str(self.out_dir / f"tts-{self._n:06d}.wav") - req = json.dumps({"text": text, "out": out, "speed": self.speed}) + if segments: + payload = { + "segments": [ + {"text": s.text, "speed": s.speed, "pitch": s.pitch} + for s in segments + ], + "out": out, + } + else: # empty/whitespace reply: keep legacy single-utterance behaviour + payload = {"text": text, "out": out, "speed": self.speed} + req = json.dumps(payload) s = time.monotonic() async with self._lock: assert self._proc and self._proc.stdin and self._proc.stdout diff --git a/wsai/backends/melo_worker.py b/wsai/backends/melo_worker.py index 16f75ce..ecc3ab9 100644 --- a/wsai/backends/melo_worker.py +++ b/wsai/backends/melo_worker.py @@ -11,10 +11,17 @@ library chatter lands on stderr instead. Protocol (one JSON object per line, on the protocol channel): <- {"text": "...", "out": "/abs/path.wav", "speed": 1.3} + <- {"segments": [{"text": "...", "speed": 1.3, "pitch": 2.0}, ...], + "out": "/abs/path.wav"} # expressive form: per-segment speed + pitch -> {"ok": true, "out": "/abs/path.wav", "ms": 123} -> {"ok": false, "error": "..."} On startup, once the model is ready, it emits exactly one line: -> {"ready": true, "ms": , "device": "cpu"} + +``pitch`` is a semitone offset applied to that segment's wav (0 == no shift) so +emotion tags can raise/lower the voice without changing the words. Segments are +synthesised independently and concatenated with a short gap so a single reply can +carry several emotions. """ import json @@ -66,8 +73,41 @@ def main() -> None: else: raise speaker_id = tts.hps.data.spk2id[lang] + sr = tts.hps.data.sampling_rate load_ms = int((time.monotonic() - t0) * 1000) + import numpy as np + import soundfile + + _GAP = np.zeros(int(sr * 0.12), dtype=np.float32) # 120 ms between segments + + def _pitch_shift(audio, semitones: float): + if not semitones: + return audio + import librosa + + return librosa.effects.pitch_shift( + audio.astype(np.float32), sr=sr, n_steps=float(semitones) + ) + + def _synth_segments(segments: list[dict], out: str) -> None: + """Synthesize each segment, pitch-shift it, and concatenate to one wav.""" + pieces = [] + for i, seg in enumerate(segments): + text = seg["text"] + if not text.strip(): + continue + speed = float(seg.get("speed", 1.0)) + pitch = float(seg.get("pitch", 0.0)) + audio = tts.tts_to_file(text, speaker_id, None, speed=speed) + audio = _pitch_shift(np.asarray(audio, dtype=np.float32), pitch) + if pieces: + pieces.append(_GAP) + pieces.append(audio) + if not pieces: + raise ValueError("no speakable segment") + soundfile.write(out, np.concatenate(pieces), sr) + # Warm up before signalling ready: the first CUDA synth pays a large lazy # cost (kernel autotune/cudnn), ~10s cold vs ~130ms hot, which would blow the # voice loop's ~1s budget on the very first reply. Do that dummy synth here so @@ -78,6 +118,11 @@ def main() -> None: os.makedirs(os.path.dirname(warm_out), exist_ok=True) w = time.monotonic() tts.tts_to_file("워밍업", speaker_id, warm_out, speed=1.3) + # Also JIT-warm librosa's pitch shifter (first call pays ~0.4s numba + # compile) so the first *emotional* reply doesn't stall. + import librosa + + librosa.effects.pitch_shift(np.zeros(sr, dtype=np.float32), sr=sr, n_steps=1.0) warmup_ms = int((time.monotonic() - w) * 1000) except Exception as exc: _log(f"[melo_worker] warmup skipped: {exc}") @@ -91,13 +136,15 @@ def main() -> None: continue try: req = json.loads(line) - text = req["text"] out = req["out"] - speed = float(req.get("speed", 1.0)) if out.startswith("/tmp") or out.startswith("/dev/shm"): raise ValueError(f"refusing RAM-backed tmpfs path: {out}") s = time.monotonic() - tts.tts_to_file(text, speaker_id, out, speed=speed) + if "segments" in req: + _synth_segments(req["segments"], out) + else: # legacy single-utterance form + speed = float(req.get("speed", 1.0)) + tts.tts_to_file(req["text"], speaker_id, out, speed=speed) ms = int((time.monotonic() - s) * 1000) _emit({"ok": True, "out": out, "ms": ms}) except Exception as exc: # keep the worker alive across bad requests