app.py

import csv
import json
import os
import pickle
import random
import string
import sys
import time
from glob import glob

import datasets
import gdown
import gradio as gr
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import torch
import torchvision
from huggingface_hub import HfApi, login, snapshot_download
from PIL import Image
import re
from fnmatch import translate


session_token = os.environ.get("SessionToken")
login(token=session_token)

csv.field_size_limit(sys.maxsize)

np.random.seed(int(time.time()))

with open("./imagenet_hard_nearest_indices.pkl", "rb") as f:
    knn_results = pickle.load(f)

with open("imagenet-labels.json") as f:
    wnid_to_label = json.load(f)

with open("id_to_label.json", "r") as f:
    id_to_labels = json.load(f)

imagenet_training_samples_path = "imagenet_samples"

bad_items = open("./ex2.txt", "r").read().split("\n")
bad_items = [x.split(".")[0] for x in bad_items]
bad_items = [int(x) for x in bad_items if x != ""]

NUMBER_OF_IMAGES = len(bad_items)


gdown.cached_download(
    url="https://huggingface.co/datasets/taesiri/imagenet_hard_review_samples/resolve/main/data.zip",
    path="./data.zip",
    quiet=False,
    md5="ece2720fed664e71799f316a881d4324",
)

# EXTRACT if needed

if not os.path.exists("./imagenet_samples") or not os.path.exists(
    "./knn_cache_for_imagenet_hard"
):
    torchvision.datasets.utils.extract_archive(
        from_path="data.zip",
        to_path="./",
        remove_finished=False,
    )

imagenet_hard = datasets.load_dataset("taesiri/imagenet-hard", split="validation")


def update_snapshot(username):
    pattern = f"*{username}*.json"

    output_dir = snapshot_download(
        repo_id="taesiri/imagenet_hard_review_data",
        allow_patterns=pattern,
        repo_type="dataset",
    )
    files = glob(f"{output_dir}/*.json")

    df = pd.DataFrame()
    columns = ["id", "user_id", "time", "decision"]
    rows = []
    for file in files:
        with open(file) as f:
            data = json.load(f)
            tdf = [data[x] for x in columns]
            rows.append(tdf)

    df = pd.DataFrame(rows, columns=columns)
    df = df[df["user_id"] == username]

    return df


def generate_dataset(username):
    global NUMBER_OF_IMAGES
    df = update_snapshot(username)

    all_images = set(bad_items)
    answered = set(df.id)
    remaining = list(all_images - answered)
    # shuffle remaining
    random.shuffle(remaining)

    NUMBER_OF_IMAGES = len(bad_items)

    print(f"NUMBER_OF_IMAGES: {NUMBER_OF_IMAGES}")
    print(f"Remaining: {len(remaining)}")

    if NUMBER_OF_IMAGES == 0:
        return []

    data = []
    for i, image in enumerate(remaining):
        data.append(
            {
                "id": remaining[i],
            }
        )
    return data


def string_to_image(text):
    text = text.replace("_", " ").lower().replace(", ", "\n")
    # Create a blank white square image
    img = np.ones((220, 75, 3))

    fig, ax = plt.subplots(figsize=(6, 2.25))
    ax.imshow(img, extent=[0, 1, 0, 1])
    ax.text(0.5, 0.75, text, fontsize=18, ha="center", va="center")
    ax.set_xticks([])
    ax.set_yticks([])
    ax.set_xticklabels([])
    ax.set_yticklabels([])
    for spine in ax.spines.values():
        spine.set_visible(False)

    return fig


all_samples = glob("./imagenet_samples/*.JPEG")
qid_to_sample = {
    int(x.split("/")[-1].split(".")[0].split("_")[0]): x for x in all_samples
}


def get_training_samples(qid):
    labels_id = imagenet_hard[int(qid)]["label"]
    samples = [qid_to_sample[x] for x in labels_id]
    return samples


