387 lines
16 KiB
Python
387 lines
16 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 . import M2Mtoken
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
# Load the .env file
|
|
load_dotenv()
|
|
|
|
# Get the client_id and client_secret
|
|
client_id = os.getenv('CLIENT_ID')
|
|
client_secret = os.getenv('CLIENT_SECRET')
|
|
|
|
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": ("STRING", {"default": "add source image url here", "display": "text"}),
|
|
},
|
|
"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")
|
|
|
|
# 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")
|
|
|
|
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://api1.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
|
|
})
|
|
|
|
# Get the access token
|
|
access_token = M2Mtoken.get_token(client_id, client_secret)
|
|
headers = {
|
|
'x-vision-env': 'pro',
|
|
'x-haiper-client': 'public',
|
|
'x-haiper-auth': 'auth0',
|
|
'authorization': f'Bearer {access_token}',
|
|
'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://api1.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://api1.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://api1.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
|
|
})
|
|
|
|
# Get the access token
|
|
access_token = M2Mtoken.get_token(client_id, client_secret)
|
|
headers = {
|
|
'x-vision-env': 'pro',
|
|
'x-haiper-client': 'public',
|
|
'x-haiper-auth': 'auth0',
|
|
'authorization': f'Bearer {access_token}',
|
|
'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://api1.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://api1.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://api1.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
|
|
})
|
|
|
|
# Get the access token
|
|
access_token = M2Mtoken.get_token(client_id, client_secret)
|
|
headers = {
|
|
'x-vision-env': 'pro',
|
|
'x-haiper-client': 'public',
|
|
'x-haiper-auth': 'auth0',
|
|
'authorization': f'Bearer {access_token}',
|
|
'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://api1.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://api1.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)
|