Upload app_hf.py

app_hf.py

@@ -0,0 +1,186 @@
+# app.py
+
+import threading
+import time
+import os, json
+import glob
+import atexit
+from nicegui import ui, app
+
+from LivePortraitIdle import generate_fixed_chunks
+from LatentSync import run_latentsync_inference
+from SparkTTS import run_tts
+from Dual_LLM import generate_darwin_response
+from PlaylistManager import idle_playlist_maker, response_playlist_maker, create_lipsync_playlist
+from ui import build_ui  # Import build_ui from the ui.py file
+
+REPO_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))  
+STREAM_LIVE_DIR = os.path.join(REPO_DIR, "stream", "live")
+STREAM_SPEECH_DIR = os.path.join(REPO_DIR, "stream", "speech")
+PLAYLIST_PATH = os.path.join("stream", "playlist", "playlist.json")
+
+# === Thread management ===
+_main_thread = None
+_thread_lock = threading.Lock()
+_shutdown_flag = False
+
+# === State Flags ===
+awaiting_response = False
+user_prompt = ""  # Store the user's prompt
+
+# Expose media folder
+app.add_static_files('/stream', os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'stream')))
+
+def get_latest_video_from_live_dir():
+    """Get the most recent video from the live directory"""
+    video_files = glob.glob(os.path.join(STREAM_LIVE_DIR, "*.mp4"))
+    if not video_files:
+        return None
+    return max(video_files, key=os.path.getctime)
+
+def update_playlist_with_single_video(video_path):
+    """Create a playlist with just the specified video"""
+    # Convert absolute path to relative path for playlist
+    try:
+        # Get the path relative to the REPO_DIR
+        relative_path = os.path.relpath(video_path, os.path.join(REPO_DIR))
+        
+        # Important fix: Remove any "stream" prefix to avoid path duplication
+        if relative_path.startswith("stream\\") or relative_path.startswith("stream/"):
+            relative_path = relative_path[7:]  # Skip past "stream/" or "stream\"
+        
+        # Convert backslashes to forward slashes for web paths
+        web_path = relative_path.replace("\\", "/")
+        
+        # Create a playlist with just this video
+        playlist = [web_path]
+        
+        # Save the playlist
+        with open(PLAYLIST_PATH, "w") as f:
+            json.dump(playlist, f)
+        
+        print(f"[PLAYLIST] Updated playlist with single video: {web_path}")
+        return True
+    except Exception as e:
+        print(f"[ERROR] Failed to update playlist: {e}")
+        return False
+
+def idle_mode():
+    print("[MAIN] Starting idle mode...")
+    idle_playlist_maker()
+    #generate_fixed_chunks(mode="talking", chunks_per_video=1, video_limit=1)
+
+    print("[MAIN] Idle mode finished, waiting for response trigger...")
+    return
+
+def response_mode():
+    global user_prompt
+    print("[MAIN] Starting response mode...")
+    
+    # Step 1: Generate response from the LLM
+    print("[LLM] Generating Darwin's response...")
+    if not user_prompt:
+        print("[WARNING] No user prompt provided, using default")
+        user_prompt = "Tell me about your theory of evolution."
+        
+    llm_response = generate_darwin_response(user_prompt)
+    print(f"[LLM] Response generated: {llm_response[:500]}...")
+    
+    # Step 2: Convert response to speech using TTS
+    print("[TTS] Converting text to speech...")
+    speech_file = run_tts(text=llm_response)
+    print(f"[TTS] Speech generated: {speech_file}")
+    
+    # Step 3: Generate lip-synced video
+    print("[SYNC] Running lip sync...")
+    success = run_latentsync_inference()
+    print(f"[SYNC] Lip sync completed with status: {success}")
+    
+    # Step 4: If successful, get the latest video and make it the only item in the playlist
+    if success:
+        # Get the latest video from the LIVE_DIR
+        latest_video = get_latest_video_from_live_dir()
+        if latest_video:
+            print(f"[PLAYBACK] Found latest video: {latest_video}")
+            # Update playlist to only include this video, forcing playback
+            update_playlist_with_single_video(latest_video)
+            # The updated UI will automatically detect playlist changes and play the video
+        else:
+            print("[ERROR] No video found in live directory after sync completion")
+    
+    # Reset the user prompt for next interaction
+    user_prompt = ""
+    
+    return success
+
+def main_loop():
+    global awaiting_response
+    thread_id = threading.get_ident()
+    print(f"[THREAD] Main loop running in thread {thread_id}")
+    
+    while not _shutdown_flag:
+        awaiting_response = False
+        idle_mode()
+
+        # Wait here for response button to be pressed
+        print("[MAIN] Idle finished, waiting for trigger")
+        while not awaiting_response and not _shutdown_flag:
+            time.sleep(1)  # Sleep without printing to reduce log spam
+        
+        # Check if we got shutdown while waiting
+        if _shutdown_flag:
+            break
+            
+        print("[MAIN] Response triggered, starting response processing")
+        response_mode()
+        
+        # Add a delay to ensure the video has time to play (the response video)
+        time.sleep(10)
+        
+        # After showing the response video, update the playlist to include other videos too
+        print("[MAIN] Updating playlist with additional videos")
+        create_lipsync_playlist()
+        
+        print("[MAIN] Response mode completed, ready for next interaction")
+    
+    print(f"[THREAD] Thread {thread_id} shutting down")
+
+def trigger_response_with_prompt(prompt):
+    global awaiting_response, user_prompt
+    user_prompt = prompt
+    awaiting_response = True
+    print(f"[UI] Response triggered by user with prompt: {prompt[:50]}...")
+
+def start_main_thread():
+    global _main_thread
+    with _thread_lock:
+        if _main_thread is None or not _main_thread.is_alive():
+            _main_thread = threading.Thread(target=main_loop, daemon=True)
+            _main_thread.start()
+            print(f"[INIT] Main thread started with ID {_main_thread.ident}")
+            return True
+        else:
+            print(f"[INIT] Main thread already running with ID {_main_thread.ident}")
+            return False
+
+def shutdown():
+    global _shutdown_flag
+    _shutdown_flag = True
+    print("[APP] Shutdown initiated, waiting for threads to terminate...")
+    if _main_thread and _main_thread.is_alive():
+        _main_thread.join(timeout=5)
+    print("[APP] Shutdown complete")
+
+# Register shutdown handler
+atexit.register(shutdown)
+
+# Only run this block when the file is executed directly
+if __name__ in {"__main__", "__mp_main__"}:  # Support multiprocessing
+    # Build the UI with our response callback
+    build_ui(trigger_response_callback=trigger_response_with_prompt)
+    
+    # Start the main thread
+    start_main_thread()
+    
+    # Start the NiceGUI server
+    ui.run(host='0.0.0.0', port=7860)