def load_sample(data, current_index):
    image_id = data[current_index]["id"]
    qimage = imagenet_hard[int(image_id)]["image"]
    # labels = data[current_index]["correct_label"]
    labels = imagenet_hard[int(image_id)]["english_label"]
    # print(f"Image ID: {image_id}")
    # print(f"Labels: {labels}")

    return qimage, labels


def preprocessing(data, current_index, history, username):
    data = generate_dataset(username)

    remaining_images = len(data)
    labeled_images = len(bad_items) - remaining_images

    if remaining_images == 0:
        fake_plot = string_to_image("No more images to review")
        empty_image = Image.new("RGB", (224, 224))
        return (
            empty_image,
            fake_plot,
            current_index,
            history,
            data,
            None,
            labeled_images,
        )

    current_index = 0
    qimage, labels = load_sample(data, current_index)
    image_id = data[current_index]["id"]
    training_samples_image = get_training_samples(image_id)
    training_samples_image = [
        Image.open(x).convert("RGB") for x in training_samples_image
    ]

    # labels is a list of labels, conver it to a string
    labels = ", ".join(labels)
    label_plot = string_to_image(labels)

    return (
        qimage,
        label_plot,
        current_index,
        history,
        data,
        training_samples_image,
        labeled_images,
    )


def update_app(decision, data, current_index, history, username):
    global NUMBER_OF_IMAGES
    if current_index == -1:
        fake_plot = string_to_image("Please Enter your username and load samples")
        empty_image = Image.new("RGB", (224, 224))
        return empty_image, fake_plot, current_index, history, data, None, 0

    if current_index == NUMBER_OF_IMAGES - 1:
        time_stamp = int(time.time())

        image_id = data[current_index]["id"]
        # convert to percentage
        dicision_dict = {
            "id": int(image_id),
            "user_id": username,
            "time": time_stamp,
            "decision": decision,
        }

        # upload the decision to the server
        temp_filename = f"results_{username}_{time_stamp}.json"
        # convert decision_dict to json and save it on the disk
        with open(temp_filename, "w") as f:
            json.dump(dicision_dict, f)

        api = HfApi()
        api.upload_file(
            path_or_fileobj=temp_filename,
            path_in_repo=temp_filename,
            repo_id="taesiri/imagenet_hard_review_data",
            repo_type="dataset",
        )

        os.remove(temp_filename)

        fake_plot = string_to_image("Thank you for your time!")
        empty_image = Image.new("RGB", (224, 224))

        remaining_images = len(data)
        labeled_images = (len(bad_items) - remaining_images) + current_index

        return (
            empty_image,
            fake_plot,
            current_index,
            history,
            data,
            None,
            labeled_images + 1,
        )

    if current_index >= 0 and current_index < NUMBER_OF_IMAGES - 1:
        time_stamp = int(time.time())

        image_id = data[current_index]["id"]
        # convert to percentage
        dicision_dict = {
            "id": int(image_id),
            "user_id": username,
            "time": time_stamp,
            "decision": decision,
        }

        # upload the decision to the server
        temp_filename = f"results_{username}_{time_stamp}.json"
        # convert decision_dict to json and save it on the disk
        with open(temp_filename, "w") as f:
            json.dump(dicision_dict, f)

        api = HfApi()
        api.upload_file(
            path_or_fileobj=temp_filename,
            path_in_repo=temp_filename,
            repo_id="taesiri/imagenet_hard_review_data",
            repo_type="dataset",
        )

        os.remove(temp_filename)

        # Load the Next Image

        current_index += 1
        qimage, labels = load_sample(data, current_index)
        image_id = data[current_index]["id"]
        training_samples_image = get_training_samples(image_id)
        training_samples_image = [
            Image.open(x).convert("RGB") for x in training_samples_image
        ]

        # labels is a list of labels, conver it to a string
        labels = ", ".join(labels)
        label_plot = string_to_image(labels)

        remaining_images = len(data)
        labeled_images = (len(bad_items) - remaining_images) + current_index

        return (
            qimage,
            label_plot,
            current_index,
            history,
            data,
            training_samples_image,
            labeled_images,
        )


