File size: 3,310 Bytes
e7c2ed2 | 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 | #!/usr/bin/env python3
"""Gradio API backend for the hosted blinded human evaluation."""
import json
import os
from pathlib import Path
import gradio as gr
try:
import spaces
except ImportError:
class _LocalSpaces:
@staticmethod
def GPU(*_args, **_kwargs):
return lambda function: function
spaces = _LocalSpaces()
from protocol import (
MAX_REQUEST_BYTES,
public_manifest,
validate_submission,
write_json_atomic,
)
APP_ROOT = Path(__file__).resolve().parent
STUDY_ROOT = APP_ROOT / "study"
RESPONSE_ROOT = Path(os.environ.get("RESPONSE_ROOT", "/data/responses")).resolve()
@spaces.GPU(duration=1)
def zero_gpu_runtime_probe():
"""Register ZeroGPU compatibility without putting study traffic on a GPU."""
return "ready"
def load_study(mode, rater_id):
"""Return only the participant-specific blinded study manifest."""
try:
return public_manifest(STUDY_ROOT, mode, rater_id)
except (KeyError, TypeError, ValueError) as exc:
raise gr.Error(str(exc)) from exc
def save_response(payload):
"""Validate and atomically persist one participant's current progress."""
try:
if isinstance(payload, str):
payload = json.loads(payload)
serialized = json.dumps(payload).encode("utf-8")
if not serialized or len(serialized) > MAX_REQUEST_BYTES:
raise ValueError("Invalid request size")
mode = payload.get("set", "")
rater_id = payload.get("rater_id", "")
manifest = public_manifest(STUDY_ROOT, mode, rater_id)
validate_submission(payload, manifest)
response_path = RESPONSE_ROOT / mode / f"{rater_id}.json"
write_json_atomic(payload, response_path)
return {
"saved": True,
"completed_items": len(payload["responses"]),
"complete": bool(payload.get("complete")),
}
except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
raise gr.Error(str(exc)) from exc
with gr.Blocks(title="Image Composition Human Evaluation API") as demo:
gr.Markdown(
"# Image Composition Human Evaluation API\n"
"This service validates and stores blinded study responses. "
"The participant interface is hosted separately."
)
probe_output = gr.Textbox(visible=False)
probe_button = gr.Button("Runtime probe", visible=False)
probe_button.click(
zero_gpu_runtime_probe,
outputs=probe_output,
api_name=False,
show_progress="hidden",
)
mode_input = gr.Textbox(value="main", visible=False)
rater_input = gr.Textbox(visible=False)
load_output = gr.JSON(visible=False)
load_button = gr.Button("Load study", visible=False)
load_button.click(
load_study,
inputs=[mode_input, rater_input],
outputs=load_output,
api_name="load_study",
show_progress="hidden",
)
payload_input = gr.JSON(visible=False)
save_output = gr.JSON(visible=False)
save_button = gr.Button("Save response", visible=False)
save_button.click(
save_response,
inputs=payload_input,
outputs=save_output,
api_name="save_response",
show_progress="hidden",
)
if __name__ == "__main__":
demo.launch()
|