Initial commit

.gitignore

@@ -0,0 +1,18 @@
+# Environments
+.venv/
+.env
+
+# Python cache
+__pycache__/
+*.pyc
+
+# Gradio cache
+.gradio_cache/
+
+# Uploaded files
+/tmp/
+
+# VSCode
+.vscode/ 
+
+save.json

.gradio/flagged/dataset1.csv

@@ -0,0 +1,2 @@
+word,letter,output,timestamp
+TEstasdsdasd,,13,2025-06-02 16:48:30.771994

app.py

@@ -0,0 +1,254 @@
+import gradio as gr
+import shutil
+import os
+import subprocess
+from src.mcp.tools import letter_counter
+from src.mcp import video_tools
+from src.llm import llm
+
+def is_ffmpeg_installed():
+    """Checks if FFmpeg is installed and accessible in the system's PATH."""
+    return shutil.which("ffmpeg") is not None
+
+def update_ffmpeg_status():
+    """Creates a colored status message indicating if FFmpeg is installed."""
+    if is_ffmpeg_installed():
+        return "<p style='color:green; font-weight:bold;'>ffmpeg is installed</p>", True
+    return "<p style='color:red; font-weight:bold;'>ffmpeg is not installed</p>", False
+
+def handle_video_upload(file_obj, ffmpeg_installed):
+    """
+    Validates an uploaded MP4 file, saves it to a temporary directory,
+    and controls the visibility of UI components based on validation success.
+    """
+    if not ffmpeg_installed:
+        return None, "Cannot process video: FFmpeg is not installed.", None, gr.update(visible=False)
+    if not file_obj:
+        return None, "Please upload a file.", None, gr.update(visible=False)
+
+    tmp_dir = "tmp"
+    if not os.path.exists(tmp_dir):
+        os.makedirs(tmp_dir)
+
+    # Basic check for file extension
+    if not file_obj.name.lower().endswith('.mp4'):
+        return None, "File must be an .mp4 file.", None, gr.update(visible=False)
+
+    # More robust check using ffprobe
+    try:
+        subprocess.run(
+            ["ffprobe", "-v", "error", "-show_format", "-show_streams", file_obj.name],
+            capture_output=True, text=True, check=True
+        )
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return None, "Could not validate file. Ensure it's a valid MP4.", None, gr.update(visible=False)
+
+    file_path = os.path.join(tmp_dir, os.path.basename(file_obj.name))
+    shutil.copy(file_obj.name, file_path)
+    
+    return file_path, "File uploaded successfully!", file_path, gr.update(visible=True)
+
+def clear_previous_outputs():
+    """Clears all video-related outputs to ensure a clean state for new uploads."""
+    return (
+        None, # video_output
+        "",   # upload_status_text
+        None, # uploaded_video_path_state
+        gr.update(visible=False), # video_tools_group
+        None, # first_frame_img
+        None, # last_frame_img
+        None, # llm_first_frame_img
+        None, # llm_last_frame_img
+        None, # llm_gif_output
+        "",   # llm_result_text
+        "",   # raw_prompt_text
+        ""    # raw_response_text
+    )
+
+def on_load_tasks():
+    """Runs on app startup to load settings and check FFmpeg status."""
+    url, model, _ = llm.load_settings()
+    ffmpeg_status_text, ffmpeg_installed = update_ffmpeg_status()
+    # The return order must match the outputs in demo.load
+    return url, model, ffmpeg_status_text, ffmpeg_installed, url # Pass url to both state and input box
+
+# --- Gradio UI ---
+with gr.Blocks() as demo:
+    # --- State Variables ---
+    ffmpeg_installed_state = gr.State(False)
+    preferred_llm_state = gr.State("")
+    ollama_url_state = gr.State("")
+    uploaded_video_path_state = gr.State("")
+
+    with gr.Tabs():
+        # --- Setup & Video Upload Tab ---
+        with gr.Tab("Setup & Video Upload"):
+            gr.Markdown("## System Status")
+            with gr.Row():
+                check_ffmpeg_btn = gr.Button("Check FFmpeg Status")
+                status_text = gr.Markdown("Status will be checked on load.")
+
+            gr.Markdown("---")
+            gr.Markdown("## Video Tools")
+            with gr.Row():
+                with gr.Column(scale=1):
+                    gr.Markdown("### Upload Video")
+                    file_input = gr.File(label="Upload MP4", file_types=[".mp4"])
+                    video_output = gr.Video(label="Preview", interactive=False)
+                    upload_status_text = gr.Textbox(label="Upload Status", interactive=False)
+                
+                with gr.Column(scale=2, visible=False) as video_tools_group:
+                    gr.Markdown("### Manual Frame Extraction")
+                    with gr.Row():
+                        get_first_frame_btn = gr.Button("Get First Frame")
+                        get_last_frame_btn = gr.Button("Get Last Frame")
+                    with gr.Row():
+                        first_frame_img = gr.Image(label="First Frame", type="filepath", interactive=False)
+                        last_frame_img = gr.Image(label="Last Frame", type="filepath", interactive=False)
+
+        # --- LLM Configuration Tab ---
+        with gr.Tab("LLM Configuration"):
+            gr.Markdown("# Ollama Endpoint Configuration")
+            with gr.Row():
+                llm_endpoint_input = gr.Textbox(
+                    label="Ollama Endpoint URL",
+                    placeholder="e.g., http://localhost:11434"
+                )
+                check_llm_button = gr.Button("Check Endpoint")
+            llm_status_text = gr.Textbox(label="Endpoint Status", interactive=False)
+            
+            with gr.Column(visible=False) as llm_selection_group:
+                with gr.Row():
+                    llm_model_select = gr.Dropdown(label="Available Models")
+                    set_llm_button = gr.Button("Set as Preferred")
+                preferred_llm_display = gr.Textbox(label="Current Preferred Model", interactive=False)
+
+        # --- MCP Integration Tab ---
+        with gr.Tab("MCP Integration"):
+            gr.Markdown("## Test MCP Tool Calls with a Local LLM")
+            with gr.Row():
+                with gr.Column(scale=1):
+                    gr.Markdown("### Video Command Center")
+                    video_prompt_input = gr.Textbox(label="Video Command Prompt", placeholder="e.g., 'Get me the first frame of the video'")
+                    run_video_prompt_btn = gr.Button("Run Command")
+                    
+                    llm_result_text = gr.Textbox(label="LLM Result", interactive=False)
+
+                    with gr.Accordion("LLM Debug Info", open=False):
+                        raw_prompt_text = gr.Textbox(label="Raw Prompt to LLM", interactive=False, lines=15)
+                        raw_response_text = gr.Textbox(label="Raw Response from LLM", interactive=False, lines=15)
+
+                with gr.Column(scale=1):
+                    gr.Markdown("### Frame Results (from LLM)")
+                    with gr.Row():
+                        llm_first_frame_img = gr.Image(label="First Frame", type="filepath", interactive=False)
+                        llm_last_frame_img = gr.Image(label="Last Frame", type="filepath", interactive=False)
+                    llm_gif_output = gr.Image(label="Generated GIF", type="filepath", interactive=False)
+
+    # --- Event Handlers ---
+
+    # --- System Events ---
+    check_ffmpeg_btn.click(
+        fn=update_ffmpeg_status,
+        outputs=[status_text, ffmpeg_installed_state]
+    )
+
+    file_input.upload(
+        fn=clear_previous_outputs,
+        outputs=[
+            video_output,
+            upload_status_text,
+            uploaded_video_path_state,
+            video_tools_group,
+            first_frame_img,
+            last_frame_img,
+            llm_first_frame_img,
+            llm_last_frame_img,
+            llm_gif_output,
+            llm_result_text,
+            raw_prompt_text,
+            raw_response_text
+        ]
+    ).then(
+        fn=handle_video_upload,
+        inputs=[file_input, ffmpeg_installed_state],
+        outputs=[video_output, upload_status_text, uploaded_video_path_state, video_tools_group]
+    )
+
+    get_first_frame_btn.click(
+        fn=video_tools.getFirstFrame,
+        inputs=uploaded_video_path_state,
+        outputs=first_frame_img
+    )
+
+    get_last_frame_btn.click(
+        fn=video_tools.getLastFrame,
+        inputs=uploaded_video_path_state,
+        outputs=last_frame_img
+    )
+
+    # --- LLM Events ---
+    check_llm_button.click(
+        fn=llm.check_ollama_endpoint,
+        inputs=[llm_endpoint_input, preferred_llm_state],
+        outputs=[llm_status_text, llm_model_select, llm_selection_group, ollama_url_state]
+    )
+
+    llm_model_select.change(
+        fn=llm.set_preferred_model,
+        inputs=[llm_model_select, ollama_url_state],
+        outputs=[preferred_llm_state, preferred_llm_display]
+    )
+    
+    run_video_prompt_btn.click(
+        fn=llm.process_video_prompt,
+        inputs=[
+            video_prompt_input, 
+            uploaded_video_path_state, 
+            ollama_url_state, 
+            preferred_llm_state,
+            llm_first_frame_img, # Pass the component itself
+            llm_last_frame_img,   # Pass the component itself
+            llm_gif_output
+        ],
+        outputs=[
+            llm_first_frame_img,
+            llm_last_frame_img,
+            llm_gif_output,
+            llm_result_text,
+            raw_prompt_text,
+            raw_response_text
+        ]
+    )
+
+    # --- App Load Event ---
+    demo.load(
+        fn=on_load_tasks,
+        outputs=[
+            ollama_url_state, 
+            preferred_llm_state, 
+            status_text, 
+            ffmpeg_installed_state,
+            llm_endpoint_input
+        ]
+    ).then(
+        fn=llm.check_on_load,
+        inputs=[ollama_url_state, preferred_llm_state],
+        outputs=[llm_status_text, llm_model_select, llm_selection_group, ollama_url_state, preferred_llm_display]
+    )
+
+# --- MCP Server Launch ---
+if __name__ == "__main__":
+    # Expose the functions we want to be available as tools
+    demo.add_tools(
+        {
+            "letter_counter": letter_counter,
+            "getFirstFrame": video_tools.getFirstFrame,
+            "getLastFrame": video_tools.getLastFrame,
+            "convert_mp4_to_gif": video_tools.convert_mp4_to_gif,
+        }
+    )
+    # Launch the Gradio app with MCP server enabled
+    demo.launch(mcp_server=True)
+
+