newcss = """
#query_image{
}

#nn_gallery {
  height: auto !important;
}

#sample_gallery {
    height: auto !important;
}


/* Set display to flex for the parent element */
.svelte-parentrowclass {
  display: flex;
}

/* Set the flex-grow property for the children elements */
.svelte-parentrowclass > #query_image {
  min-width: min(400px, 40%);
  flex : 1;
  flex-grow: 0; !important;
  border-style: solid;
  height: auto !important;
}

.svelte-parentrowclass > .svelte-rightcolumn {
  flex: 2;
  flex-grow: 0; !important;
  min-width: min(600px, 60%);
}



"""

with gr.Blocks(css=newcss, theme=gr.themes.Soft()) as demo:
    data_gr = gr.State({})
    current_index = gr.State(-1)
    history = gr.State({})

    gr.Markdown("# Help Us to Clean `ImageNet-Hard`!")

    gr.Markdown("## Instructions")
    gr.Markdown(
        "Please enter your username and press `Load Samples`. The loading process might take up to a minute. Once the loading is done, you can start reviewing the samples."
    )
    gr.Markdown(
        """For each image, please select one of the following options: `Accept`, `Not Sure!`, `Reject`.
        - If you think any of the labels are correct, please select `Accept`.
        - If you think none of the labels matching the image, please select `Reject`. 
        - If you are not sure about the label, please select `Not Sure!`. 

        You can refer to `Training samples` if you are not sure about the target label.
        """
    )

    random_str = "".join(
        random.choice(string.ascii_lowercase + string.digits) for _ in range(5)
    )

    with gr.Column():
        with gr.Row():
            username = gr.Textbox(label="Username", value=f"user-{random_str}")
            labeled_images = gr.Textbox(label="Labeled Images", value="0")
            total_images = gr.Textbox(label="Total Images", value=len(bad_items))

        prepare_btn = gr.Button(value="Load Samples")

    with gr.Column():
        with gr.Row():
            accept_btn = gr.Button(value="Accept")
            myabe_btn = gr.Button(value="Not Sure!")
            reject_btn = gr.Button(value="Reject")
        with gr.Row(elem_id="parent_row", elem_classes="svelte-parentrowclass"):
            query_image = gr.Image(type="pil", label="Query", elem_id="query_image")
            with gr.Column(
                elem_id="samples_col",
                elem_classes="svelte-rightcolumn",
            ):
                label_plot = gr.Plot(
                    label="Is this a correct label for this image?", type="fig"
                )
                training_samples = gr.Gallery(
                    type="pil", label="Training samples", elem_id="sample_gallery"
                )

    accept_btn.click(
        update_app,
        inputs=[accept_btn, data_gr, current_index, history, username],
        outputs=[
            query_image,
            label_plot,
            current_index,
            history,
            data_gr,
            training_samples,
            labeled_images,
        ],
    )
    myabe_btn.click(
        update_app,
        inputs=[myabe_btn, data_gr, current_index, history, username],
        outputs=[
            query_image,
            label_plot,
            current_index,
            history,
            data_gr,
            training_samples,
            labeled_images,
        ],
    )

    reject_btn.click(
        update_app,
        inputs=[reject_btn, data_gr, current_index, history, username],
        outputs=[
            query_image,
            label_plot,
            current_index,
            history,
            data_gr,
            training_samples,
            labeled_images,
        ],
    )

    prepare_btn.click(
        preprocessing,
        inputs=[data_gr, current_index, history, username],
        outputs=[
            query_image,
            label_plot,
            current_index,
            history,
            data_gr,
            training_samples,
            labeled_images,
        ],
    )


demo.launch(debug=False)