Create app.py

app.py

@@ -0,0 +1,502 @@
+# change the eval ftn to take a list of lists
+import gradio as gr
+import pandas as pd
+import time
+import torch
+import os
+import torchvision.transforms as transforms
+from torchvision import datasets
+import torch.nn.functional as F
+from torch.utils.data import DataLoader
+import subprocess
+# from dummy_eval import foo
+import zipfile
+import shutil
+import numpy as np
+import importlib.util
+import inspect
+from huggingface_hub import HfApi
+from datasets import load_dataset, Dataset
+from huggingface_hub import login
+import requests
+import matplotlib
+matplotlib.use("Agg")
+
+def fetch_required_files(exp_config):
+    # os.makedirs("temp_data", exist_ok=True)
+    for key in exp_config:
+        file_path = exp_config[key]['file']
+        url = f"https://saraht14-server.hf.space/file/{file_path}.txt"
+
+        filename_only = os.path.basename(file_path) + ".txt"
+        local_path = os.path.join("./", filename_only)
+
+        downloaded = download_file(url, local_path)
+        if not downloaded:
+            raise Exception(f"Could not download file: {file_path}")
+
+        exp_config[key]["local_file"] = local_path
+
+    return exp_config
+
+
+def call_flask_server(username):
+    url = "https://saraht14-server.hf.space/"
+    
+    try:
+        response = requests.get(url)
+        result = response.json()
+        print("Flask response:", result)
+        return result.get("result", "No result")
+    except Exception as e:
+        print("Failed to contact Flask server:", e)
+        return f"Error contacting server: {e}"
+call_flask_server("sarah")
+def download_file(url, local_path):
+    try:
+        r = requests.get(url, headers={"Authorization": f"Bearer {HF_TOKEN}"})
+        r.raise_for_status()
+        with open(local_path, 'wb') as f:
+            f.write(r.content)
+        return local_path
+    except Exception as e:
+        print(f"Error downloading file from {url}: {e}")
+        return None
+# def log_submission_request(username, zip_file):
+#     try:
+#         requests_ds = load_dataset("IndoorOutdoor/requests", split="train")
+#     except Exception as e:
+#         print("Could not load requests dataset, creating a new one.", e)
+#         requests_ds = Dataset.from_dict({"username": [], "timestamp": [], "zip_filename": []})
+    
+#     new_entry = {"username": username,
+#                  "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
+#                  "zip_filename": os.path.basename(zip_file.name)}
+    
+#     updated_requests = requests_ds.add_item(new_entry)
+    
+#     updated_requests.push_to_hub("IndoorOutdoor/requests", token=HF_TOKEN)
+#     print("Logged submission request to the requests dataset.")
+# def update_results_dataset(leaderboard_df):
+repo_id = "saraht14/responses"
+def update_results_dataset(new_row_df):
+    repo_id = "saraht14/responses"
+    
+    try:
+        leaderboard_dataset = load_dataset(repo_id, split="train", token=HF_TOKEN)
+        leaderboard_df = leaderboard_dataset.to_pandas()
+        updated_df = pd.concat([leaderboard_df, new_row_df], ignore_index=True)
+        updated_dataset = Dataset.from_pandas(updated_df)
+        updated_dataset.push_to_hub(repo_id, token=HF_TOKEN)
+        print("New row(s) added to existing leaderboard dataset.")
+        return updated_dataset
+    except Exception as e:
+        print("Dataset not found or failed to load, creating a new one.")
+        try:
+            new_dataset = Dataset.from_pandas(new_row_df)
+            new_dataset.push_to_hub(repo_id, token=HF_TOKEN)
+            return new_dataset
+            print("New leaderboard dataset created and uploaded.")
+        except Exception as inner_e:
+            print("Failed to create and push new leaderboard dataset:", inner_e)
+        
+
+    
+# Info to change for your repository
+# ----------------------------------
+HF_TOKEN = os.environ.get("HF_TOKEN") # A read/write token for your org
+print(f"{HF_TOKEN}")
+OWNER = "IndoorOutdoor" # Change to your org - don't forget to create a results and request dataset, with the correct format!
+# ----------------------------------
+
+REPO_ID = f"{OWNER}/leaderboard"
+QUEUE_REPO = f"{OWNER}/requests"
+RESULTS_REPO = f"{OWNER}/results"
+global_error_message = "Ready for submission!"
+# def set_error_message(message):
+#     global global_error_message
+#     global_error_message = message
+#     print("ERROR UPDATED:", global_error_message)  # Debugging
+    
+def get_error_message():
+    return global_error_message
+
+def install_requirements(file_path):
+    try:
+        with open(file_path, "r") as file:
+            requirements = file.readlines()
+        
+        for req in requirements:
+            package = req.strip()
+            if package:
+                subprocess.run(["pip", "install", package], check=True)
+                print(f"Installed: {package}")
+
+        print("All requirements installed successfully.")
+
+    except FileNotFoundError:
+        print(f"Error: {file_path} not found.")
+    except subprocess.CalledProcessError as e:
+        print(f"Installation failed: {e}")
+HEADERS = ["Username", "Execution Time (s)", "Accuracy", "TP", "FP", "FN", "TN", "Status"]
+BASE = {'ottawa':(45.30326753851309,-75.93640391349997),
+        'ali_home':(37.88560412289598,-122.30218612514359),
+        'josh_home':(37.8697406, -122.30218612514359),
+        'cory':(37.8697406,-122.281570)}
+
+def get_base(filename):
+    if "home" in filename:
+        return BASE["ali_home"]
+    elif "ottawa" in filename:
+        return BASE["ottawa"]
+    elif "josh" in filename:
+        return BASE["josh_home"]
+    else:
+        return BASE["cory"]
+
+metadata_path = "metadata.csv" 
+dir = ""
+df = pd.read_csv(metadata_path)
+
+print(df.head())
+def fetch_lb():
+    try:
+        leaderboard_dataset = load_dataset("saraht14/responses", split="train", token=HF_TOKEN)
+        leaderboard_data = leaderboard_dataset.to_pandas()
+        leaderboard_data = leaderboard_data[HEADERS]  # keep it ordered
+        leaderboard_data = leaderboard_data.sort_values(by=["Accuracy", "Execution Time (s)"], ascending=[False, True])
+    except Exception as e:
+        print(f"Error loading leaderboard:", e)
+        leaderboard_data = pd.DataFrame(columns=HEADERS)
+    
+    print(f"THIS IS THE LEADERBOARD:\n{leaderboard_data}")
+    return leaderboard_data
+    
+
+leaderboard_data = fetch_lb()
+
+def compute_stats_sector(sectors_model, sector_gt):
+    TP = FP = FN = TN = 0
+    ignored = 0
+    for i in range(len(sector_gt)):
+        if sector_gt[i]  == 1:
+            if sectors_model[i] > 0 or sectors_model[(i+1) % 8] > 0 or sectors_model[(i-1) % 8] > 0 :
+                TP += 1
+            else:
+                FN += 1
+        else:
+            if sectors_model[i] > 0:
+                if sector_gt[(i-1) % 8]  > 0 or sector_gt[(i+1) % 8]  > 0:
+                    TP += 1
+                    continue
+                FP += 1
+            else:
+                TN += 1   
+    NUM_SECTORS = 8 - ignored      
+    return [TP / NUM_SECTORS, FP / NUM_SECTORS, FN / NUM_SECTORS, TN / NUM_SECTORS]
+
+#Compare the model output with ground truth
+#return TP, FP, FN, TN
+#This fuction compute stats when the model is binary i.e., outputs only indoor vs outdoor
+def compute_stats_in_out(sectors_model, indoor_gt):
+    if indoor_gt: #if groundtruth is indoor
+        for i in range(len(sectors_model)):
+            if sectors_model[i]:
+                return [0,1,0,0]
+        return [0,0,0,1]
+    else: #if outdoor
+        for i in range(len(sectors_model)):
+            if sectors_model[i]:
+                return [1,0,0,0]
+        return [0,0,1,0]
+       
+def read_configuration(filename):
+    print("read config")
+    with open(filename, 'r') as file:
+        data = file.read().split('\n')
+    data = data[1:] #ignore the header
+    print("head", data)
+    exp = {}
+    for line in data:
+        if len(line) == 0:
+            continue
+        tokens =line.split(',')
+        file = tokens[0]
+        scenario = tokens[1]
+        indoor = True if tokens[2] == "TRUE" else 0
+
+        exp[scenario] = {'sectors':[1 if x == "TRUE" else 0 for x in tokens[3:]], 'indoor':indoor, "file":file}
+    return exp
+
+
+
+
+def evaluate_model(username, file):
+    print("evaluating...")
+    global leaderboard_data
+
+    username = username.strip()
+    if not username:
+        return leaderboard_data.values.tolist()
+
+    script_path = f"submissions/{username}.py"
+    os.makedirs("submissions", exist_ok=True)
+    
+    # # Get the file path from the NamedString object
+    # file_path = file.name  # Get the actual file path
+    # print("file_path:", file_path)
+    # with open(script_path, "wb") as f:
+    #     with open(file_path, "rb") as uploaded_file:
+    #         f.write(uploaded_file.read())
+
+
+    
+    # script_path = f"submissions/{username}.py"
+    # os.makedirs("submissions", exist_ok=True)
+    # with open(script_path, "wb") as f:
+    #     f.write(file.read())
+
+    try:
+
+        exp = read_configuration("metadata.csv") 
+        print(f"FIRST: {len(exp)}")
+        exp = fetch_required_files(exp)
+        print(f"SECOND: {len(exp)}")
+        
+        start_time = time.time()
+        stats_model_sectors = []
+        stats_model_in_out = []
+
+        for key in exp:
+            filename = exp[key]['file']
+            indoor_gt = exp[key]['indoor']
+            sectors_gt = exp[key]["sectors"]
+
+            # file_path = os.path.join(dataset_directory, filename)
+            # print(file_path)
+            filename = filename + ".txt"
+            print("FILE TO PROCESS:", filename)
+            # filename_url = f"https://saraht14-server.hf.space/file/{filename}"
+            # local_txt_path = f"./{filename}.txt"
+            # os.makedirs("temp_data", exist_ok=True)
+            local_file_path = exp[key]["local_file"]
+            # downloaded = download_file(filename_url, local_txt_path)
+            
+            # if not downloaded:
+                # raise Exception("Failed to fetch remote file.")
+            # sectors_model = subprocess.run(["python", script_path,filename], capture_output=True, text=True, timeout=300)
+            # hello = foo()
+            # print(f"HELLO: {hello}")
+            # import 
+            sectors_model = import_and_run_function(file, "evaluate", local_file_path)
+            try:
+                os.remove(local_file_path)
+            except Exception as e:
+                print(f"Warning: Couldn't delete {local_file_path} — {e}")
+            # print(status)
+            print(f"TYPE: {type(sectors_model), {type(sectors_model[0])}}")
+            
+            
+            print("SECTORS MODEL: ", sectors_model)
+            # sectors_model = eval(filename)
+            # print(sectors_model)
+            # sectors_model = model_based_clustering(dataset_directory, filename)
+
+            stats_model_sectors.append(compute_stats_sector(sectors_model, sectors_gt))
+            stats_model_in_out.append(compute_stats_in_out(sectors_model, indoor_gt))
+
+        execution_time = round(time.time() - start_time, 4)
+        print("calculating summary stats")
+        TP = np.mean([x[0] for x in stats_model_sectors])
+        FP = np.mean([x[1] for x in stats_model_sectors])
+        FN = np.mean([x[2] for x in stats_model_sectors])
+        TN = np.mean([x[3] for x in stats_model_sectors])
+        print("calculating exec stats")
+        
+        accuracy = round((TP + TN) / (TP + TN + FP + FN), 2)
+
+        status = "Success" if accuracy > 0 else "Incorrect Model"
+# ["Username", "Execution Time (s)", "Accuracy", "True Positive", "False Positive", "False Negative", "False Positive", "Status"]
+    except Exception as e:
+        leaderboard_data = pd.concat([leaderboard_data, pd.DataFrame([[username, float("inf"), 0,-1,-1,-1,-1, f"Model Error: {str(e)}"]], 
+                                 columns=HEADERS)], ignore_index=True)
+        return leaderboard_data.values.tolist()
+    print("calculating new entry")
+    
+    new_entry = pd.DataFrame([[username, execution_time, accuracy, TP, FP, FN, TN, status]], 
+                                 columns=HEADERS)
+    print("updating new entry")
+    leaderboard_data = update_results_dataset(new_entry)
+    # leaderboard_data = pd.concat([leaderboard_data, new_entry], ignore_index=True)
+    leaderboard_data = leaderboard_data.to_pandas() if leaderboard_data is not None else None
+    if leaderboard_data is not None:
+        leaderboard_data = leaderboard_data.sort_values(by=["Accuracy", "Execution Time (s)"], ascending=[False, True]).reset_index(drop=True)
+        print(f"DATA: {leaderboard_data}")
+        return leaderboard_data.values.tolist()
+
+    
+def import_and_run_function(script_path, function_name, filename):
+
+        
+    if not os.path.exists(script_path):
+        set_error_message(f"Error: {script_path} not found.")
+        return None
+        
+
+    if not script_path.endswith(".py"):
+        set_error_message("Error: Provided file is not a Python script.")
+        return None
+        
+    module_name = os.path.splitext(os.path.basename(script_path))[0]
+
+    try:
+        spec = importlib.util.spec_from_file_location(module_name, script_path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+    except SyntaxError as e:
+        set_error_message(f"Error: Syntax error in the script - {e}")
+        return None
+    except ImportError as e:
+        set_error_message(f"Error: Import issue in the script - {e}")
+        return None
+    except Exception as e:
+        set_error_message(f"Error: Failed to import script - {e}")
+        return None
+
+    
+    if not hasattr(module, function_name):
+        set_error_message(f"Error: Function '{function_name}' not found in '{script_path}'.")
+        return None
+    
+    function_to_run = getattr(module, function_name)
+
+    try:
+        sig = inspect.signature(function_to_run)
+        params = list(sig.parameters.values())
+        if len(params) != 1 or params[0].kind not in [inspect.Parameter.POSITIONAL_OR_KEYWORD]:
+            set_error_message(f"Error: Function '{function_name}' must have exactly one parameter (filepath).")
+            return None
+    except Exception as e:
+        set_error_message(f"Error: Unable to inspect function signature - {e}")
+        return None
+
+    try:
+        result = function_to_run(filename)
+        print(f"TYPE: {type(result), {type(result[0])}}, RESULT: {result}")
+    except Exception as e:
+        set_error_message(f"Error: Function '{function_name}' raised an error during execution - {e}")
+        return None
+    
+    if not isinstance(result, list):
+        set_error_message(f"Error: Function '{function_name}' must return a list.")
+        return None
+
+    if len(result) != 8:
+        set_error_message(f"Error: Function '{function_name}' must return a list of exactly 8 elements.")
+        return None
+        
+    if not all(isinstance(x, int) and x in [0, 1] for x in result):
+        return f"Error: Function '{function_name}' must return a list of 8 integers, each 0 or 1.", None
+    
+    print(f"Function '{function_name}' executed successfully. Output: {result}")
+    # set_error_message(f"Function '{function_name}' executed successfully.")
+    return result
+   
+
+
+
+def update_leaderboard(username, zip_file):
+    if not zip_file:
+        set_error_message("No file uploaded.")
+        return get_error_message(), None
+
+    zip_path = zip_file.name 
+    extract_path = os.path.join("", username)
+   # if not os.path.exists(extract_path):
+    #    os.makedirs(extract_path)
+        
+    try: 
+        if not os.path.exists(extract_path):
+            os.makedirs(extract_path)
+
+    except OSError:
+        set_error_message("Error creating directory for extraction.")
+        return get_error_message(), None
+
+    try:
+        with zipfile.ZipFile(zip_path, "r") as zip_ref:
+            zip_ref.extractall(extract_path)
+    except zipfile.BadZipFile:
+        return "Invalid ZIP file.", None
+
+    except Exception as e:
+        return f"Error extracting ZIP file: {str(e)}", None
+
+    
+    extracted_files = os.listdir(extract_path)
+    print("EXTRACTED FILES:", extracted_files)
+    
+    req_file = os.path.join(extract_path, "user_reqs.txt")
+    
+    if "user_reqs.txt" not in extracted_files:
+        return "Missing user_reqs.txt in ZIP file.", None
+    try:
+        install_requirements(req_file)
+    except Exception as e:
+        return f"Error installing dependencies: {str(e)}", None
+
+
+    
+    # for file in os.listdir(extract_path):
+    #     if file.endswith(".py"):
+    #         python_script = os.path.join(extract_path, file)
+    #         break
+    python_script = os.path.join(extract_path, "main.py")
+
+    if "main.py" not in extracted_files:
+        return "No Python script (main.py) found in ZIP.", None
+
+    
+    # if not python_script:
+    #     return "No Python script found in ZIP."
+
+    if "main.py" not in extracted_files:
+        return "No Python script (main.py) found in ZIP.", None
+    
+    try:
+        updated_leaderboard = evaluate_model(username, python_script)
+    except Exception as e:
+        print("Error in eval mode:", str(e))
+        return f"Error evaluating model: {str(e)}", None
+
+    # log_submission_request(username, zip_file)
+    return "Submission successful!", updated_leaderboard
+
+
+
+with gr.Blocks() as demo:
+    
+    gr.Markdown("# 🚀 Model Submission & Leaderboard (Hugging Face Spaces)")
+
+    with gr.Row():
+        username_input = gr.Textbox(label="Username")
+        file_input = gr.File(label="Upload Zip File")
+        submit_button = gr.Button("Submit File")
+
+    status_output = gr.Textbox(label="Status", interactive=False)
+    
+    with gr.Row():
+        leaderboard_display = gr.Dataframe(
+        headers=HEADERS,
+        value=fetch_lb, 
+        label="Leaderboard"
+    )
+
+   
+    submit_button.click(fn=update_leaderboard, 
+                        inputs=[username_input, file_input], 
+                        outputs=[status_output, leaderboard_display])
+
+    status_output.change(fn=get_error_message, inputs=[], outputs=status_output)
+
+demo.launch()