packages.txt

@@ -0,0 +1 @@
+ffmpeg

requirements.txt

@@ -0,0 +1,2 @@
+gradio
+requests

src/__init__.py

@@ -0,0 +1 @@
+# This file makes the 'src' directory a Python package. 

src/llm/__init__.py

@@ -0,0 +1 @@
+# This file makes the 'llm' directory a Python package. 

src/llm/llm.py

@@ -0,0 +1,368 @@
+import os
+import json
+import requests
+import gradio as gr
+from ..mcp import video_tools
+
+SAVE_FILE = "save.json"
+
+def save_settings(url=None, model_list=None, preferred_model=None):
+    """Saves endpoint-specific settings to the save file."""
+    settings = {}
+    if os.path.exists(SAVE_FILE):
+        try:
+            with open(SAVE_FILE, 'r') as f:
+                settings = json.load(f)
+        except (json.JSONDecodeError, IOError):
+            settings = {} # Start with fresh settings if file is corrupt
+
+    # Ensure the settings object has the correct structure
+    if 'endpoints' not in settings or not isinstance(settings['endpoints'], list):
+        settings['endpoints'] = []
+    if 'last_active_url' not in settings:
+        settings['last_active_url'] = ""
+
+    if url:
+        # Normalize the URL before saving or looking up
+        if not (url.startswith("http://") or url.startswith("https://")):
+            url = "http://" + url
+
+        settings['last_active_url'] = url
+        endpoint_data = next((e for e in settings['endpoints'] if e['url'] == url), None)
+        if not endpoint_data:
+            endpoint_data = {'url': url, 'models': [], 'preferred_model': ''}
+            settings['endpoints'].append(endpoint_data)
+
+        if model_list is not None:
+            endpoint_data['models'] = model_list
+        if preferred_model is not None:
+            endpoint_data['preferred_model'] = preferred_model
+            
+    with open(SAVE_FILE, 'w') as f:
+        json.dump(settings, f, indent=4)
+
+def load_settings():
+    """Loads settings for the last active endpoint."""
+    if not os.path.exists(SAVE_FILE):
+        return "", "", ""
+    try:
+        with open(SAVE_FILE, "r") as f:
+            data = json.load(f)
+            last_url = data.get("last_active_url", "")
+            if not last_url:
+                return "", "", ""
+            
+            endpoint_data = next((e for e in data['endpoints'] if e['url'] == last_url), None)
+            if not endpoint_data:
+                return last_url, "", ""
+
+            model = endpoint_data.get("preferred_model", "")
+            model_display_text = f"Preferred model set to: {model}" if model else ""
+            return last_url, model, model_display_text
+    except (json.JSONDecodeError, IOError):
+        return "", "", ""
+
+def parse_param_size(size_str: str) -> float:
+    """Converts a model size string (e.g., '7B') to a float."""
+    if not isinstance(size_str, str):
+        return 0.0
+    size_str = size_str.upper().strip()
+    if size_str.endswith('B'):
+        try:
+            return float(size_str[:-1])
+        except (ValueError, TypeError):
+            return 0.0
+    return 0.0
+
+def get_model_details(url: str, model_name: str) -> dict:
+    """Inspects a model's file for details like tool use and parameter size."""
+    details = {
+        'name': model_name,
+        'supports_tool_calling': False,
+        'parameter_size': 0.0,
+        'parameter_size_str': "N/A"
+    }
+    try:
+        res = requests.post(
+            f"{url.rstrip('/')}/api/show",
+            json={"name": model_name},
+            timeout=10
+        )
+        res.raise_for_status()
+        data = res.json()
+        modelfile = data.get("modelfile", "")
+        
+        keywords = ["tool", "function", "available_tools", "function_call"]
+        details['supports_tool_calling'] = any(keyword in modelfile.lower() for keyword in keywords)
+        
+        param_size_str = data.get("details", {}).get("parameter_size", "0B")
+        details['parameter_size_str'] = param_size_str
+        details['parameter_size'] = parse_param_size(param_size_str)
+        
+        return details
+    except (requests.exceptions.RequestException, json.JSONDecodeError):
+        return details
+
+def check_ollama_endpoint(url, preferred_model=None):
+    """
+    Checks an Ollama endpoint, gets a list of available models with their capabilities,
+    sorts them, and updates the UI accordingly.
+    """
+    if not url or not url.strip():
+        return "Please enter a URL.", gr.update(visible=False), gr.update(visible=False), url
+
+    if not (url.startswith("http://") or url.startswith("https://")):
+        url = "http://" + url
+    
+    api_url = f"{url.rstrip('/')}/api/tags"
+
+    try:
+        response = requests.get(api_url, timeout=5)
+        response.raise_for_status()
+        
+        models_data = response.json().get("models", [])
+        if not models_data:
+            return "Connected, but no models found.", gr.update(visible=False), gr.update(visible=False), url
+
+        detailed_models = [get_model_details(url, m['name']) for m in models_data]
+
+        # Sort models: 1. Tool support (desc), 2. Param size (desc), 3. Name (asc)
+        detailed_models.sort(key=lambda m: (not m['supports_tool_calling'], -m['parameter_size'], m['name']))
+
+        save_settings(url=url, model_list=detailed_models)
+        
+        dropdown_choices = []
+        for m in detailed_models:
+            tool_text = "Tools: Yes" if m['supports_tool_calling'] else "Tools: No"
+            name_display = f"🛠️ {m['name']}" if m['supports_tool_calling'] else m['name']
+            dropdown_choices.append(f"{name_display} ({tool_text}, {m['parameter_size_str']})")
+
+        status_message = f"Success! Found and sorted {len(detailed_models)} models."
+        
+        default_choice = dropdown_choices[0] if dropdown_choices else None
+        if preferred_model:
+            matching_choice = next((choice for choice in dropdown_choices if preferred_model in choice), None)
+            if matching_choice:
+                default_choice = matching_choice
+
+        return status_message, gr.update(choices=dropdown_choices, value=default_choice, visible=True), gr.update(visible=True), url
+
+    except requests.exceptions.RequestException:
+        error_message = "Connection Error: Is the address correct and Ollama running?"
+        return error_message, gr.update(visible=False), gr.update(visible=False), url
+
+def set_preferred_model(model_selection, current_url):
+    """Stores the selected model for the current endpoint and saves it."""
+    # Parse model name from variants like "🛠️ model-name (Tools: Yes, 7B)"
+    if not model_selection:
+        return "", "No model selected."
+    model_name = model_selection.split(" (")[0].strip().replace("🛠️ ", "")
+    save_settings(url=current_url, preferred_model=model_name)
+    return model_name, f"Preferred model for {current_url} set to: {model_name}"
+
+def check_on_load(url, preferred_model):
+    """
+    Wrapper to trigger endpoint check on load if a URL exists,
+    otherwise sets a neutral status. Also sets the dropdown to the preferred model.
+    """
+    if not url or not url.strip():
+        # The empty string for preferred_llm_display ensures it's cleared on a failed load
+        return "Enter an endpoint URL and click 'Check' to begin.", gr.update(visible=False), gr.update(visible=False), None, ""
+    
+    # Unpack the results from the main checking function, now passing the preferred model
+    status, dropdown_update, button_update, current_url = check_ollama_endpoint(url, preferred_model)
+    
+    # Construct the display text for the textbox.
+    model_display_text = f"Preferred model set to: {preferred_model}" if preferred_model else ""
+
+    # The last value returned now correctly populates the preferred_llm_display textbox.
+    return status, dropdown_update, button_update, current_url, model_display_text
+
+def process_video_prompt(
+    prompt: str, 
+    video_path: str, 
+    ollama_url: str, 
+    model_name: str, 
+    first_frame_component: gr.Image,
+    last_frame_component: gr.Image,
+    gif_component: gr.Image
+):
+    """
+    Sends a prompt to an Ollama model with video tools and executes the response.
+    Returns updates for the Gradio UI components.
+    """
+    if not all([prompt, video_path, ollama_url, model_name]):
+        error_msg = "Error: Missing prompt, video path, or LLM configuration."
+        return first_frame_component, last_frame_component, gif_component, None, error_msg, "N/A", "N/A"
+
+    api_url = f"{ollama_url.rstrip('/')}/api/chat" # Use the /api/chat endpoint for tool calling
+
+    # Define the tools available to the LLM based on our functions
+    tools = [
+        {
+            "type": "function",
+            "function": {
+                "name": "getFirstFrame",
+                "description": "Extracts the very first frame from a video file.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "video_path": {
+                            "type": "string",
+                            "description": "The path to the video file to process."
+                        }
+                    },
+                    "required": ["video_path"]
+                }
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "getLastFrame",
+                "description": "Extracts the very last frame from a video file.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "video_path": {
+                            "type": "string",
+                            "description": "The path to the video file to process."
+                        }
+                    },
+                    "required": ["video_path"]
+                }
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "convert_mp4_to_gif",
+                "description": "Converts a full MP4 video into a high-quality animated GIF.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "video_path": {
+                            "type": "string",
+                            "description": "The path to the video file to convert."
+                        },
+                        "maxResolution": {
+                            "type": "integer",
+                            "description": "Optional. The maximum dimension (width or height) for the GIF. Must be between 16 and 500. Defaults to 500.",
+                            "default": 500
+                        },
+                        "fps": {
+                            "type": "integer",
+                            "description": "Optional. Frames per second for the GIF. Must be between 14 and 50 due to browser limitations. Defaults to 15.",
+                            "default": 15
+                        },
+                        "pingpong": {
+                            "type": "boolean",
+                            "description": "Optional. If true, the GIF will play forwards then backwards. Defaults to false.",
+                            "default": False
+                        }
+                    },
+                    "required": ["video_path"]
+                }
+            }
+        }
+    ]
+
+    # Map tool names to actual Python functions
+    available_tools = {
+        "getFirstFrame": video_tools.getFirstFrame,
+        "getLastFrame": video_tools.getLastFrame,
+        "convert_mp4_to_gif": video_tools.convert_mp4_to_gif,
+    }
+
+    # For /api/chat, we provide a user message. The model should infer the context.
+    prompt_with_context = f"{prompt}\n\nThe video for this prompt is located at the following path: '{video_path}'"
+    
+    raw_response_str = ""
+    try:
+        # Request to the LLM with the messages and tools
+        response = requests.post(
+            api_url,
+            json={
+                "model": model_name,
+                "messages": [{"role": "user", "content": prompt_with_context}],
+                "stream": False,
+                "tools": tools,
+            },
+            timeout=60,
+        )
+        response.raise_for_status()
+        response_data = response.json()
+        raw_response_str = json.dumps(response_data, indent=2)
+        
+        # Check if the model decided to use a tool, using the /api/chat response structure.
+        message = response_data.get("message", {})
+        tool_calls = message.get("tool_calls")
+
+        if not tool_calls:
+            status = message.get("content", "The model did not call a tool.")
+            return first_frame_component, last_frame_component, gif_component, status, prompt_with_context, raw_response_str
+
+        # --- Loop through all tool calls from the LLM ---
+        first_frame_update = first_frame_component
+        last_frame_update = last_frame_component
+        gif_update = gif_component # Start with the original component state
+        execution_statuses = []
+
+        for tool_call in tool_calls:
+            function_call = tool_call.get("function", {})
+            tool_name_from_llm = function_call.get("name")
+            
+            if not tool_name_from_llm:
+                execution_statuses.append("Error: Could not parse tool name from a tool call.")
+                continue
+            
+            # Make the check case-insensitive
+            tool_name_lower = tool_name_from_llm.lower()
+            available_tools_lower = {k.lower(): v for k, v in available_tools.items()}
+
+            if tool_name_lower in available_tools_lower:
+                original_tool_name = list(available_tools.keys())[list(available_tools_lower.keys()).index(tool_name_lower)]
+                tool_function = available_tools[original_tool_name]
+                
+                arguments = function_call.get("arguments", {})
+                
+                # Prepare arguments for the tool function
+                tool_args = {}
+                if "video_path" in arguments:
+                    tool_args["video_path"] = arguments["video_path"]
+                else:
+                    # Fallback for safety, though path is required in schema
+                    tool_args["video_path"] = video_path 
+                
+                if original_tool_name == "convert_mp4_to_gif":
+                    if "maxResolution" in arguments:
+                        tool_args["maxResolution"] = arguments["maxResolution"]
+                    if "fps" in arguments:
+                        tool_args["fps"] = arguments["fps"]
+                    if "pingpong" in arguments:
+                        tool_args["pingpong"] = arguments["pingpong"]
+
+                result_path = tool_function(**tool_args)
+
+                if "Error:" in result_path:
+                    execution_statuses.append(f"Tool '{original_tool_name}' failed: {result_path}")
+                else:
+                    execution_statuses.append(f"Successfully executed tool: {original_tool_name}")
+                    if original_tool_name == "getFirstFrame":
+                        first_frame_update = gr.update(value=result_path)
+                    elif original_tool_name == "getLastFrame":
+                        last_frame_update = gr.update(value=result_path)
+                    elif original_tool_name == "convert_mp4_to_gif":
+                        gif_update = gr.update(value=result_path)
+            else:
+                execution_statuses.append(f"Error: The model tried to call an unknown tool: {tool_name_from_llm}")
+
+        final_status = "\n".join(execution_statuses)
+        return first_frame_update, last_frame_update, gif_update, final_status, prompt_with_context, raw_response_str
+
+    except requests.exceptions.RequestException as e:
+        error_msg = f"API Error: Could not connect to Ollama. {e}"
+        return first_frame_component, last_frame_component, gif_component, error_msg, prompt_with_context, getattr(e.response, 'text', 'No response text')
+    except (KeyError, IndexError, json.JSONDecodeError) as e:
+        error_msg = f"Error processing LLM response: {e}"
+        return first_frame_component, last_frame_component, gif_component, error_msg, prompt_with_context, raw_response_str

