from flask import Flask, request, jsonify, send_from_directory import torch import shutil import os import sys from src.utils.preprocess import CropAndExtract from src.test_audio2coeff import Audio2Coeff from src.facerender.animate import AnimateFromCoeff from src.generate_batch import get_data from src.generate_facerender_batch import get_facerender_data import tempfile from elevenlabs import set_api_key, generate, play, clone, Voice, VoiceSettings from flask_cors import CORS, cross_origin import uuid import time from PIL import Image import moviepy.editor as mp import requests import json import pickle # from videoretalking import inference_function # import base64 # import gfpgan_enhancer # from time import strftime # from argparse import Namespace # from argparse import ArgumentParser # from flask_swagger_ui import get_swaggerui_blueprint # import threading # import elevenlabs # from src.utils.init_path import init_path class AnimationConfig: def __init__(self, driven_audio_path, source_image_path, result_folder,pose_style,expression_scale,enhancer,still,preprocess,ref_pose_video_path, image_hardcoded): self.driven_audio = driven_audio_path self.source_image = source_image_path self.ref_eyeblink = None self.ref_pose = ref_pose_video_path self.checkpoint_dir = './checkpoints' self.result_dir = result_folder self.pose_style = pose_style self.batch_size = 8 self.expression_scale = expression_scale # self.input_yaw = [-15,0,10,5,0,-10,-5,0, 5,10] self.input_yaw = None self.input_pitch = None self.input_roll = None self.enhancer = enhancer self.background_enhancer = None self.cpu = False self.face3dvis = False self.still = still self.preprocess = preprocess self.verbose = False self.old_version = False self.net_recon = 'resnet50' self.init_path = None self.use_last_fc = False self.bfm_folder = './checkpoints/BFM_Fitting/' self.bfm_model = 'BFM_model_front.mat' self.focal = 1015. self.center = 112. self.camera_d = 10. self.z_near = 5. self.z_far = 15. self.device = 'cuda' self.image_hardcoded = image_hardcoded app = Flask(__name__) CORS(app) TEMP_DIR = None start_time = None VIDEO_DIRECTORY = None preprocessed_data = None args = None unique_id = None app.config['temp_response'] = None app.config['generation_thread'] = None app.config['text_prompt'] = None app.config['final_video_path'] = None app.config['final_video_duration'] = None # Global paths dir_path = os.path.dirname(os.path.realpath(__file__)) current_root_path = dir_path path_of_lm_croper = os.path.join(current_root_path, 'checkpoints', 'shape_predictor_68_face_landmarks.dat') path_of_net_recon_model = os.path.join(current_root_path, 'checkpoints', 'epoch_20.pth') dir_of_BFM_fitting = os.path.join(current_root_path, 'checkpoints', 'BFM_Fitting') wav2lip_checkpoint = os.path.join(current_root_path, 'checkpoints', 'wav2lip.pth') audio2pose_checkpoint = os.path.join(current_root_path, 'checkpoints', 'auido2pose_00140-model.pth') audio2pose_yaml_path = os.path.join(current_root_path, 'src', 'config', 'auido2pose.yaml') audio2exp_checkpoint = os.path.join(current_root_path, 'checkpoints', 'auido2exp_00300-model.pth') audio2exp_yaml_path = os.path.join(current_root_path, 'src', 'config', 'auido2exp.yaml') free_view_checkpoint = os.path.join(current_root_path, 'checkpoints', 'facevid2vid_00189-model.pth.tar') # Function for running the actual task (using preprocessed data) def process_chunk(audio_chunk, args): print("Entered Process Chunk Function") global audio2pose_checkpoint, audio2pose_yaml_path, audio2exp_checkpoint, audio2exp_yaml_path, wav2lip_checkpoint global path_of_lm_croper, path_of_net_recon_model, dir_of_BFM_fitting global free_view_checkpoint if args.preprocess == 'full': mapping_checkpoint = os.path.join(current_root_path, 'checkpoints', 'mapping_00109-model.pth.tar') facerender_yaml_path = os.path.join(current_root_path, 'src', 'config', 'facerender_still.yaml') else: mapping_checkpoint = os.path.join(current_root_path, 'checkpoints', 'mapping_00229-model.pth.tar') facerender_yaml_path = os.path.join(current_root_path, 'src', 'config', 'facerender.yaml') # first_coeff_path = preprocessed_data["first_coeff_path"] # crop_pic_path = preprocessed_data["crop_pic_path"] # crop_info_path = "/home/user/app/preprocess_data/crop_info.json" # with open(crop_info_path , "rb") as f: # crop_info = json.load(f) first_frame_dir = os.path.join(args.result_dir, 'first_frame_dir') os.makedirs(first_frame_dir, exist_ok=True) preprocess_model = CropAndExtract(path_of_lm_croper, path_of_net_recon_model, dir_of_BFM_fitting, args.device) first_coeff_path, crop_pic_path, crop_info = preprocess_model.generate(args.source_image, first_frame_dir, args.preprocess, source_image_flag=True) print(f"Loaded existing preprocessed data") print("first_coeff_path",first_coeff_path) print("crop_pic_path",crop_pic_path) print("crop_info",crop_info) torch.cuda.empty_cache() if args.ref_pose is not None: ref_pose_videoname = os.path.splitext(os.path.split(args.ref_pose)[-1])[0] ref_pose_frame_dir = os.path.join(args.result_dir, ref_pose_videoname) os.makedirs(ref_pose_frame_dir, exist_ok=True) ref_pose_coeff_path, _, _ = preprocess_model.generate(args.ref_pose, ref_pose_frame_dir) print('ref_eyeblink_coeff_path',ref_pose_coeff_path) else: ref_pose_coeff_path = None print('ref_eyeblink_coeff_path',ref_pose_coeff_path) batch = get_data(first_coeff_path, audio_chunk, args.device, ref_eyeblink_coeff_path=None, still=args.still) audio_to_coeff = Audio2Coeff(audio2pose_checkpoint, audio2pose_yaml_path, audio2exp_checkpoint, audio2exp_yaml_path, wav2lip_checkpoint, args.device) coeff_path = audio_to_coeff.generate(batch, args.result_dir, args.pose_style, ref_pose_coeff_path) # Further processing with animate_from_coeff using the coeff_path animate_from_coeff = AnimateFromCoeff(free_view_checkpoint, mapping_checkpoint, facerender_yaml_path, args.device) torch.cuda.empty_cache() data = get_facerender_data(coeff_path, crop_pic_path, first_coeff_path, audio_chunk, args.batch_size, args.input_yaw, args.input_pitch, args.input_roll, expression_scale=args.expression_scale, still_mode=args.still, preprocess=args.preprocess) torch.cuda.empty_cache() print("Will Enter Animation") result, base64_video, temp_file_path, _ = animate_from_coeff.generate(data, args.result_dir, args.source_image, crop_info, enhancer=args.enhancer, background_enhancer=args.background_enhancer, preprocess=args.preprocess) # video_clip = mp.VideoFileClip(temp_file_path) # duration = video_clip.duration app.config['temp_response'] = base64_video app.config['final_video_path'] = temp_file_path # app.config['final_video_duration'] = duration torch.cuda.empty_cache() return base64_video, temp_file_path def create_temp_dir(): return tempfile.TemporaryDirectory() def save_uploaded_file(file, filename,TEMP_DIR): unique_filename = str(uuid.uuid4()) + "_" + filename file_path = os.path.join(TEMP_DIR.name, unique_filename) file.save(file_path) return file_path def custom_cleanup(temp_dir, exclude_dir): # Iterate over the files and directories in TEMP_DIR for filename in os.listdir(temp_dir): file_path = os.path.join(temp_dir, filename) # Skip the directory we want to exclude if file_path != exclude_dir: try: if os.path.isdir(file_path): shutil.rmtree(file_path) else: os.remove(file_path) print(f"Deleted: {file_path}") except Exception as e: print(f"Failed to delete {file_path}. Reason: {e}") def generate_audio(voice_cloning, voice_gender, text_prompt,user_voice_path,language): print("generate_audio") # Map language → ElevenLabs voice/language codes language_mapping = { "en": "en-IN", # Indian English "hi": "hi-IN", # Hindi "ta": "ta-IN", # Tamil "te": "te-IN", # Telugu "ml": "ml-IN", # Malayalam "bn": "bn-IN", # Bengali "gu": "gu-IN", # Gujarati "mr": "mr-IN", # Marathi "kn": "kn-IN", # Kannada } selected_language = language_mapping.get(language, "en-IN") print("TTS Language Selected:", selected_language) if voice_cloning == 'no': if voice_gender == 'male': voice = 'echo' print('Entering Audio creation using elevenlabs') set_api_key('sk_e823e586aa0c238fdfae02466faad9472bb668fd04431fca') #audio_old = generate(text = text_prompt, voice = "Daniel", model = "eleven_multilingual_v2",stream=True, latency=4) audio = generate( text=text_prompt, voice="Daniel", model="eleven_multilingual_v2", stream=True, latency=4 ) with tempfile.NamedTemporaryFile(suffix=".mp3", prefix="text_to_speech_",dir=TEMP_DIR.name, delete=False) as temp_file: for chunk in audio: temp_file.write(chunk) driven_audio_path = temp_file.name print('driven_audio_path',driven_audio_path) print('Audio file saved using elevenlabs') else: voice = 'nova' print('Entering Audio creation using whisper') response = client.audio.speech.create(model="tts-1-hd", voice=voice, input = text_prompt) print('Audio created using whisper') with tempfile.NamedTemporaryFile(suffix=".wav", prefix="text_to_speech_",dir=TEMP_DIR.name, delete=False) as temp_file: driven_audio_path = temp_file.name response.write_to_file(driven_audio_path) print('Audio file saved using whisper') elif voice_cloning == 'yes': set_api_key('sk_e823e586aa0c238fdfae02466faad9472bb668fd04431fca') print('Input in user_voice_path: ',user_voice_path) # user_voice_path = '/home/user/app/images/AUDIO-2024-10-04-09-51-34.m4a' voice = clone(name = "User Cloned Voice", files = [user_voice_path] ) # DeZH4ash9IU9gUcNjVXh marc # KR6RRu8YgfxrhYocGuOc sachin print("voice data",voice) # voice = Voice(voice_id="DeZH4ash9IU9gUcNjVXh",name="Marc",settings=VoiceSettings( # stability=0.71, similarity_boost=0.9, style=0.0, use_speaker_boost=True),) #audio_old = generate(text = text_prompt, voice = voice, model = "eleven_multilingual_v2",stream=True, latency=4) audio = generate( text=text_prompt, voice=voice, model="eleven_multilingual_v2", stream=True, latency=4 ) with tempfile.NamedTemporaryFile(suffix=".mp3", prefix="cloned_audio_",dir=TEMP_DIR.name, delete=False) as temp_file: for chunk in audio: temp_file.write(chunk) driven_audio_path = temp_file.name print('driven_audio_path',driven_audio_path) # audio_duration = get_audio_duration(driven_audio_path) # print('Total Audio Duration in seconds',audio_duration) return driven_audio_path def run_preprocessing(args): global path_of_lm_croper, path_of_net_recon_model, dir_of_BFM_fitting first_frame_dir = os.path.join(args.result_dir, 'first_frame_dir') os.makedirs(first_frame_dir, exist_ok=True) fixed_temp_dir = "/home/user/app/preprocess_data/" os.makedirs(fixed_temp_dir, exist_ok=True) preprocessed_data_path = os.path.join(fixed_temp_dir, "preprocessed_data.pkl") if os.path.exists(preprocessed_data_path) and args.image_hardcoded == "yes": print("Loading preprocessed data...") with open(preprocessed_data_path, "rb") as f: preprocessed_data = pickle.load(f) print("Loaded existing preprocessed data from:", preprocessed_data_path) return preprocessed_data @app.route("/run", methods=['POST']) def generate_video(): global start_time, VIDEO_DIRECTORY start_time = time.time() global TEMP_DIR TEMP_DIR = create_temp_dir() print('request:',request.method) try: if request.method == 'POST': source_image = request.files['source_image'] #new code # image_path = '/home/user/app/images/marc_smile_enhanced.jpg' old code # source_image = Image.open(image_path) old code text_prompt = request.form['text_prompt'] print('Input text prompt: ',text_prompt) language = request.form.get('language', 'en') # default English print("Selected language:", language) text_prompt = text_prompt.strip() if not text_prompt: return jsonify({'error': 'Input text prompt cannot be blank'}), 400 voice_cloning = request.form.get('voice_cloning', 'yes') image_hardcoded = request.form.get('image_hardcoded', 'yes') target_language = request.form.get('target_language', 'original_text') print('target_language',target_language) pose_style = int(request.form.get('pose_style', 1)) expression_scale = float(request.form.get('expression_scale', 1)) enhancer = request.form.get('enhancer', None) voice_gender = request.form.get('voice_gender', 'male') still_str = request.form.get('still', 'False') still = still_str.lower() == 'false' print('still', still) preprocess = request.form.get('preprocess', 'crop') print('preprocess selected: ',preprocess) ref_pose_video = request.files.get('ref_pose', None) # Handle user uploaded voice file dynamically user_voice = request.files.get('user_voice') if user_voice: user_voice_path = save_uploaded_file(user_voice, user_voice.filename, TEMP_DIR) print(f"User voice file saved at: {user_voice_path}") else: user_voice_path = None print("No user voice file provided — skipping cloning.") app.config['text_prompt'] = text_prompt print('Final output text prompt using openai: ',text_prompt) if 'source_image' not in request.files: return jsonify({'error': 'No source image provided'}), 400 source_image = request.files['source_image'] source_image_path = save_uploaded_file(source_image, source_image.filename, TEMP_DIR) print(f"Source image saved at: {source_image_path}") driven_audio_path = generate_audio(voice_cloning, voice_gender, text_prompt,user_voice_path,language) #driven_audio_path_not_use = user_voice_path print(f"driven audio path: {driven_audio_path}") save_dir = tempfile.mkdtemp(dir=TEMP_DIR.name) result_folder = os.path.join(save_dir, "results") os.makedirs(result_folder, exist_ok=True) ref_pose_video_path = None if ref_pose_video: with tempfile.NamedTemporaryFile(suffix=".mp4", prefix="ref_pose_",dir=TEMP_DIR.name, delete=False) as temp_file: ref_pose_video_path = temp_file.name ref_pose_video.save(ref_pose_video_path) print('ref_pose_video_path',ref_pose_video_path) except Exception as e: app.logger.error(f"An error occurred: {e}") return "An error occurred", 500 # Example of using the class with some hypothetical paths args = AnimationConfig(driven_audio_path=driven_audio_path, source_image_path=source_image_path, result_folder=result_folder, pose_style=pose_style, expression_scale=expression_scale,enhancer=enhancer,still=still,preprocess=preprocess,ref_pose_video_path=ref_pose_video_path, image_hardcoded=image_hardcoded) if torch.cuda.is_available() and not args.cpu: args.device = "cuda" else: args.device = "cpu" try: # preprocessed_data = run_preprocessing(args) base64_video, temp_file_path = process_chunk(driven_audio_path, args) final_video_path = app.config['final_video_path'] print('final_video_path',final_video_path) if temp_file_path and temp_file_path.endswith('.mp4'): filename = os.path.basename(temp_file_path) os.makedirs('videos', exist_ok=True) VIDEO_DIRECTORY = os.path.abspath('videos') print("VIDEO_DIRECTORY: ",VIDEO_DIRECTORY) destination_path = os.path.join(VIDEO_DIRECTORY, filename) shutil.copy(temp_file_path, destination_path) video_url = f"/videos/{filename}" if final_video_path and os.path.exists(final_video_path): os.remove(final_video_path) print("Deleted video file:", final_video_path) preprocess_dir = os.path.join("/tmp", "preprocess_data") custom_cleanup(TEMP_DIR.name, preprocess_dir) print("Temporary files cleaned up, but preprocess_data is retained.") end_time = time.time() time_taken = end_time - start_time print(f"Time taken for endpoint: {time_taken:.2f} seconds") return jsonify({ "message": "Video processed and saved successfully.", "video_url": video_url, "time_taken": time_taken, "status": "success" }) else: return jsonify({ "message": "Failed to process the video.", "status": "error" }), 500 except Exception as e: return jsonify({'status': 'error', 'message': str(e)}), 500 @app.route("/videos/", methods=['GET']) def serve_video(filename): global VIDEO_DIRECTORY return send_from_directory(VIDEO_DIRECTORY, filename, as_attachment=False) @app.route("/health", methods=["GET"]) def health_status(): response = {"online": "true"} return jsonify(response) if __name__ == '__main__': app.run(debug=True)