| |
|
|
| 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. |
| """ |
| |
| 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 |
|
|
| |
| payload = { |
| "inputs": { |
| "video": encoded_video, |
| "analysis_type": analysis_type |
| } |
| } |
|
|
| |
| print(f"Sending request for '{analysis_type}' analysis to endpoint: {api_url}") |
| response = requests.post(api_url, headers=headers, json=payload) |
|
|
| |
| 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...") |
|
|
| |
| 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}'") |
|
|
| |
| 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 |
| ) |
