Update AI Twin chat client to use cloned voice from /local-tts

chat_twin.py

@@ -1,4 +1,4 @@
-#!/home/jason/anaconda3/envs/env_twin/bin/python3
+#!/usr/bin/env python3
 import sys
 import os
 
@@ -20,6 +20,7 @@ import subprocess
 import threading
 import queue
 import time
+import shutil
 import tempfile
 
 # --- ANSI Terminal Colors ---
@@ -38,7 +39,7 @@ RECORD_FILE = "/tmp/chat_twin_record.wav"
 
 # Speech/TTS/STT Configuration
 tts_config = {
-    "voice": "male1",    # Options: male1, male2, male3, female1, female2, female3, child_male, child_female
+    "voice": "cloned",   # Options: cloned, male1, male2, male3, female1, female2, female3, child_male, child_female
     "rate": "0",         # -100 to 100
     "pitch": "0",        # -100 to 100
     "volume": "0",       # -100 to 100
@@ -46,6 +47,36 @@ tts_config = {
     "fallback_espeak": False
 }
 
+# Voice Cloning & Cloned Speaker Paths (/local-tts)
+CLONED_VOICE_CANDIDATES = [
+    "/home/jason/local-tts/cloned_output.wav",
+    "/home/jason/local-tts/my_voice_clean.wav",
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "cloned_output.wav"),
+    os.path.join(os.path.dirname(os.path.abspath(__file__)), "my_voice_clean.wav"),
+]
+
+def get_cloned_voice_path():
+    """Locate the cloned speaker reference audio file."""
+    for p in CLONED_VOICE_CANDIDATES:
+        if os.path.exists(p) and os.path.getsize(p) > 1000:
+            return p
+    return None
+
+TTS_CMD_CANDIDATES = [
+    "/home/jason/miniconda3/envs/env_twin/bin/tts",
+    "/home/jason/.local/bin/tts",
+    "/home/jason/local-tts/tts-env/bin/tts",
+    "/home/jason/local-tts/bin/tts",
+]
+
+def get_tts_cmd():
+    """Locate the TTS binary for voice cloning synthesis."""
+    for c in TTS_CMD_CANDIDATES:
+        if c and os.path.exists(c):
+            return c
+    return shutil.which("tts")
+
+
 # --- Lazy-Loaded Speech-to-Text (STT) ---
 asr_pipeline = None
 
@@ -90,9 +121,6 @@ def clean_markdown(text):
     text = re.sub(r'\s+', ' ', text)
     return text.strip()
 
-CLONED_SPEAKER_WAV = "/home/jason/local-tts/cloned_output.wav"
-TTS_CMD = "/home/jason/anaconda3/envs/tts-backend/bin/tts"
-
 def speak_text(text):
     """Executes the TTS system commands to say the text using cloned voice."""
     if not tts_config["enabled"]:
@@ -102,39 +130,52 @@ def speak_text(text):
     if not clean_text:
         return
 
