ghostsInTheMachine's picture
Update app.py
2f18b37 verified
raw
history blame
8.01 kB
import spaces # Import the spaces module for ZeroGPU
import gradio as gr
import torch
import os
import tempfile
import shutil
import time
import ffmpeg
import numpy as np
from PIL import Image
import moviepy.editor as mp
from infer import lotus, load_models, pipe_g, pipe_d # Import the global models
import logging
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Device will be set inside GPU-decorated functions
device = 'cuda' # Use 'cuda' as placeholder
# Load models once inside a GPU context
task_name = 'depth'
@spaces.GPU
def initialize_models():
load_models(task_name, device)
# Call the function to load models
initialize_models()
# Preprocess the video to adjust frame rate
def preprocess_video(video_path, target_fps=24):
"""Preprocess the video to adjust its frame rate."""
video = mp.VideoFileClip(video_path)
# Adjust FPS if target_fps is specified
if target_fps > 0:
video = video.set_fps(target_fps)
return video
# Resize image while preserving aspect ratio and adding padding
def resize_and_pad(image, target_size):
"""Resize and pad an image to the target size while preserving aspect ratio."""
# Calculate the new size preserving aspect ratio
image.thumbnail(target_size, Image.LANCZOS)
# Create a new image with the target size and black background
new_image = Image.new("RGB", target_size)
new_image.paste(
image, ((target_size[0] - image.width) // 2, (target_size[1] - image.height) // 2)
)
return new_image
@spaces.GPU
def process_frame(frame, seed=0, target_size=(512, 512)):
"""Process a single frame and return depth map."""
try:
# Convert frame to PIL Image
image = Image.fromarray(frame).convert('RGB')
# Resize and pad image
input_image = resize_and_pad(image, target_size)
# Run inference
depth_map = lotus(input_image, 'depth', seed, device)
# Crop the output depth map back to original image size
width, height = image.size
left = (target_size[0] - width) // 2
top = (target_size[1] - height) // 2
right = left + width
bottom = top + height
depth_map_cropped = depth_map.crop((left, top, right, bottom))
return depth_map_cropped
except Exception as e:
logger.error(f"Error processing frame: {e}")
return None
def process_video(video_path, fps=0, seed=0):
"""Process video frames individually and generate depth maps."""
# Create a persistent temporary directory
temp_dir = tempfile.mkdtemp()
try:
start_time = time.time()
# Preprocess the video
video = preprocess_video(video_path, target_fps=fps)
# Use original video FPS if not specified
if fps == 0:
fps = video.fps
frames = list(video.iter_frames())
total_frames = len(frames)
logger.info(f"Processing {total_frames} frames at {fps} FPS...")
# Create directory for frame sequence and outputs
frames_dir = os.path.join(temp_dir, "frames")
os.makedirs(frames_dir, exist_ok=True)
# Process frames individually
for i, frame in enumerate(frames):
# Process each frame with GPU allocation
depth_map = process_frame(frame, seed)
if depth_map is not None:
# Save frame
frame_path = os.path.join(frames_dir, f"frame_{i:06d}.png")
depth_map.save(frame_path, format='PNG', compress_level=0)
# Update live preview every 10% progress
if i % max(1, total_frames // 10) == 0:
elapsed_time = time.time() - start_time
progress = (i / total_frames) * 100
yield depth_map, None, None, f"Processed {i}/{total_frames} frames... ({progress:.2f}%) Elapsed: {elapsed_time:.2f}s"
else:
logger.error(f"Error processing frame {i}")
logger.info("Creating output files...")
# Create ZIP of frame sequence
zip_filename = f"depth_frames_{int(time.time())}.zip"
zip_path = os.path.join(temp_dir, zip_filename)
shutil.make_archive(zip_path[:-4], 'zip', frames_dir)
# Create MP4 video
video_filename = f"depth_video_{int(time.time())}.mp4"
output_video_path = os.path.join(temp_dir, video_filename)
try:
# FFmpeg settings for high-quality MP4
input_pattern = os.path.join(frames_dir, 'frame_%06d.png')
(
ffmpeg
.input(input_pattern, pattern_type='sequence', framerate=fps)
.output(output_video_path, vcodec='libx264', pix_fmt='yuv420p', crf=17, preset='slow')
.run(overwrite_output=True, quiet=True)
)
logger.info("MP4 video created successfully!")
except ffmpeg.Error as e:
logger.error(f"Error creating video: {e.stderr.decode() if e.stderr else str(e)}")
output_video_path = None
total_time = time.time() - start_time
logger.info("Processing complete!")
# Yield the file paths
yield None, zip_path, output_video_path, f"Processing complete! Total time: {total_time:.2f} seconds"
except Exception as e:
logger.error(f"Error: {e}")
yield None, None, None, f"Error processing video: {e}"
finally:
# Clean up temporary directory if necessary
pass
def process_wrapper(video, fps=0, seed=0):
if video is None:
raise gr.Error("Please upload a video.")
try:
outputs = []
# Initialize models within the Gradio request context
initialize_models()
# Use video directly, since it's the file path
for output in process_video(video, fps, seed):
outputs.append(output)
yield output
return outputs[-1]
except Exception as e:
raise gr.Error(f"Error processing video: {str(e)}")
# Custom CSS for styling (unchanged)
custom_css = """
.title-container {
text-align: center;
padding: 10px 0;
}
#title {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
font-size: 36px;
font-weight: bold;
color: #000000;
padding: 10px;
border-radius: 10px;
display: inline-block;
background: linear-gradient(
135deg,
#e0f7fa, #e8f5e9, #fff9c4, #ffebee,
#f3e5f5, #e1f5fe, #fff3e0, #e8eaf6
);
background-size: 400% 400%;
animation: gradient-animation 15s ease infinite;
}
@keyframes gradient-animation {
0% { background-position: 0% 50%; }
50% { background-position: 100% 50%; }
100% { background-position: 0% 50%; }
}
"""
# Gradio Interface
with gr.Blocks(css=custom_css) as demo:
gr.HTML('''
<div class="title-container">
<div id="title">Video Depth Estimation</div>
</div>
''')
with gr.Row():
with gr.Column():
video_input = gr.Video(label="Upload Video", interactive=True)
fps_slider = gr.Slider(minimum=0, maximum=60, step=1, value=0, label="Output FPS (0 for original)")
seed_slider = gr.Number(value=0, label="Seed")
btn = gr.Button("Process Video")
with gr.Column():
preview_image = gr.Image(label="Live Preview")
output_frames_zip = gr.File(label="Download Frame Sequence (ZIP)")
output_video = gr.File(label="Download Video (MP4)")
time_textbox = gr.Textbox(label="Status", interactive=False)
btn.click(
fn=process_wrapper,
inputs=[video_input, fps_slider, seed_slider],
outputs=[preview_image, output_frames_zip, output_video, time_textbox]
)
demo.queue()
if __name__ == "__main__":
demo.launch(debug=True)