Livepeer Youtube Video Generator Swarm

import os
import requests

def save_to_file(content: str, file_name: str, file_path: str) -> None:
    """
    Saves the given content to a file at the specified path with the specified file name.

    :param content: The content to be written to the file.
    :param file_name: The name of the file to save the content in.
    :param file_path: The path where the file should be saved.
    """

    # ensure the directory exists
    os.makedirs(file_path, exist_ok=True)

    # construct the full file path
    full_path = os.path.join(file_path, file_name)

    # write the content to the file
    with open(full_path, "w") as file:
        file.write(content)

    print(f"File saved to {full_path}")


def read_from_file(file_path: str) -> str:
    """
    Reads the content from a file at the specified path.

    :param file_path: The path of the file to read.
    :return: The content of the file.
    """
    with open(file_path, "r") as file:
        return file.read()
    
def list_files(file_path: str) -> list[str]:
    """
    Lists the files in the specified directory.

    :param file_path: The path of the directory to list.
    :return: A list of files in the directory.
    """
    return os.listdir(file_path)



def download_from_url(url: str, file_name: str,file_path: str) -> str:

    """
    Saves the given content to a file at the specified path with the specified file name.

    :param url: Url to download the image from.
    :param file_name: Name of the file to save the image to.
    :param file_path: Path to save the image to.
    :return: Path to the saved image.
    """
    response = requests.get(url)
    if response.status_code == 200:
        # Create the full directory path
        os.makedirs(file_path, exist_ok=True)
        
        full_path = os.path.join(file_path, file_name)
        
        # Write the image content to the file
        with open(full_path, "wb") as f:
            f.write(response.content)
        
        return full_path
    else:
        raise Exception(f"Failed to download image: HTTP {response.status_code}")

from livepeer_ai import Livepeer
from dotenv import load_dotenv
import os

load_dotenv()

livepeer_api_key= os.getenv("LIVEPEER_API_KEY")

text2img_model_list = [
    'SG161222/RealVisXL_V4.0_Lightning',
    'ByteDance/SDXL-Lightning',
    'SG161222/Realistic_Vision_V6.0_B1_noVAE',
    'stabilityai/stable-diffusion-xl-base-1.0',
    'runwayml/stable-diffusion-v1-5',
    'prompthero/openjourney-v4',
    'SG161222/RealVisXL_V4.0',
    'stabilityai/sd-turbo',
    'stabilityai/sdxl-turbo',
    'stabilityai/stable-diffusion-3-medium-diffusers'
]

img2video_model_list = [
    'stable-video-diffusion-img2vid-xt',
    'stabilityai/stable-video-diffusion-img2vid-xt-1-1',   
]

img2img_model_list = [
    'timbrooks/instruct-pix2pix',
    'ByteDance/SDXL-Lightning',
    'SG161222/RealVisXL_V4.0',
    'SG161222/RealVisXL_V4.0_Lightning',
    'stabilityai/sd-turbo',
    'stabilityai/sdxl-turbo'
]

segment_model_list = [
"facebook/sam2-hiera-large"
"facebook/sam2-hiera-base-plus",
"facebook/sam2-hiera-small",
"facebook/sam2-hiera-tiny"
]

def text_to_image(prompt: str, model_id: str = text2img_model_list[1]):
    """
    This function generates an image from a text prompt using the Livepeer AI API.

    Args:
        prompt (str): The text prompt to guide image generation.
        
    Returns:
        str: The URL of the generated image.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    res = s.generate.text_to_image(request={
        "prompt": prompt,
        "model_id": model_id,
        # Optional fields
        # "loras": { "latent-consistency/lcm-lora-sdxl": 1.0, "nerijs/pixel-art-xl": 1.2 },  # Low-Rank Adaptation models and weights
        # "height": 1080,                              # Height of the generated image in pixels
        # "width": 1920,                               # Width of the generated image in pixels
        # "guidance_scale": 7.5,                       # Degree to which the model follows the prompt (higher values = closer to prompt)
        # "negative_prompt": "bad quality",            # Text prompt to exclude from image generation
        # "safety_check": True,                        # Perform a safety check to filter offensive content
        # "seed": 42,                                  # Set seed for reproducibility
        # "num_inference_steps": 50,                   # Number of denoising steps for improved quality
        # "num_images_per_prompt": 1                   # Number of images generated per prompt
    })
    
    print(res)
    
    if res.image_response is not None and res.image_response.images:
        generated_image_url = res.image_response.images[0].url
        return generated_image_url
    else:
        return None
    
def image_to_video(image_path: str, model_id: str = img2video_model_list[1]):
    """
    This function generates a video from an image using the Livepeer AI API.

    Args:
        image_path (str): The path to the image to be used for video generation.
        
    Returns:
        str: The URL of the generated video.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.image_to_video(request={
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "height": 1080,                             # Height of the generated video in pixels
        # "width": 1920,                              # Width of the generated video in pixels
        #"fps": 24,                                  # Frames per second for the generated video
        # "motion_bucket_id": 5,                      # Conditions motion amount (higher values = more motion)
        # "noise_aug_strength": 0.5,                  # Amount of noise added, reduces resemblance to original image and increases motion
        # "safety_check": True,                       # Enable safety checks to filter harmful content
        # "seed": 42,                                 # Set seed for reproducibility
        # "num_inference_steps": 50                   # Number of denoising steps for better quality
    })
    
    print(res)
    
    if res.video_response is not None and res.video_response.images:
        generated_video_url = res.video_response.images[0].url
        return generated_video_url
    else:
        return None
    
