app_no_ui.py

import asyncio
import os
import shutil
import subprocess
import tempfile
from typing import List
import logging

from moviepy import concatenate_videoclips, VideoFileClip
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai.providers.google_gla import GoogleGLAProvider
from config import api_key
import nest_asyncio
nest_asyncio.apply()

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Configure your Gemini API key
gemini_llm = GeminiModel(
    'gemini-2.0-flash', provider=GoogleGLAProvider(api_key=api_key)
)


class ChapterDescription(BaseModel):
    """Describes a chapter in the video."""
    title: str = Field(description="Title of the chapter.")
    explanation: str = Field(description="Detailed explanation of the chapter's content, including how Manim should visualize it. Be very specific with Manim instructions, including animations, shapes, positions, colors, and timing. Include LaTeX for mathematical formulas. Specify scene transitions. Example: 'Create a number line. Animate a point moving along the number line to illustrate addition. Use Transform to show the equation changing. Transition to a new Scene.'")

class VideoOutline(BaseModel):
    """Describes the outline of the video."""
    title: str = Field(description="Title of the entire video.")
    chapters: List[ChapterDescription] = Field(description="List of chapters in the video.")

class ManimCode(BaseModel):
    """Describes the Manim code for a chapter."""
    code: str = Field(description="Complete Manim code for the chapter. Include all necessary imports. The code should create a single scene. Add comments to explain the code. Do not include any comments that are not valid Python comments. Ensure the code is runnable.")

outline_agent = Agent(
    model=gemini_llm,
    result_type=VideoOutline,
    system_prompt="""
    You are a video script writer. Your job is to create a clear and concise outline for an educational video explaining a concept.
    The video should have a title and a list of chapters (maximum 3). Each chapter should have a title and a detailed explanation.
    The explanation should be very specific about how the concept should be visualized using Manim. Include detailed instructions
    for animations, shapes, positions, colors, and timing. Use LaTeX for mathematical formulas. Specify scene transitions.
    Do not include code, only explanations.
    """
)

manim_agent = Agent(
    model=gemini_llm,
    result_type=ManimCode,
    system_prompt="""
    You are a Manim code generator. Your job is to create Manim code for a single chapter of a video, given a detailed explanation of the chapter's content and how it should be visualized.
    The code should be complete and runnable. Include all necessary imports. The code should create a single scene. Add comments to explain the code.
    Do not include any comments that are not valid Python comments. Ensure the code is runnable. Do not include any text outside of the code block.

    """
)

code_fixer_agent = Agent(
    model=gemini_llm,
    result_type=ManimCode,
    system_prompt="""
    You are a Manim code debugging expert. You will receive Manim code that failed to execute and the error message.
    Your task is to analyze the code and the error, identify the issue, and provide corrected, runnable Manim code.
    Ensure the corrected code addresses the error and still aims to achieve the visualization described in the original code.
    Include all necessary imports and ensure the code creates a single scene. Add comments to explain the changes you made.
    Do not include any comments that are not valid Python comments. Ensure the code is runnable. Do not include any text outside of the code block.
    """
)

def generate_manim_code(chapter_description: ChapterDescription) -> str:
    """Generates initial Manim code for a single chapter."""
    logging.info(f"Generating Manim code for chapter: {chapter_description.title}")
    result = manim_agent.run_sync(f"title: {chapter_description.title}. Explanation: {chapter_description.explanation}")
    return result.data.code

def fix_manim_code(error: str, current_code: str) -> str:
    """Attempts to fix the Manim code that resulted in an error."""
    logging.info(f"Attempting to fix Manim code due to error: {error}")
    result = code_fixer_agent.run_sync(f"Error: {error}\nCurrent Code: {current_code}")
    return result.data.code

def generate_video_outline(concept: str) -> VideoOutline:
    """Generates the video outline."""
    logging.info(f"Generating video outline for concept: {concept}")
    result = outline_agent.run_sync(concept)
    return result.data

