openclaw/tools/voice-surface/train_james.py

#!/usr/bin/env python3
"""
Train a custom "James" wake word model using openWakeWord embeddings + sklearn.

Steps:
  1. Synthesize positive samples ("James", "Hey James", …) via edge-tts
  2. Synthesize negative samples (common words that aren't the wake word)
  3. Extract 96-dim embeddings with openWakeWord's built-in embedding model
  4. Train LogisticRegression classifier
  5. Save to models/james_classifier.pkl  (loaded by wake_detector.py)

Usage:  python3 train_james.py
"""
import os, sys, subprocess, tempfile, pathlib, warnings
warnings.filterwarnings("ignore")
os.environ["ORT_LOGGING_LEVEL"] = "3"

import numpy as np
import joblib
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report
from openwakeword.utils import AudioFeatures

MODELS_DIR = pathlib.Path(__file__).parent / "models"
MODELS_DIR.mkdir(exist_ok=True)
OUT_MODEL = MODELS_DIR / "james_classifier.pkl"

# ---- edge-tts voices to vary accent / timbre ----
VOICES = [
    "en-US-GuyNeural",
    "en-US-DavisNeural",
    "en-US-ChristopherNeural",
    "en-US-EricNeural",
    "en-GB-RyanNeural",
    "en-AU-WilliamNeural",
    "en-IN-PrabhatNeural",
]

# ---- wake word phrases (positive class) ----
POSITIVES = [
    "James",
    "Hey James",
    "James please",
    "OK James",
    "James listen",
    "James hello",
    "James wake up",
    "Hi James",
]

# ---- negative phrases (things that sound vaguely similar) ----
NEGATIVES = [
    "Jane",
    "Games",
    "Frames",
    "Names",
    "Flames",
    "Claims",
    "Aims",
    "Hey",
    "OK",
    "Hello",
    "Stop",
    "What time is it",
    "Turn off the lights",
    "Play music",
    "Set a timer",
    "Good morning",
    "Thank you",
    "Yes please",
    "No thanks",
    "How are you",
    "Silence",
    "Computer",
    "Alexa",
    "Siri",
    "Google",
]


def synth(text: str, voice: str, tmpdir: str) -> str | None:
    """Synthesize text with edge-tts → 16kHz mono WAV. Returns path or None."""
    name = text.replace(" ", "_").replace("'", "") + "_" + voice.split("-")[1]
    mp3_path = os.path.join(tmpdir, name + ".mp3")
    wav_path = os.path.join(tmpdir, name + ".wav")

    r = subprocess.run(
        ["edge-tts", "--voice", voice, "--text", text, "--write-media", mp3_path],
        capture_output=True, timeout=15,
    )
    if r.returncode != 0:
        return None

    r2 = subprocess.run(
        ["ffmpeg", "-y", "-i", mp3_path, "-ar", "16000", "-ac", "1", wav_path],
        capture_output=True, timeout=10,
    )
    if r2.returncode != 0:
        return None

    os.unlink(mp3_path)
    return wav_path


EMBED_SAMPLES = 16000 * 2  # 2 seconds at 16kHz — pad/trim each clip to this length

def load_audio(path: str) -> np.ndarray | None:
    """Load WAV as 16kHz mono int16, padded/trimmed to EMBED_SAMPLES."""
    import soundfile as sf
    data, _ = sf.read(path, dtype="int16")
    if data.ndim > 1:
        data = data[:, 0]
    # Pad or trim to fixed length
    if len(data) < EMBED_SAMPLES:
        data = np.pad(data, (0, EMBED_SAMPLES - len(data)))
    else:
        data = data[:EMBED_SAMPLES]
    return data


def extract_embeddings(paths: list[str], af: AudioFeatures) -> np.ndarray:
    """Load WAV files and extract 96-dim embedding per clip."""
    clips = []
    good_paths = []
    for p in paths:
        try:
            clips.append(load_audio(p))
            good_paths.append(p)
        except Exception as ex:
            print(f"  skip {os.path.basename(p)}: {ex}")

    if not clips:
        return np.zeros((0, 96))

    # Stack into (N, samples) array — required by embed_clips
    X = np.stack(clips, axis=0)   # shape: (N, EMBED_SAMPLES)
    embs = af.embed_clips(X)      # returns (N, frames, 96) — mean over time → (N, 96)
    embs = np.array(embs)
    if embs.ndim == 3:
        embs = embs.mean(axis=1)  # (N, frames, 96) → (N, 96)
    return embs


def main():
    # Check soundfile is available
    try:
        import soundfile  # noqa
    except ImportError:
        print("Installing soundfile…")
        subprocess.run([sys.executable, "-m", "pip", "install", "soundfile",
                        "--break-system-packages", "-q"])
        import soundfile  # noqa

    af = AudioFeatures()

    with tempfile.TemporaryDirectory() as tmpdir:
        print(f"\n=== Synthesizing {len(POSITIVES)} positive × {len(VOICES)} voices ===")
        pos_paths = []
        for phrase in POSITIVES:
            for voice in VOICES:
                p = synth(phrase, voice, tmpdir)
                if p:
                    pos_paths.append(p)
                    print(f"  + {phrase!r} [{voice.split('-')[1]}]")
                else:
                    print(f"  SKIP {phrase!r} [{voice}]")

        print(f"\n=== Synthesizing {len(NEGATIVES)} negative × {len(VOICES)} voices ===")
        neg_paths = []
        for phrase in NEGATIVES:
            for voice in VOICES:
                p = synth(phrase, voice, tmpdir)
                if p:
                    neg_paths.append(p)
                else:
                    print(f"  SKIP {phrase!r}")

        print(f"\n=== Extracting embeddings ({len(pos_paths)} pos, {len(neg_paths)} neg) ===")
        X_pos = extract_embeddings(pos_paths, af)
        X_neg = extract_embeddings(neg_paths, af)

        print(f"Positive embeddings: {X_pos.shape}")
        print(f"Negative embeddings: {X_neg.shape}")

        if len(X_pos) == 0:
            print("ERROR: No positive samples extracted. Check edge-tts and ffmpeg.")
            sys.exit(1)

    X = np.vstack([X_pos, X_neg])
    y = np.array([1] * len(X_pos) + [0] * len(X_neg))

    # Shuffle
    idx = np.random.permutation(len(X))
    X, y = X[idx], y[idx]

    print(f"\n=== Training classifier (total {len(X)} samples) ===")
    clf = Pipeline([
        ("scaler", StandardScaler()),
        ("lr", LogisticRegression(C=1.0, max_iter=1000, class_weight="balanced")),
    ])
    clf.fit(X, y)

    # Quick eval on training set (no held-out set — small data)
    y_pred = clf.predict(X)
    print(classification_report(y, y_pred, target_names=["not-james", "james"]))

    # Check probability threshold
    probs = clf.predict_proba(X[y == 1])[:, 1]
    print(f"Positive sample probs — min: {probs.min():.2f}  mean: {probs.mean():.2f}  max: {probs.max():.2f}")

    joblib.dump(clf, OUT_MODEL)
    print(f"\n✅ Model saved to {OUT_MODEL}")
    print("Restart the voice-surface server to activate it.")


if __name__ == "__main__":
    main()