rtaa.py

#!/usr/bin/env python3
"""
Realtime Agent Launcher
Starts:
 - assemblyai_multistream_to_log.py (streams a stereo file to AssemblyAI and logs END-OF-TURN lines to assemblyai.log)
 - log_agents.py (tails assemblyai.log and writes transcripts.log, customer.log, recommendations.log)

Shuts down both on Ctrl+C.
"""

import os
import sys
import signal
import subprocess
import time
import argparse
import threading
from dotenv import load_dotenv

load_dotenv()

def main(input_source: str, mode: str, filepath: str):
    python_bin = sys.executable or "python3"

    # Verify API key early
    if not os.getenv("ASSEMBLYAI_API_KEY"):
        print("❌ ASSEMBLYAI_API_KEY not found in environment variables")
        print("   Make sure your .env contains ASSEMBLYAI_API_KEY and the env is loaded.")
        # Continue anyway; child scripts may also handle this

    procs = []

    try:
        # Start AssemblyAI transcription
        cmd = [python_bin, "-u", "transcription.py", "--mode", mode, "-i", input_source]
        if mode == "file":
            cmd += ["-f", filepath]
        env = os.environ.copy()
        env["PYTHONUNBUFFERED"] = "1"
        # Let transcription.py write directly to this console (no buffering/alteration)
        p1 = subprocess.Popen(
            cmd,
            env=env,
        )
        # Do not pipe its stdout; it will print directly like when run alone

        # Give it a short head start to create assemblyai.log
        time.sleep(0.5)

        # Start log agents orchestrator
        p2 = subprocess.Popen(
            [python_bin, "-u", "agents.py"],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
            env=env,
        )
        procs.append(("agents", p2))

        print("🚀 Realtime agent started. Press Ctrl+C to stop.")

        # Start dedicated reader threads for child stdout (agents only)
        def pump_output(name: str, proc: subprocess.Popen):
            try:
                if not proc.stdout:
                    return
                # Line-based for agents
                while True:
                    line = proc.stdout.readline()
                    if not line:
                        break
                    print(f"[{name}] {line}", end="")
            except Exception:
                pass

        readers = []
        for name, p in procs:
            t = threading.Thread(target=pump_output, args=(name, p), daemon=True)
            t.start()
            readers.append(t)

        # Wait for processes to exit
        while procs or (p1 and p1.poll() is None):
            alive = []
            for name, p in procs:
                if p.poll() is None:
                    alive.append((name, p))
                else:
                    print(f"[{name}] exited with code {p.returncode}")
            procs = alive
            time.sleep(0.1)

    except KeyboardInterrupt:
        print("\n🛑 Ctrl+C received. Stopping...")
    finally:
        # Terminate children gracefully, then kill if needed
        for name, p in procs:
            try:
                p.terminate()
            except Exception:
                pass
        # Give them a moment
        time.sleep(1.0)
        for name, p in procs:
            if p.poll() is None:
                try:
                    p.kill()
                except Exception:
                    pass
        print("✅ Shutdown complete")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Realtime Agent Launcher")
    parser.add_argument("-i", "--input", choices=["blackhole", "microphone"], default="blackhole", help="Audio input source (default: blackhole)")
    parser.add_argument("--mode", "-m", choices=["live", "file"], default="file", help="Streaming mode: file-based stereo (default) or live input")
    parser.add_argument("-f", "--filepath", default="./sample_call.wav", help="Stereo WAV file path for --mode file (default: sample_call.wav)")
    args = parser.parse_args()

    # Ensure child processes also receive SIGINT
    try:
        # On POSIX, set process group so children get signals
        if hasattr(os, 'setpgrp'):
            os.setpgrp()
    except Exception:
        pass
    main(args.input, args.mode, args.filepath)