def upscale_image(prompt: str, image_path: str, model_id: str = 'stabilityai/stable-diffusion-x4-upscaler'):
    """
    This function upscales an image using the Livepeer AI API.

    Args:
        prompt (str): The text prompt to guide the upscaled image generation.
        image_path (str): The path to the image to be upscaled.
        
    Returns:
        str: The URL of the upscaled image.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.upscale(request={
        "prompt": prompt,
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "safety_check": True,   # Perform a safety check to filter offensive content
        # "seed": 42,             # Set seed for reproducible results
        # "num_inference_steps": 50,  # Number of denoising steps (higher = better quality but slower)
    })
    
    print(res)
    
    if res.image_response is not None and res.image_response.images:
        upscaled_image_url = res.image_response.images[0].url
        return upscaled_image_url
    else:
        return None
    
def image_to_image(prompt: str, image_path: str, model_id: str = img2img_model_list[0]):
    """
    This function transforms an image based on a text prompt using the Livepeer AI API.

    Args:
        prompt (str): The text prompt to guide image generation.
        image_path (str): The path to the image to be modified.
        
    Returns:
        str: The URL of the transformed image.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.image_to_image(request={
        "prompt": prompt,
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "loras": { "latent-consistency/lcm-lora-sdxl": 1.0, "nerijs/pixel-art-xl": 1.2 },  # Low-Rank Adaptation models and weights
        # "strength": 0.75,                             # Degree of transformation (0 to 1)
        # "guidance_scale": 7.5,                        # Pushes model towards text prompt (higher values = closer to text prompt)
        # "image_guidance_scale": 1.0,                  # Degree to which generated image is influenced by the original image
        # "negative_prompt": "bad quality",             # What to exclude from image generation
        # "safety_check": True,                         # Enable safety checks to filter out harmful content
        # "seed": 42,                                   # Set a seed for reproducible results
        # "num_inference_steps": 50,                    # Number of denoising steps for better quality
        # "num_images_per_prompt": 1                    # Number of images generated per prompt
    })
    
    print(res)
    
    if res.image_response is not None and res.image_response.images:
        generated_image_url = res.image_response.images[0].url
        return generated_image_url
    else:
        return None
    
def segment_anything(image_path: str, model_id: str = segment_model_list[0]):
    """
    This function segments an image using the Livepeer AI API.

    Args:
        image_path (str): The path to the image to be segmented.
        
    Returns:
        dict: The segmentation response including masks or other segmentation outputs.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.segment_anything2(request={
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "point_coords": [[100, 200], [300, 400]],  # Nx2 array for point prompts in (X, Y) pixel format
        # "point_labels": [1, 0],                   # Labels for points (1 = foreground, 0 = background)
        # "box": [50, 50, 300, 300],                # Box prompt in XYXY format
        # "mask_input": "previous_mask_data",       # Low-res mask from a previous iteration (1xHxW)
        # "multimask_output": True,                 # If true, returns multiple masks for ambiguous prompts
        # "return_logits": True,                    # If true, returns un-thresholded mask logits
        # "normalize_coords": True                  # If true, normalizes point coordinates to [0,1] range
    })
    
    print(res)
    
    if res.masks_response is not None:
        return res.masks_response.masks
    else:
        return None
    
def audio_to_text(audio_path: str, model_id: str = "openai/whisper-large-v3"):
    """
    This function transcribes an audio file using the Livepeer AI API.

    Args:
        audio_path (str): The path to the audio file to be transcribed.        
    Returns:
        str: The transcribed text from the audio file.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(audio_path, "rb") as audio_file:
        audio_content = audio_file.read()

    res = s.generate.audio_to_text(request={
        "audio": {
            "file_name": audio_path.split('/')[-1],
            "content": audio_content,
        },
        "model_id": model_id,
    })
    
    print(res)
    
    if res.text_response is not None:
        return res.text_response.text
    else:
        return None

