| |
| """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() |
|
|
