import json import os import random import re import sys import time from datetime import datetime from glob import glob from pathlib import Path from typing import List, Optional from uuid import uuid4 import gradio as gr import numpy as np import pandas as pd import requests from datasets import load_dataset from huggingface_hub import ( CommitScheduler, HfApi, InferenceClient, login, snapshot_download, ) from PIL import Image cached_latest_posts_df = None last_fetched = None from moviepy.editor import VideoFileClip import numpy as np def extract_frames_moviepy(video_url, num_frames=10): try: print(f"Extracting {num_frames} frames from {video_url}") # Load the video clip = VideoFileClip(video_url) # Calculate the times to extract frames duration = clip.duration frame_times = np.linspace(0, duration, num_frames, endpoint=False) frameas = [] # Extract and save frames for i, time in enumerate(frame_times): frame = clip.get_frame(time) frame_image = Image.fromarray(frame) frameas.append(frame_image) return frameas except e as Exception: raise gr.Error(f"Error extracting frames from video: {e}") def get_latest_pots(): global cached_latest_posts_df global last_fetched # make sure we don't fetch data too often, limit to 1 request per 10 minutes now_time = datetime.now() if last_fetched is not None and (now_time - last_fetched).seconds < 600: print("Using cached data") return cached_latest_posts_df last_fetched = now_time url = "https://www.reddit.com/r/GamePhysics/.json" headers = {"User-Agent": "Mozilla/5.0"} response = requests.get(url, headers=headers) if response.status_code != 200: return [] data = response.json() # Extract posts from the data posts = data["data"]["children"] for post in posts: title = post["data"]["title"] post_id = post["data"]["id"] print(f"ID: {post_id}, Title: {title}") # create [post_id, title] list examples = [[post["data"]["id"], post["data"]["title"]] for post in posts] # make a dataframe examples = pd.DataFrame(examples, columns=["post_id", "title"]) cached_latest_posts_df = examples return examples def row_selected(evt: gr.SelectData): global cached_latest_posts_df row = evt.index[0] post_id = cached_latest_posts_df.iloc[row]["post_id"] return post_id def load_video(url): # Regular expression pattern for r/GamePhysics URLs and IDs pattern = r"https://www\.reddit\.com/r/GamePhysics/comments/([0-9a-zA-Z]+).*|([0-9a-zA-Z]+)" # Match the URL or ID against the pattern match = re.match(pattern, url) if match: # Extract the post ID from the URL post_id = match.group(1) or match.group(2) print(f"Valid GamePhysics post ID: {post_id}") else: post_id = url video_url = f"https://huggingface.co/datasets/asgaardlab/GamePhysicsDailyDump/resolve/main/data/videos/{post_id}.mp4?download=true" # make sure file exists before returning, make a request without downloading the file r = requests.head(video_url) if r.status_code != 200 and r.status_code != 302: raise gr.Error( f"Video is not in the repo, please try another post. - {r.status_code }" ) return video_url with gr.Blocks() as demo: gr.Markdown("## Preview GamePhysics") dummt_title = gr.Textbox(visible=False) with gr.Row(): with gr.Column(): reddit_id = gr.Textbox( lines=1, placeholder="Post url or id here", label="URL or Post ID" ) load_btn = gr.Button("Load") video_player = gr.Video(interactive=False) with gr.Column(): gr.Markdown("## Latest Posts") latest_post_dataframe = gr.Dataframe() get_latest_pots_btn = gr.Button("Refresh Latest Posts") with gr.Column(): gr.Markdown("## Sampled Frames from Video") with gr.Row(): num_frames = gr.Slider(minimum=1, maximum=20, step=1, value=10) sample_btn = gr.Button("Sample") sampled_frames = gr.Gallery() sample_btn.click( extract_frames_moviepy, inputs=[video_player, num_frames], outputs=[sampled_frames], ) load_btn.click(load_video, inputs=[reddit_id], outputs=[video_player]) get_latest_pots_btn.click(get_latest_pots, outputs=[latest_post_dataframe]) demo.load(get_latest_pots, outputs=[latest_post_dataframe]) latest_post_dataframe.select(fn=row_selected, outputs=[reddit_id]).then( load_video, inputs=[reddit_id], outputs=[video_player] ) demo.launch()