app.py

import gradio as gr
import json
from datetime import datetime
import random
import os
from huggingface_hub import Repository
random.seed(20240128)

import subprocess
# Set Git user information
subprocess.run(["git", "config", "--global", "user.email", "wuqixu2015@gmail.com"])
subprocess.run(["git", "config", "--global", "user.name", "peterw2333"])
subprocess.run(["git", "pull"])


hf_token = os.getenv("HF_TOKEN")
print("HF Token is none?", hf_token is None)

# Initialize the repository
DATASET_REPO_URL = "https://huggingface.co/datasets/peterw2333/test"
repo = Repository(
    local_dir="user_responses",
    clone_from=DATASET_REPO_URL,
    use_auth_token=hf_token
)

def prepare_test_cases():
    # json_path = "videos/videos.json"
    json_path = "videos/rir.json"
    with open(json_path, "r") as f:
        video_dict = json.load(f)
    
    video_ids = list(video_dict.keys())
    for video_id in video_ids:
        if random.random() > 0.5:
            video_list = [video_dict[video_id]['25k'], video_dict[video_id]['50k']]
        else: 
            video_list = [video_dict[video_id]['50k'], video_dict[video_id]['25k']]

        random.shuffle(video_list)

        # video_dict[video_id]['prompt'] = video_dict[video_id]['prompt']
        video_dict[video_id]['Video 1'] = video_list[0]
        video_dict[video_id]['Video 2'] = video_list[1]

    return video_dict

video_dict = prepare_test_cases()
video_ids = list(video_dict.keys())  
random.shuffle(video_ids)


questions = [
    # "Between Group 1 and Group 2, which one's results are more diverse?",
    "Between Video 1 (left) and Video 2 (right), which one's results are more accurate according to the text prompt?",
    "Between Video 1 (left) and Video 2 (right), which one's results are more has a higher quality of human-human interaction?"
]
submissions_file = "user_responses/response_202507.jsonl"

def has_already_submitted(user_id):
    if os.path.exists(submissions_file):
        with open(submissions_file, "r") as f:
            for line in f:
                submission = json.loads(line)
                if submission.get("u_id") == user_id:
                    return True
    return False

# Save responses
def save_responses(unique_submission, *responses):
    timestamp = datetime.now().isoformat()
    info = responses[-1]
    responses = responses[:-1]
    unique_id = info["session_id"]
    user_id = f"{unique_id}"

    # Check for unique submission
    if unique_submission and has_already_submitted(user_id):
        return "You have already submitted responses. Thank you for participating!"

    # Initialize the result dictionary
    result = {
        "u_id": user_id,
        "timestamp": timestamp,
        "responses": []
    }

    for index in range(len(video_ids)):
        start_idx = index * len(questions)
        end_idx = start_idx + len(questions)

        response = responses[start_idx:end_idx]
        if any(r is None for r in response):
            return "Please answer all questions before submitting."

        video_id = video_ids[index]
        pair_response = {
            video_id: {
                # 'diversity': "before" if "before" in video_dict[video_id][response[0]][0] else "after",
                'accuracy': "25k" if "25k" in video_dict[video_id][response[0]][0] else "50k",
                'quality': "25k" if "25k" in video_dict[video_id][response[1]][0] else "50k",
            }
        }
        result["responses"].append(pair_response)

    result["responses"] = sorted(result["responses"], key=lambda x: x.keys())
    
    # Save response locally and push to Hugging Face Hub
    with open(submissions_file, "a") as f:
        f.write(json.dumps(result) + "\n")
    
    # Push changes to the Hugging Face dataset repo
    repo.push_to_hub()

    return "All responses saved! Thank you for participating!"




def create_interface(unique_submission=False):
    with gr.Blocks() as demo:
        gr.Markdown("# Human Preference Study: Two Person Interaction")
        gr.Markdown("""
            ## In each of the following pairs, you will be presented with two videos.\n
            ## For each pair, please first read the text prompt carefully, then examine the videos and answer the following questions.
        """)

        # Display video pairs and questions
        responses = []
        for index, video_id in enumerate(video_ids):
            video_prompt = video_dict[video_id]['prompt']
            video1_list = video_dict[video_id]['Video 1']
            video2_list = video_dict[video_id]['Video 2']

            gr.Markdown(f"# Video Pair {index + 1}")
            gr.Markdown(f"## Text Prompt: {video_prompt}")
            # gr.Markdown(f"### Group 1")
            with gr.Row():
                for video in video1_list:
                    gr.Video(video, label="Video 1")
                
                for video in video2_list:
                    gr.Video(video, label="Video 2")
            
            # gr.Markdown(f"### Group 2")
            # with gr.Row():
                

            with gr.Row():
                responses.append(gr.Radio(["Video 1", "Video 2"], label=questions[0], value=None))
            with gr.Row():
                responses.append(gr.Radio(["Video 1", "Video 2"], label=questions[1], value=None))
            # with gr.Row():
            #     responses.append(gr.Radio(["Video 1", "Video 2"], label=questions[2], value=None))

            gr.Markdown("---")

        info = gr.JSON(visible=False)
        demo.load(predict, None, info)

        submit_btn = gr.Button("Submit")
        result_message = gr.Textbox(label="Message (please only submit once)", interactive=False)

        submit_btn.click(
            fn=lambda *args: save_responses(unique_submission, *args),
            inputs=responses+[info],
            outputs=result_message
        )

    return demo

def predict(request: gr.Request):
    headers = request.headers
    host = request.client.host
    user_agent = request.headers["user-agent"]
    session_id = request.session_hash
    return {
        "ip": host,
        "user_agent": user_agent,
        "headers": headers,
        "session_id": session_id
    }

if __name__ == "__main__":    
    # Launch with unique_submission set based on `--unique` flag
    demo = create_interface(unique_submission=True)
    demo.launch(share=True)