worker/models_worker/call_model.py

import os
import sys
import base64
import argparse
from pathlib import Path
from openai import OpenAI

def encode_image(image_path: str) -> str:
    """Read a local image file and return its base64 data URI."""
    path = Path(image_path)
    if not path.is_file():
        raise FileNotFoundError(f"Image file not found: {image_path}")
    
    # Determine MIME type from file extension
    suffix = path.suffix.lower()
    mime_type = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp",
    }.get(suffix, "application/octet-stream")
    
    with open(path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")
    return f"data:{mime_type};base64,{image_data}"

def describe_image(
    image_source: str,
    model: str = "MiniMaxAI/MiniMax-M3",
    api_key: str = None,
    base_url: str = "https://api.featherless.ai/v1",
) -> str:
    """
    Send an image (URL or local path) to the Featherless AI vision API
    and return the generated description.
    """
    # Use provided API key or fall back to environment variable
    api_key = api_key or os.getenv("FEATHERLESS_API_KEY")
    if not api_key:
        raise ValueError("API key must be provided or set in FEATHERLESS_API_KEY env var")

    client = OpenAI(base_url=base_url, api_key=api_key)

    # Determine if image_source is a URL or a local file
    if image_source.startswith(("http://", "https://")):
        image_url = image_source
    else:
        # Assume it's a local file; encode to data URI
        image_url = encode_image(image_source)

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe the image in one paragraph."},
                {"type": "image_url", "image_url": {"url": image_url}},
            ],
        }
    ]

    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            # You can add more parameters, e.g., temperature, max_tokens
        )
        return response.choices[0].message.content
    except Exception as e:
        # Catch and re-raise with a user-friendly message
        raise RuntimeError(f"API call failed: {e}") from e

def main():
    parser = argparse.ArgumentParser(
        description="Describe an image using Featherless AI vision models."
    )
    parser.add_argument(
        "image",
        help="Image URL or local file path.",
    )
    parser.add_argument(
        "--model",
        default="MiniMaxAI/MiniMax-M3",
        help="Model name (default: MiniMaxAI/MiniMax-M3).",
    )
    parser.add_argument(
        "--api-key",
        help="Featherless API key (overrides FEATHERLESS_API_KEY env var).",
    )
    parser.add_argument(
        "--base-url",
        default="https://api.featherless.ai/v1",
        help="Base URL for the API (default: https://api.featherless.ai/v1).",
    )
    args = parser.parse_args()

    try:
        description = describe_image(
            image_source=args.image,
            model=args.model,
            api_key=args.api_key,
            base_url=args.base_url,
        )
        print(description)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()




"""
Bro, sif you are here then please add ahere video support and make this as a function as a whole!
"""