replay_event_test.py 3.13 KB
import json
import os

from football_replay_video_event_by_llm import FootballReplayVideoEvent
from qwen_asr_util import QwenAsr
def batch_with_asr():
    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
    qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B",temperature=0.7,
                           api_key='no_key')

    videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
    for video_name in sorted(os.listdir(videos_dir)):
        replays_dir = os.path.join(videos_dir, video_name, 'replays')
        replay_videos_dir = os.path.join(replays_dir, 'videos')
        replay_goals_dir = os.path.join(replays_dir, 'goals')
        replay_asr_dir = os.path.join(replays_dir, 'asr')

        for replay_video_name in sorted(os.listdir(replay_videos_dir)):
            replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
            replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
            replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.json")
            if os.path.exists(replay_goals_path):
                print("skip", replay_video_path)
                continue
            asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
            event_json = fbrv.video_event(replay_video_name, asr_text)
            print(event_json)

def batch_without_asr():
    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
    # qwen_asr = QwenAsr(base_url="http://192.168.1.59:8101/v1", model="Qwen/Qwen3-ASR-1.7B", temperature=0.7,
    #                    api_key='no_key')

    videos_dir = rf"R:\wdx\202604\20260420_download_football_video\finished"
    for video_name in sorted(os.listdir(videos_dir)):
        replays_dir = os.path.join(videos_dir, video_name, 'replays')
        replay_videos_dir = os.path.join(replays_dir, 'videos')
        replay_goals_dir = os.path.join(replays_dir, 'goals')
        replay_asr_dir = os.path.join(replays_dir, 'asr')

        for replay_video_name in sorted(os.listdir(replay_videos_dir)):
            replay_video_path = os.path.join(replay_videos_dir, replay_video_name)
            replay_goals_path = os.path.join(replay_goals_dir, f"{replay_video_name}.json")
            replay_asr_path = os.path.join(replay_asr_dir, f"{replay_video_name}.json")
            if os.path.exists(replay_goals_path):
                print("skip", replay_video_path)
                continue
            # asr_text = qwen_asr.asr(replay_video_path, cache_path=replay_asr_path)
            event_json = fbrv.video_event(replay_video_path, None, cache_path=replay_goals_path)
            print(replay_video_name)
            print(event_json)

def one_video_test(video_path):
    fbrv = FootballReplayVideoEvent(base_url="http://192.168.1.59:11434", model="qwen3.6:35b-a3b-q8_0", temperature=0.7)
    event_json = fbrv.video_event(video_path, None)
    print(event_json)

if __name__ == '__main__':
    # one_video_test(rf"D:\Code\py260417\src\llm_demo\lc_t\ftb\replay_1.mp4")
    batch_without_asr()