import sys
from moviepy.editor import VideoFileClip, concatenate_videoclips, CompositeVideoClip, vfx

def video_editor(video_files: list[str], output_filename: str):
    """
    This function takes a list of video files, resizes and crops them to fit YouTube Shorts dimensions, and concatenates them into a single video file.

    :param video_files: A list of paths to video files.
    :param output_filename: The path to the output video file.
    :return: The path to the output video file.
    """

    clips = []
    for video in video_files:
        try:
            # Load the video file
            clip = VideoFileClip(video)

            # Set target dimensions for YouTube Shorts
            target_width = 1080
            target_height = 1920

            # Determine the scaling factor to cover the target dimensions
            clip_width, clip_height = clip.size
            width_ratio = target_width / clip_width
            height_ratio = target_height / clip_height
            scaling_factor = max(width_ratio, height_ratio)

            # Resize the clip to cover the target area
            clip_resized = clip.resize(height=int(clip.h * scaling_factor))

            # Center crop the clip to target dimensions
            clip_cropped = clip_resized.crop(
                x_center=clip_resized.w / 2,
                y_center=clip_resized.h / 2,
                width=target_width,
                height=target_height
            )

            # Optionally limit the clip duration to 60 seconds
            # clip_cropped = clip_cropped.subclip(0, min(clip_cropped.duration, 60))

            clips.append(clip_cropped)
        except Exception as e:
            print(f"Error processing {video}: {e}")
            continue

    if not clips:
        print("No valid video clips were processed.")
        sys.exit(1)

    # Concatenate clips
    final_clip = concatenate_videoclips(clips, method='compose')

    # Write the final video file
    final_clip.write_videofile(
        output_filename,
        fps=30,
        codec='libx264',
        audio_codec='aac',
        preset='medium',
        threads=4
    )

    return output_filename

import http.client
import httplib2
import os
import random
import sys
import time
import pickle
import logging
from typing import Optional, List

from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload

from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request

# Explicitly tell the underlying HTTP transport library not to retry, since
# we are handling retry logic ourselves.
httplib2.RETRIES = 1

# Maximum number of times to retry before giving up.
MAX_RETRIES = 10

# Always retry when these exceptions are raised.
from http.client import NotConnected, IncompleteRead, ImproperConnectionState, \
    CannotSendRequest, CannotSendHeader, ResponseNotReady, BadStatusLine

RETRIABLE_EXCEPTIONS = (httplib2.HttpLib2Error, IOError, NotConnected,
                        IncompleteRead, ImproperConnectionState,
                        CannotSendRequest, CannotSendHeader,
                        ResponseNotReady, BadStatusLine)

# Always retry when a googleapiclient.errors.HttpError with one of these status codes is raised.
RETRIABLE_STATUS_CODES = [500, 502, 503, 504]

# This OAuth 2.0 access scope allows an application to upload files to the authenticated user's YouTube channel.
YOUTUBE_UPLOAD_SCOPE = "https://www.googleapis.com/auth/youtube.upload"
YOUTUBE_API_SERVICE_NAME = "youtube"
YOUTUBE_API_VERSION = "v3"

VALID_PRIVACY_STATUSES = ("public", "private", "unlisted")


def upload_video(
    file: str,
    title: str,
    description: str,
    category: str = "22",
    keywords: Optional[str] = "",
    privacyStatus: str = "unlisted",
    client_secrets_file: str = "client_secret.json",
    credentials_file: str = "token.pickle"
) -> None:
    """
    Uploads a video file to YouTube.

    Args:
        file (str): Path to the video file to upload.
        title (str): Video title.
        description (str): Video description.
        keywords (str): Comma-separated list of video keywords.
    """
    if not os.path.exists(file):
        raise FileNotFoundError(f"Please specify a valid file. '{file}' does not exist.")

    youtube = get_authenticated_service(client_secrets_file, credentials_file)

    tags = None
    if keywords:
        tags = keywords.split(",")

    body = dict(
        snippet=dict(
            title=title,
            description=description,
            tags=tags,
            categoryId=category
        ),
        status=dict(
            privacyStatus=privacyStatus
        )
    )

    # Call the API's videos.insert method to create and upload the video.
    insert_request = youtube.videos().insert(
        part=",".join(body.keys()),
        body=body,
        media_body=MediaFileUpload(file, chunksize=-1, resumable=True)
    )

    resumable_upload(insert_request)


