import os import sys import json import socket import tempfile import pathlib from dataclasses import dataclass from typing import List, Dict, Any import pickle import cv2 import torch import numpy as np import imageio import gradio as gr import spaces import smplx import pyrender import trimesh import trimesh.transformations as tra import requests from download_precomputed import download_and_extract_precomputed # if intermediate_results/8_kid_crossing doesn't exist, download it if not os.path.exists('intermediate_results/8_kid_crossing'): download_and_extract_precomputed() # Constants and configuration CHECKPOINT_PATH = './models/checkpoint_150.pt' NUM_SAMPLES = 50 DISPLAYED_PREDS = 3 FRAME_LIMIT = 30 FPS = 60 DESCRIPTION = "# SkeletonDiffusion Demo" # Create necessary directories for dir_name in ['downloads', 'predictions', 'vis', 'intermediate_results', 'outputs', 'assets']: os.makedirs(dir_name, exist_ok=True) # Create a simple loading image if it doesn't exist LOADING_IMAGE_PATH = os.path.join('assets', 'loading.png') if not os.path.exists(LOADING_IMAGE_PATH): # Create a simple loading image with text img = np.zeros((200, 400, 3), dtype=np.uint8) img.fill(255) # White background cv2.putText(img, "Processing...", (50, 100), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 0), 2) cv2.putText(img, "Please wait", (50, 150), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 0), 2) cv2.imwrite(LOADING_IMAGE_PATH, img) # Available options for displayed predictions DISPLAYED_PREDS_OPTIONS = [2, 3, 4, 5, 6] # Fix display and setup issues os.environ['PYOPENGL_PLATFORM'] = 'egl' os.system('export IMAGEMAGICK_BINARY=./magick') os.system('bash ./SkeletonDiffusion_demo/setup_headless.bash') # Download ImageMagick if not present MAGICK_PATH = "./magick" if not os.path.exists(MAGICK_PATH): response = requests.get("https://imagemagick.org/archive/binaries/magick", stream=True) if response.status_code == 200: with open(MAGICK_PATH, "wb") as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) print(f"Download completed and saved as '{MAGICK_PATH}'") else: print(f"Download failed with status code: {response.status_code}") os.system('chmod +x ./magick') # Setup device device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Local imports import SkeletonDiffusion_demo.plot_several_meshes as plot_several_meshes import SkeletonDiffusion_demo.combine_video as combine from SkeletonDiffusion.src.data.skeleton import create_skeleton from SkeletonDiffusion.src.eval_prepare_model import get_prediction, load_model_config_exp, prepare_model from src_joints2smpl_demo.joints2smpl.convert_joints2smpl import process_motion from SkeletonDiffusion.src.metrics.ranking import get_closest_and_nfurthest_maxapd @dataclass class SMPLParams: """Data structure to hold SMPL parameters.""" global_orient: torch.Tensor body_pose: torch.Tensor betas: torch.Tensor transl: torch.Tensor joints3d: torch.Tensor def handle_video_input(video_file: str) -> str: """Handle video input from either a local file or YouTube URL. Args: video_file: Path to local video file Returns: str: Path to the video file """ if video_file: return video_file return None def correct_vertices(vertices: np.ndarray) -> np.ndarray: """Correct SMPL vertices to convert from SMPL to renderer coordinate system. Applies a rotation about the Y-axis by 180 degrees so that the original +X axis (face direction) is transformed to the -Z axis. Args: vertices: SMPL vertices in shape (1, N, 3) Returns: np.ndarray: Corrected vertices in shape (1, N, 3) """ angle = np.radians(180) R = tra.rotation_matrix(angle, [1, 0, 0]) vertices_homo = np.hstack([vertices[0], np.ones((vertices[0].shape[0], 1))]) vertices_corrected = (R @ vertices_homo.T).T return vertices_corrected[:, :3].reshape(1, -1, 3) def render_smpl(vertices: np.ndarray, width: int, height: int) -> np.ndarray: """Render SMPL 3D model using PyRender. Args: vertices: SMPL vertices in shape (1, N, 3) width: Output image width height: Output image height Returns: np.ndarray: Rendered image in BGR format """ device = torch.device("cuda" if torch.cuda.is_available() else "cpu") smpl_model = smplx.create("./models/SMPL_NEUTRAL.pkl", model_type="smpl", gender="neutral").to(device) vertices_corrected = correct_vertices(vertices) mesh = trimesh.Trimesh(vertices_corrected[0], smpl_model.faces) scene = pyrender.Scene(bg_color=[1.0, 1.0, 1.0, 0.9]) mesh_node = pyrender.Mesh.from_trimesh(mesh) scene.add(mesh_node) camera = pyrender.OrthographicCamera(xmag=1.0, ymag=1.0) camera_pose = np.eye(4) camera_pose[:3, 3] = [0, 0, 5.0] # distance = 5.0 scene.add(camera, pose=camera_pose) renderer = pyrender.OffscreenRenderer(width, height) color, _ = renderer.render(scene) return cv2.cvtColor(color, cv2.COLOR_RGB2BGR) def save_intermediate_results(video_name: str, results: Dict[str, Any], displayed_preds: int = DISPLAYED_PREDS): """Save intermediate results for a video. Args: video_name: Name of the video file results: Dictionary containing intermediate results displayed_preds: Number of displayed predictions """ base_name = os.path.splitext(os.path.basename(video_name))[0] results_dir = os.path.join('intermediate_results', base_name) os.makedirs(results_dir, exist_ok=True) # Save results with displayed_preds in filename results_path = os.path.join(results_dir, f'results_{displayed_preds}.pkl') with open(results_path, 'wb') as f: pickle.dump(results, f) def load_intermediate_results(video_name: str, displayed_preds: int = DISPLAYED_PREDS) -> Dict[str, Any]: """Load intermediate results for a video. Args: video_name: Name of the video file displayed_preds: Number of displayed predictions Returns: Dictionary containing intermediate results or None if not found """ base_name = os.path.splitext(os.path.basename(video_name))[0] results_path = os.path.join('intermediate_results', base_name, f'results_{displayed_preds}.pkl') if os.path.exists(results_path): # Set default tensor type to CPU before loading torch.set_default_tensor_type(torch.FloatTensor) with open(results_path, 'rb') as f: # Use map_location to ensure tensors are loaded on CPU results = torch.load(f, map_location='cpu') # Reset default tensor type torch.set_default_tensor_type(torch.cuda.FloatTensor) return results return None @spaces.GPU(duration=60) def process_video_gpu(video_file: str, displayed_preds: int = DISPLAYED_PREDS) -> tuple: """GPU version of process_video that does the actual processing.""" import time start_time = time.time() # Load models model_start_time = time.time() smpl_model = smplx.create("./models/SMPL_NEUTRAL.pkl", model_type="smpl", gender="neutral").to(device) nlf_model = torch.jit.load("./models/nlf_l_multi.torchscript").to(device).eval() print(f"Time for model loading: {time.time() - model_start_time:.2f}s") # Handle video input input_start_time = time.time() input_path = handle_video_input(video_file) if not input_path: return None, None print(f"Time for video input handling: {time.time() - input_start_time:.2f}s") # Create output path in outputs directory base_name = os.path.splitext(os.path.basename(video_file))[0] output_path = os.path.join('outputs', f'{base_name}_smpl_{displayed_preds}.gif') # Process frames frame_start_time = time.time() cap = cv2.VideoCapture(input_path) frame_count = 0 smpl_params_list = [] nlf_frames = [] # Store NLF detection frames # Initialize time counters total_nlf_time = 0 total_smpl_time = 0 total_frame_time = 0 while cap.isOpened() and frame_count < FRAME_LIMIT: frame_process_start = time.time() ret, frame = cap.read() if not ret: break frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) image_tensor = torch.from_numpy(frame_rgb).permute(2, 0, 1).int().to(device) # SMPL detection nlf_start_time = time.time() with torch.inference_mode(): pred = nlf_model.detect_smpl_batched(image_tensor.unsqueeze(0)) pose_params = pred["pose"][0].cpu().numpy() betas = pred["betas"][0].cpu().numpy() transl = pred["trans"][0].cpu().numpy() joints3d = pred['joints3d'][0].cpu().numpy() nlf_time = time.time() - nlf_start_time total_nlf_time += nlf_time print(f"Time for NLF model inference: {nlf_time:.2f}s") if pose_params.shape[0] == 0: print(f"No SMPL detected in frame {frame_count}") nlf_frames.append(frame_rgb) continue if pose_params.shape[0] > 0: # SMPL model render_start_time = time.time() smpl_param = SMPLParams( global_orient=torch.tensor(pose_params[:, :3]).to(device), body_pose=torch.tensor(pose_params).to(device), betas=torch.tensor(betas).to(device), transl=torch.tensor(transl).to(device), joints3d=torch.tensor(joints3d[:, 0:22, :3]).to(device), ) output_smpl = smpl_model( global_orient=torch.tensor(pose_params[:, :3]).to(device), body_pose=torch.tensor(pose_params[:, 3:]).to(device), betas=torch.tensor(betas).to(device), transl=torch.tensor(transl).to(device), joints3d=torch.tensor(joints3d[:, 0:66]).to(device), ) smpl_time = time.time() - render_start_time total_smpl_time += smpl_time print(f"Time for SMPL model: {smpl_time:.2f}s") smpl_params_list.append(smpl_param) nlf_frames.append(frame_rgb) # Store original frame for NLF visualization frame_count += 1 frame_time = time.time() - frame_process_start total_frame_time += frame_time print(f"Total time for frame {frame_count}: {frame_time:.2f}s") cap.release() gr.Info("Video-to-motion processing completed!") print(f"\nTime statistics for {frame_count} frames:") print(f"Average NLF model time per frame: {total_nlf_time/frame_count:.2f}s") print(f"Average SMPL model time per frame: {total_smpl_time/frame_count:.2f}s") print(f"Average total time per frame: {total_frame_time/frame_count:.2f}s") print(f"Total time for all frame processing: {time.time() - frame_start_time:.2f}s") # Serialize SMPL parameters serial_start_time = time.time() smpl_params_serialized = [ { "global_orient": p.global_orient.tolist(), "body_pose": p.body_pose.tolist(), "betas": p.betas.tolist(), "transl": p.transl.tolist(), "joints3d": p.joints3d.tolist(), } for p in smpl_params_list ] print(f"Time for parameter serialization: {time.time() - serial_start_time:.2f}s") print(f"Total time: {time.time() - start_time:.2f}s") # Save SMPL params as JSON with open('smpl_params.json', 'w') as f: json.dump(smpl_params_serialized, f) # Save intermediate results results = { 'output_path': output_path, 'smpl_params_serialized': smpl_params_serialized, 'nlf_frames': nlf_frames, 'smpl_params_list': smpl_params_list } save_intermediate_results(video_file, results, displayed_preds) return output_path, smpl_params_serialized @spaces.GPU(duration=200) def generate_motion_video_gpu(smpl_params_json: List[Dict[str, Any]], video_file: str, displayed_preds: int = DISPLAYED_PREDS) -> str: """GPU version of generate_motion_video that does the actual processing.""" import time start_time = time.time() # Setup device device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Create output path in outputs directory with displayed_preds in filename base_name = os.path.splitext(os.path.basename(video_file))[0] output_path = os.path.join('outputs', f'{base_name}_motion_{displayed_preds}.gif') # Load input video frames input_frames = [] cap = cv2.VideoCapture(video_file) frame_count = 0 while cap.isOpened() and frame_count < FRAME_LIMIT: ret, frame = cap.read() if not ret: break input_frames.append(frame) frame_count += 1 cap.release() # Ensure we have exactly FRAME_LIMIT frames if len(input_frames) < FRAME_LIMIT: # Pad with the last frame if needed last_frame = input_frames[-1] if input_frames else np.zeros((480, 640, 3), dtype=np.uint8) input_frames.extend([last_frame] * (FRAME_LIMIT - len(input_frames))) elif len(input_frames) > FRAME_LIMIT: # Trim to FRAME_LIMIT frames input_frames = input_frames[:FRAME_LIMIT] # Deserialize JSON back into SMPLParams objects smpl_params_list = [ SMPLParams( global_orient=torch.tensor(p["global_orient"]).to(device), body_pose=torch.tensor(p["body_pose"]).to(device), betas=torch.tensor(p["betas"]).to(device), transl=torch.tensor(p["transl"]).to(device), joints3d=torch.tensor(p["joints3d"]).to(device), ) for p in smpl_params_json ] print(f"Time for deserialization: {time.time() - start_time:.2f}s") # Calculate number of frames for 0.5 seconds # frames_for_half_second = int(FPS * 0.5) frames_for_half_second = 30 # Collect last 0.5 seconds of frames obs = torch.stack([p.joints3d[0] for p in smpl_params_list[-frames_for_half_second:]]) # Pad with first frame if needed if len(obs) < frames_for_half_second: padding = torch.stack([obs[0]] * (frames_for_half_second - len(obs))) obs = torch.cat([padding, obs], dim=0) # Load model and prepare data model_start_time = time.time() config, exp_folder = load_model_config_exp(CHECKPOINT_PATH) config['checkpoint_path'] = CHECKPOINT_PATH skeleton = create_skeleton(**config) model, device, *_ = prepare_model(config, skeleton, **config) # Convert from mm to m and prepare input obs = obs / 1000.0 obs = obs.reshape(1, frames_for_half_second, 22, 3).to(device) obs = torch.stack([-obs[..., 2], obs[..., 0], -obs[..., 1]], dim=-1) obs = obs - obs[..., 0:1, :] obs_in = skeleton.tranform_to_input_space(obs).to(device) # Get predictions pred = get_prediction(obs_in, model, num_samples=NUM_SAMPLES, **config) pred = torch.cat((torch.zeros(1, NUM_SAMPLES, pred.shape[2], 1, 3).to(device), pred), dim=3) obs_in = torch.cat((torch.zeros(1, frames_for_half_second, 1, 3).to(device), obs_in), dim=2) print(f"Time for model inference: {time.time() - model_start_time:.2f}s") # Convert predictions to SMPL parameters smpl_start_time = time.time() print(f"Time for SMPL conversion: {time.time() - smpl_start_time:.2f}s") # Prepare data for visualization obs_in_50 = obs_in.unsqueeze(1).repeat(1, NUM_SAMPLES, 1, 1, 1) obs_and_pred = torch.cat((obs_in_50, pred), dim=2) # Save predictions pred_np = obs_and_pred.cpu().numpy() os.makedirs('predictions', exist_ok=True) np.save('predictions/joints3d.npy', pred_np) print(f"Joints3D data saved to predictions/joints3d.npy") # Calculate metrics and select best samples metric_start_time = time.time() from SkeletonDiffusion.src.metrics.body_realism import limb_stretching_normed_rmse limbstretching = limb_stretching_normed_rmse( pred[..., 1:, :], target=obs[0, ..., 1:, :].unsqueeze(0), limbseq=skeleton.get_limbseq(), reduction='persample', obs_as_target=True ) # Sort samples by limb stretching and take the half with smallest values limbstretching_sorted, indices = torch.sort(limbstretching.squeeze(1), dim=-1, descending=False) half_size = len(indices[0]) // 2 best_half_indices = indices[0, :15] # Get predictions for best half y_pred = pred.squeeze(0)[best_half_indices] y_gt = y_pred[0].unsqueeze(0) # Use get_closest_and_nfurthest_maxapd to select diverse samples _, _, top_indices = get_closest_and_nfurthest_maxapd(y_pred, y_gt, nsamples=displayed_preds) print(f"Selected {len(best_half_indices)} samples with smallest limb stretching") print(f"Selected {len(top_indices)} diverse samples from best half") # Generate visualization vis_start_time = time.time() with torch.no_grad(): # Create video-specific obj directory obj_dir = os.path.join('outputs', f'{base_name}_obj/') os.makedirs(obj_dir, exist_ok=True) process_motion("smpl_params.json", "predictions/joints3d.npy", device=device, sorted_idx=top_indices, output_dir=obj_dir) print(f"Checking obj_dir contents: {obj_dir}") if os.path.exists(obj_dir): print("Contents of obj_dir:") for root, dirs, files in os.walk(obj_dir): print(f"Directory: {root}") print(f"Files: {files}") print(f"Subdirectories: {dirs}") plot_several_meshes.main(obj_dir, displayed_preds) # Show completion message before combining video gr.Info("Step 3/3: Combine GIFs") # Save the motion video to the output path with input frames as picture-in-picture output_path = combine.combine_video(obj_dir, output_path, input_frames, displayed_preds) return output_path def process_video(video_file: str, displayed_preds: int = DISPLAYED_PREDS) -> tuple: """Process input video to extract SMPL parameters and generate visualization.""" # Check if we have pre-computed results pre_computed = load_intermediate_results(video_file, displayed_preds) if pre_computed is not None: print("Using pre-computed results") return pre_computed['output_path'], pre_computed['smpl_params_serialized'] # If no pre-computed results, use GPU processing return process_video_gpu(video_file, displayed_preds) def generate_motion_video(smpl_params_json: List[Dict[str, Any]], video_file: str = None, displayed_preds: int = DISPLAYED_PREDS) -> str: """Generate a motion video from SMPL parameters.""" if smpl_params_json is None: raise ValueError("No SMPL parameters provided. Please process a video first.") # Check if we have pre-computed results if video_file: pre_computed = load_intermediate_results(video_file, displayed_preds) if pre_computed is not None and 'motion_video_path' in pre_computed: print("Using pre-computed motion video") return pre_computed['motion_video_path'] # If no pre-computed results, use GPU processing return generate_motion_video_gpu(smpl_params_json, video_file, displayed_preds) def video_to_gif(video_path, gif_path, frame_limit=30): """Convert video to GIF with specified frame limit and smooth looping.""" frames = [] # lower resolution reader = imageio.get_reader(video_path) for i, frame in enumerate(reader): if i >= frame_limit: break frames.append(frame) # Set FPS to 15 for smooth looping of 30 frames (2 seconds per loop) imageio.mimsave(gif_path, frames, fps=15, loop=0) # loop=0 means infinite loop return gif_path def concat_gifs_side_by_side(gif1_path, gif2_path, output_path): """Pad both GIFs to the same (max) height, center them vertically, then concatenate side by side.""" gif1 = imageio.mimread(gif1_path) gif2 = imageio.mimread(gif2_path) # Ensure both GIFs have the same number of frames if len(gif1) != len(gif2): print(f"Warning: GIFs have different frame counts ({len(gif1)} vs {len(gif2)}). Adjusting to match.") # Use the shorter length min_frames = min(len(gif1), len(gif2)) gif1 = gif1[:min_frames] gif2 = gif2[:min_frames] frames = [] for f1, f2 in zip(gif1, gif2): # Convert both frames to RGB if needed (handle RGBA with alpha channel) if f1.shape[2] == 4: f1 = f1[..., :3] if f2.shape[2] == 4: f2 = f2[..., :3] h1, w1, c1 = f1.shape h2, w2, c2 = f2.shape max_h = max(h1, h2) # Pad f1 to max_h, vertically centered pad_top1 = (max_h - h1) // 2 pad_bot1 = max_h - h1 - pad_top1 f1_pad = np.pad(f1, ((pad_top1, pad_bot1), (0, 0), (0, 0)), mode='constant', constant_values=255) # Pad f2 to max_h, vertically centered pad_top2 = (max_h - h2) // 2 pad_bot2 = max_h - h2 - pad_top2 f2_pad = np.pad(f2, ((pad_top2, pad_bot2), (0, 0), (0, 0)), mode='constant', constant_values=255) # Concatenate horizontally frame = np.concatenate([f1_pad, f2_pad], axis=1) frames.append(frame) imageio.mimsave(output_path, frames, fps=15, loop=0) return output_path def create_gradio_interface(): """Create and configure the Gradio interface.""" with gr.Blocks(css="style.css") as demo: gr.Markdown(DESCRIPTION) # Add user instructions gr.Markdown(""" Demo for the CVPR2025 paper "Nonisotropic Gaussian Diffusion for Realistic 3D Human Motion Prediction", available [here](https://ceveloper.github.io/publications/skeletondiffusion/). Codebase released on [GitHub](https://github.com/Ceveloper/SkeletonDiffusion/tree/main). SkeletonDiffusion takes as input a sequence of 3D body joints coordinates, which not everyone has at disposal. In this demo, we use a publicly available model, Neural Localizer Fields ([NLF](https://istvansarandi.com/nlf/)) to extract 3D poses from a given input video. We feed the extracted poses to SkeletonDiffusion to generate corresponding future motions. Note that the poses extracted from the video are noisy and imperfect, but SkeletonDiffusion has been trained only with precise sensor data obtained in laboratory settings. Despite never having seen noisy data and various real-world actions (ballet, basketball, etc.), SkeletonDiffusion can handle most cases reasonably! ### Instructions 1. Upload a video or select from examples 2. Choose whether to use precomputed results (if available) 3. Select the number of motion predictions to display (2-6) 4. Click "Run Skeleton Diffusion" to start **Note:** - SkeletonDiffusion requires less than half a second for a forward pass, but extracting the poses from RGB and rendering the output are time consuming - Only the first 30 frames of the input video will be used - The first 0.5 seconds of motion will be used to predict future motion - Processing time depends on video length and selected number of predictions - Precomputed results will be much faster if available """) with gr.Tabs(): with gr.Tab("Video Processing"): with gr.Row(): input_video = gr.Video(label="Input Video", height=600) with gr.Row(): gr.Examples( examples=sorted(pathlib.Path("downloads").glob("*.mp4")), inputs=input_video, cache_examples=False, ) with gr.Row(): use_precomputed = gr.Checkbox( label="Use precomputed results if available", value=True, info="If checked, will use existing results instead of processing the video again" ) displayed_preds = gr.Dropdown( choices=DISPLAYED_PREDS_OPTIONS, value=DISPLAYED_PREDS, label="Number of displayed predictions", info="Select how many motion predictions to display (2-6)" ) with gr.Row(): process_btn = gr.Button("Run Skeleton Diffusion") # Two-column output: left=input, right=output with gr.Row(): with gr.Column(): input_video_display = gr.Image(label="Input Video (Preview GIF)", height=600) with gr.Column(): output_video = gr.Image(label="Generated Motion", height=600) # Download buttons with gr.Row(): with gr.Column(): download_motion_btn = gr.Button("Download Motion Video") download_motion_btn.click( fn=lambda video_file, displayed_preds: os.path.join('outputs', f'{os.path.splitext(os.path.basename(video_file))[0]}_motion_{displayed_preds}.gif') if video_file else None, inputs=[input_video, displayed_preds], outputs=[gr.File(label="Download Motion Video")] ) with gr.Column(): download_data_btn = gr.Button("Download Motion Data (SMPL + Joints)") download_data_btn.click( fn=lambda video_file, displayed_preds: os.path.join('intermediate_results', os.path.splitext(os.path.basename(video_file))[0], f'results_{displayed_preds}.pkl') if video_file else None, inputs=[input_video, displayed_preds], outputs=[gr.File(label="Download Motion Data")] ) def process_video_with_notification(video_file, use_precomputed, displayed_preds): # Step 1: Show input video as GIF immediately with exactly 30 frames gr.Info("Converting input video to preview GIF...") gif_path = os.path.join(tempfile.gettempdir(), f"input_preview_{os.path.splitext(os.path.basename(video_file))[0]}.gif") video_to_gif(video_file, gif_path, frame_limit=30) # Explicitly set to 30 frames yield gif_path, LOADING_IMAGE_PATH base_name = os.path.splitext(os.path.basename(video_file))[0] output_path = os.path.join('outputs', f'{base_name}_motion_{displayed_preds}.gif') obs_gif_path = os.path.join('outputs', f'{base_name}_obj', 'shadow_gif', 'obs_obj_tranp.gif') concat_gif_path = os.path.join(tempfile.gettempdir(), f"concat_{base_name}.gif") # If using precomputed and obs_obj_tranp.gif exists, show concatenated GIF immediately if use_precomputed and os.path.exists(obs_gif_path): concat_gifs_side_by_side(gif_path, obs_gif_path, concat_gif_path) # Replace the preview with the concatenated GIF yield concat_gif_path, LOADING_IMAGE_PATH # ... continue with rest of workflow # (The rest of the workflow remains unchanged) if os.path.exists(output_path): gr.Info("Found precomputed video.\nUsing existing results...") yield concat_gif_path, output_path return # ... existing code ... # If not precomputed, after obs_obj_tranp.gif is generated, show concatenated GIF # Continue with the rest of the workflow as before # Case 1: Not using precomputed results if not use_precomputed: gr.Info("Starting video processing with GPU...\nThis may take a few minutes.") gr.Info("Step 1/3: Extracting SMPL parameters from video...") _, smpl_params = process_video_gpu(video_file, displayed_preds) # After SMPL extraction, check if obs_obj_tranp.gif exists and show concatenated GIF if os.path.exists(obs_gif_path): concat_gifs_side_by_side(gif_path, obs_gif_path, concat_gif_path) yield concat_gif_path, LOADING_IMAGE_PATH gr.Info("Step 2/3: Generating motion predictions...") motion_gif = generate_motion_video_gpu(smpl_params, video_file, displayed_preds) yield concat_gif_path, motion_gif return # If no precomputed video, check if we have enough GIFs to generate one shadow_gif_dir = os.path.join('outputs', f'{base_name}_obj', 'shadow_gif') if os.path.exists(shadow_gif_dir): existing_gifs = [f for f in os.listdir(shadow_gif_dir) if f.endswith('_tranp.gif') and not f.startswith('obs')] if len(existing_gifs) >= displayed_preds: gr.Info(f"Found {len(existing_gifs)} existing GIFs.\nGenerating video from existing predictions...") # Load input video frames for picture-in-picture input_frames = [] cap = cv2.VideoCapture(video_file) frame_count = 0 while cap.isOpened() and frame_count < FRAME_LIMIT: ret, frame = cap.read() if not ret: break input_frames.append(frame) frame_count += 1 cap.release() # Ensure we have exactly FRAME_LIMIT frames if len(input_frames) < FRAME_LIMIT: # Pad with the last frame if needed last_frame = input_frames[-1] if input_frames else np.zeros((480, 640, 3), dtype=np.uint8) input_frames.extend([last_frame] * (FRAME_LIMIT - len(input_frames))) elif len(input_frames) > FRAME_LIMIT: # Trim to FRAME_LIMIT frames input_frames = input_frames[:FRAME_LIMIT] # If obs_obj_tranp.gif exists, show concatenated GIF if os.path.exists(obs_gif_path): concat_gifs_side_by_side(gif_path, obs_gif_path, concat_gif_path) gif_path = concat_gif_path yield concat_gif_path, LOADING_IMAGE_PATH # Generate video from existing GIFs gr.Info("Combining predictions into final video...") motion_gif = combine.combine_video( os.path.join('outputs', f'{base_name}_obj'), output_path, input_frames, displayed_preds ) gr.Info("Processing complete!") yield gif_path, motion_gif return # If we don't have enough GIFs, proceed with full processing gr.Info("No precomputed results found.\nStarting full video processing...") gr.Info("Step 1/3: Extracting SMPL parameters from video...") _, smpl_params = process_video_gpu(video_file, displayed_preds) gr.Info("Step 2/3: Generating motion predictions...") motion_gif = generate_motion_video_gpu(smpl_params, video_file, displayed_preds) # After SMPL extraction, check if obs_obj_tranp.gif exists and show concatenated GIF if os.path.exists(obs_gif_path): concat_gifs_side_by_side(gif_path, obs_gif_path, concat_gif_path) yield concat_gif_path, LOADING_IMAGE_PATH yield concat_gif_path, motion_gif return process_btn.click( fn=process_video_with_notification, inputs=[input_video, use_precomputed, displayed_preds], outputs=[input_video_display, output_video] ) return demo if __name__ == "__main__": # Get local IP address s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) print(s.getsockname()[0]) s.close() # Create and launch interface demo = create_gradio_interface() print(demo.get_api_info()) demo.launch(server_name="0.0.0.0", share=True)