homeai/homeai-visual/scripts/test-expressions.py

#!/usr/bin/env python3
"""
Test script for VTube Studio Expression Bridge.

Usage:
    python3 test-expressions.py              # test all expressions
    python3 test-expressions.py --auth       # run auth flow first
    python3 test-expressions.py --lipsync    # test lip sync parameter
    python3 test-expressions.py --latency    # measure round-trip latency

Requires the vtube-bridge to be running on port 8002.
"""

import argparse
import json
import sys
import time
import urllib.request

BRIDGE_URL = "http://localhost:8002"

EXPRESSIONS = ["idle", "listening", "thinking", "speaking", "happy", "sad", "surprised", "error"]


def _post(path: str, data: dict | None = None) -> dict:
    body = json.dumps(data or {}).encode()
    req = urllib.request.Request(
        f"{BRIDGE_URL}{path}",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return json.loads(resp.read())


def _get(path: str) -> dict:
    req = urllib.request.Request(f"{BRIDGE_URL}{path}")
    with urllib.request.urlopen(req, timeout=10) as resp:
        return json.loads(resp.read())


def check_bridge():
    """Verify bridge is running and connected."""
    try:
        status = _get("/status")
        print(f"Bridge status: connected={status['connected']}, authenticated={status['authenticated']}")
        print(f"  Expressions: {', '.join(status.get('expressions', []))}")
        if not status["connected"]:
            print("\n  WARNING: Not connected to VTube Studio. Is it running?")
        if not status["authenticated"]:
            print("  WARNING: Not authenticated. Run with --auth to initiate auth flow.")
        return status
    except Exception as e:
        print(f"ERROR: Cannot reach bridge at {BRIDGE_URL}: {e}")
        print("  Is vtube-bridge.py running?")
        sys.exit(1)


def run_auth():
    """Initiate auth flow — user must click Allow in VTube Studio."""
    print("Requesting authentication token...")
    print("  >>> Click 'Allow' in VTube Studio when prompted <<<")
    result = _post("/auth")
    print(f"  Result: {json.dumps(result, indent=2)}")
    return result


def test_expressions(delay: float = 2.0):
    """Cycle through all expressions with a pause between each."""
    print(f"\nCycling through {len(EXPRESSIONS)} expressions ({delay}s each):\n")
    for expr in EXPRESSIONS:
        print(f"  → {expr}...", end=" ", flush=True)
        t0 = time.monotonic()
        result = _post("/expression", {"event": expr})
        dt = (time.monotonic() - t0) * 1000
        if result.get("ok"):
            print(f"OK ({dt:.0f}ms)")
        else:
            print(f"FAILED: {result.get('error', 'unknown')}")
        time.sleep(delay)

    # Return to idle
    _post("/expression", {"event": "idle"})
    print("\n  Returned to idle.")


def test_lipsync(duration: float = 3.0):
    """Simulate lip sync by sweeping MouthOpen 0→1→0."""
    import math
    print(f"\nTesting lip sync (MouthOpen sweep, {duration}s)...\n")

    fps = 20
    frames = int(duration * fps)
    for i in range(frames):
        t = i / frames
        # Sine wave for smooth open/close
        value = abs(math.sin(t * math.pi * 4))
        value = round(value, 3)
        _post("/parameter", {"name": "MouthOpen", "value": value})
        print(f"\r  MouthOpen = {value:.3f}", end="", flush=True)
        time.sleep(1.0 / fps)

    _post("/parameter", {"name": "MouthOpen", "value": 0.0})
    print("\r  MouthOpen = 0.000 (done)          ")


def test_latency(iterations: int = 20):
    """Measure expression trigger round-trip latency."""
    print(f"\nMeasuring latency ({iterations} iterations)...\n")
    times = []
    for i in range(iterations):
        expr = "thinking" if i % 2 == 0 else "idle"
        t0 = time.monotonic()
        _post("/expression", {"event": expr})
        dt = (time.monotonic() - t0) * 1000
        times.append(dt)
        print(f"  {i+1:2d}. {expr:10s} → {dt:.1f}ms")

    avg = sum(times) / len(times)
    mn = min(times)
    mx = max(times)
    print(f"\n  Avg: {avg:.1f}ms  Min: {mn:.1f}ms  Max: {mx:.1f}ms")
    if avg < 100:
        print("  PASS: Average latency under 100ms target")
    else:
        print("  WARNING: Average latency exceeds 100ms target")

    # Return to idle
    _post("/expression", {"event": "idle"})


def main():
    parser = argparse.ArgumentParser(description="VTube Studio Expression Bridge Tester")
    parser.add_argument("--auth", action="store_true", help="Run auth flow")
    parser.add_argument("--lipsync", action="store_true", help="Test lip sync parameter sweep")
    parser.add_argument("--latency", action="store_true", help="Measure round-trip latency")
    parser.add_argument("--delay", type=float, default=2.0, help="Delay between expressions (default: 2s)")
    parser.add_argument("--all", action="store_true", help="Run all tests")
    args = parser.parse_args()

    print("VTube Studio Expression Bridge Tester")
    print("=" * 42)

    status = check_bridge()

    if args.auth:
        run_auth()
        print()
        status = check_bridge()

    if not status.get("authenticated") and not args.auth:
        print("\nNot authenticated — skipping expression tests.")
        print("Run with --auth to authenticate, or start VTube Studio first.")
        return

    if args.all:
        test_expressions(args.delay)
        test_lipsync()
        test_latency()
    elif args.lipsync:
        test_lipsync()
    elif args.latency:
        test_latency()
    else:
        test_expressions(args.delay)


if __name__ == "__main__":
    main()