#!/usr/bin/env python3 """ Upload a scene flow challenge submission (Python 3.8+, standard library only). export SCENEFLOW_API_KEY=sfk_... # from your profile page python sceneflow_submit.py submission.zip --method-name "MyMethod" \ --training-data {single,multi} [--project-url URL] [--notes TEXT] [--unsupervised] [--long-range] [--private] [--wait] python sceneflow_submit.py --status # check the key and today's quota The zip is sent in chunks (default 50 MB) with a SHA-256 checksum each. If the upload is interrupted, run the same command again: the server keeps the received chunks for 24 hours and only the missing ones are sent. --long-range publishes the 35-70 m results too (overall 0-70 m ranking); leave it off if your model crops the point cloud (e.g. at 51.2 m). You can change it later on the website. Before uploading, the zip layout ({scene_id}/{timestamp}.feather at the top level) is checked locally so a malformed zip is not uploaded for nothing. """ import argparse import hashlib import json import os import re import sys import time import urllib.error import urllib.request import zipfile from pathlib import Path DEFAULT_SERVER = "https://sceneflow.kth-rpl.se" USER_AGENT = "sceneflow-submit/1.0" STORAGE_WAIT_SECONDS, STORAGE_WAIT_ATTEMPTS = 120, 16 # wait up to ~30 min if the server is full FEATHER_RE = re.compile(r"^[A-Za-z0-9_-]{1,128}/[0-9]{1,32}\.feather$") STATE_DIR = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "sceneflow_submit" class ApiError(Exception): def __init__(self, status, message): super().__init__(message) self.status = status def request(server, key, method, path, body=None, headers=None, timeout=300): data = None hdrs = {"X-API-Key": key, "User-Agent": USER_AGENT, "Accept": "application/json"} if isinstance(body, (dict, list)): data = json.dumps(body).encode() hdrs["Content-Type"] = "application/json" elif body is not None: data = body hdrs["Content-Type"] = "application/octet-stream" hdrs.update(headers or {}) req = urllib.request.Request(server.rstrip("/") + path, data=data, method=method, headers=hdrs) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read() return json.loads(raw) if raw else {} except urllib.error.HTTPError as e: raw = e.read() try: detail = json.loads(raw).get("detail", raw.decode(errors="replace")) except ValueError: detail = raw.decode(errors="replace")[:300] or e.reason raise ApiError(e.code, f"HTTP {e.code}: {detail}") from None def retrying(fn, what, attempts=6): for attempt in range(1, attempts + 1): try: return fn() except ApiError as e: if e.status < 500 and e.status not in (408, 429) or attempt == attempts: raise err = e except (urllib.error.URLError, OSError) as e: # network hiccup, timeout, reset if attempt == attempts: raise err = e wait = min(60, 3 * 2 ** (attempt - 1)) print(f"\n {what} failed ({err}); retry {attempt}/{attempts - 1} in {wait}s", file=sys.stderr) time.sleep(wait) def check_zip(path): try: with zipfile.ZipFile(path) as zf: names = zf.namelist() except zipfile.BadZipFile: sys.exit(f"error: {path} is not a valid zip file") feathers = [n for n in names if n.endswith(".feather")] good = [n for n in feathers if FEATHER_RE.match(n)] if not good: if feathers: top = feathers[0].rsplit("/", 2)[0] sys.exit(f"error: the .feather files are inside an extra folder ({top}/). Zip the contents of the " "submission folder instead:\n cd submission && zip -r -0 ../submission.zip .") sys.exit("error: no {scene_id}/{timestamp}.feather files in the zip") scenes = {n.split("/", 1)[0] for n in good} print(f"zip check: {len(good)} frames in {len(scenes)} scenes" + (f" ({len(feathers) - len(good)} nested .feather files will be ignored)" if len(good) < len(feathers) else "")) def state_file(server, path): st = path.stat() ident = f"{server}|{path.resolve()}|{st.st_size}|{int(st.st_mtime)}" return STATE_DIR / (hashlib.sha256(ident.encode()).hexdigest()[:24] + ".json") def fmt_bytes(n): return f"{n / 1024**3:.2f} GB" if n >= 1024**3 else f"{n / 1024**2:.1f} MB" def show_scores(sub): s = sub.get("scores") or {} labels = [("[0-35] mean Dynamic", "mean Dynamic 0-35 m"), ("[35-70] mean Dynamic", "mean Dynamic 35-70 m"), ("mean Dynamic", "mean Dynamic 0-70 m"), ("[0-35] mean Static", "mean Static 0-35 m"), ("[0-35] Threeway EPE", "Threeway EPE 0-35 m")] for k, label in labels: if s.get(k) is not None: print(f" {label:<22s} {s[k]:.4f}") if "show_long_range" in sub: print(" 35-70 m results are " + ("published (overall 0-70 m ranking)" if sub["show_long_range"] else "only visible to you (change on the website)")) def wait_for(server, key, sub_id): print("waiting for the evaluation (Ctrl+C stops waiting; the evaluation continues on the server)...") last = None while True: sub = retrying(lambda: request(server, key, "GET", f"/api/v1/submissions/{sub_id}"), "status check") state = sub["status"] + (f" (queue position {sub['queue_position']})" if sub.get("queue_position") else "") if state != last: print(f" [{time.strftime('%H:%M:%S')}] {state}") last = state if sub["status"] in ("done", "failed", "rejected", "deleted"): if sub.get("message"): print(f" {sub['message']}") if sub["status"] == "done": show_scores(sub) print(f"details: {sub['url']}") return 0 if sub["status"] == "done" else 1 time.sleep(30) def main(): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("zip", nargs="?", type=Path, help="submission.zip") p.add_argument("--method-name", help="name shown on the leaderboard (max 80 characters)") p.add_argument("--project-url", default="", help="paper or code URL") p.add_argument("--notes", default="", help="training data, backbone, test-time optimization, ...") p.add_argument("--unsupervised", action="store_true", help="unsupervised track: no human labels of any real-world dataset were used (self-supervised " "losses, synthetic data with simulated labels and foundation models are allowed)") p.add_argument("--training-data", choices=("single", "multi"), help="single: trained on one dataset only (e.g. only AV2); multi: trained on several datasets " "(list them in --notes). Shown as a label on the leaderboard.") p.add_argument("--long-range", action="store_true", help="publish the 35-70 m results and enter the overall 0-70 m ranking " "(leave off if your model crops the point cloud, e.g. at 51.2 m; changeable on the website)") p.add_argument("--private", action="store_true", help="keep the result off the public leaderboard") p.add_argument("--wait", action="store_true", help="wait for the evaluation and print the scores") p.add_argument("--status", action="store_true", help="only check the API key and today's quota") p.add_argument("--server", default=os.environ.get("SCENEFLOW_SERVER", DEFAULT_SERVER)) p.add_argument("--api-key", default=os.environ.get("SCENEFLOW_API_KEY"), help="default: $SCENEFLOW_API_KEY (prefer the variable: arguments end up in shell history)") args = p.parse_args() if not args.api_key: sys.exit("error: set SCENEFLOW_API_KEY (create a key on your profile page)") server, key = args.server, args.api_key me = request(server, key, "GET", "/api/v1/me") q = me["quota"] print(f"team: {me['team']} · {q['used']} of {q['limit']} submissions used in the last 24 h") if args.status: return 0 if args.zip is None or not args.method_name: p.error("the zip file and --method-name are required") if not args.zip.is_file(): sys.exit(f"error: {args.zip} not found") check_zip(args.zip) size = args.zip.stat().st_size sf = state_file(server, args.zip) up, received = None, set() if sf.exists(): try: saved = json.loads(sf.read_text())["upload_id"] st = request(server, key, "GET", f"/api/v1/uploads/{saved}") if st["status"] == "open": up, received = st, set(st["received"]) print(f"resuming upload {saved}: {len(received)}/{st['n_chunks']} chunks already on the server") except (ApiError, ValueError, KeyError): pass if up is None: sf.unlink(missing_ok=True) if up is None and not args.training_data: print("note: --training-data single|multi not given; you can set it later on the submission page", file=sys.stderr) if up is None: for attempt in range(1, STORAGE_WAIT_ATTEMPTS + 1): try: up = request(server, key, "POST", "/api/v1/uploads", { "filename": args.zip.name, "size": size, "method_name": args.method_name, "project_url": args.project_url, "notes": args.notes, "is_private": args.private, "is_unsupervised": args.unsupervised, "show_long_range": args.long_range, "training_data": args.training_data or ""}) break except ApiError as e: # 507: server storage busy (e.g. many uploads right before a deadline) if e.status != 507 or attempt == STORAGE_WAIT_ATTEMPTS: raise print(f"{e}\n waiting {STORAGE_WAIT_SECONDS}s before trying again " f"({attempt}/{STORAGE_WAIT_ATTEMPTS - 1}; Ctrl+C to stop)", file=sys.stderr) time.sleep(STORAGE_WAIT_SECONDS) STATE_DIR.mkdir(parents=True, exist_ok=True) sf.write_text(json.dumps({"upload_id": up["upload_id"]})) upload_id, chunk, n = up["upload_id"], up["chunk_size"], up["n_chunks"] todo = [i for i in range(n) if i not in received] print(f"uploading {fmt_bytes(size)} in {n} chunks of {fmt_bytes(chunk)} ({len(todo)} to send)") t0, sent = time.time(), 0 with open(args.zip, "rb") as f: for k, i in enumerate(todo, 1): f.seek(i * chunk) data = f.read(chunk) digest = hashlib.sha256(data).hexdigest() retrying(lambda: request(server, key, "PUT", f"/api/v1/uploads/{upload_id}/chunks/{i}", data, {"X-Chunk-SHA256": digest}), f"chunk {i}") sent += len(data) rate = sent / max(time.time() - t0, 1e-3) eta = (len(todo) - k) * chunk / max(rate, 1) print(f"\r {n - len(todo) + k}/{n} chunks · {rate / 1024**2:.1f} MB/s · ~{eta / 60:.0f} min left ", end="", flush=True) print() sub = retrying(lambda: request(server, key, "POST", f"/api/v1/uploads/{upload_id}/complete"), "complete") sf.unlink(missing_ok=True) print(f"submission #{sub['submission_id']} queued: {sub['url']}") return wait_for(server, key, sub["submission_id"]) if args.wait else 0 if __name__ == "__main__": try: sys.exit(main()) except ApiError as e: sys.exit(f"error: {e}") except KeyboardInterrupt: sys.exit("\ninterrupted; run the same command again to resume the upload")