| """Minimal Gradio REST helper (no gradio_client dependency). |
| Implements: upload file, call endpoint, poll SSE result.""" |
| import os, json, pathlib, requests |
|
|
| def token(): |
| return (os.environ.get("HF_TOKEN") |
| or pathlib.Path("~/.cache/huggingface/token").expanduser().read_text().strip()) |
|
|
| def _headers(): |
| return {"Authorization": f"Bearer {token()}"} |
|
|
| def upload(space_root, filepath): |
| """Upload a local file, return the gradio FileData dict to reference it.""" |
| with open(filepath, "rb") as f: |
| r = requests.post(f"{space_root}/gradio_api/upload", |
| headers=_headers(), |
| files={"files": (pathlib.Path(filepath).name, f)}, |
| timeout=300) |
| r.raise_for_status() |
| server_path = r.json()[0] |
| return {"path": server_path, "meta": {"_type": "gradio.FileData"}, |
| "orig_name": pathlib.Path(filepath).name, "url": None, |
| "size": None, "mime_type": None, "is_stream": False} |
|
|
| def call(space_root, api_name, data, timeout=900): |
| """Call /gradio_api/call/{api_name} with ordered `data` list; poll SSE; return data list.""" |
| r = requests.post(f"{space_root}/gradio_api/call/{api_name}", |
| headers={**_headers(), "Content-Type": "application/json"}, |
| data=json.dumps({"data": data}), timeout=60) |
| r.raise_for_status() |
| event_id = r.json()["event_id"] |
| |
| with requests.get(f"{space_root}/gradio_api/call/{api_name}/{event_id}", |
| headers=_headers(), stream=True, timeout=timeout) as s: |
| s.raise_for_status() |
| event = None |
| for raw in s.iter_lines(decode_unicode=True): |
| if raw is None or raw == "": |
| continue |
| if raw.startswith("event:"): |
| event = raw.split(":", 1)[1].strip() |
| elif raw.startswith("data:"): |
| payload = raw.split(":", 1)[1].strip() |
| if event == "complete": |
| return json.loads(payload) |
| if event == "error": |
| raise RuntimeError(f"Gradio error: {payload}") |
| raise RuntimeError("SSE stream ended without a 'complete' event") |
|
|
| def download(url_or_data, dest, space_root=None): |
| """Download a gradio result (url string or FileData dict) to dest.""" |
| if isinstance(url_or_data, dict): |
| url = url_or_data.get("url") |
| if not url and url_or_data.get("path") and space_root: |
| url = f"{space_root}/gradio_api/file={url_or_data['path']}" |
| else: |
| url = url_or_data |
| if url and url.startswith("/") and space_root: |
| url = space_root + url |
| r = requests.get(url, headers=_headers(), stream=True, timeout=600) |
| r.raise_for_status() |
| with open(dest, "wb") as f: |
| for chunk in r.iter_content(chunk_size=1 << 16): |
| f.write(chunk) |
| return dest |
|
|
