Spaces:
Running on Zero
Running on Zero
| #!/usr/bin/env python3 | |
| """Gradio API backend for the hosted study (deployed as its own Hugging Face Space). | |
| Two public operations, both called by the static participant interface: | |
| start_session(participant_id, linked_to) -> booklet assignment, trial order, saved progress | |
| save_session(payload) -> validates and stores one participant's session | |
| There is no operation that lists or aggregates responses. Responses are written to | |
| RESPONSE_ROOT, which must be persistent storage (a private Storage Bucket mounted at | |
| /data on the Space). | |
| """ | |
| import os | |
| from pathlib import Path | |
| import gradio as gr | |
| try: | |
| import spaces | |
| except ImportError: # Local runs and CPU Spaces. | |
| class _Spaces: | |
| def GPU(*_args, **_kwargs): | |
| return lambda function: function | |
| spaces = _Spaces() | |
| from protocol import StudyService, read_json | |
| APP_ROOT = Path(__file__).resolve().parent | |
| MANIFEST = read_json(APP_ROOT / "study" / "manifest.json") | |
| RESPONSE_ROOT = Path(os.environ.get("RESPONSE_ROOT", "/data/responses")) | |
| SERVICE = StudyService(MANIFEST, RESPONSE_ROOT) | |
| def check_storage(): | |
| """On a Space, refuse to work unless responses land on the mounted bucket. | |
| Without the mount, files would go to the container's own disk and vanish at the | |
| next restart, so it is better that nobody can start. | |
| """ | |
| folders = [RESPONSE_ROOT, *RESPONSE_ROOT.parents[:-1]] | |
| if os.environ.get("SPACE_ID") and not any(os.path.ismount(folder) for folder in folders): | |
| raise gr.Error("Saving is not available right now. Please try again later.") | |
| def zero_gpu_runtime_probe(): | |
| """Registers ZeroGPU compatibility; study traffic never touches a GPU.""" | |
| return "ready" | |
| def start_session(participant_id, linked_to): | |
| check_storage() | |
| try: | |
| return SERVICE.start_session(participant_id, linked_to or None) | |
| except (KeyError, TypeError, ValueError) as error: | |
| raise gr.Error(str(error)) from error | |
| def save_session(payload): | |
| check_storage() | |
| try: | |
| return SERVICE.save_session(payload) | |
| except (KeyError, TypeError, ValueError) as error: | |
| raise gr.Error(str(error)) from error | |
| with gr.Blocks(title="Study API") as demo: | |
| gr.Markdown( | |
| f"### Study API\nStores responses for `{MANIFEST['study_id']}` " | |
| f"(build `{MANIFEST['build_id']}`). The participant interface is hosted separately." | |
| ) | |
| probe_out = gr.Textbox(visible=False) | |
| gr.Button("probe", visible=False).click( | |
| zero_gpu_runtime_probe, outputs=probe_out, api_name=False, show_progress="hidden" | |
| ) | |
| participant = gr.Textbox(visible=False) | |
| linked = gr.Textbox(visible=False) | |
| session_out = gr.JSON(visible=False) | |
| gr.Button("start", visible=False).click( | |
| start_session, inputs=[participant, linked], outputs=session_out, | |
| api_name="start_session", show_progress="hidden", | |
| ) | |
| payload = gr.JSON(visible=False) | |
| saved_out = gr.JSON(visible=False) | |
| gr.Button("save", visible=False).click( | |
| save_session, inputs=payload, outputs=saved_out, | |
| api_name="save_session", show_progress="hidden", | |
| ) | |
| if __name__ == "__main__": | |
| demo.queue(default_concurrency_limit=1).launch() | |