# Copyright 2025 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import numpy as np
import os
import re
from datetime import datetime
from dataclasses import dataclass, field
from typing import Optional
from datasets import load_dataset, load_from_disk, Dataset, DatasetDict
from transformers import Qwen2VLForConditionalGeneration
from math_verify import parse, verify
from src.open_r1.trainer import Qwen2VLGRPOTrainer_Video as Qwen2VLGRPOTrainer
from src.open_r1.trainer import Qwen2VLGRPOVLLMTrainer_Video as Qwen2VLGRPOVLLMTrainer
from trl import GRPOConfig, GRPOTrainer, ModelConfig, ScriptArguments, TrlParser, get_peft_config
from src.open_r1.trainer import Qwen2VLGRPOTrainer_Video_GT_Soft as Qwen2VLGRPOTrainer_GT_Soft
from tqdm import tqdm
import torch
import json
import random
@dataclass
class GRPOScriptArguments(ScriptArguments):
"""
Script arguments for the GRPO training script.
Args:
reward_funcs (`list[str]`):
List of reward functions. Possible values: 'iou', 'format'.
"""
reward_funcs: list[str] = field(
default_factory=lambda: ["iou"],
metadata={"help": "List of reward functions. Possible values: 'iou', 'format'"},
)
max_pixels: Optional[int] = field(
default=12845056,
metadata={"help": "Maximum number of pixels for the image"},
)
min_pixels: Optional[int] = field(
default=3136,
metadata={"help": "Minimum number of pixels for the image"},
)
train_data_path: str = field(
default="/share/wy/Video/Charades/charades_annotation/train.json",
metadata={"help": "Path to the training data JSON file."},
)
eval_data_path: str = field(
default="/share/wy/Video/Charades/charades_annotation/val.json",
metadata={"help": "Path to the evaluation data JSON file."},
)
video_folder: str = field(
default="/share/wy/Video/Charades/Charades_v1", # Replace with your actual video folder path
metadata={"help": "Path to the folder containing video files."},
)
preprocessed_data_path: Optional[str] = field( # Add preprocessed_data_path argument
default="",
metadata={"help": "Path to the preprocessed dataset directory. If provided, load preprocessed data instead of raw videos."},
)
def parse_timestamp_output(output_string):
"""Parses timestamp output, similar to the example code."""
# 1. Find all ... blocks.
answer_matches = re.findall(r"(.*?)", output_string, re.DOTALL)
if not answer_matches:
return None # No tags found.
# 2. Use the content of the *last* block.
last_answer_content = answer_matches[-1]
print('last_answer_content:', last_answer_content)
matches = re.findall(r"(\d+\.?\d*) (to|and) (\d+\.?\d*)", last_answer_content, re.IGNORECASE)
if not matches:
return None
last_match = matches[-1]
start_time = float(last_match[0])
end_time = float(last_match[2])
return start_time, end_time
def iou_timestamp_reward(completions, solution, durations, **kwargs): # Modified reward function name and arguments
"""Reward function that calculates IoU between predicted and ground truth timestamps."""
# print(completions, solution, durations)
# contents = [completion[0]["content"] for completion in completions]
rewards = []
# print(completions, solution, durations, **kwargs)
current_time = datetime.now().strftime("%d-%H-%M-%S-%f")
for content, sol, duration in zip(completions, solution, durations): # Added video_durations
reward = 0.0
parsed_times = parse_timestamp_output(content)
start_time, end_time = 0, 0
gt_start, gt_end = sol
# s, e = gt_start / duration, gt_end / duration
s, e = gt_start, gt_end
if parsed_times:
start_time, end_time = parsed_times
from_number = start_time
to_number = end_time
intersection = max(0, min(to_number, e) - max(from_number, s))
union = max(to_number, e) - min(from_number, s)
iou = 0.0
if union > 0:
iou = intersection / union
reward = iou
print('gt second:', gt_start, gt_end)
print('pred second:', start_time, end_time)
print(f"------------- {current_time} IoU reward: {reward} -------------\n")
rewards.append(reward)
if os.getenv("DEBUG_MODE") == "true":
log_path = os.getenv("LOG_PATH")
with open(log_path, "a") as f:
f.write(f"Content: {content}\n")
f.write(f"pred second: {str(start_time)}, {str(end_time)}\n")
f.write(f"gt second: {str(gt_start)}, {str(gt_end)}\n")
f.write(f"------------- {current_time} IoU reward: {reward} -------------\n") # Modified log message
return rewards
def format_reward(completions, **kwargs):
"""Reward function that checks if the completion has tags with timestamp format."""
pattern = re.compile(r'\s*\d+\.?\d*\s+to\s+\d+\.?\d*\s*', re.DOTALL)
matches = [re.search(pattern, content) for content in completions]
print('format matches:', matches)
return [1.0 if match else 0.0 for match in matches]
reward_funcs_registry = {
"iou": iou_timestamp_reward, # Modified registry to use iou_timestamp_reward
"format": format_reward,
}
QUESTION_TEMPLATE = """To accurately pinpoint the event "[EVENT]" in the video, determine the precise time period of the event.
Output your thought process within the tags, including analysis with either specific timestamps (xx.xx) or time ranges (xx.xx to xx.xx) in tags.
Then, provide the start and end times (in seconds, precise to two decimal places) in the format "start time to end time" within the tags. For example: "12.54 to 17.83"."""
def load_json_dataset(train_data_path, eval_data_path, video_folder, preprocessed_data_path=None): # Modified to accept preprocessed_data_path
def create_dataset_from_json(file_path, split_name):
with open(file_path, 'r') as f:
data = json.load(f)
examples = []
for video_id, video_data in tqdm(data.items()):
for sentence_id, (timestamps, sentence) in enumerate(zip(video_data['timestamps'], video_data['sentences'])):
sentence = sentence.strip().lower()
if sentence.endswith("."):
sentence = sentence[:-1]
video_filename_base = video_id
video_path = None
for ext in ['mp4', 'mkv', 'webm']:
candidate_path = os.path.join(video_folder, f"{video_filename_base}.{ext}")
if os.path.isfile(candidate_path):
video_path = candidate_path
break
if video_path is None:
print(f"Warning: Video file not found for ID: {video_id}")
continue
example = {
"problem": sentence,
"solution": (timestamps[0], timestamps[1]),
"video_path": video_path,
"durations": video_data['duration'],
"preprocessed_path": "" # Initialize preprocessed_path as None
}
if preprocessed_data_path != "": # If preprocessed data path is provided, construct the path
example["preprocessed_path"] = os.path.join(preprocessed_data_path, split_name, f"{video_id}_{sentence_id}")
examples.append(example)
random.shuffle(examples)
print(len(examples))
print(examples[:5])
dataset = Dataset.from_list(examples)
def __getitem__(self, idx): # Define getitem within the scope where dataset is available
example = dataset[idx]
# return example
data_to_return = {k: v for k, v in example.items()} # Create a copy to avoid modifying original dataset
# print(data_to_return)
# print("preprocessed_path:", example["preprocessed_path"])
if example["preprocessed_path"] != "": # Check if preprocessed path exists
try:
# data_to_return["image_inputs"] = [torch.load(os.path.join(example["preprocessed_path"][0], "image_inputs.pt"))]
data_to_return["video_inputs"] = [torch.load(os.path.join(example["preprocessed_path"][0], "video_inputs.pt"))]
with open(os.path.join(example["preprocessed_path"][0], "video_kwargs.json"), 'r') as f:
data_to_return["video_kwargs"] = [json.load(f)]
data_to_return["use_preprocessed"] = [True] # Flag to indicate preprocessed data is used
except Exception as e:
print(f"Warning: Error loading preprocessed data from {example['preprocessed_path'][0]}, falling back to video_path. Error: {e}")
data_to_return["use_preprocessed"] = [False] # Fallback to video_path if loading fails
else:
data_to_return["use_preprocessed"] = [False] # No preprocessed data to use or path invalid
return data_to_return
dataset.__getitem__ = __getitem__.__get__(dataset, Dataset) # Bind getitem to the dataset
return dataset
train_dataset = create_dataset_from_json(train_data_path, "train")
eval_dataset = create_dataset_from_json(eval_data_path, "eval")
return DatasetDict({"train": train_dataset, "eval": eval_dataset})
def main(script_args, training_args, model_args):
# Get reward functions
reward_funcs = [reward_funcs_registry[func] for func in script_args.reward_funcs]
# # Load the dataset
# dataset = load_dataset(script_args.dataset_name, name=script_args.dataset_config)
# Load the dataset, now handles both raw and preprocessed data
dataset = load_json_dataset(
script_args.train_data_path,
script_args.eval_data_path,
script_args.video_folder,
script_args.preprocessed_data_path # Pass preprocessed_data_path
)
# Format into conversation
# QUESTION_TEMPLATE = "{Question} Output the thinking process in and final answer (number) in tags."
# def make_conversation_image(example):
# return {
# "prompt": [
# {
# "role": "user",
# "content": [
# {"type": "image"},
# {"type": "text", "text": QUESTION_TEMPLATE.format(Question=example["problem"])},
# ],
# },
# ],
# }
# trainer_cls = Qwen2VLGRPOTrainer if not training_args.use_vllm else Qwen2VLGRPOVLLMTrainer
trainer_cls = Qwen2VLGRPOTrainer_GT_Soft
print("using: ", trainer_cls)
# from peft import LoraConfig, get_peft_model
# lora_config = LoraConfig(
# task_type="CAUSAL_LM",
# target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
# inference_mode=False,
# r=64,
# lora_alpha=16,
# lora_dropout=0.05,
# bias="none",
# )
# Initialize the GRPO trainer
trainer = trainer_cls(
model=model_args.model_name_or_path,
reward_funcs=reward_funcs,
args=training_args,
train_dataset=dataset[script_args.dataset_train_split],
eval_dataset=dataset[script_args.dataset_test_split] if training_args.eval_strategy != "no" else None,
peft_config=get_peft_config(model_args),
attn_implementation=model_args.attn_implementation,
max_pixels=script_args.max_pixels,
min_pixels=script_args.min_pixels,
)
# Train and push the model to the Hub
trainer.train()
# Save and push to hub
trainer.save_model(training_args.output_dir)
if training_args.push_to_hub:
trainer.push_to_hub(dataset_name=script_args.dataset_name)
if __name__ == "__main__":
parser = TrlParser((GRPOScriptArguments, GRPOConfig, ModelConfig))
script_args, training_args, model_args = parser.parse_args_and_config()
main(script_args, training_args, model_args)