def create_video_from_code(code: str, chapter_number: int) -> str:
    """Creates a video from Manim code and returns the video file path using subprocess.Popen."""
    with open("temp.py", "w") as temp_file:
        temp_file.write(code)
        temp_file_name = temp_file.name

    process = None
    try:
        command = ["manim", temp_file_name, "-ql", "--disable_caching"]
        process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, text=True)
        stdout, stderr = process.communicate(timeout=60)  # Add a timeout to prevent indefinite blocking

        if process.returncode == 0:
            logging.info(f"Manim execution successful for chapter {chapter_number}.")
            logging.debug(f"Manim stdout:\n{stdout}")
            logging.debug(f"Manim stderr:\n{stderr}")
        else:
            error_msg = f"Manim execution failed for chapter {chapter_number} with return code {process.returncode}:\nStdout:\n{stdout}\nStderr:\n{stderr}"
            logging.error(str(error_msg).split('\n')[-1])
            raise subprocess.CalledProcessError(process.returncode, command, output=stdout.encode(), stderr=stderr.encode())

    except subprocess.TimeoutExpired:
        logging.error(f"Manim process timed out for chapter {chapter_number}.")
        if process:
            process.kill()
        raise
    except FileNotFoundError:
        logging.error("Error: The 'manim' command was not found. Ensure Manim is installed and in your system's PATH.")
        raise
    finally:
        pass
        # if os.path.exists(temp_file_name):
        #     os.remove(temp_file_name)

    # Construct the video file name. Manim names the file based on the class name.
    # Extract the class name from the python code.
    import re
    match = re.search(r"class\s+(\w+)\(Scene\):", code)
    if match:
        class_name = match.group(1)
        video_file_name = f"{class_name}.mp4"
        return video_file_name
    else:
        raise ValueError(f"Could not extract class name from Manim code for chapter {chapter_number}")

async def main(concept: str):
    """Generates a video explanation for a given concept using Manim with error correction."""
    logging.info(f"Generating video for concept: {concept}")
    outline = generate_video_outline(concept)
    logging.info(f"Video outline: {outline}")

    video_files = []
    for i, chapter in enumerate(outline.chapters):
        logging.info(f"Processing chapter {i + 1}: {chapter.title}")
        manim_code = generate_manim_code(chapter)
        logging.debug(f"Generated Manim code for chapter {i + 1}:\n{manim_code}")

        success = False
        attempts = 0
        max_attempts = 2  # Try fixing the code once

        while attempts < max_attempts and not success:
            try:
                video_file = create_video_from_code(manim_code, i + 1)
                video_files.append(video_file)
                logging.info(f"Video file created for chapter {i + 1}: {video_file}")
                success = True
            except subprocess.CalledProcessError as e:
                attempts += 1
                logging.error(f"Manim execution failed for chapter {i + 1} (Attempt {attempts}): {e}")
                logging.info(f"Attempting to fix the code...")
                manim_code = fix_manim_code(str(e), manim_code)
                logging.debug(f"Fixed Manim code (Attempt {attempts}):\n{manim_code}")
            except ValueError as e:
                logging.error(f"Error processing Manim code for chapter {i + 1}: {e}")
                return  # Stop processing if a critical error occurs with code structure
            except FileNotFoundError:
                logging.error("Manim not found. Please ensure it's installed and in your PATH.")
                return
            except subprocess.TimeoutExpired:
                logging.error(f"Manim process timed out for chapter {i + 1}. Attempting to fix...")
                manim_code = fix_manim_code(f"Manim process timed out.", manim_code)
                logging.debug(f"Fixed Manim code (Attempt {attempts}):\n{manim_code}")

        if not success:
            logging.error(f"Failed to generate video for chapter {i + 1} after {max_attempts} attempts. Skipping chapter.")
            continue

    # Combine the video files
    if video_files:
        logging.info("Combining video files...")
        clips = [VideoFileClip("./media/videos/temp/480p15/"+video_file) for video_file in video_files]
        final_video_path = f"final.mp4"
        final_clip = concatenate_videoclips(clips)
        final_clip.write_videofile(final_video_path, codec="libx264", audio_codec="aac")
        final_clip.close()

        logging.info(f"Final video created: {final_video_path}")

        # Clean up intermediate video files
        for video_file in video_files:
            try:
                os.remove(video_file)
                logging.info(f"Deleted intermediate video file: {video_file}")
            except Exception as e:
                logging.error(f"Error deleting intermediate video file {video_file}: {e}")
    else:
        logging.warning("No video files to combine.")

if __name__ == "__main__":
    concept = input("Enter your Prompt: ")  # Replace with your concept
    asyncio.run(main(concept))