Workflow

APIs for create workflow and examples of how to get result of workflow.

Overlay Video template creates a video that have animated media with background video. Here is an Example Video of Overlay Video template.

The input is your image to animate and URLs of the background video.

Example code of How to get Overlay Video video.

import requests
import mimetypes
import async

image_path = "path/to/image.png"
image_name = os.path.basename(image_path)

background_urls = [URL1, URL2]

options = {
    "background_urls": ','.join(background_urls),
    "prompt": 'YOUR_PROMPT_HERE',
    "motion_type": "Auto"
}

payload = {
    "title": "Overlay Video Example",
    "description": "Overlay Video Example",
    "options": json.dumps(options),
}

## Create file input request
content_type, _ = mimetypes.guess_type(image_path)
imageBody =  open(image_path, 'rb')
files=[
    ('image_files',(image_name, imageBody, content_type))
]

headers = {
    'X-Api-Key': "YOUR_API_KEY"
}

try:
    response = await fetch("POST", f"{BASE_URL}/workflow", headers=headers, data=payload, files=files)
    workflow_data = response.json().get('data', None)
except:
    print(response.text)
    print("Please submit again!")
    exit()

imageBody.close()

# Check if workflow is available
if not workflow_data:
    print("Please submit again!")
    exit()

# Extract workflow ID from the response
workflow_id = workflow_data.get("id")

await asyncio.sleep(10)
# Get workflow detail to check status
response = await fetch("GET", f"{BASE_URL}/workflow/{workflow_id}", headers=headers)
print(f"{workflow_id}: {response.text}")
processing_status = response.json().get("data").get("processing_status")
processing_percentage = response.json().get("data").get("progress_percentage")

# Poll the processing status until it is either 'success' or 'fail'
while processing_status not in ['success', 'fail']:
    await asyncio.sleep(20)  # Wait for 20 seconds before checking again
    try:
        # Check the processing status again
        response = await fetch("GET", f"{BASE_URL}/workflow/{workflow_id}", headers=headers)
        processing_status = response.json().get("data").get("processing_status")
        processing_percentage = response.json().get("data").get("progress_percentage")
    except:
        print("Warning: Failed to retrieve status.")

# If processing failed, notify the user
if processing_status == 'fail':
    print("Please submit again!")
    exit()

# Get final workflow
try:
    # response = requests.request("GET", f"{BASE_URL}/workflow/{workflow_id}", headers=headers, timeout=100)
    response = await fetch("GET", f"{BASE_URL}/workflow/{workflow_id}", headers=headers)
    workflow_data = response.json().get('data', None)
except:
    print("Please submit again!")
    return ''

url_all = ''

for index, result_item in enumerate(workflow_data['result']):
    output_path_name = output_video_path[:-4]+'_'+str(index+1)+output_video_path[-4:]
    response = requests.request("GET", result_item['url'], headers=headers)
    url_all = url_all + result_item['url'] + ' '

    try:
        # Save the downloaded video to the specified path
        with open(output_path_name, 'wb') as file:
            file.write(response.content)
        print(f"File downloaded as {output_path_name}")
    except:
        print("Failed to download video")
        return ''

print(f"Video generated: {output_video_path}")