def get_authenticated_service(client_secrets_file: str, credentials_file: str):
    credentials = None
    # The file token.pickle stores the user's access and refresh tokens.
    if os.path.exists(credentials_file):
        with open(credentials_file, 'rb') as token:
            credentials = pickle.load(token)
    # If there are no valid credentials, let the user log in.
    if not credentials or not credentials.valid:
        if credentials and credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())
        else:
            if not os.path.exists(client_secrets_file):
                raise FileNotFoundError(f"Missing client_secrets.json file: {client_secrets_file}")
            flow = InstalledAppFlow.from_client_secrets_file(
                client_secrets_file, scopes=[YOUTUBE_UPLOAD_SCOPE])
            
            credentials = flow.run_local_server(
                    host='localhost',
                    port=8088,
                    authorization_prompt_message='Please visit this URL: {url}',
                    success_message='The auth flow is complete; you may close this window.',
                    open_browser=True)
        # Save the credentials for the next run.
        with open(credentials_file, 'wb') as token:
            pickle.dump(credentials, token)
    return build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=credentials)


def resumable_upload(insert_request):
    response = None
    error = None
    retry = 0
    while response is None:
        try:
            print("Uploading file...")
            status, response = insert_request.next_chunk()
            if response is not None:
                if 'id' in response:
                    print("Video id '%s' was successfully uploaded." %
                          response['id'])
                else:
                    exit("The upload failed with an unexpected response: %s" % response)
        except HttpError as e:
            if e.resp.status in RETRIABLE_STATUS_CODES:
                error = "A retriable HTTP error %d occurred:\n%s" % (e.resp.status,
                                                                     e.content)
            else:
                raise
        except RETRIABLE_EXCEPTIONS as e:
            error = "A retriable error occurred: %s" % e

        if error is not None:
            print(error)
            retry += 1
            if retry > MAX_RETRIES:
                exit("No longer attempting to retry.")

            max_sleep = 2 ** retry
            sleep_seconds = random.random() * max_sleep
            print("Sleeping %f seconds and then retrying..." % sleep_seconds)
            time.sleep(sleep_seconds)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

[model]
model = "gpt-4o"

[environment]
type = "dev"

[timeout]
llm = 60

[Script_Writer_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 30
instruction= """
You are a script writer on a video production team. You should write a script for a video that is engaging and informative. 
The video should be 15 seconds long.
The video should be in the style of a youtube video. 
"""

[Scene_Writer_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are a scene writer agent on a video production team. You should write a scene for a video that is engaging and informative. You should define each scene in terms of the following attributes:
- Scene Number
- Scene Heading
- Scene Description
- Visual Aids
Every scene should be 3 seconds long. You should write 5 scenes with very detailed descriptions.
"""


[Scene_Prompt_Generator_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are an expert scene prompt generator agent on a video production team. Your role is to create highly detailed and effective prompts for the image generation pipeline. When given scene scripts:

1. Analyze each scene thoroughly, focusing on visual elements, mood, and key details.
2. For each scene, create:
   - A concise Scene Prompt summarizing the overall visual concept
   - A detailed Image Prompt optimized for AI image generation:
     • Use vivid, descriptive language
     • Specify important elements like composition, lighting, colors, and style
     • Include relevant artistic references if applicable
   - A Negative Prompt to avoid unwanted elements or styles
3. Tailor your prompts to the specific requirements of AI image generation:
   - Use clear, unambiguous language
   - Prioritize visual elements over abstract concepts
   - Balance detail with room for creative interpretation
4. Format your output clearly:
   - Scene Number
   - Scene Prompt
   - Image Prompt
   - Negative Prompt

Remember, the quality and specificity of your prompts directly impact the final images. Strive for prompts that will result in visually compelling and accurate representations of each scene.
"""

[Scene_Image_Generator_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are a scene image generator agent on a video production team. You should use the prompts from the scene prompt generator agent and generate images for each scene. Collect all urls returned from text_to_image tool in a list and return the list to your manager.
"""
tools=[
    { module = "app.tools", name = "text_to_image" }
]

[Scene_Image_to_Video_Generator_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are a scene image to video generator agent on a video production team. You should use the images from the scene image generator agent and generate videos for each scene. Collect all urls returned from image_to_video tool in a list and return the list to your manager.
"""
tools=[
    { module = "app.tools", name = "image_to_video" }
]

[Video_Editor_Agent]
model = "gpt-4o"
environment="dev"
timeout = 60
instruction= """
You are a video editor agent on a video production team. You should use the videos from the video generator agent and edit them to create a final video.
"""
tools=[
    { module = "app.tools", name = "video_editor" }
]

[Youtube_Upload_Agent]
model = "gpt-4o"
environment="dev"
timeout = 60
instruction= """
You are a youtube upload agent on a video production team. You should use the videos from the video editor agent and upload them to youtube. You should also create a title, description and keywords for the video.
"""
tools=[
    { module = "app.tools", name = "upload_video" }
]