515 lines
21 KiB
Python
515 lines
21 KiB
Python
import requests
|
|
import json
|
|
import os
|
|
import time
|
|
import math
|
|
from uuid import uuid4
|
|
from PIL import Image
|
|
import torch
|
|
import shutil
|
|
import numpy as np
|
|
import comfy
|
|
from dotenv import load_dotenv
|
|
import os
|
|
import base64
|
|
import io
|
|
|
|
# Load the .env file
|
|
load_dotenv()
|
|
|
|
# Get haiper key
|
|
haiper_key = os.getenv('HAIPER_KEY')
|
|
|
|
def image2base64(image):
|
|
image = 255.0 * image.cpu().numpy()
|
|
image = Image.fromarray(np.clip(image, 0, 255).astype(np.uint8))
|
|
|
|
# Save PIL Image to buffer
|
|
buffer = io.BytesIO()
|
|
image.save(buffer, format="PNG")
|
|
buffer.seek(0)
|
|
|
|
base64_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
|
|
return "data:image/png;base64," + base64_image
|
|
|
|
|
|
class I2VPipelineNode:
|
|
def __init__(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
"""
|
|
Return a dictionary which contains config for all input fields.
|
|
"""
|
|
return {
|
|
"required": {
|
|
"prompt": ("STRING", {"default": "add prompt here", "display": "text"}),
|
|
"source_image": ("IMAGE",),
|
|
},
|
|
"optional": {
|
|
"duration": ("INT", {"default": 6, "min": 1, "max": 6, "step": 1, "display": "number"}),
|
|
"seed": ("INT", {"default": -1, "step": 1, "display": "number"}),
|
|
"resolution": ("INT", {"default": 720, "step": 1, "display": "number"}),
|
|
"is_public": ("BOOLEAN", {"default": False}),
|
|
"is_enable_prompt_enhancer": ("BOOLEAN", {"default": False}),
|
|
"guidance_scale": ("INT", {"default": 50, "step": 10, "display": "number"})
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("output_video_path",)
|
|
|
|
FUNCTION = "run"
|
|
|
|
CATEGORY = "HaiperAI"
|
|
|
|
def run(self, prompt, source_image, duration=None, seed=None, resolution=None, is_public=None, is_enable_prompt_enhancer=None, guidance_scale=None):
|
|
random_id = uuid4()
|
|
directory = f"/tmp/{random_id}/"
|
|
os.makedirs(directory, exist_ok=True)
|
|
output_path = os.path.join(directory, "output-i2v.mp4")
|
|
|
|
source_image = image2base64(source_image[0])
|
|
|
|
# Use the user-defined prompt instead of a hardcoded prompt
|
|
run_status = get_video_by_i2v_pipeline(prompt, source_image, duration, seed, resolution, is_public, is_enable_prompt_enhancer, guidance_scale, output_path)
|
|
|
|
if run_status:
|
|
return (output_path,)
|
|
else:
|
|
raise RuntimeError("Run video generation failed")
|
|
|
|
class T2VPipelineNode:
|
|
def __init__(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
"""
|
|
Return a dictionary which contains config for all input fields.
|
|
"""
|
|
return {
|
|
"required": {
|
|
"prompt": ("STRING", {"default": "add prompt here", "display": "text"}),
|
|
},
|
|
"optional": {
|
|
"negative_prompt": ("STRING", {"default": "bad, slow", "display": "text"}),
|
|
"seed": ("INT", {"default": -1, "step": 1, "display": "number"}),
|
|
"aspect_ratio": ("STRING", {"default": "16:9", "display": "text"}),
|
|
"resolution": ("INT", {"default": 720, "step": 1, "display": "number"}),
|
|
"duration": ("INT", {"default": 6, "min": 1, "max": 6, "step": 1, "display": "number"}),
|
|
"is_public": ("BOOLEAN", {"default": False}),
|
|
"use_ff_cond": ("BOOLEAN", {"default": True}),
|
|
"is_enable_prompt_enhancer": ("BOOLEAN", {"default": True})
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("output_video_path",)
|
|
|
|
FUNCTION = "run"
|
|
|
|
CATEGORY = "HaiperAI"
|
|
|
|
def run(self, prompt, negative_prompt=None, seed=None, aspect_ratio=None, resolution=None, duration=None, is_public=None, use_ff_cond=None, is_enable_prompt_enhancer=None):
|
|
random_id = uuid4()
|
|
directory = f"/tmp/{random_id}/"
|
|
os.makedirs(directory, exist_ok=True)
|
|
|
|
output_path = os.path.join(directory, "output-t2v.mp4")
|
|
|
|
# Use the user-defined prompt instead of a hardcoded prompt
|
|
run_status = get_video_by_t2v_pipeline(prompt, negative_prompt, seed, aspect_ratio, resolution, duration, is_public, use_ff_cond, is_enable_prompt_enhancer, output_path)
|
|
|
|
if run_status:
|
|
return (output_path,)
|
|
else:
|
|
raise RuntimeError("Run video generation failed")
|
|
|
|
class T2IPipelineNode:
|
|
def __init__(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
"""
|
|
Return a dictionary which contains config for all input fields.
|
|
"""
|
|
return {
|
|
"required": {
|
|
"prompt": ("STRING", {"default": "add prompt here", "display": "text"}),
|
|
},
|
|
"optional": {
|
|
"negative_prompt": ("STRING", {"default": "bad, slow", "display": "text"}),
|
|
"seed": ("INT", {"default": -1, "step": 1, "display": "number"}),
|
|
"aspect_ratio": ("STRING", {"default": "16:9", "display": "text"}),
|
|
"resolution": ("INT", {"default": 720, "step": 1, "display": "number"}),
|
|
"is_public": ("BOOLEAN", {"default": False}),
|
|
"is_enable_prompt_enhancer": ("BOOLEAN", {"default": True})
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "STRING")
|
|
RETURN_NAMES = ("output_image_dir_path", "output_image_0_path", "output_image_1_path", "output_image_2_path", "output_image_3_path")
|
|
|
|
FUNCTION = "run"
|
|
|
|
CATEGORY = "HaiperAI"
|
|
|
|
def run(self, prompt, negative_prompt=None, seed=None, aspect_ratio=None, resolution=None, is_public=None, is_enable_prompt_enhancer=None):
|
|
random_id = uuid4()
|
|
directory = f"/tmp/{random_id}/"
|
|
os.makedirs(directory, exist_ok=True)
|
|
|
|
output_image_dir_path = directory
|
|
output_image_0_path = os.path.join(directory, "output-image-0.jpg")
|
|
output_image_1_path = os.path.join(directory, "output-image-1.jpg")
|
|
output_image_2_path = os.path.join(directory, "output-image-2.jpg")
|
|
output_image_3_path = os.path.join(directory, "output-image-3.jpg")
|
|
|
|
|
|
# Use the user-defined prompt instead of a hardcoded prompt
|
|
run_status = get_video_by_t2i_pipeline(prompt, negative_prompt, seed, aspect_ratio, resolution, is_public, is_enable_prompt_enhancer, output_image_0_path, output_image_1_path, output_image_2_path, output_image_3_path)
|
|
|
|
if run_status:
|
|
return (output_image_dir_path, output_image_0_path, output_image_1_path, output_image_2_path, output_image_3_path)
|
|
else:
|
|
raise RuntimeError("Run image generation failed")
|
|
|
|
class KFCPipelineNode:
|
|
def __init__(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
"""
|
|
Return a dictionary which contains config for all input fields.
|
|
"""
|
|
return {
|
|
"required": {
|
|
"frame_1": ("IMAGE",),
|
|
"frame_2": ("IMAGE",),
|
|
"frame_indices_str": ("STRING", {"default": "0, 31"}),
|
|
"image_width": ("INT", {"default": 1280, "step": 1, "display": "number"}),
|
|
"image_height": ("INT", {"default": 720, "step": 1, "display": "number"}),
|
|
"is_public": ("BOOLEAN", {"default": False}),
|
|
"prompt": ("STRING", {"default": "a smooth consecutive video"}),
|
|
"negative_prompt": ("STRING", {"default": "bad, slow"}),
|
|
"seed": ("INT", {"default": -1, "step": 1, "display": "number"}),
|
|
"duration": ("INT", {"default": 4, "step": 1, "display": "number"}),
|
|
},
|
|
"optional": {
|
|
"frame_3": ("IMAGE",),
|
|
"frame_4": ("IMAGE",),
|
|
"frame_5": ("IMAGE",),
|
|
"frame_6": ("IMAGE",),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("output_video_path",)
|
|
|
|
FUNCTION = "run"
|
|
|
|
CATEGORY = "haiper"
|
|
|
|
def run(self, frame_1, frame_2, frame_indices_str, image_width, image_height, is_public, prompt, negative_prompt, seed, duration, frame_3=None, frame_4=None, frame_5=None, frame_6=None):
|
|
# Convert the frame indices string to a list of integers
|
|
frame_indices = [int(item.strip()) for item in frame_indices_str.split(",")]
|
|
|
|
frames = [frame[0] for frame in [frame_1, frame_2, frame_3, frame_4, frame_5, frame_6] if frame is not None]
|
|
source_images = []
|
|
for frame in frames:
|
|
source_images.append(image2base64(frame))
|
|
|
|
# Check that the number of source_images matches the frame indices
|
|
if len(source_images) != len(frame_indices):
|
|
raise ValueError("Source Images need to align with the length of Frame Indices")
|
|
|
|
random_id = uuid4()
|
|
directory = f"/tmp/{random_id}/"
|
|
os.makedirs(directory, exist_ok=True)
|
|
output_path = os.path.join(directory, "output.mp4")
|
|
|
|
# Use the user-defined prompt instead of a hardcoded prompt
|
|
run_status = get_video_by_kfc_pipeline(source_images, frame_indices, image_width, image_height, is_public, prompt, negative_prompt, seed, duration, output_path)
|
|
|
|
if run_status:
|
|
return (output_path,)
|
|
else:
|
|
raise RuntimeError("Run video generation failed")
|
|
|
|
def get_video_by_i2v_pipeline(prompt, source_image, duration, seed, resolution, is_public, is_enable_prompt_enhancer, guidance_scale, output_path):
|
|
pbar = comfy.utils.ProgressBar(100)
|
|
pbar.update_absolute(0, 100)
|
|
|
|
try:
|
|
# Replace with your actual API endpoint
|
|
api_url = 'https://api.haiper.ai/v1/jobs/gen2/image2video'
|
|
|
|
payload = json.dumps({
|
|
"prompt": prompt, # Use the prompt provided by the user
|
|
"config": {
|
|
"source_image": source_image,
|
|
},
|
|
"settings": {
|
|
"duration": duration,
|
|
"seed": seed,
|
|
"resolution": resolution,
|
|
"guidance_scale": guidance_scale
|
|
},
|
|
"is_public": is_public,
|
|
"is_enable_prompt_enhancer": is_enable_prompt_enhancer
|
|
})
|
|
|
|
headers = {
|
|
'authorization': f'Bearer {haiper_key}',
|
|
'content-type': 'application/json'
|
|
}
|
|
|
|
gen_response = requests.request("POST", api_url, headers=headers, data=payload).json()
|
|
generation_id = gen_response['value']["generation_id"]
|
|
|
|
# Poll the status until the video is ready
|
|
while True:
|
|
time.sleep(20) # Wait for 20 seconds before checking the status
|
|
status_url = f'https://api.haiper.ai/v1/jobs/{generation_id}/status'
|
|
status_response = requests.request("GET", status_url, headers=headers)
|
|
status_data = status_response.json().get('value')
|
|
print(status_data)
|
|
|
|
status = status_data.get('status')
|
|
progress = status_data.get('progress', 0)
|
|
|
|
pbar.update_absolute(math.ceil(progress * 100), 100)
|
|
|
|
if status == 'succeed':
|
|
# Get watermark free video url
|
|
get_watermark_free_video_url = f'https://api.haiper.ai/v1/creation/{generation_id}/watermark-free-url'
|
|
url_response = requests.request("POST", get_watermark_free_video_url, headers=headers).json()
|
|
watermark_free_url = url_response['value']['url']
|
|
|
|
# Download watermark free video from URL
|
|
video_response = requests.get(watermark_free_url, stream=True)
|
|
with open(output_path, 'wb') as video_file:
|
|
shutil.copyfileobj(video_response.raw, video_file)
|
|
|
|
return True
|
|
|
|
elif status == 'processing' or status == 'pending' or status == 'post_processing':
|
|
print(f"Video is still processing (Job ID: {generation_id}). Waiting...")
|
|
continue
|
|
else:
|
|
print(f"Unexpected status '{status}'. Exiting.")
|
|
return False
|
|
|
|
except Exception as error:
|
|
print(error)
|
|
|
|
def get_video_by_t2v_pipeline(prompt, negative_prompt, seed, aspect_ratio, resolution, duration, is_public, use_ff_cond, is_enable_prompt_enhancer, output_path):
|
|
pbar = comfy.utils.ProgressBar(100)
|
|
pbar.update_absolute(0, 100)
|
|
|
|
try:
|
|
# Replace with your actual API endpoint
|
|
api_url = 'https://api.haiper.ai/v1/jobs/gen2/text2video'
|
|
|
|
payload = json.dumps({
|
|
"prompt": prompt,
|
|
"negative_prompt": negative_prompt,
|
|
"settings": {
|
|
"seed": seed,
|
|
"aspect_ratio": aspect_ratio,
|
|
"resolution": resolution,
|
|
"duration": duration
|
|
},
|
|
"is_public": is_public,
|
|
"use_ff_cond": use_ff_cond,
|
|
"is_enable_prompt_enhancer": is_enable_prompt_enhancer
|
|
})
|
|
|
|
headers = {
|
|
'authorization': f'Bearer {haiper_key}',
|
|
'content-type': 'application/json'
|
|
}
|
|
|
|
gen_response = requests.request("POST", api_url, headers=headers, data=payload).json()
|
|
generation_id = gen_response['value']["generation_id"]
|
|
|
|
# Poll the status until the video is ready
|
|
while True:
|
|
time.sleep(20) # Wait for 20 seconds before checking the status
|
|
status_url = f'https://api.haiper.ai/v1/jobs/{generation_id}/status'
|
|
status_response = requests.request("GET", status_url, headers=headers)
|
|
status_data = status_response.json().get('value')
|
|
print(status_data)
|
|
|
|
status = status_data.get('status')
|
|
progress = status_data.get('progress', 0)
|
|
|
|
pbar.update_absolute(math.ceil(progress * 100), 100)
|
|
|
|
if status == 'succeed':
|
|
# Get watermark free video url
|
|
generate_watermark_free_video_url = f'https://api.haiper.ai/v1/creation/{generation_id}/watermark-free-url'
|
|
url_response = requests.request("POST", generate_watermark_free_video_url, headers=headers).json()
|
|
watermark_free_url = url_response['value']['url']
|
|
|
|
# Download watermark free video from URL
|
|
video_response = requests.get(watermark_free_url, stream=True)
|
|
with open(output_path, 'wb') as video_file:
|
|
shutil.copyfileobj(video_response.raw, video_file)
|
|
|
|
return True
|
|
|
|
elif status == 'processing' or status == 'pending' or status == 'post_processing':
|
|
print(f"Video is still processing (Job ID: {generation_id}). Waiting...")
|
|
continue
|
|
else:
|
|
print(f"Unexpected status '{status}'. Exiting.")
|
|
return False
|
|
|
|
except Exception as error:
|
|
print(error)
|
|
|
|
def get_video_by_t2i_pipeline(prompt, negative_prompt, seed, aspect_ratio, resolution, is_public, is_enable_prompt_enhancer, output_image_0_path, output_image_1_path, output_image_2_path, output_image_3_path):
|
|
pbar = comfy.utils.ProgressBar(100)
|
|
pbar.update_absolute(0, 100)
|
|
|
|
try:
|
|
# Replace with your actual API endpoint
|
|
api_url = 'https://api.haiper.ai/v1/jobs/gen2/text2image'
|
|
|
|
payload = json.dumps({
|
|
"prompt": prompt,
|
|
"negative_prompt": negative_prompt,
|
|
"settings": {
|
|
"seed": seed,
|
|
"aspect_ratio": aspect_ratio,
|
|
"resolution": resolution
|
|
},
|
|
"is_public": is_public,
|
|
"is_enable_prompt_enhancer": is_enable_prompt_enhancer
|
|
})
|
|
|
|
headers = {
|
|
'authorization': f'Bearer {haiper_key}',
|
|
'content-type': 'application/json'
|
|
}
|
|
|
|
gen_response = requests.request("POST", api_url, headers=headers, data=payload).json()
|
|
generation_id = gen_response['value']["generation_id"]
|
|
|
|
# Poll the status until the video is ready
|
|
while True:
|
|
time.sleep(20) # Wait for 20 seconds before checking the status
|
|
status_url = f'https://api.haiper.ai/v1/jobs/{generation_id}/status'
|
|
status_response = requests.request("GET", status_url, headers=headers)
|
|
status_data = status_response.json().get('value')
|
|
print(status_data)
|
|
|
|
status = status_data.get('status')
|
|
progress = status_data.get('progress', 0)
|
|
|
|
pbar.update_absolute(math.ceil(progress * 100), 100)
|
|
|
|
if status == 'succeed':
|
|
get_creation_detail_url = f'https://api.haiper.ai/v1/creation/{generation_id}'
|
|
# Get the image URLs
|
|
gen_response = requests.request("GET", get_creation_detail_url, headers=headers, data=payload).json()
|
|
output_image_0_url = gen_response['value']["outputs"][0]['media_url']
|
|
output_image_1_url = gen_response['value']["outputs"][1]['media_url']
|
|
output_image_2_url = gen_response['value']["outputs"][2]['media_url']
|
|
output_image_3_url = gen_response['value']["outputs"][3]['media_url']
|
|
|
|
# Download image from URL
|
|
image_response = requests.get(output_image_0_url, stream=True)
|
|
with open(output_image_0_path, 'wb') as image_file:
|
|
shutil.copyfileobj(image_response.raw, image_file)
|
|
image_response = requests.get(output_image_1_url, stream=True)
|
|
with open(output_image_1_path, 'wb') as image_file:
|
|
shutil.copyfileobj(image_response.raw, image_file)
|
|
image_response = requests.get(output_image_2_url, stream=True)
|
|
with open(output_image_2_path, 'wb') as image_file:
|
|
shutil.copyfileobj(image_response.raw, image_file)
|
|
image_response = requests.get(output_image_3_url, stream=True)
|
|
with open(output_image_3_path, 'wb') as image_file:
|
|
shutil.copyfileobj(image_response.raw, image_file)
|
|
return True
|
|
|
|
elif status == 'processing' or status == 'pending' or status == 'post_processing':
|
|
print(f"Images are still processing (Job ID: {generation_id}). Waiting...")
|
|
continue
|
|
else:
|
|
print(f"Unexpected status '{status}'. Exiting.")
|
|
return False
|
|
|
|
except Exception as error:
|
|
print(error)
|
|
|
|
|
|
def get_video_by_kfc_pipeline(source_images, frame_indices, image_width, image_height, is_public, prompt, negative_prompt, seed, duration, output_path):
|
|
pbar = comfy.utils.ProgressBar(100)
|
|
pbar.update_absolute(0, 100)
|
|
|
|
try:
|
|
api_url = 'https://api.haiper.ai/v1/jobs/gen2/afc'
|
|
|
|
payload = json.dumps({
|
|
"config": {
|
|
"source_images": source_images,
|
|
"frame_indices": frame_indices,
|
|
"input_width": image_width,
|
|
"input_height": image_height,
|
|
},
|
|
"is_public": is_public,
|
|
"prompt": prompt, # Use the prompt provided by the user
|
|
"negative_prompt": negative_prompt,
|
|
"settings": {
|
|
"seed": seed,
|
|
"duration": duration,
|
|
}
|
|
})
|
|
|
|
headers = {
|
|
'authorization': f'Bearer {haiper_key}',
|
|
'content-type': 'application/json'
|
|
}
|
|
|
|
gen_response = requests.request("POST", api_url, headers=headers, data=payload).json()
|
|
generation_id = gen_response['value']["generation_id"]
|
|
|
|
# Poll the status until the video is ready
|
|
while True:
|
|
time.sleep(20) # Wait for 20 seconds before checking the status
|
|
status_url = f'https://api.haiper.ai/v1/jobs/{generation_id}/status'
|
|
status_response = requests.request("GET", status_url, headers=headers)
|
|
status_data = status_response.json().get('value')
|
|
|
|
status = status_data.get('status')
|
|
progress = status_data.get('progress', 0)
|
|
|
|
pbar.update_absolute(math.ceil(progress * 100), 100)
|
|
if status == 'succeed':
|
|
# Get watermark free video url
|
|
generate_watermark_free_video_url = f'https://api.haiper.ai/v1/creation/{generation_id}/watermark-free-url'
|
|
url_response = requests.request("POST", generate_watermark_free_video_url, headers=headers).json()
|
|
watermark_free_url = url_response['value']['url']
|
|
# Download watermark free video from URL
|
|
video_response = requests.get(watermark_free_url, stream=True)
|
|
with open(output_path, 'wb') as video_file:
|
|
shutil.copyfileobj(video_response.raw, video_file)
|
|
|
|
return True
|
|
|
|
elif status == 'processing' or status == 'pending' or status == 'post_processing':
|
|
print(f"Video is still processing (Job ID: {generation_id}). Waiting...")
|
|
continue
|
|
else:
|
|
print(f"Unexpected status '{status}'. Exiting.")
|
|
return False
|
|
|
|
except Exception as error:
|
|
print(error)
|