File size: 3,376 Bytes
281711d 8efd41a b20af37 281711d 51aaefc 281711d 51aaefc fb695b7 51aaefc b20af37 281711d 8efd41a 281711d 8efd41a 281711d 0c144dc 281711d 0c144dc 8efd41a f3c8f85 8efd41a f3c8f85 b20af37 8efd41a 0c144dc 2a0354e 0c144dc 8efd41a b20af37 8efd41a 0c144dc 8efd41a 0c144dc b20af37 281711d b20af37 281711d 0c144dc 281711d b20af37 281711d 8efd41a 281711d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
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)
# write to dataset
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()
# then do eval
local_path = read_boundary(filename)
try:
# result = evaluate_problem(problem_type, str(tmp_boundary_path))
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()
|