|
|
import pathlib |
|
|
import tempfile |
|
|
from typing import BinaryIO, Literal |
|
|
import json |
|
|
|
|
|
import gradio as gr |
|
|
from datasets import load_dataset, Dataset |
|
|
from huggingface_hub import upload_file, hf_hub_download |
|
|
from evaluation import evaluate_problem |
|
|
from datetime import datetime |
|
|
import os |
|
|
from huggingface_hub import HfApi |
|
|
|
|
|
PROBLEM_TYPES = ["geometrical", "simple_to_build", "mhd_stable"] |
|
|
TOKEN = os.environ.get("HF_TOKEN") |
|
|
CACHE_PATH=os.getenv("HF_HOME", ".") |
|
|
API = HfApi(token=TOKEN) |
|
|
submissions_repo = "cgeorgiaw/constellaration-submissions" |
|
|
|
|
|
def submit_boundary( |
|
|
problem_type: Literal["geometrical", "simple_to_build", "mhd_stable"], |
|
|
boundary_file: BinaryIO, |
|
|
) -> str: |
|
|
|
|
|
file_path = boundary_file.name |
|
|
if not file_path: |
|
|
return "Error: Uploaded file object does not have a valid file path." |
|
|
|
|
|
path_obj = pathlib.Path(file_path) |
|
|
timestamp = datetime.utcnow().isoformat() |
|
|
|
|
|
with ( |
|
|
path_obj.open("rb") as f_in, |
|
|
tempfile.NamedTemporaryFile(delete=False, suffix=".json") as tmp_boundary, |
|
|
): |
|
|
file_content = f_in.read() |
|
|
tmp_boundary.write(file_content) |
|
|
tmp_boundary_path = pathlib.Path(tmp_boundary.name) |
|
|
|
|
|
|
|
|
filename = f"{problem_type}/{timestamp.replace(':', '-')}_{problem_type}.json" |
|
|
record = { |
|
|
"submission_time": timestamp, |
|
|
"problem_type": problem_type, |
|
|
"boundary_json": file_content.decode("utf-8"), |
|
|
"evaluated": False, |
|
|
} |
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as tmp: |
|
|
json.dump(record, tmp, indent=2) |
|
|
tmp.flush() |
|
|
tmp_name = tmp.name |
|
|
|
|
|
API.upload_file( |
|
|
path_or_fileobj=tmp_name, |
|
|
path_in_repo=filename, |
|
|
repo_id=submissions_repo, |
|
|
repo_type="dataset", |
|
|
commit_message=f"Add submission for {problem_type} at {timestamp}" |
|
|
) |
|
|
pathlib.Path(tmp_name).unlink() |
|
|
|
|
|
|
|
|
local_path = read_boundary(filename) |
|
|
|
|
|
try: |
|
|
|
|
|
result = evaluate_problem(problem_type, local_path) |
|
|
output = str(result) |
|
|
except Exception as e: |
|
|
output = f"Error during evaluation:\n{e}" |
|
|
finally: |
|
|
tmp_boundary_path.unlink() |
|
|
|
|
|
return output |
|
|
|
|
|
def read_boundary(filename): |
|
|
local_path = hf_hub_download( |
|
|
repo_id=submissions_repo, |
|
|
repo_type="dataset", |
|
|
filename=filename, |
|
|
) |
|
|
return local_path |
|
|
|
|
|
def gradio_interface() -> gr.Blocks: |
|
|
with gr.Blocks() as demo: |
|
|
gr.Markdown( |
|
|
""" |
|
|
# Plasma Boundary Evaluation App |
|
|
Upload your plasma boundary JSON and select the problem type to get your score. |
|
|
""" |
|
|
) |
|
|
with gr.Row(): |
|
|
problem_type = gr.Dropdown( |
|
|
PROBLEM_TYPES, label="Problem Type", value="geometrical" |
|
|
) |
|
|
boundary_file = gr.File(label="Boundary JSON File (.json)") |
|
|
|
|
|
boundary_file |
|
|
output = gr.Textbox(label="Evaluation Result", lines=10) |
|
|
submit_btn = gr.Button("Evaluate") |
|
|
submit_btn.click( |
|
|
submit_boundary, |
|
|
inputs=[problem_type, boundary_file], |
|
|
outputs=output, |
|
|
) |
|
|
return demo |
|
|
|
|
|
|
|
|
if __name__ == "__main__": |
|
|
gradio_interface().launch() |
|
|
|
