|
|
| """The submission ledger, kept in a private Hugging Face dataset.
|
|
|
| A Space's container filesystem is ephemeral. Every rebuild, restart or sleep wipes it,
|
| and with it every submission and every score. That is survivable for a demo and not
|
| survivable for a contest with a prize attached, so the ledger lives outside the container
|
| in a dataset repo that the entrant's work outlives the service.
|
|
|
| Layout - one file per record, never appended to:
|
|
|
| submissions/<id>.json written by the web service when an entry is accepted
|
| results/<id>.json written by the GPU worker when it finishes scoring
|
| leaderboard.json rolled up by the worker so the page reads one file
|
|
|
| One file per record is what makes concurrent writers safe. Two entrants submitting at the
|
| same moment touch different paths, so neither commit can clobber the other - which an
|
| append to a shared JSONL absolutely would.
|
|
|
| The rollup exists because a page load must not fan out into one request per entry. The
|
| worker already holds every score at the moment it writes one, so it is the natural place
|
| to rebuild the table.
|
| """
|
| import base64
|
| import json
|
| import os
|
| import threading
|
| import time
|
| import urllib.error
|
| import urllib.request
|
|
|
| REPO = os.environ.get("OMC_DATASET", "FINAL-Bench/omc-submissions")
|
| TOKEN = os.environ.get("HF_TOKEN", "")
|
| API = "https://huggingface.co/api/datasets/%s" % REPO
|
| RESOLVE = "https://huggingface.co/datasets/%s/resolve/main" % REPO
|
|
|
|
|
| def _hdr(extra=None):
|
| h = {"User-Agent": "VIDRAFT-OMC/1.0"}
|
| if TOKEN:
|
| h["Authorization"] = "Bearer " + TOKEN
|
| if extra:
|
| h.update(extra)
|
| return h
|
|
|
|
|
| def _get_json(url, timeout=30):
|
| req = urllib.request.Request(url, headers=_hdr())
|
| with urllib.request.urlopen(req, timeout=timeout) as r:
|
| return json.loads(r.read().decode())
|
|
|
|
|
| def read(path, default=None):
|
| """Fetch one record. Missing files are a normal state, not an error."""
|
| try:
|
| return _get_json("%s/%s" % (RESOLVE, path))
|
| except urllib.error.HTTPError as e:
|
| if e.code in (404, 401, 403):
|
| return default
|
| raise
|
| except Exception:
|
| return default
|
|
|
|
|
| def write(path, obj, summary=None):
|
| """Commit one record. NDJSON is the format this endpoint takes - a plain JSON body
|
| is accepted and then quietly does nothing, which is a long way to debug."""
|
| blob = base64.b64encode(json.dumps(obj, ensure_ascii=False).encode()).decode()
|
| lines = [
|
| json.dumps({"key": "header", "value": {"summary": summary or ("write " + path)}}),
|
| json.dumps({"key": "file", "value": {"path": path, "content": blob,
|
| "encoding": "base64"}}),
|
| ]
|
| body = ("\n".join(lines) + "\n").encode()
|
| req = urllib.request.Request(API + "/commit/main", data=body,
|
| headers=_hdr({"Content-Type": "application/x-ndjson"}))
|
| with urllib.request.urlopen(req, timeout=60) as r:
|
| return json.loads(r.read().decode())
|
|
|
|
|
| MAX_PAGES = int(os.environ.get("OMC_MAX_PAGES", "200"))
|
|
|
|
|
| def _tree_pages(prefix):
|
| """Every page of a tree listing, following the Link cursor.
|
|
|
| The Hub caps a page at 1,000 entries. Reading only the first page was silent while
|
| the dataset was small and started dropping records the moment it was not."""
|
| url = "%s/tree/main/%s" % (API, prefix)
|
| seen = 0
|
| for _ in range(MAX_PAGES):
|
| req = urllib.request.Request(url, headers=_hdr())
|
| with urllib.request.urlopen(req, timeout=60) as r:
|
| page = json.loads(r.read().decode())
|
| link = r.headers.get("Link") or ""
|
| yield page
|
| seen += len(page)
|
| nxt = ""
|
| for part in link.split(","):
|
| if 'rel="next"' in part and "<" in part:
|
| nxt = part[part.index("<") + 1:part.index(">")]
|
| if not nxt:
|
| return
|
| url = nxt
|
| raise RuntimeError("tree listing exceeded %d pages at %s (%d entries)"
|
| % (MAX_PAGES, prefix, seen))
|
|
|
|
|
| def listdir(prefix):
|
| """Record ids under a prefix. Absent directory means nothing has been written yet."""
|
| out = []
|
| try:
|
| for page in _tree_pages(prefix):
|
| for e in page:
|
| q = e.get("path", "")
|
| if q.endswith(".json") and not os.path.basename(q).startswith("_"):
|
| out.append(os.path.basename(q)[:-5])
|
| except urllib.error.HTTPError as e:
|
| if e.code == 404:
|
| return []
|
| raise
|
| except Exception:
|
|
|
|
|
| raise
|
| return out
|
|
|
|
|
| class Cached:
|
| """A small TTL cache so a page refresh does not become a round trip to the Hub.
|
|
|
| Staleness is bounded and harmless here: the worst case is a leaderboard a few seconds
|
| behind, and the page polls anyway.
|
|
|
| **Single flight.** One refresher at a time; everyone else is served the value already
|
| held, so an expiry does not send every concurrent request to the Hub at once.
|
|
|
| **Stale beats blocking, and stale beats an error.** A leaderboard a minute old is a
|
| working page. A timeout is not.
|
| """
|
|
|
| def __init__(self, ttl=20):
|
| self.ttl = ttl
|
| self._v = {}
|
| self._locks = {}
|
| self._guard = threading.Lock()
|
|
|
| def _lock_for(self, key):
|
| with self._guard:
|
| lk = self._locks.get(key)
|
| if lk is None:
|
| lk = self._locks[key] = threading.Lock()
|
| return lk
|
|
|
| def get(self, key, produce):
|
| hit = self._v.get(key)
|
| if hit and time.time() - hit[0] < self.ttl:
|
| return hit[1]
|
|
|
| lk = self._lock_for(key)
|
|
|
|
|
| if not lk.acquire(blocking=(hit is None)):
|
| return hit[1]
|
| try:
|
| hit = self._v.get(key)
|
| if hit and time.time() - hit[0] < self.ttl:
|
| return hit[1]
|
| try:
|
| val = produce()
|
| except Exception:
|
| if hit:
|
| return hit[1]
|
| raise
|
| self._v[key] = (time.time(), val)
|
| return val
|
| finally:
|
| lk.release()
|
|
|
| def drop(self, key=None):
|
| if key is None:
|
| self._v.clear()
|
| else:
|
| self._v.pop(key, None)
|
|
|
