|
|
import gradio as gr |
|
|
import requests |
|
|
import io |
|
|
from PIL import Image, ImageOps |
|
|
import base64 |
|
|
import time |
|
|
|
|
|
def load_image_from_url(url): |
|
|
try: |
|
|
response = requests.get(url) |
|
|
response.raise_for_status() |
|
|
image = Image.open(io.BytesIO(response.content)) |
|
|
return image |
|
|
except Exception as e: |
|
|
return None, f"Error: {e}" |
|
|
|
|
|
def send_to_api(key, prompt, image_url, mask_base64, path_points): |
|
|
"""Send the image and mask to the API endpoint.""" |
|
|
url = "https://api.goapi.ai/api/v1/task" |
|
|
payload = { |
|
|
"model": "kling", |
|
|
"task_type": "video_generation", |
|
|
"input": { |
|
|
"prompt": prompt, |
|
|
"negative_prompt": "", |
|
|
"cfg_scale": 0.5, |
|
|
"duration": 5, |
|
|
"image_url": image_url, |
|
|
"image_tail_url": "", |
|
|
"mode": "std", |
|
|
"version": "1.0", |
|
|
"motion_brush": { |
|
|
"mask_url": f"data:image/png;base64,{mask_base64}", |
|
|
"static_masks": [{"points": []}], |
|
|
"dynamic_masks": [{"points": path_points}] |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
headers = { |
|
|
"x-api-key": key |
|
|
} |
|
|
|
|
|
response = requests.post(url, headers=headers, json=payload) |
|
|
if response.status_code == 200: |
|
|
data = response.json() |
|
|
task_id = data.get("data", {}).get("task_id") |
|
|
return task_id if task_id else None |
|
|
else: |
|
|
return f"Request failed, status code: {response.status_code}", None |
|
|
|
|
|
def fetch_api(task_id, key): |
|
|
"""Fetch task status and return video URL, retrying every 20 seconds until task is completed.""" |
|
|
url = f"https://api.goapi.ai/api/v1/task/{task_id}" |
|
|
headers = { |
|
|
"x-api-key": key |
|
|
} |
|
|
|
|
|
while True: |
|
|
response = requests.get(url, headers=headers) |
|
|
if response.status_code == 200: |
|
|
data = response.json() |
|
|
status = data.get("data", {}).get("status", "") |
|
|
if status == "completed": |
|
|
video_url = data.get("data", {}).get("output", {}).get("video_url", "Error video URL") |
|
|
return video_url |
|
|
else: |
|
|
print(f"Task status is '{status}'. Retrying in 10 seconds...") |
|
|
else: |
|
|
return f"Request failed, status code: {response.status_code}", None |
|
|
|
|
|
time.sleep(10) |
|
|
|
|
|
def image_to_base64(image): |
|
|
"""Convert a PIL Image to a base64-encoded PNG string.""" |
|
|
buffered = io.BytesIO() |
|
|
image.save(buffered, format="PNG") |
|
|
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8") |
|
|
return img_base64 |
|
|
|
|
|
def generate_mask_and_path(editor_value, path_direction, key, prompt, original_image_url): |
|
|
layers = editor_value.get("layers", []) |
|
|
if len(layers) < 3: |
|
|
return None |
|
|
|
|
|
green_layer = layers[0] |
|
|
green_mask = ImageOps.colorize( |
|
|
ImageOps.grayscale(green_layer), black="black", white="green" |
|
|
) |
|
|
|
|
|
black_layer = layers[1] |
|
|
black_mask = ImageOps.colorize( |
|
|
ImageOps.grayscale(green_layer), black="black", white="green" |
|
|
) |
|
|
|
|
|
width, height = green_mask.size |
|
|
composite_image = Image.new("RGB", (width, height), "white") |
|
|
composite_image.paste(green_mask, mask=green_layer) |
|
|
composite_image.paste(black_mask, mask=black_layer) |
|
|
|
|
|
path_layer = layers[2] |
|
|
path_array = path_layer.load() |
|
|
path_points = [] |
|
|
|
|
|
|
|
|
if path_direction == "Left to Right": |
|
|
for y in range(height): |
|
|
for x in range(width): |
|
|
if path_array[x, y] == (255, 255, 255, 255): |
|
|
path_points.append({"x": x, "y": y}) |
|
|
elif path_direction == "Right to Left": |
|
|
for y in range(height): |
|
|
for x in range(width - 1, -1, -1): |
|
|
if path_array[x, y] == (255, 255, 255, 255): |
|
|
path_points.append({"x": x, "y": y}) |
|
|
elif path_direction == "Top to Bottom": |
|
|
for x in range(width): |
|
|
for y in range(height): |
|
|
if path_array[x, y] == (255, 255, 255, 255): |
|
|
path_points.append({"x": x, "y": y}) |
|
|
elif path_direction == "Bottom to Top": |
|
|
for x in range(width): |
|
|
for y in range(height - 1, -1, -1): |
|
|
if path_array[x, y] == (255, 255, 255, 255): |
|
|
path_points.append({"x": x, "y": y}) |
|
|
|
|
|
selected_points = [] |
|
|
if path_points: |
|
|
step = max(len(path_points) // 10, 1) |
|
|
selected_points = path_points[::step][:10] |
|
|
|
|
|
original_image = original_image_url |
|
|
mask_base64 = image_to_base64(composite_image) |
|
|
|
|
|
task_id = send_to_api(key, prompt, original_image, mask_base64, selected_points) |
|
|
video_url = fetch_api(task_id, key) |
|
|
|
|
|
return composite_image, selected_points, task_id, video_url |
|
|
|
|
|
with gr.Blocks() as interface: |
|
|
gr.Markdown("# Video Motion Generation Tool") |
|
|
|
|
|
gr.Markdown("---") |
|
|
gr.Markdown("### 1. Input Background Image URL") |
|
|
with gr.Row(): |
|
|
url_input = gr.Textbox(label="Input Background Image URL", placeholder="Enter the image URL") |
|
|
load_image_btn = gr.Button("Load Image") |
|
|
|
|
|
gr.Markdown("---") |
|
|
gr.Markdown("### 2. Use the Brush Tool to Edit the Image") |
|
|
gr.Markdown("Layer 1 will generate a dynamic mask, Layer 2 is a static mask, and Layer 3 will generate path points.") |
|
|
with gr.Row(): |
|
|
image_editor = gr.ImageEditor( |
|
|
type="pil", |
|
|
brush=gr.Brush(default_size=20, colors=["#FFFFFF"], color_mode="fixed"), |
|
|
layers=True, |
|
|
interactive=True, |
|
|
label="Drawing Tool to Generate Mask and Path", |
|
|
height=700, |
|
|
) |
|
|
with gr.Row(): |
|
|
prompt_input = gr.Textbox(label="Prompt", placeholder="Enter Prompt") |
|
|
|
|
|
with gr.Row(): |
|
|
key_input = gr.Textbox(label="API Key", placeholder="Enter PiAPI Key") |
|
|
|
|
|
|
|
|
with gr.Row(): |
|
|
direction_input = gr.Dropdown( |
|
|
choices=["Left to Right", "Right to Left", "Top to Bottom", "Bottom to Top"], label="Select Path Direction" |
|
|
) |
|
|
submit_btn = gr.Button("Generate") |
|
|
|
|
|
with gr.Row(): |
|
|
output_composite_image = gr.Image(label="Generated Composite Image") |
|
|
output_path_points = gr.Textbox(label="Path Point Data") |
|
|
output_task_id = gr.Textbox(label="Task ID") |
|
|
output_video = gr.Video(label="Generated Video Link") |
|
|
|
|
|
|
|
|
load_image_btn.click( |
|
|
fn=load_image_from_url, |
|
|
inputs=[url_input], |
|
|
outputs=[image_editor], |
|
|
) |
|
|
|
|
|
submit_btn.click( |
|
|
fn=generate_mask_and_path, |
|
|
inputs=[image_editor, direction_input, key_input, prompt_input, url_input], |
|
|
outputs=[output_composite_image, output_path_points, output_task_id, output_video], |
|
|
) |
|
|
|
|
|
interface.launch() |
|
|
|
