File size: 4,070 Bytes
51a005e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | # 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
) |