Spaces:

AvatarResurrectionChatbot
/

DarwinChatbotFinal

Paused

J Z

Upload app_hf.py

399c8e8 verified over 1 year ago

6.96 kB

# 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)

	# 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)