call_endpoint.py

# File: call_endpoint.py

import requests
import base64
import argparse
import os
import json
import pandas as pd


def call_emotion_endpoint(video_path: str, output_prefix: str, analysis_type: str, api_url: str, api_token: str):
    """
    Calls the multimodal emotion analysis endpoint.

    Args:
        video_path (str): Path to the input video file.
        output_prefix (str): Prefix for saving output files (e.g., 'my_analysis').
        analysis_type (str): The type of analysis to perform ('audio', 'facial', 'text').
        api_url (str): The URL of the inference endpoint.
        api_token (str): Your Hugging Face API token.
    """
    # 1. Prepare headers and read/encode video
    headers = {"Authorization": f"Bearer {api_token}", "Content-Type": "application/json"}
    try:
        with open(video_path, "rb") as f:
            video_bytes = f.read()
        encoded_video = base64.b64encode(video_bytes).decode("utf-8")
        print(f"Successfully read and encoded '{video_path}'")
    except FileNotFoundError:
        print(f"Error: Input file not found at '{video_path}'")
        return

    # 2. Construct the JSON payload
    payload = {
        "inputs": {
            "video": encoded_video,
            "analysis_type": analysis_type
        }
    }

    # 3. Make the POST request
    print(f"Sending request for '{analysis_type}' analysis to endpoint: {api_url}")
    response = requests.post(api_url, headers=headers, json=payload)

    # 4. Process the response
    if response.status_code == 200:
        try:
            response_data = response.json()

            if response_data.get("status") == "error":
                print(f"Endpoint returned an error: {response_data.get('message')}")
                return

            print("Success! Processing response...")

            # Save any DataFrame results to CSV/XLSX
            for key, value in response_data.items():
                if key.endswith("_data") and value:
                    df = pd.read_json(value, orient='split')
                    output_path = f"{output_prefix}_{key}.xlsx"
                    df.to_excel(output_path, index=False)
                    print(f"Saved DataFrame to '{output_path}'")

            # Save any base64 encoded files
            if "processed_video" in response_data:
                video_b64 = response_data["processed_video"]
                decoded_video_bytes = base64.b64decode(video_b64)
                output_path = f"{output_prefix}_processed_video.mp4"
                with open(output_path, "wb") as f:
                    f.write(decoded_video_bytes)
                print(f"Saved processed video to '{output_path}'")

        except (requests.exceptions.JSONDecodeError, KeyError, TypeError) as e:
            print(f"Error processing the response: {e}")
            print("Response content:", response.text)
    else:
        print(f"Error: Endpoint returned status code {response.status_code}")
        print("Response content:", response.text)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Call the multimodal emotion analysis endpoint.")
    parser.add_argument("video_path", type=str, help="Path to the input video file.")
    parser.add_argument("analysis_type", type=str, choices=['audio', 'facial', 'text'], help="Type of analysis to run.")
    parser.add_argument("--output_prefix", type=str, default="analysis_result", help="Prefix for output files.")
    parser.add_argument("--api_url", type=str, required=True, help="The URL of the inference endpoint.")
    parser.add_argument("--api_token", type=str, default=os.environ.get("HF_API_TOKEN"),
                        help="Your HF API token (or set HF_API_TOKEN env var).")

    args = parser.parse_args()

    if not args.api_token:
        raise ValueError("Hugging Face API token is required.")

    call_emotion_endpoint(
        video_path=args.video_path,
        output_prefix=args.output_prefix,
        analysis_type=args.analysis_type,
        api_url=args.api_url,
        api_token=args.api_token
    )