File size: 4,293 Bytes
e25b41e 7e100e2 e25b41e b325d6c e25b41e b325d6c e25b41e 7e100e2 e25b41e 7efbaa8 e25b41e c45ea05 e25b41e 7e100e2 e25b41e 903aff6 9e25a21 e25b41e c45ea05 6b32057 e25b41e 03da17a e25b41e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 |
import torch
from diffusers import AnimateDiffPipeline, LCMScheduler, MotionAdapter
from diffusers.utils import export_to_video
from flask import Flask, request, jsonify
from flask_cors import CORS
import base64
import tempfile
import os
import threading
import traceback
app = Flask(__name__)
CORS(app)
pipe = None
app.config['temp_response'] = None
app.config['generation_thread'] = None
def download_pipeline():
global pipe
try:
print('Downloading the model weights')
# Download and initialize the animation pipeline
adapter = MotionAdapter.from_pretrained("wangfuyun/AnimateLCM", torch_dtype=torch.float16)
pipe = AnimateDiffPipeline.from_pretrained("emilianJR/epiCRealism", motion_adapter=adapter, torch_dtype=torch.float16)
pipe.scheduler = LCMScheduler.from_config(pipe.scheduler.config, beta_schedule="linear")
pipe.load_lora_weights("wangfuyun/AnimateLCM", weight_name="AnimateLCM_sd15_t2v_lora.safetensors", adapter_name="lcm-lora")
pipe.set_adapters(["lcm-lora"], [0.8])
pipe.enable_vae_slicing()
pipe.enable_model_cpu_offload()
return True
except Exception as e:
print(f"Error downloading pipeline: {e}")
return False
def generate_and_export_animation(prompt):
global pipe
# Ensure the animation pipeline is initialized
if pipe is None:
if not download_pipeline():
return None, "Failed to initialize animation pipeline"
try:
# Generate animation frames
print('Generating Video frames')
output = pipe(
prompt=prompt,
negative_prompt="bad quality, worse quality, low resolution, blur",
num_frames=16,
guidance_scale=2.0,
num_inference_steps=6
)
print('Video frames generated')
# Export frames to a temporary video file
with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as temp_file:
temp_video_path = temp_file.name
print('temp_video_path', temp_video_path)
export_to_video(output.frames[0], temp_video_path)
with open(temp_video_path, 'rb') as video_file:
video_binary = video_file.read()
video_base64 = base64.b64encode(video_binary).decode('utf-8')
os.remove(temp_video_path)
response_data = {'video_base64': '','status':None}
response_data['video_base64'] = video_base64
print('response_data',response_data)
return response_data
except Exception as e:
print(f"Error generating animation: {e}")
# return None, "Failed to generate animation"
traceback.print_exc() # Print exception details to console
return jsonify({"error": f"Failed to generate animation: {str(e)}"}), 500
def background(prompt):
with app.app_context():
temp_response = generate_and_export_animation(prompt)
# json_content = temp_response.get_json()
app.config['temp_response'] = temp_response
@app.route('/run', methods=['POST'])
def handle_animation_request():
prompt = request.form.get('prompt')
if prompt:
generation_thread = threading.Thread(target=background, args=(prompt,))
app.config['generation_thread'] = generation_thread
generation_thread.start()
response_data = {"message": "Video generation started", "process_id": generation_thread.ident}
return jsonify(response_data)
else:
return jsonify({"message": "Please provide a valid text prompt."}), 400
@app.route('/status', methods=['GET'])
def check_animation_status():
process_id = request.args.get('process_id',None)
if process_id:
generation_thread = app.config.get('generation_thread')
if generation_thread and generation_thread.is_alive():
return jsonify({"status": "in_progress"}), 200
elif app.config.get('temp_response'):
print('final',app.config.get('temp_response'))
# app.config['temp_response']['status'] = 'completed'
final_response = app.config['temp_response']
final_response['status'] = 'completed'
return jsonify(final_response)
if __name__ == '__main__':
app.run(debug=True)
|