ollama_minicpm/test.py

import os
import subprocess
import sys
import argparse
import json
import time
from tqdm import tqdm
output_dir = "output_0823"

parser = argparse.ArgumentParser()
parser.add_argument("--model", type=str, default="LFM")
args = parser.parse_args()
output_dir = os.path.join(output_dir, args.model)
os.makedirs(output_dir, exist_ok=True)
VIDEO_FILE_DIR = "/mnt/data/xiuying/Code/test/videos"

# API服务器的URL
API_URL = "http://127.0.0.1:8008/video-inference/"


PROMPT = "Summarize the key observable events in this 1-minute convenience store video clip. Focus strictly on the physical actions and interactions of the people. Describe only what you can see; do not interpret intentions, relationships, or work efficiency. Avoid all repetitive descriptions of the store's layout or shelves."

files = os.listdir(VIDEO_FILE_DIR)
files.sort()
total_output = {}
cur_time = time.strftime("%Y%m%d_%H%M%S", time.localtime())
output_file_path = os.path.join(output_dir, cur_time, f"{VIDEO_FILE_DIR.split('/')[-1]}.json")
os.makedirs(os.path.join(output_dir, cur_time), exist_ok=True)
for file in tqdm(files):
    video_file_path = os.path.join(VIDEO_FILE_DIR, file)
    start_time = time.time()
    command = (
        f"curl -v -X POST '{API_URL}' "
        f"-F \"prompt={PROMPT}\" "
        f"-F \"video_path={video_file_path}\" "
        f"-F \"sampling_method=UNIFORM\" "
        f"-F \"sampling_rate=30\" "
    )

    print("将要执行以下 cURL 命令:")
    print("---------------------------------")
    print(command)
    print("---------------------------------")
    print("\n正在执行...\n")

    return_result = subprocess.check_output(command, shell=True)
    response = json.loads(return_result)
    total_output[file] = response
    end_time = time.time()
    total_output[file]["request_time"] = end_time - start_time
    with open(output_file_path, "w") as f:
        json.dump(total_output, f, indent=4)
    
print("\n\n✅ 测试脚本执行完毕。")