Duplicate from taesiri/CLIPxGamePhysics

Co-authored-by: taesiri <taesiri@users.noreply.huggingface.co>

.gitattributes

@@ -0,0 +1,27 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bin.* filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zstandard filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

README.md

@@ -0,0 +1,39 @@
+---
+title: CLIP GamePhysics
+emoji: 🚚
+colorFrom: red
+colorTo: blue
+sdk: gradio
+sdk_version: 3.0.5
+app_file: app.py
+pinned: false
+duplicated_from: taesiri/CLIPxGamePhysics
+---
+
+# Configuration
+
+`title`: _string_  
+Display title for the Space
+
+`emoji`: _string_  
+Space emoji (emoji-only character allowed)
+
+`colorFrom`: _string_  
+Color for Thumbnail gradient (red, yellow, green, blue, indigo, purple, pink, gray)
+
+`colorTo`: _string_  
+Color for Thumbnail gradient (red, yellow, green, blue, indigo, purple, pink, gray)
+
+`sdk`: _string_  
+Can be either `gradio` or `streamlit`
+
+`sdk_version` : _string_  
+Only applicable for `streamlit` SDK.  
+See [doc](https://hf.co/docs/hub/spaces) for more info on supported versions.
+
+`app_file`: _string_  
+Path to your main application file (which contains either `gradio` or `streamlit` Python code).  
+Path is relative to the root of the repository.
+
+`pinned`: _boolean_  
+Whether the Space stays on top of your list.

SimSearch.py

@@ -0,0 +1,46 @@
+import faiss
+import numpy as np
+
+class FaissNeighbors:
+  def __init__(self):
+    self.index = None
+    self.y = None
+
+  def fit(self, X, y):
+    self.index = faiss.IndexFlatL2(X.shape[1])
+    self.index.add(X.astype(np.float32))
+    self.y = y
+  
+  def get_distances_and_indices(self, X, top_K=1000):
+    distances, indices = self.index.search(X.astype(np.float32), k=top_K)
+    return np.copy(distances), np.copy(indices), np.copy(self.y[indices])
+  
+  def get_nearest_labels(self, X, top_K=1000):
+    distances, indices = self.index.search(X.astype(np.float32), k=top_K)
+    return np.copy(self.y[indices])
+
+
+class FaissCosineNeighbors:
+  def __init__(self):
+    self.cindex = None
+    self.y = None
+
+  def fit(self, X, y):
+    self.cindex = faiss.index_factory(X.shape[1], "Flat", faiss.METRIC_INNER_PRODUCT)
+    X = np.copy(X)
+    X = X.astype(np.float32)
+    faiss.normalize_L2(X)
+    self.cindex.add(X)
+    self.y = y
+  
+  def get_distances_and_indices(self, Q, topK):
+    Q = np.copy(Q)
+    faiss.normalize_L2(Q)
+    distances, indices = self.cindex.search(Q.astype(np.float32), k=topK)
+    return np.copy(distances), np.copy(indices), np.copy(self.y[indices])
+  
+  def get_nearest_labels(self, Q, topK=1000):
+    Q = np.copy(Q)
+    faiss.normalize_L2(Q)
+    distances, indices = self.cindex.search(Q.astype(np.float32), k=topK)
+    return np.copy(self.y[indices])

app.py

@@ -0,0 +1,253 @@
+import csv
+import os
+import pickle
+import random
+import sys
+from collections import Counter
+from glob import glob
+
+import clip
+import gdown
+import gradio as gr
+import numpy as np
+import psutil
+import torch
+import torchvision
+from datasets import load_dataset
+from tqdm import tqdm
+
+from SimSearch import FaissCosineNeighbors
+
+csv.field_size_limit(sys.maxsize)
+
+# Download Embeddings
+gdown.cached_download(
+    url="https://huggingface.co/datasets/taesiri/GTA_V_CLIP_Embeddings/resolve/main/mini-GTA-V-Embeddings.zip",
+    path="./GTA-V-Embeddings.zip",
+    quiet=False,
+    md5="b1228503d5a89eef7e35e2cbf86b2fc0",
+)
+
+# EXTRACT
+torchvision.datasets.utils.extract_archive(
+    from_path="GTA-V-Embeddings.zip",
+    to_path="Embeddings/VIT32/",
+    remove_finished=False,
+)
+
+# Load videos from Dataset
+gta_v_videos = load_dataset("taesiri/GamePhysics_Grand_Theft_Auto_V")
+post_id_to_video_path = {
+    os.path.splitext(os.path.basename(x))[0]: x
+    for x in gta_v_videos["Grand_Theft_Auto_V"][:]["video_file_path"]
+}
+# Initialize CLIP model
+clip.available_models()
+
+
+# Log runtime environment info
+def log_runtime_information():
+    print(f"CPU Count: {psutil.cpu_count()}")
+    print(f"Virtual Memory: {psutil.virtual_memory()}")
+    print(f"Swap Memory: {psutil.swap_memory()}")
+
+
+# # Searcher
+class GamePhysicsSearcher:
+    def __init__(self, CLIP_MODEL, GAME_NAME, EMBEDDING_PATH="./Embeddings/VIT32/"):
+        self.CLIP_MODEL = CLIP_MODEL
+        self.GAME_NAME = GAME_NAME
+        self.simsearcher = FaissCosineNeighbors()
+
+        self.all_embeddings = glob(f"{EMBEDDING_PATH}{self.GAME_NAME}/*.npy")
+
+        self.filenames = [os.path.basename(x) for x in self.all_embeddings]
+        self.file_to_class_id = {x: i for i, x in enumerate(self.filenames)}
+        self.class_id_to_file = {i: x for i, x in enumerate(self.filenames)}
+        self.build_index()
+
+    def read_features(self, file_path):
+        with open(file_path, "rb") as f:
+            video_features = pickle.load(f)
+        return video_features
+
+    def read_all_features(self):
+        features = {}
+        filenames_extended = []
+
+        X_train = []
+        y_train = []
+
+        for i, vfile in enumerate(tqdm(self.all_embeddings)):
+            vfeatures = self.read_features(vfile)
+            features[vfile.split("/")[-1]] = vfeatures
+            X_train.extend(vfeatures)
+            y_train.extend([i] * vfeatures.shape[0])
+            filenames_extended.extend(vfeatures.shape[0] * [vfile.split("/")[-1]])
+
+        X_train = np.asarray(X_train)
+        y_train = np.asarray(y_train)
+
+        return X_train, y_train
+
+    def build_index(self):
+        X_train, y_train = self.read_all_features()
+        self.simsearcher.fit(X_train, y_train)
+
+    def text_to_vector(self, query):
+        text_tokens = clip.tokenize(query)
+        with torch.no_grad():
+            text_features = self.CLIP_MODEL.encode_text(text_tokens).float()
+        text_features /= text_features.norm(dim=-1, keepdim=True)
+        return text_features
+
+    # Source: https://stackoverflow.com/a/480227
+    def f7(self, seq):
+        seen = set()
+        seen_add = seen.add  # This is for performance improvement, don't remove
+        return [x for x in seq if not (x in seen or seen_add(x))]
+
+    def search_top_k(self, q, k=5, pool_size=1000, search_mod="Majority"):
+        q = self.text_to_vector(q)
+        nearest_data_points = self.simsearcher.get_nearest_labels(q, pool_size)
+
+        if search_mod == "Majority":
+            topKs = [x[0] for x in Counter(nearest_data_points[0]).most_common(k)]
+        elif search_mod == "Top-K":
+            topKs = list(self.f7(nearest_data_points[0]))[:k]
+
+        video_filename = [
+            post_id_to_video_path[self.class_id_to_file[x].replace(".npy", "")]
+            for x in topKs
+        ]
+
+        return video_filename
+
+
+################ SEARCH CORE ################
+# CRAETE CLIP MODEL
+vit_model, vit_preprocess = clip.load("ViT-B/32")
+vit_model.eval()
+
+saved_searchers = {}
+
+
+def gradio_search(query, game_name, selected_model, aggregator, pool_size, k=6):
+    # print(query, game_name, selected_model, aggregator, pool_size)
+    if f"{game_name}_{selected_model}" in saved_searchers.keys():
+        searcher = saved_searchers[f"{game_name}_{selected_model}"]
+    else:
+        if selected_model == "ViT-B/32":
+            model = vit_model
+            searcher = GamePhysicsSearcher(CLIP_MODEL=model, GAME_NAME=game_name)
+        else:
+            raise
+
+        saved_searchers[f"{game_name}_{selected_model}"] = searcher
+
+    results = []
+    relevant_videos = searcher.search_top_k(
+        query, k=k, pool_size=pool_size, search_mod=aggregator
+    )
+
+    params = ", ".join(
+        map(str, [query, game_name, selected_model, aggregator, pool_size])
+    )
+    results.append(params)
+
+    for v in relevant_videos:
+        results.append(v)
+        sid = v.split("/")[-1].split(".")[0]
+        results.append(
+            f'<a href="https://www.reddit.com/r/GamePhysics/comments/{sid}/" target="_blank">Link to the post</a>'
+        )
+
+    print(f"found {len(results)} results")
+    return results
+
+
+def main():
+    list_of_games = ["Grand Theft Auto V"]
+
+    # GRADIO APP
+    main = gr.Interface(
+        fn=gradio_search,
+        inputs=[
+            gr.Textbox(
+                lines=1,
+                placeholder="Search Query",
+                value="A person flying in the air",
+                label="Query",
+            ),
+            gr.Radio(list_of_games, label="Game To Search"),
+            gr.Radio(["ViT-B/32"], label="MODEL"),
+            gr.Radio(["Majority", "Top-K"], label="Aggregator"),
+            gr.Slider(300, 2000, label="Pool Size", value=1000),
+        ],
+        outputs=[
+            gr.Textbox(type="auto", label="Search Params"),
+            gr.Video(type="mp4", label="Result 1"),
+            gr.Markdown(),
+            gr.Video(type="mp4", label="Result 2"),
+            gr.Markdown(),
+            gr.Video(type="mp4", label="Result 3"),
+            gr.Markdown(),
+            gr.Video(type="mp4", label="Result 4"),
+            gr.Markdown(),
+            gr.Video(type="mp4", label="Result 5"),
+            gr.Markdown(),
+        ],
+        examples=[
+            ["A red car", list_of_games[0], "ViT-B/32", "Top-K", 1000],
+            ["A person wearing pink", list_of_games[0], "ViT-B/32", "Top-K", 1000],
+            ["A car flying in the air", list_of_games[0], "ViT-B/32", "Majority", 1000],
+            [
+                "A person flying in the air",
+                list_of_games[0],
+                "ViT-B/32",
+                "Majority",
+                1000,
+            ],
+            [
+                "A car in vertical position",
+                list_of_games[0],
+                "ViT-B/32",
+                "Majority",
+                1000,
+            ],
+            ["A bike inside a car", list_of_games[0], "ViT-B/32", "Majority", 1000],
+            ["A bike on a wall", list_of_games[0], "ViT-B/32", "Majority", 1000],
+            ["A car stuck in a rock", list_of_games[0], "ViT-B/32", "Majority", 1000],
+            ["A car stuck in a tree", list_of_games[0], "ViT-B/32", "Majority", 1000],
+        ],
+    )
+
+    blocks = gr.Blocks()
+    with blocks:
+        gr.Markdown(
+            """
+        # CLIP + GamePhysics - Searching dataset of Gameplay bugs
+        
+        This demo shows how to use the CLIP model to search for gameplay bugs in a video game.
+        
+        Enter your query and select the game you want to search for.
+        """
+        )
+
+        gr.Markdown(
+            """
+            [Website](https://asgaardlab.github.io/CLIPxGamePhysics/) - [Paper](https://arxiv.org/abs/2203.11096)
+            """
+        )
+
+        gr.TabbedInterface([main], ["GTA V Demo"])
+
+    blocks.launch(
+        debug=True,
+        enable_queue=True,
+    )
+
+
+if __name__ == "__main__":
+    log_runtime_information()
+    main()

requirements.txt

@@ -0,0 +1,9 @@
+torch
+torchvision
+tqdm
+gdown
+pandas
+clip @ git+https://github.com/openai/CLIP.git@573315e83f07b53a61ff5098757e8fc885f1703e
+faiss-cpu==1.7.2
+psutil
+numpy