-    # Attempt voice-cloned TTS using XTTS v2 and cloned_output.wav
-    if os.path.exists(TTS_CMD) and os.path.exists(CLONED_SPEAKER_WAV) and not tts_config.get("fallback_espeak", False):
-        try:
-            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_file:
-                tmp_wav = tmp_file.name
-
-            cmd = [
-                TTS_CMD,
-                "--model_name", "tts_models/multilingual/multi-dataset/xtts_v2",
-                "--text", clean_text,
-                "--speaker_wav", CLONED_SPEAKER_WAV,
-                "--language_idx", "en",
-                "--out_path", tmp_wav,
-                "--use_cuda", "true"
-            ]
-            res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=45)
-            if res.returncode == 0 and os.path.exists(tmp_wav):
-                play_res = subprocess.run(["paplay", tmp_wav], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-                if play_res.returncode != 0:
-                    subprocess.run(["aplay", tmp_wav], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-                try:
-                    os.remove(tmp_wav)
-                except Exception:
-                    pass
-                return
-        except Exception:
-            pass
-
-    # Prepare command for spd-say
+    # 1. Attempt Voice-Cloned TTS using XTTS v2 and cloned audio from /local-tts
+    if tts_config["voice"] == "cloned" and not tts_config.get("fallback_espeak", False):
+        cloned_wav = get_cloned_voice_path()
+        tts_cmd = get_tts_cmd()
+        if tts_cmd and cloned_wav:
+            tmp_wav = None
+            try:
+                with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_file:
+                    tmp_wav = tmp_file.name
+
+                cmd = [
+                    tts_cmd,
+                    "--model_name", "tts_models/multilingual/multi-dataset/xtts_v2",
+                    "--text", clean_text,
+                    "--speaker_wav", cloned_wav,
+                    "--language_idx", "en",
+                    "--out_path", tmp_wav,
+                    "--use_cuda", "true"
+                ]
+                env = os.environ.copy()
+                env["COQUI_TOS_AGREED"] = "1"
+                res = subprocess.run(cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=45)
+                if res.returncode == 0 and os.path.exists(tmp_wav) and os.path.getsize(tmp_wav) > 1000:
+                    play_res = subprocess.run(["paplay", tmp_wav], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+                    if play_res.returncode != 0:
+                        subprocess.run(["aplay", "-q", tmp_wav], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+                    try:
+                        os.remove(tmp_wav)
+                    except Exception:
+                        pass
+                    return
+            except Exception:
+                pass
+            finally:
+                if tmp_wav and os.path.exists(tmp_wav):
+                    try:
+                        os.remove(tmp_wav)
+                    except Exception:
+                        pass
+
+    # 2. Prepare command for spd-say (standard voice fallback)
     if not tts_config["fallback_espeak"]:
         cmd = ["spd-say", "-w"]  # -w waits until speaking is finished
-        if tts_config["voice"]:
-            cmd.extend(["-t", tts_config["voice"]])
+        voice_target = tts_config["voice"] if tts_config["voice"] != "cloned" else "male1"
+        if voice_target:
+            cmd.extend(["-t", voice_target])
         if tts_config["rate"]:
             cmd.extend(["-r", tts_config["rate"]])
         if tts_config["pitch"]:
@@ -144,19 +185,22 @@ def speak_text(text):
         cmd.append(clean_text)
         
         try:
-            subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            if res.returncode == 0:
+                return
         except Exception:
             # Fallback to espeak-ng if spd-say fails
             tts_config["fallback_espeak"] = True
             speak_text(text)
+            return
     else:
-        # Fallback to espeak-ng
+        # 3. Fallback to espeak-ng
         cmd = ["espeak-ng"]
         try:
             rate_val = int(tts_config["rate"])
             wpm = 175 + int(rate_val * 0.8)
             cmd.extend(["-s", str(wpm)])
-        except:
+        except Exception:
             pass
         cmd.append(clean_text)
         try:
@@ -165,7 +209,7 @@ def speak_text(text):
             print(f"\n{C_YELLOW}[TTS Error: {e}]{C_RESET}")
 
 def stop_speech():
-    """Cancels any ongoing speech and flushes the queue."""
+    """Cancels any ongoing speech, terminates playback processes, and flushes the queue."""
     # Clear the queue
     while not tts_queue.empty():
         try:
@@ -174,16 +218,17 @@ def stop_speech():
         except queue.Empty:
             break
     
+    # Kill any audio players or ongoing tts synthesis processes
+    for proc in ["paplay", "aplay", "tts", "espeak-ng"]:
+        try:
+            subprocess.run(["pkill", "-x", proc], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        except Exception:
+            pass
+            
     # Send cancel to speech dispatcher
     try:
         subprocess.run(["spd-say", "-C"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-    except:
-        pass
-    
-    # Terminate any running espeak-ng processes
-    try:
-        subprocess.run(["pkill", "-x", "espeak-ng"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-    except:
+    except Exception:
         pass
 
 def tts_worker():
@@ -254,7 +299,7 @@ def transcribe_audio(filename=RECORD_FILE):
 def print_help():
     print(f"\n{C_MAGENTA}{C_BOLD}--- Digital Twin Help Menu ---{C_RESET}")
     print(f"  {C_BOLD}/stop{C_RESET} or {C_BOLD}/s{C_RESET}       : Stop current speaking immediately.")
-    print(f"  {C_BOLD}/voice <type>{C_RESET}   : Set voice (male1, male2, male3, female1, female2, female3, child_male, child_female).")
+    print(f"  {C_BOLD}/voice <type>{C_RESET}   : Set voice (cloned [from /local-tts], male1, male2, male3, female1, female2, female3, child_male, child_female).")
     print(f"  {C_BOLD}/rate <value>{C_RESET}   : Set speech rate (-100 to 100, e.g. -20, 0, 20).")
     print(f"  {C_BOLD}/clear{C_RESET} or {C_BOLD}/c{C_RESET}     : Clear screen and reset conversation history.")
     print(f"  {C_BOLD}/tts <on|off>{C_RESET}   : Enable or disable Text-to-Speech.")
@@ -308,6 +353,11 @@ def main():
         return
 
     print(f"Twin model '{C_BOLD}{MODEL_NAME}{C_RESET}' loaded.")
+    cloned_path = get_cloned_voice_path()
+    if cloned_path and tts_config["voice"] == "cloned":
+        print(f"Voice Profile: {C_BOLD}{C_GREEN}Jason's Cloned Voice ({cloned_path}){C_RESET}")
+    else:
+        print(f"Voice Profile: {C_BOLD}{C_MAGENTA}{tts_config['voice']}{C_RESET}")
     print(f"Two-Way Voice active. Type text OR press {C_BOLD}[ENTER]{C_RESET} to speak. (Type {C_BOLD}/help{C_RESET} for commands)\n")
     
     start_tts_system()
@@ -325,9 +375,7 @@ def main():
             
             # If input is empty, enter Voice Input Mode
             if not user_input:
-                if os.path.exists("/home/jason/coral/mic_active.wav"):
-                    subprocess.run(["aplay", "-q", "/home/jason/coral/mic_active.wav"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-                print(f"{C_YELLOW}[🎙️ Mic Active - Recording... Press ENTER to stop recording]{C_RESET}", end="", flush=True)
+                print(f"{C_YELLOW}[Recording... Press ENTER to stop recording]{C_RESET}", end="", flush=True)
                 
                 # Start recording
                 record_proc = record_audio()
@@ -384,14 +432,22 @@ def main():
                 elif cmd == "/voice":
                     if len(cmd_parts) > 1:
                         new_voice = cmd_parts[1].lower()
-                        valid_voices = ["male1", "male2", "male3", "female1", "female2", "female3", "child_male", "child_female"]
+                        valid_voices = ["cloned", "male1", "male2", "male3", "female1", "female2", "female3", "child_male", "child_female"]
                         if new_voice in valid_voices:
                             tts_config["voice"] = new_voice
-                            print(f"{C_MAGENTA}Voice updated to: {new_voice}{C_RESET}")
+                            if new_voice == "cloned":
+                                p = get_cloned_voice_path()
+                                print(f"{C_MAGENTA}Voice updated to: Jason's Cloned Voice ({p}){C_RESET}")
+                            else:
+                                print(f"{C_MAGENTA}Voice updated to: {new_voice}{C_RESET}")
                         else:
                             print(f"{C_YELLOW}Invalid voice. Select from: {', '.join(valid_voices)}{C_RESET}")
                     else:
-                        print(f"{C_MAGENTA}Current voice: {tts_config['voice']}{C_RESET}")
+                        if tts_config['voice'] == 'cloned':
+                            p = get_cloned_voice_path()
+                            print(f"{C_MAGENTA}Current voice: Jason's Cloned Voice ({p}){C_RESET}")
+                        else:
+                            print(f"{C_MAGENTA}Current voice: {tts_config['voice']}{C_RESET}")
                     continue
                 elif cmd == "/rate":
                     if len(cmd_parts) > 1: