Spaces:
Runtime error
Runtime error
Update app.py
Browse files
app.py
CHANGED
|
@@ -5,7 +5,6 @@ from transformers import AutoModelForImageSegmentation
|
|
| 5 |
import torch
|
| 6 |
from torchvision import transforms
|
| 7 |
import moviepy.editor as mp
|
| 8 |
-
from pydub import AudioSegment
|
| 9 |
from PIL import Image
|
| 10 |
import numpy as np
|
| 11 |
import os
|
|
@@ -28,7 +27,19 @@ transform_image = transforms.Compose(
|
|
| 28 |
)
|
| 29 |
|
| 30 |
BATCH_SIZE = 3
|
| 31 |
-
executor = ThreadPoolExecutor(max_workers=4)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 32 |
|
| 33 |
@spaces.GPU
|
| 34 |
def fn(vid, bg_type="Color", bg_image=None, bg_video=None, color="#00FF00", fps=0, video_handling="slow_down"):
|
|
@@ -38,66 +49,62 @@ def fn(vid, bg_type="Color", bg_image=None, bg_video=None, color="#00FF00", fps=
|
|
| 38 |
audio = video.audio
|
| 39 |
except AttributeError:
|
| 40 |
audio = None
|
|
|
|
| 41 |
if fps == 0:
|
| 42 |
fps = video.fps
|
| 43 |
-
|
| 44 |
frames = video.iter_frames(fps=fps)
|
| 45 |
processed_frames = []
|
| 46 |
-
yield gr.update(visible=True), gr.update(visible=False)
|
| 47 |
|
| 48 |
if bg_type == "Video":
|
| 49 |
background_video = mp.VideoFileClip(bg_video)
|
| 50 |
-
|
| 51 |
if background_video.duration < video.duration and video_handling == "slow_down":
|
| 52 |
slow_down_factor = video.duration / background_video.duration
|
| 53 |
else:
|
| 54 |
slow_down_factor = 1
|
| 55 |
background_frames = list(background_video.iter_frames(fps=fps))
|
| 56 |
-
|
| 57 |
else:
|
| 58 |
background_frames = None
|
| 59 |
-
slow_down_factor = None
|
| 60 |
-
|
| 61 |
|
| 62 |
bg_frame_index = 0
|
| 63 |
frame_batch = []
|
| 64 |
|
| 65 |
-
|
| 66 |
for i, frame in enumerate(frames):
|
| 67 |
frame_batch.append(frame)
|
| 68 |
-
|
| 69 |
if len(frame_batch) == BATCH_SIZE or i == int(video.fps * video.duration) - 1:
|
| 70 |
-
|
| 71 |
pil_images = [Image.fromarray(f) for f in frame_batch]
|
| 72 |
|
| 73 |
if bg_type == "Video":
|
| 74 |
processed_images = list(executor.map(process, pil_images, [get_background_image(bg_type, bg_image, background_frames, bg_frame_index + j, video_handling, slow_down_factor) for j in range(len(pil_images))]))
|
| 75 |
bg_frame_index += len(frame_batch)
|
| 76 |
elif bg_type == "Color":
|
| 77 |
-
processed_images = list(executor.map(process, pil_images, [color] * len(pil_images)))
|
| 78 |
elif bg_type == "Image":
|
| 79 |
-
processed_images = list(executor.map(process, pil_images, [bg_image] * len(pil_images)))
|
| 80 |
else:
|
| 81 |
-
processed_images = pil_images
|
| 82 |
|
| 83 |
for processed_image in processed_images:
|
| 84 |
processed_frames.append(np.array(processed_image))
|
| 85 |
-
yield processed_image, None
|
| 86 |
frame_batch = []
|
| 87 |
|
|
|
|
| 88 |
processed_video = mp.ImageSequenceClip(processed_frames, fps=fps)
|
| 89 |
if audio:
|
| 90 |
processed_video = processed_video.set_audio(audio)
|
| 91 |
|
|
|
|
| 92 |
temp_dir = "temp"
|
| 93 |
os.makedirs(temp_dir, exist_ok=True)
|
| 94 |
unique_filename = str(uuid.uuid4()) + ".mp4"
|
| 95 |
temp_filepath = os.path.join(temp_dir, unique_filename)
|
| 96 |
-
|
| 97 |
processed_video.write_videofile(temp_filepath, codec="libx264", logger=None)
|
| 98 |
|
| 99 |
-
|
| 100 |
-
yield
|
|
|
|
| 101 |
|
| 102 |
except Exception as e:
|
| 103 |
print(f"Error: {e}")
|
|
@@ -105,30 +112,30 @@ def fn(vid, bg_type="Color", bg_image=None, bg_video=None, color="#00FF00", fps=
|
|
| 105 |
yield None, f"Error processing video: {e}"
|
| 106 |
|
| 107 |
|
|
|
|
|
|
|
| 108 |
def process(image, bg):
|
| 109 |
image_size = image.size
|
| 110 |
input_images = transform_image(image).unsqueeze(0).to("cuda")
|
| 111 |
-
# Prediction
|
| 112 |
with torch.no_grad():
|
| 113 |
preds = birefnet(input_images)[-1].sigmoid().cpu()
|
| 114 |
pred = preds[0].squeeze()
|
| 115 |
pred_pil = transforms.ToPILImage()(pred)
|
| 116 |
mask = pred_pil.resize(image_size)
|
| 117 |
|
| 118 |
-
if isinstance(bg, str) and bg.startswith("#"):
|
| 119 |
color_rgb = tuple(int(bg[i:i+2], 16) for i in (1, 3, 5))
|
| 120 |
-
background = Image.new("RGBA", image_size, color_rgb + (255,))
|
| 121 |
elif isinstance(bg, Image.Image):
|
| 122 |
-
background = bg.convert("RGBA").resize(image_size)
|
| 123 |
-
else:
|
| 124 |
-
background = Image.open(bg).convert("RGBA").resize(image_size)
|
| 125 |
|
| 126 |
-
# Composite the image onto the background using the mask
|
| 127 |
image = Image.composite(image, background, mask)
|
| 128 |
-
|
| 129 |
return image
|
| 130 |
|
| 131 |
|
|
|
|
| 132 |
with gr.Blocks(theme=gr.themes.Ocean()) as demo:
|
| 133 |
with gr.Row():
|
| 134 |
in_video = gr.Video(label="Input Video", interactive=True)
|
|
|
|
| 5 |
import torch
|
| 6 |
from torchvision import transforms
|
| 7 |
import moviepy.editor as mp
|
|
|
|
| 8 |
from PIL import Image
|
| 9 |
import numpy as np
|
| 10 |
import os
|
|
|
|
| 27 |
)
|
| 28 |
|
| 29 |
BATCH_SIZE = 3
|
| 30 |
+
executor = ThreadPoolExecutor(max_workers=4) # Adjust as needed
|
| 31 |
+
|
| 32 |
+
def get_background_image(bg_type, bg_image, background_frames, current_frame_index, video_handling, slow_down_factor):
|
| 33 |
+
if bg_type == "Video":
|
| 34 |
+
if video_handling == "slow_down":
|
| 35 |
+
frame_index = int(current_frame_index / slow_down_factor)
|
| 36 |
+
else:
|
| 37 |
+
frame_index = current_frame_index
|
| 38 |
+
return Image.fromarray(background_frames[frame_index % len(background_frames)])
|
| 39 |
+
elif bg_type == "Image":
|
| 40 |
+
return bg_image # Directly returns the image path
|
| 41 |
+
else: # bg_type == "Color"
|
| 42 |
+
return bg_image # bg_image here is the color string
|
| 43 |
|
| 44 |
@spaces.GPU
|
| 45 |
def fn(vid, bg_type="Color", bg_image=None, bg_video=None, color="#00FF00", fps=0, video_handling="slow_down"):
|
|
|
|
| 49 |
audio = video.audio
|
| 50 |
except AttributeError:
|
| 51 |
audio = None
|
| 52 |
+
|
| 53 |
if fps == 0:
|
| 54 |
fps = video.fps
|
|
|
|
| 55 |
frames = video.iter_frames(fps=fps)
|
| 56 |
processed_frames = []
|
| 57 |
+
yield gr.update(visible=True), gr.update(visible=False) # Update Gradio display
|
| 58 |
|
| 59 |
if bg_type == "Video":
|
| 60 |
background_video = mp.VideoFileClip(bg_video)
|
|
|
|
| 61 |
if background_video.duration < video.duration and video_handling == "slow_down":
|
| 62 |
slow_down_factor = video.duration / background_video.duration
|
| 63 |
else:
|
| 64 |
slow_down_factor = 1
|
| 65 |
background_frames = list(background_video.iter_frames(fps=fps))
|
|
|
|
| 66 |
else:
|
| 67 |
background_frames = None
|
| 68 |
+
slow_down_factor = None # Not needed for image or color backgrounds
|
|
|
|
| 69 |
|
| 70 |
bg_frame_index = 0
|
| 71 |
frame_batch = []
|
| 72 |
|
|
|
|
| 73 |
for i, frame in enumerate(frames):
|
| 74 |
frame_batch.append(frame)
|
|
|
|
| 75 |
if len(frame_batch) == BATCH_SIZE or i == int(video.fps * video.duration) - 1:
|
|
|
|
| 76 |
pil_images = [Image.fromarray(f) for f in frame_batch]
|
| 77 |
|
| 78 |
if bg_type == "Video":
|
| 79 |
processed_images = list(executor.map(process, pil_images, [get_background_image(bg_type, bg_image, background_frames, bg_frame_index + j, video_handling, slow_down_factor) for j in range(len(pil_images))]))
|
| 80 |
bg_frame_index += len(frame_batch)
|
| 81 |
elif bg_type == "Color":
|
| 82 |
+
processed_images = list(executor.map(process, pil_images, [color] * len(pil_images))) # Use color directly
|
| 83 |
elif bg_type == "Image":
|
| 84 |
+
processed_images = list(executor.map(process, pil_images, [bg_image] * len(pil_images))) # Use image path directly
|
| 85 |
else:
|
| 86 |
+
processed_images = pil_images # No processing needed
|
| 87 |
|
| 88 |
for processed_image in processed_images:
|
| 89 |
processed_frames.append(np.array(processed_image))
|
| 90 |
+
yield processed_image, None # Update Gradio with processed images
|
| 91 |
frame_batch = []
|
| 92 |
|
| 93 |
+
|
| 94 |
processed_video = mp.ImageSequenceClip(processed_frames, fps=fps)
|
| 95 |
if audio:
|
| 96 |
processed_video = processed_video.set_audio(audio)
|
| 97 |
|
| 98 |
+
# Save processed video to a temporary file
|
| 99 |
temp_dir = "temp"
|
| 100 |
os.makedirs(temp_dir, exist_ok=True)
|
| 101 |
unique_filename = str(uuid.uuid4()) + ".mp4"
|
| 102 |
temp_filepath = os.path.join(temp_dir, unique_filename)
|
|
|
|
| 103 |
processed_video.write_videofile(temp_filepath, codec="libx264", logger=None)
|
| 104 |
|
| 105 |
+
|
| 106 |
+
yield gr.update(visible=False), gr.update(visible=True) # Update Gradio display
|
| 107 |
+
yield processed_image, temp_filepath # Return final output
|
| 108 |
|
| 109 |
except Exception as e:
|
| 110 |
print(f"Error: {e}")
|
|
|
|
| 112 |
yield None, f"Error processing video: {e}"
|
| 113 |
|
| 114 |
|
| 115 |
+
|
| 116 |
+
|
| 117 |
def process(image, bg):
|
| 118 |
image_size = image.size
|
| 119 |
input_images = transform_image(image).unsqueeze(0).to("cuda")
|
|
|
|
| 120 |
with torch.no_grad():
|
| 121 |
preds = birefnet(input_images)[-1].sigmoid().cpu()
|
| 122 |
pred = preds[0].squeeze()
|
| 123 |
pred_pil = transforms.ToPILImage()(pred)
|
| 124 |
mask = pred_pil.resize(image_size)
|
| 125 |
|
| 126 |
+
if isinstance(bg, str) and bg.startswith("#"): # If bg is a color
|
| 127 |
color_rgb = tuple(int(bg[i:i+2], 16) for i in (1, 3, 5))
|
| 128 |
+
background = Image.new("RGBA", image_size, color_rgb + (255,)) # Create image with color
|
| 129 |
elif isinstance(bg, Image.Image):
|
| 130 |
+
background = bg.convert("RGBA").resize(image_size) #Resize if bg is an image
|
| 131 |
+
else: #If bg is an image path
|
| 132 |
+
background = Image.open(bg).convert("RGBA").resize(image_size) # Open and resize image
|
| 133 |
|
|
|
|
| 134 |
image = Image.composite(image, background, mask)
|
|
|
|
| 135 |
return image
|
| 136 |
|
| 137 |
|
| 138 |
+
|
| 139 |
with gr.Blocks(theme=gr.themes.Ocean()) as demo:
|
| 140 |
with gr.Row():
|
| 141 |
in_video = gr.Video(label="Input Video", interactive=True)
|