tuimble/tests/tools/voice-smoke

#!/usr/bin/env python3
"""Headless voice smoke test against a live Mumble server.

Connects to a server, starts the audio pipeline, captures mic for a few
seconds, sends encoded frames, listens for incoming audio, then disconnects.
No TUI needed — pure terminal output.

Usage:
  voice-smoke HOST [--port PORT] [--user NAME] [--seconds SEC]
  voice-smoke --help
"""

from __future__ import annotations

import argparse
import sys
import time

RST = "\033[0m"
DIM = "\033[2m"
GRN = "\033[38;5;108m"
RED = "\033[38;5;131m"
YEL = "\033[38;5;179m"
CYN = "\033[38;5;109m"
BLD = "\033[1m"


def ok(msg: str) -> None:
    print(f"  {GRN}\u2713{RST} {msg}")


def fail(msg: str) -> None:
    print(f"  {RED}\u2717{RST} {msg}")


def warn(msg: str) -> None:
    print(f"  {YEL}\u26a0{RST} {msg}")


def info(msg: str) -> None:
    print(f"  {DIM}{msg}{RST}")


def heading(msg: str) -> None:
    print(f"\n{CYN}{msg}{RST}")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Headless voice smoke test",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("host", help="mumble server hostname")
    parser.add_argument("--port", type=int, default=64738, help="server port")
    parser.add_argument("--user", default="tuimble-test", help="username")
    parser.add_argument(
        "--seconds", type=float, default=5.0,
        help="capture+send duration (default: 5)",
    )
    parser.add_argument(
        "--version", action="version", version="voice-smoke 0.1.0",
    )
    args = parser.parse_args()

    print(f"{CYN}voice smoke test{RST}  {DIM}{args.host}:{args.port}{RST}")

    # -- dependencies --------------------------------------------------------

    heading("Prerequisites")
    try:
        import sounddevice as sd  # noqa: F401

        ok("sounddevice + PortAudio")
    except OSError as e:
        fail(f"PortAudio: {e}")
        return 1

    try:
        import opuslib  # noqa: F401

        ok("opuslib")
    except ImportError:
        fail("opuslib not installed")
        return 1

    from tuimble.audio import AudioPipeline
    from tuimble.client import MumbleClient

    ok("tuimble imports")

    # -- connect -------------------------------------------------------------

    heading("Connection")
    info(f"connecting to {args.host}:{args.port} as {args.user}...")

    client = MumbleClient(
        host=args.host, port=args.port, username=args.user,
    )

    rx_stats = {"frames": 0, "bytes": 0}

    def on_sound(_user, pcm_data: bytes) -> None:
        rx_stats["frames"] += 1
        rx_stats["bytes"] += len(pcm_data)

    client.on_sound_received = on_sound

    try:
        client.connect()
    except Exception as e:
        fail(f"connect: {e}")
        return 1

    ok(f"connected (channels={len(client.channels)} users={len(client.users)})")

    for cid, ch in client.channels.items():
        n_users = sum(
            1 for u in client.users.values() if u.channel_id == cid
        )
        info(f"  {ch.name} ({n_users} users)")

    # -- audio pipeline ------------------------------------------------------

    heading("Audio Pipeline")
    audio = AudioPipeline()

    try:
        audio.start()
        ok("streams opened (input + output)")
    except Exception as e:
        fail(f"audio start: {e}")
        client.disconnect()
        return 1

    # -- capture + send ------------------------------------------------------

    heading(f"Transmit ({args.seconds:.1f}s)")
    info("capturing mic + encoding + sending to server...")

    audio.capturing = True
    tx_frames = 0
    deadline = time.monotonic() + args.seconds

    try:
        while time.monotonic() < deadline:
            frame = audio.get_capture_frame()
            if frame is not None:
                client.send_audio(frame)
                tx_frames += 1
            else:
                time.sleep(0.005)
    except KeyboardInterrupt:
        pass

    audio.capturing = False
    expected = int(args.seconds * 48000 / 960)
    pct = (tx_frames / expected * 100) if expected else 0

    if tx_frames == 0:
        warn(f"sent 0 frames (expected ~{expected})")
        info("mic may be muted or no input device")
    elif pct < 80:
        warn(f"sent {tx_frames} frames ({pct:.0f}% of expected ~{expected})")
    else:
        ok(f"sent {tx_frames} frames ({pct:.0f}% of ~{expected})")

    # -- listen for incoming -------------------------------------------------

    heading("Receive (2s)")
    info("listening for incoming voice...")
    time.sleep(2.0)

    if rx_stats["frames"] > 0:
        ok(f"received {rx_stats['frames']} frames "
           f"({rx_stats['bytes']} bytes)")
    else:
        info("no incoming audio (normal if alone on server)")

    # -- cleanup -------------------------------------------------------------

    heading("Cleanup")
    audio.stop()
    ok("audio stopped")
    client.disconnect()
    ok("disconnected")

    # -- summary -------------------------------------------------------------

    heading("Summary")
    results = []
    if tx_frames > 0:
        results.append(f"tx={tx_frames}")
    if rx_stats["frames"] > 0:
        results.append(f"rx={rx_stats['frames']}")

    if results:
        ok(f"voice pipeline working ({', '.join(results)})")
    else:
        warn("no audio transmitted or received")
        info("check: mic not muted, other users in channel")

    return 0


if __name__ == "__main__":
    sys.exit(main())