src/mcp/__init__.py

@@ -0,0 +1 @@
+# This file makes the 'mcp' directory a Python package. 

src/mcp/tools.py

@@ -0,0 +1,11 @@
+def letter_counter(word, letter):
+    """Count the occurrences of a specific letter in a word.
+    
+    Args:
+        word: The word or phrase to analyze
+        letter: The letter to count occurrences of
+        
+    Returns:
+        The number of times the letter appears in the word
+    """
+    return word.lower().count(letter.lower()) 

src/mcp/video_tools.py

@@ -0,0 +1,174 @@
+import os
+import subprocess
+
+def getFirstFrame(video_path: str) -> str:
+    """
+    Extracts the first frame from an MP4 video file using FFmpeg.
+
+    Args:
+        video_path: The relative path to the MP4 file (e.g., 'tmp/my_video.mp4').
+
+    Returns:
+        The path to the extracted JPEG image frame, or an error message string.
+    """
+    if not video_path or not os.path.exists(video_path):
+        return "Error: Video file not found at the specified path."
+
+    # Ensure the output directory exists
+    output_dir = os.path.dirname(video_path)
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
+
+    # Define the output path for the frame
+    base_name = os.path.basename(video_path)
+    file_name_without_ext = os.path.splitext(base_name)[0]
+    output_frame_path = os.path.join(output_dir, f"{file_name_without_ext}_frame.jpg")
+
+    # Construct and run the ffmpeg command
+    command = [
+        "ffmpeg",
+        "-i", video_path,      # Input file
+        "-vframes", "1",       # Extract only one frame
+        "-q:v", "2",           # Output quality (2 is high)
+        "-y",                  # Overwrite output file if it exists
+        output_frame_path
+    ]
+
+    try:
+        # Use subprocess.run to execute the command
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            text=True,
+            check=True  # This will raise a CalledProcessError if ffmpeg returns a non-zero exit code
+        )
+        return output_frame_path
+    except FileNotFoundError:
+        return "Error: ffmpeg is not installed or not found in the system's PATH."
+    except subprocess.CalledProcessError as e:
+        # Provide the stderr from ffmpeg for easier debugging
+        return f"Error during frame extraction: {e.stderr}"
+
+def getLastFrame(video_path: str) -> str:
+    """
+    Extracts the last frame from an MP4 video file using FFmpeg.
+
+    Args:
+        video_path: The relative path to the MP4 file (e.g., 'tmp/my_video.mp4').
+
+    Returns:
+        The path to the extracted JPEG image frame, or an error message string.
+    """
+    if not video_path or not os.path.exists(video_path):
+        return "Error: Video file not found at the specified path."
+
+    # Ensure the output directory exists
+    output_dir = os.path.dirname(video_path)
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
+
+    # Define the output path for the frame
+    base_name = os.path.basename(video_path)
+    file_name_without_ext = os.path.splitext(base_name)[0]
+    output_frame_path = os.path.join(output_dir, f"{file_name_without_ext}_last_frame.jpg")
+
+    # Construct and run the ffmpeg command
+    # -sseof -1 seeks to 1 second before the end of the file to grab the last frame.
+    command = [
+        "ffmpeg",
+        "-sseof", "-1",        # Seek to 1s before the end.
+        "-i", video_path,      # Input file
+        "-vframes", "1",       # Extract only one frame
+        "-q:v", "2",           # Output quality (2 is high)
+        "-y",                  # Overwrite output file if it exists
+        output_frame_path
+    ]
+
+    try:
+        # Use subprocess.run to execute the command
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        return output_frame_path
+    except FileNotFoundError:
+        return "Error: ffmpeg is not installed or not found in the system's PATH."
+    except subprocess.CalledProcessError as e:
+        # Provide the stderr from ffmpeg for easier debugging
+        return f"Error: Could not extract last frame. {e}"
+
+def convert_mp4_to_gif(video_path: str, maxResolution: int = 500, fps: int = 15, pingpong: bool = False) -> str:
+    """
+    Converts a video file to a high-quality animated GIF using a two-pass ffmpeg method.
+
+    Args:
+        video_path: The path to the input video file.
+        maxResolution: The maximum dimension (width or height) of the output GIF.
+                       Value must be between 16 and 500. Defaults to 500.
+        fps: Frames per second for the output GIF. Must be between 14 and 50.
+        pingpong: If True, the GIF will play forwards and then reverse.
+
+    Returns:
+        The path to the generated GIF file, or an error string.
+    """
+    try:
+        maxResolution = int(maxResolution)
+        fps = int(fps)
+        # Handle boolean from LLM which might be a string
+        if isinstance(pingpong, str):
+            pingpong = pingpong.lower() in ['true', '1', 't', 'y', 'yes']
+    except (ValueError, TypeError):
+        return f"Error: maxResolution and fps must be valid integers."
+
+    if not 16 <= maxResolution <= 500:
+        return f"Error: maxResolution must be between 16 and 500, but was {maxResolution}."
+    if not 14 <= fps <= 50:
+        return f"Error: fps must be between 14 and 50, but was {fps}."
+        
+    if not os.path.exists(video_path):
+        return f"Error: Video file not found at {video_path}"
+
+    base_path, _ = os.path.splitext(video_path)
+    output_gif_path = f"{base_path}.gif"
+    palette_path = os.path.join(os.path.dirname(video_path), "palette.png")
+
+    try:
+        scaling_filter = f"scale={maxResolution}:{maxResolution}:force_original_aspect_ratio=decrease:flags=lanczos"
+        
+        if pingpong:
+            vf_options = f"[0:v]split[original][copy];[copy]reverse[reversed];[original][reversed]concat=n=2:v=1:a=0,fps={fps},{scaling_filter}"
+        else:
+            vf_options = f"fps={fps},{scaling_filter}"
+
+        # Pass 1: Generate the color palette
+        palettegen_cmd = [
+            "ffmpeg",
+            "-i", video_path,
+            "-vf", f"{vf_options},palettegen",
+            "-y",
+            palette_path
+        ]
+        subprocess.run(palettegen_cmd, check=True, capture_output=True, text=True)
+
+        # Pass 2: Use the palette to create the GIF
+        gif_conversion_cmd = [
+            "ffmpeg",
+            "-i", video_path,
+            "-i", palette_path,
+            "-lavfi", f"{vf_options} [x]; [x][1:v] paletteuse",
+            "-y",
+            output_gif_path
+        ]
+        subprocess.run(gif_conversion_cmd, check=True, capture_output=True, text=True)
+
+        return output_gif_path
+
+    except subprocess.CalledProcessError as e:
+        # Provide more specific error from ffmpeg's stderr
+        return f"Error during GIF conversion: {e.stderr}"
+    finally:
+        # Clean up the temporary palette file
+        if os.path.exists(palette_path):
+            os.remove(palette_path)