#!/usr/bin/env python3 """FaceNinja from the command line, for people and AI agents: photos in, MetaHuman head out. Standard library only, Python 3.9+, Linux, macOS and Windows (`python faceninja.py ...`). faceninja.py login [--no-wait] sign in: approve a code in the browser, a new API key is saved faceninja.py whoami | balance who is signed in, credits left faceninja.py make PHOTO... [--name N] [--formats glb,zip] [--out DIR] [--no-wait] [--max-wait S] faceninja.py status ID [--wait] [--max-wait S] faceninja.py download ID [--formats glb,fbx,zip] [--out DIR] faceninja.py batch DIR [--per-person-subfolders] [--yes] ... faceninja.py list | logout Add --json for one machine-readable JSON object on stdout (progress goes to stderr). The API key comes from FACENINJA_API_KEY, else from the credentials file `login` writes (~/.config/faceninja/credentials.json, %APPDATA%\\FaceNinja\\credentials.json on Windows). Keys are never printed. Exit codes: 0 ok, 1 error, 2 bad usage, 3 needs the user's confirmation (--yes), 4 not signed in or sign-in refused, 5 not enough credits, 6 the head failed (credits returned), 75 still running or still waiting (run the command again). """ import argparse import hashlib import json import mimetypes import os import platform import re import socket import sys import time import urllib.error import urllib.parse import urllib.request import uuid import webbrowser from pathlib import Path __version__ = "1.0.0" DEFAULT_BASE = "https://faceninja.net" MAX_PHOTOS = 15 MAX_PHOTO_BYTES = 25 * 1024 * 1024 BATCH_CONFIRM_ABOVE = 5 FORMATS = {"glb": "glb", "fbx": "fbx", "obj": "obj", "stl": "stl", "ply": "ply", "zip": "metahuman_zip", "metahuman_zip": "metahuman_zip"} ALL_FORMATS = ["glb", "fbx", "obj", "stl", "ply", "metahuman_zip"] IMAGE_TYPES = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp", ".heic": "image/heic", ".heif": "image/heif", ".avif": "image/avif"} FINAL = ("completed", "failed", "deleted") STATE_FILE = ".faceninja.json" EXIT_OK, EXIT_ERROR, EXIT_USAGE, EXIT_CONFIRM, EXIT_AUTH, EXIT_CREDITS, EXIT_FAILED, EXIT_PENDING = 0, 1, 2, 3, 4, 5, 6, 75 class FNError(Exception): def __init__(self, message, code="error", hint=None, exit_code=EXIT_ERROR, status=None, request_id=None, data=None): super().__init__(message) self.message, self.code, self.hint, self.exit_code = message, code, hint, exit_code self.status, self.request_id, self.data = status, request_id, data # ---- output ---------------------------------------------------------------------------------------------------------- class Out: """Human lines on stdout, or (with --json) progress on stderr and one JSON object on stdout.""" def __init__(self, as_json): self.json = as_json def say(self, *parts): print(" ".join(str(p) for p in parts), file=sys.stderr if self.json else sys.stdout, flush=True) def result(self, data, human=None): if self.json: print(json.dumps(data, indent=2), flush=True) elif human: print(human, flush=True) def num(n): return "{:,}".format(int(n or 0)) def usd(credits, per_usd=100): return "${:,.2f}".format((credits or 0) / float(per_usd or 100)) def key_prefix(key): return (key or "")[:12] + "..." # ---- configuration and credentials ----------------------------------------------------------------------------- def config_dir(): if os.environ.get("FACENINJA_CONFIG_DIR"): return Path(os.environ["FACENINJA_CONFIG_DIR"]) if os.name == "nt": return Path(os.environ.get("APPDATA") or Path.home() / "AppData" / "Roaming") / "FaceNinja" return Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "faceninja" def write_private(path, data): """JSON readable by the owner only (0600); the folder is made 0700 if we create it.""" if not path.parent.exists(): path.parent.mkdir(parents=True) try: os.chmod(path.parent, 0o700) except OSError: pass tmp = path.with_name(path.name + ".tmp") fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) try: os.chmod(str(tmp), 0o600) except OSError: pass os.replace(str(tmp), str(path)) def read_json(path): try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return None def norm_base(base): return (base or DEFAULT_BASE).strip().rstrip("/") def load_credentials(base): """(key, source): the environment first, then the credentials file if it was made for this server.""" env = (os.environ.get("FACENINJA_API_KEY") or "").strip() if env: return env, "FACENINJA_API_KEY" path = config_dir() / "credentials.json" c = read_json(path) if c and c.get("api_key") and norm_base(c.get("base_url")) == base: return c["api_key"], str(path) return None, None def detect_client(): e = os.environ if e.get("FACENINJA_CLIENT"): agent = e["FACENINJA_CLIENT"] elif e.get("CLAUDECODE") or e.get("CLAUDE_CODE_ENTRYPOINT"): agent = "Claude Code" elif any(k.startswith("CODEX_") for k in e): agent = "Codex" elif e.get("CURSOR_AGENT") or e.get("CURSOR_TRACE_ID"): agent = "Cursor" elif e.get("GEMINI_CLI"): agent = "Gemini CLI" elif any(k.startswith("OPENCLAW") for k in e): agent = "OpenClaw" elif any(k.startswith("AIDER_") for k in e): agent = "Aider" else: agent = "FaceNinja CLI" host = (socket.gethostname() or "").split(".")[0][:40] or None return {"agent": agent, "host": host, "os": platform.system() or None, "version": __version__} # ---- HTTP ------------------------------------------------------------------------------------------------------------ class Api: def __init__(self, base, key=None, out=None): self.base, self.key, self.out = base, key, out self.ua = "faceninja-skill/{} (Python {}; {})".format(__version__, platform.python_version(), platform.system()) def call(self, method, path, body=None, raw=None, content_type=None, headers=None, auth=True, timeout=60, retries=4, ok=(200, 201, 202)): """Returns (status, data, headers). Retries rate limits and transient failures; raises FNError otherwise.""" url = path if path.startswith("http") else self.base + path h = {"Accept": "application/json", "User-Agent": self.ua} h.update(headers or {}) data = None if body is not None: data = json.dumps(body).encode("utf-8") h["Content-Type"] = "application/json" elif raw is not None: data = raw h["Content-Type"] = content_type or "application/octet-stream" if auth: if not self.key: raise not_signed_in(self.base) h["Authorization"] = "Bearer " + self.key attempt = 0 while True: attempt += 1 req = urllib.request.Request(url, data=data, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=timeout) as res: text = res.read().decode("utf-8", "replace") return res.status, (json.loads(text) if text.strip() else None), res.headers except urllib.error.HTTPError as e: text = e.read().decode("utf-8", "replace") try: payload = json.loads(text) if text.strip() else None except ValueError: payload = None if e.code in ok: return e.code, payload, e.headers err = (payload or {}).get("error") if isinstance(payload, dict) else None err = err if isinstance(err, dict) else {} retry_after = _int(e.headers.get("Retry-After"), 0) if attempt <= retries and (e.code == 429 or e.code in (500, 502, 503, 504)): wait = min(max(retry_after, 2 ** attempt), 60) if self.out: self.out.say(" server answered {} ({}); retrying in {} s".format(e.code, err.get("code") or "busy", wait)) time.sleep(wait) continue raise FNError(err.get("message") or "HTTP {} from {}".format(e.code, path), code=err.get("code") or "http_{}".format(e.code), hint=err.get("hint"), status=e.code, request_id=err.get("request_id"), data={"retry_after": retry_after}) except (urllib.error.URLError, socket.timeout, ConnectionError) as e: if attempt <= retries: time.sleep(min(2 ** attempt, 30)) continue raise FNError("Cannot reach {}: {}".format(self.base, getattr(e, "reason", e)), code="network", hint="Check the connection, or FACENINJA_API_BASE if you changed the server.") def download(self, url, dest, expect_bytes=None, expect_sha=None): """GET a signed link (never with the API key) into dest, checking size and SHA-256.""" req = urllib.request.Request(url, headers={"User-Agent": self.ua}) tmp = dest.with_name(dest.name + ".part") h = hashlib.sha256() n = 0 try: with urllib.request.urlopen(req, timeout=300) as res, open(str(tmp), "wb") as f: while True: chunk = res.read(1 << 20) if not chunk: break f.write(chunk) h.update(chunk) n += len(chunk) except (urllib.error.URLError, socket.timeout, ConnectionError) as e: tmp.unlink() if tmp.exists() else None raise FNError("Download failed: {}".format(getattr(e, "reason", e)), code="download_failed", hint="Run download again; links are fresh each time.") if (expect_bytes and n != expect_bytes) or (expect_sha and h.hexdigest() != expect_sha): tmp.unlink() raise FNError("Downloaded file is incomplete or corrupted ({} bytes).".format(n), code="download_corrupt", hint="Run download again.") os.replace(str(tmp), str(dest)) return n, h.hexdigest() def _int(v, d): try: return int(v) except (TypeError, ValueError): return d def not_signed_in(base): return FNError("Not signed in to {}.".format(base), code="not_signed_in", exit_code=EXIT_AUTH, hint="Run: python faceninja.py login (or set FACENINJA_API_KEY)") def api_for(args, out, need_key=True): key, source = load_credentials(args.base) if need_key and not key: raise not_signed_in(args.base) api = Api(args.base, key, out) api.key_source = source return api def account(api): try: return api.call("GET", "/v1/account")[1] except FNError as e: if e.status == 401: raise FNError("The saved API key is not valid any more (revoked?).", code="unauthorized", exit_code=EXIT_AUTH, hint="Run: python faceninja.py login --force") raise def head_price(acct): p = acct.get("prices") or {} return int(p.get("head") or 1500), int(p.get("credits_per_usd") or 100) # ---- login / whoami / balance / logout ------------------------------------------------------------------------------ def cmd_login(args, out): api = Api(args.base, None, out) cdir = config_dir() pending_path = cdir / "pending-login.json" if os.environ.get("FACENINJA_API_KEY"): out.say("Note: FACENINJA_API_KEY is set and wins over the saved sign-in; unset it to use the key saved here.") key, source = load_credentials(args.base) if key and source != "FACENINJA_API_KEY" and not args.force: try: a = account(Api(args.base, key, out)) return out.result({"ok": True, "status": "signed_in", "email": a.get("email"), "key_prefix": key_prefix(key), "credits": a.get("credits")}, "Already signed in to {} as {} (key {}). Use --force to sign in again.".format(args.base, a.get("email"), key_prefix(key))) except FNError as e: if e.exit_code != EXIT_AUTH: raise p = read_json(pending_path) fresh = False if not (p and norm_base(p.get("base_url")) == args.base and p.get("expires_at", 0) > time.time() + 15) or args.force_new: client = detect_client() name = args.client_name or "{}{}".format(client["agent"], " on " + client["host"] if client.get("host") else "") _, d, _ = api.call("POST", "/v1/device/code", body={"client_name": name, "client": client}, auth=False) p = {"base_url": args.base, "device_code": d["device_code"], "user_code": d["user_code"], "verification_uri": d["verification_uri"], "verification_uri_complete": d["verification_uri_complete"], "interval": d.get("interval", 5), "expires_at": time.time() + int(d.get("expires_in", 600)), "client_name": d.get("client_name", name)} write_private(pending_path, p) fresh = True left = int(p["expires_at"] - time.time()) info = {"ok": True, "status": "pending", "user_code": p["user_code"], "verification_uri": p["verification_uri"], "verification_uri_complete": p["verification_uri_complete"], "expires_in": left, "client_name": p["client_name"]} msg = ("To sign in, open {}\nand approve the code {} (it expires in {} min). The page asks you to sign in to FaceNinja first if needed.\n" "The new API key will be named \"{}\".").format(p["verification_uri_complete"], p["user_code"], max(1, left // 60), p["client_name"]) if fresh and not args.no_browser: try: if webbrowser.open(p["verification_uri_complete"]): msg += "\n(Opened the page in your browser.)" except Exception: pass if args.no_wait: return out.result(dict(info, next="Run `faceninja.py login` again to wait for the approval."), msg + "\nWhen it is approved, run: python faceninja.py login") out.say(msg) out.say("Waiting for approval...") interval = int(p.get("interval") or 5) deadline = time.time() + args.max_wait if args.max_wait else p["expires_at"] + 5 while True: if time.time() + interval > deadline: raise FNError("Still waiting for the code {} to be approved.".format(p["user_code"]), code="authorization_pending", exit_code=EXIT_PENDING, hint="Approve it at {}, then run login again.".format(p["verification_uri_complete"])) time.sleep(interval) try: _, tok, _ = api.call("POST", "/v1/device/token", body={"device_code": p["device_code"]}, auth=False, retries=2) break except FNError as e: if e.code == "authorization_pending": continue if e.code == "slow_down": interval = max(interval + 5, (e.data or {}).get("retry_after") or 0) continue if e.code in ("access_denied", "expired_token", "invalid_grant"): pending_path.unlink() if pending_path.exists() else None e.exit_code = EXIT_AUTH e.hint = "Run login again to get a new code." if e.code != "access_denied" else "The sign-in was denied in the browser." raise cred = {"base_url": args.base, "api_key": tok["access_token"], "key_id": tok.get("key_id"), "key_prefix": tok.get("key_prefix"), "key_name": tok.get("key_name"), "email": tok.get("email"), "created_at": tok.get("created_at")} path = cdir / "credentials.json" write_private(path, cred) pending_path.unlink() if pending_path.exists() else None out.result({"ok": True, "status": "signed_in", "email": tok.get("email"), "key_prefix": key_prefix(tok["access_token"]), "key_name": tok.get("key_name"), "credentials_file": str(path)}, "Signed in as {}. API key {} (\"{}\") saved to {}".format(tok.get("email"), key_prefix(tok["access_token"]), tok.get("key_name"), path)) def cmd_whoami(args, out): api = api_for(args, out) a = account(api) price, per = head_price(a) out.result({"ok": True, "email": a.get("email"), "account_id": a.get("id"), "key_prefix": key_prefix(api.key), "key_source": api.key_source, "base_url": args.base, "credits": a.get("credits"), "held_credits": a.get("held_credits"), "head_price": price}, "Signed in to {} as {}\nKey {} from {}\nCredits: {} ({} heads at {} credits each)".format( args.base, a.get("email"), key_prefix(api.key), api.key_source, num(a.get("credits")), (a.get("credits") or 0) // price, num(price))) def cmd_balance(args, out): a = account(api_for(args, out)) price, per = head_price(a) c = int(a.get("credits") or 0) out.result({"ok": True, "credits": c, "held_credits": a.get("held_credits"), "head_price": price, "head_price_usd": price / per, "heads_affordable": c // price, "buy_url": args.base + "/factory#/billing"}, "{} credits ({}): enough for {} head(s). A head costs {} credits ({}).{}\nBuy credits: {}/factory#/billing".format( num(c), usd(c, per), c // price, num(price), usd(price, per), "\n{} credits are held by heads in progress.".format(num(a.get("held_credits"))) if a.get("held_credits") else "", args.base)) def cmd_logout(args, out): cdir = config_dir() path = cdir / "credentials.json" c = read_json(path) revoked = False if c and c.get("api_key") and not args.keep_key: try: Api(norm_base(c.get("base_url")), c["api_key"], out).call("DELETE", "/v1/keys/current", retries=1) revoked = True except FNError as e: out.say("Could not revoke the key on the server ({}); revoke it in Factory > API keys.".format(e.message)) for p in (path, cdir / "pending-login.json"): if p.exists(): p.unlink() note = " FACENINJA_API_KEY is still set in your environment." if os.environ.get("FACENINJA_API_KEY") else "" out.result({"ok": True, "removed": bool(c), "revoked": revoked, "env_key_set": bool(os.environ.get("FACENINJA_API_KEY"))}, ("Signed out: the saved key was {} and deleted.".format("revoked" if revoked else "kept on the server") if c else "No saved sign-in.") + note) # ---- heads ------------------------------------------------------------------------------------------------------------ def parse_formats(s): out = [] for f in (s or "").replace(" ", "").lower().split(","): if not f: continue if f == "all": out.extend(ALL_FORMATS) elif f in FORMATS: out.append(FORMATS[f]) else: raise FNError("Unknown format {!r}. Use glb, fbx, obj, stl, ply, zip (MetaHuman ZIP) or all.".format(f), code="invalid_request", exit_code=EXIT_USAGE) return list(dict.fromkeys(out)) def check_photos(paths): photos = [] for p in paths: p = Path(p).expanduser() if not p.is_file(): raise FNError("No such photo: {}".format(p), code="invalid_request", exit_code=EXIT_USAGE) ext = p.suffix.lower() if ext not in IMAGE_TYPES: raise FNError("{}: use JPEG, PNG, WebP, HEIC/HEIF or AVIF.".format(p.name), code="unsupported_media_type", exit_code=EXIT_USAGE) size = p.stat().st_size if size < 1000 or size > MAX_PHOTO_BYTES: raise FNError("{}: {} bytes; a photo must be between 1 KB and 25 MB.".format(p.name, num(size)), code="invalid_request", exit_code=EXIT_USAGE, hint="Downscale large photos to about 4000 px on the long side.") photos.append(p) if not 1 <= len(photos) <= MAX_PHOTOS: raise FNError("A head takes 1 to {} photos of one person; got {}.".format(MAX_PHOTOS, len(photos)), code="invalid_request", exit_code=EXIT_USAGE) return photos def fingerprint(photos, name): h = hashlib.sha256((name or "").encode("utf-8")) for p in sorted(hashlib.sha256(p.read_bytes()).hexdigest() for p in photos): h.update(p.encode()) return h.hexdigest()[:32] def load_state(out_dir): s = read_json(out_dir / STATE_FILE) or {} s.setdefault("heads", {}) return s def save_state(out_dir, state): out_dir.mkdir(parents=True, exist_ok=True) (out_dir / STATE_FILE).write_text(json.dumps(state, indent=2), encoding="utf-8") def create_head(api, out, photos, name): ids = [] for p in photos: q = urllib.parse.urlencode({"filename": p.name[:120]}) _, ph, _ = api.call("POST", "/v1/photos?" + q, raw=p.read_bytes(), content_type=IMAGE_TYPES[p.suffix.lower()], timeout=180) ids.append(ph["id"]) out.say(" uploaded {} -> {}".format(p.name, ph["id"])) body = {"photo_ids": ids} if name: body["name"] = name[:120] try: _, head, _ = api.call("POST", "/v1/heads", body=body, headers={"Idempotency-Key": str(uuid.uuid4())}) except FNError as e: if e.code == "insufficient_credits": e.exit_code = EXIT_CREDITS e.hint = "Buy credits at {}/factory#/billing".format(api.base) raise return head def describe(h): q = h.get("queue") or {} bits = [h.get("status")] if h.get("stage"): bits.append(h["stage"]) if h.get("progress") is not None and h.get("status") == "processing": bits.append("{}%".format(h["progress"])) if h.get("status") == "queued" and q.get("position"): bits.append("position {} in line".format(q["position"])) if q.get("eta_s"): bits.append("about {} min left".format(max(1, int(q["eta_s"]) // 60))) if q.get("builder") == "offline": bits.append("builder offline: it waits in line") if h.get("message"): bits.append(h["message"]) return ", ".join(str(b) for b in bits if b) def wait_head(api, out, head_id, max_wait=0): """Long-poll until the head is final. Raises EXIT_PENDING when max_wait seconds pass first.""" start = time.time() last = None while True: left = max_wait - (time.time() - start) if max_wait else 60 if max_wait and left <= 0: raise FNError("Head {} is still running ({}).".format(head_id, last), code="still_running", exit_code=EXIT_PENDING, hint="Run: python faceninja.py status {} --wait".format(head_id), data={"id": head_id}) _, h, _ = api.call("GET", "/v1/heads/{}?wait={}".format(head_id, int(max(1, min(60, left)))), timeout=95) d = describe(h) if d != last: out.say(" {} {}".format(time.strftime("%H:%M:%S"), d)) last = d if h.get("status") in FINAL: return h def head_failure(h, base): err = h.get("error") or {} code = err.get("code") or h.get("status") hint = { "no_face": "Use a sharp photo where one face is clearly visible.", "multiple_people": "Crop the photos to one person.", "not_same_person": "Every photo must show the same person.", "unreadable_photo": "Re-save the photo as JPEG or PNG.", "content_rejected": "The photo was refused by the content check (people under 18, nudity, or not a real photo).", }.get(code, "See the photo guide: {}/guidelines".format(base)) credits = (h.get("credits") or {}).get("state") msg = "Head {} {}: {}{}".format(h.get("id"), h.get("status"), err.get("message") or code, " Credits returned." if credits == "released" and "returned" not in (err.get("message") or "") else "") return FNError(msg, code=code, hint=hint, exit_code=EXIT_FAILED, data={"head": h}) def safe_name(s): s = re.sub(r"[^\w.-]+", "_", s or "").strip("._") return s[:120] or "file" def download_files(api, out, head_id, formats, out_dir, arkit=None): out_dir.mkdir(parents=True, exist_ok=True) saved, skipped = [], [] for fmt in formats: body = {"format": fmt} if arkit is not None and fmt in ("glb", "fbx"): body["arkit"] = arkit try: _, link, _ = api.call("POST", "/v1/heads/{}/files".format(head_id), body=body) except FNError as e: if e.code == "head_not_ready": e.exit_code, e.hint = EXIT_PENDING, "Wait first: python faceninja.py status {} --wait".format(head_id) raise if e.code in ("variant_not_available", "field_not_applicable"): out.say(" skipped {}: {}".format(fmt, e.message)) skipped.append({"format": fmt, "reason": e.message}) continue raise dest = out_dir / safe_name(link.get("filename") or "{}.{}".format(head_id, fmt)) n, sha = api.download(link["url"], dest, link.get("bytes"), link.get("sha256")) out.say(" saved {} ({:.1f} MB)".format(dest, n / 1e6)) saved.append({"format": fmt, "arkit": link.get("arkit"), "path": str(dest), "bytes": n, "sha256": sha}) return saved, skipped def quote(api, out, count): a = account(api) price, per = head_price(a) credits = int(a.get("credits") or 0) total = price * count out.say("Cost: {} head(s) x {} credits = {} credits ({}). Balance: {} credits ({}).".format( count, num(price), num(total), usd(total, per), num(credits), usd(credits, per))) return {"heads": count, "credits_per_head": price, "total_credits": total, "total_usd": total / per, "balance": credits, "balance_after": credits - total, "sufficient": credits >= total} def cmd_make(args, out): photos = check_photos(args.photos) formats = parse_formats(args.formats) out_dir = Path(args.out).expanduser().resolve() api = api_for(args, out) name = args.name or photos[0].stem state = load_state(out_dir) fp = fingerprint(photos, name) head = None prev = state["heads"].get(fp) if prev and not args.new: try: _, head, _ = api.call("GET", "/v1/heads/{}".format(prev["id"])) if head.get("status") in ("failed", "deleted"): head = None else: out.say("Resuming head {} made from these photos earlier ({}). Use --new for another head.".format(head["id"], head["status"])) except FNError as e: if e.status != 404: raise head = None q = None if head is None: q = quote(api, out, 1) if args.dry_run: return out.result(dict(ok=True, dry_run=True, quote=q, photos=[str(p) for p in photos], name=name, formats=formats), "Dry run: nothing uploaded or spent.") if not q["sufficient"]: raise FNError("Not enough credits: a head costs {} credits, the balance is {}.".format(num(q["credits_per_head"]), num(q["balance"])), code="insufficient_credits", exit_code=EXIT_CREDITS, hint="Buy credits at {}/factory#/billing".format(args.base)) head = create_head(api, out, photos, name) state["heads"][fp] = {"id": head["id"], "name": name, "photos": [str(p) for p in photos], "created_at": head.get("created_at")} save_state(out_dir, state) out.say("Created head {} ({}). {} credits held.".format(head["id"], describe(head), num(q["credits_per_head"]))) result = {"ok": True, "id": head["id"], "name": name, "status": head.get("status"), "quote": q, "out": str(out_dir)} if not args.wait: return out.result(dict(result, next="python faceninja.py status {} --wait".format(head["id"])), "Head {} is {}. Follow it: python faceninja.py status {} --wait".format(head["id"], head.get("status"), head["id"])) if head.get("status") not in FINAL: out.say("Waiting for {} (usually 5 to 15 minutes; longer if the builder has a queue)...".format(head["id"])) head = wait_head(api, out, head["id"], args.max_wait) if head.get("status") != "completed": raise head_failure(head, args.base) saved, skipped = download_files(api, out, head["id"], formats, out_dir, args.arkit) result.update(status="completed", files=saved, skipped=skipped) out.result(result, "Done: head {} saved to {}".format(head["id"], out_dir)) def cmd_status(args, out): api = api_for(args, out) if args.wait: h = wait_head(api, out, args.id, args.max_wait) else: _, h, _ = api.call("GET", "/v1/heads/{}".format(args.id)) files = [f.get("format") + ("+arkit" if f.get("arkit") else "") for f in (h.get("files") or [])] out.result({"ok": True, "head": h}, "{} {}: {}{}".format(h["id"], "(" + h["name"] + ")" if h.get("name") else "", describe(h), "\nFiles: " + ", ".join(files) if files else "") + ("\nError: {} ({})".format((h.get("error") or {}).get("message"), (h.get("error") or {}).get("code")) if h.get("error") else "")) if args.wait and h.get("status") == "failed": sys.exit(EXIT_FAILED) def cmd_download(args, out): api = api_for(args, out) saved, skipped = download_files(api, out, args.id, parse_formats(args.formats), Path(args.out).expanduser().resolve(), args.arkit) out.result({"ok": True, "id": args.id, "files": saved, "skipped": skipped}, "Saved {} file(s) to {}".format(len(saved), Path(args.out).expanduser().resolve())) def cmd_list(args, out): api = api_for(args, out) q = {"limit": args.limit} if args.status: q["status"] = args.status _, page, _ = api.call("GET", "/v1/heads?" + urllib.parse.urlencode(q)) rows = page.get("data") or [] lines = ["{} {:<10} {} {}".format(h["id"], h.get("status"), (h.get("created_at") or "")[:16].replace("T", " "), h.get("name") or "") for h in rows] out.result({"ok": True, "heads": rows, "next_cursor": page.get("next_cursor")}, "\n".join(lines) if lines else "No heads yet.") def people_in(folder, per_person): folder = Path(folder).expanduser() if not folder.is_dir(): raise FNError("No such folder: {}".format(folder), code="invalid_request", exit_code=EXIT_USAGE) imgs = lambda d: sorted(p for p in d.iterdir() if p.is_file() and p.suffix.lower() in IMAGE_TYPES) if per_person: people = [(d.name, imgs(d)) for d in sorted(folder.iterdir()) if d.is_dir() and not d.name.startswith(".")] people = [(n, ph) for n, ph in people if ph] else: people = [(p.stem, [p]) for p in imgs(folder)] if not people: raise FNError("No photos found in {}{}.".format(folder, " subfolders" if per_person else ""), code="invalid_request", exit_code=EXIT_USAGE, hint=None if per_person else "One photo = one head. For several photos per person use --per-person-subfolders.") for n, ph in people: if len(ph) > MAX_PHOTOS: raise FNError("{}: {} photos; at most {} per head.".format(n, len(ph), MAX_PHOTOS), code="too_many_photos", exit_code=EXIT_USAGE) check_photos(ph) return people def cmd_batch(args, out): people = people_in(args.folder, args.per_person_subfolders) formats = parse_formats(args.formats) out_dir = Path(args.out).expanduser().resolve() api = api_for(args, out) state = load_state(out_dir) plan = [] for name, photos in people: fp = fingerprint(photos, name) prev = state["heads"].get(fp) plan.append({"name": name, "photos": [str(p) for p in photos], "fp": fp, "id": prev["id"] if prev else None}) # heads made by an earlier run of this batch are resumed, not paid again (unless they failed) for p in plan: if p["id"]: try: _, h, _ = api.call("GET", "/v1/heads/{}".format(p["id"])) if h.get("status") in ("failed", "deleted"): p["id"] = None except FNError as e: if e.status != 404: raise p["id"] = None new = [p for p in plan if not p["id"]] out.say("Batch: {} person(s), {} new head(s) to create, {} resumed from an earlier run.".format(len(plan), len(new), len(plan) - len(new))) for p in plan: out.say(" {:<30} {} photo(s){}".format(p["name"][:30], len(p["photos"]), " (resume " + p["id"] + ")" if p["id"] else "")) q = quote(api, out, len(new)) if args.dry_run: return out.result({"ok": True, "dry_run": True, "quote": q, "plan": plan}, "Dry run: nothing uploaded or spent.") if len(new) > BATCH_CONFIRM_ABOVE and not args.yes: raise FNError("This batch creates {} heads for {} credits ({}). Ask the user to confirm, then run again with --yes.".format( len(new), num(q["total_credits"]), "${:,.2f}".format(q["total_usd"])), code="confirmation_required", exit_code=EXIT_CONFIRM, data={"quote": q}) if not q["sufficient"]: raise FNError("Not enough credits for {} heads: {} needed, {} available.".format(len(new), num(q["total_credits"]), num(q["balance"])), code="insufficient_credits", exit_code=EXIT_CREDITS, hint="Buy credits at {}/factory#/billing, or make fewer heads.".format(args.base)) for p in new: out.say("Creating {}...".format(p["name"])) h = create_head(api, out, [Path(x) for x in p["photos"]], p["name"]) p["id"] = h["id"] state["heads"][p["fp"]] = {"id": h["id"], "name": p["name"], "photos": p["photos"], "created_at": h.get("created_at")} save_state(out_dir, state) results = {p["id"]: {"name": p["name"], "id": p["id"], "status": None} for p in plan} if args.wait: start = time.time() pending = set(results) while pending: for hid in sorted(pending): _, h, _ = api.call("GET", "/v1/heads/{}".format(hid)) r = results[hid] if describe(h) != r.get("last"): out.say(" {} {:<24} {}".format(time.strftime("%H:%M:%S"), r["name"][:24], describe(h))) r["last"] = describe(h) r["status"] = h.get("status") if h.get("status") in FINAL: pending.discard(hid) if h["status"] == "completed": r["files"], r["skipped"] = download_files(api, out, hid, formats, out_dir / safe_name(r["name"]), args.arkit) else: r["error"] = h.get("error") if not pending: break if args.max_wait and time.time() - start > args.max_wait: break time.sleep(15) for r in results.values(): r.pop("last", None) done = [r for r in results.values() if r["status"] == "completed"] failed = [r for r in results.values() if r["status"] in ("failed", "deleted")] running = [r for r in results.values() if r["status"] not in FINAL] out.result({"ok": not failed and not running, "heads": list(results.values()), "completed": len(done), "failed": len(failed), "running": len(running), "out": str(out_dir)}, "Batch: {} completed, {} failed, {} still running. Files in {}{}".format( len(done), len(failed), len(running), out_dir, "\nRun the same batch command again to keep waiting; nothing is created twice." if running else "")) if running: sys.exit(EXIT_PENDING) if failed: sys.exit(EXIT_FAILED) # ---- command line ----------------------------------------------------------------------------------------------------- def main(argv=None): common = argparse.ArgumentParser(add_help=False) common.add_argument("--json", action="store_true", default=argparse.SUPPRESS, help="one JSON object on stdout, progress on stderr") common.add_argument("--base", default=argparse.SUPPRESS, help="API server (default $FACENINJA_API_BASE or {})".format(DEFAULT_BASE)) ap = argparse.ArgumentParser(prog="faceninja.py", description="FaceNinja: photos in, MetaHuman head out.", parents=[common]) ap.add_argument("--version", action="version", version="faceninja.py " + __version__) sub = ap.add_subparsers(dest="cmd", metavar="COMMAND") s = sub.add_parser("login", parents=[common], help="sign in by approving a code in the browser") s.add_argument("--no-wait", action="store_true", help="print the code and exit; run login again to wait for the approval") s.add_argument("--max-wait", type=int, default=0, help="stop waiting after this many seconds (exit 75)") s.add_argument("--no-browser", action="store_true", help="do not open the approval page") s.add_argument("--client-name", help="name of the API key this creates (default: agent and host)") s.add_argument("--force", action="store_true", help="sign in again even if a working key is saved") s.add_argument("--force-new", action="store_true", help="ask for a new code instead of resuming a pending one") s.set_defaults(fn=cmd_login) sub.add_parser("whoami", parents=[common], help="who is signed in").set_defaults(fn=cmd_whoami) sub.add_parser("balance", parents=[common], help="credits left and what a head costs").set_defaults(fn=cmd_balance) def head_opts(p): p.add_argument("--formats", default="glb,zip", help="what to download: glb,fbx,obj,stl,ply,zip (MetaHuman ZIP) or all (default glb,zip)") p.add_argument("--out", default="faceninja-out", help="folder for the files (default ./faceninja-out)") w = p.add_mutually_exclusive_group() w.add_argument("--wait", dest="wait", action="store_true", default=True, help="wait and download (default)") w.add_argument("--no-wait", dest="wait", action="store_false", help="create and return at once") p.add_argument("--max-wait", type=int, default=0, help="stop waiting after this many seconds (exit 75; run again to resume)") a = p.add_mutually_exclusive_group() a.add_argument("--arkit", dest="arkit", action="store_true", default=None, help="GLB/FBX with the 52 ARKit blendshapes (the default)") a.add_argument("--static", dest="arkit", action="store_false", help="GLB/FBX without blendshapes") p.add_argument("--dry-run", action="store_true", help="show the cost and balance, upload and spend nothing") s = sub.add_parser("make", parents=[common], help="make one head from 1-15 photos of one person") s.add_argument("photos", nargs="+", metavar="PHOTO") s.add_argument("--name", help="a name for the head (default: the first photo's file name)") s.add_argument("--new", action="store_true", help="make a new head even if these photos already made one in --out") head_opts(s) s.set_defaults(fn=cmd_make) s = sub.add_parser("status", parents=[common], help="where a head stands") s.add_argument("id") s.add_argument("--wait", action="store_true", help="wait until it is completed or failed") s.add_argument("--max-wait", type=int, default=0, help="stop waiting after this many seconds (exit 75)") s.set_defaults(fn=cmd_status) s = sub.add_parser("download", parents=[common], help="download a finished head's files") s.add_argument("id") s.add_argument("--formats", default="glb,zip") s.add_argument("--out", default="faceninja-out") a = s.add_mutually_exclusive_group() a.add_argument("--arkit", dest="arkit", action="store_true", default=None) a.add_argument("--static", dest="arkit", action="store_false") s.set_defaults(fn=cmd_download) s = sub.add_parser("batch", parents=[common], help="one head per photo, or per subfolder with --per-person-subfolders") s.add_argument("folder") s.add_argument("--per-person-subfolders", action="store_true", help="each subfolder holds 1-15 photos of one person") s.add_argument("--yes", action="store_true", help="the user confirmed the cost (needed above {} heads)".format(BATCH_CONFIRM_ABOVE)) head_opts(s) s.set_defaults(fn=cmd_batch) s = sub.add_parser("list", parents=[common], help="your heads, newest first") s.add_argument("--limit", type=int, default=20) s.add_argument("--status") s.set_defaults(fn=cmd_list) s = sub.add_parser("logout", parents=[common], help="revoke and forget the saved key") s.add_argument("--keep-key", action="store_true", help="forget the key here but leave it valid on the server") s.set_defaults(fn=cmd_logout) args = ap.parse_args(argv) args.json = getattr(args, "json", False) args.base = norm_base(getattr(args, "base", None) or os.environ.get("FACENINJA_API_BASE")) out = Out(args.json) if not getattr(args, "fn", None): ap.print_help() return EXIT_USAGE try: args.fn(args, out) return EXIT_OK except FNError as e: if args.json: err = {"code": e.code, "message": e.message} if e.hint: err["hint"] = e.hint if e.request_id: err["request_id"] = e.request_id extra = {k: v for k, v in (e.data or {}).items() if k != "retry_after"} print(json.dumps(dict({"ok": False, "error": err}, **extra), indent=2, default=str), flush=True) else: print("Error: {}".format(e.message), file=sys.stderr) if e.hint: print(" {}".format(e.hint), file=sys.stderr) if e.request_id: print(" (request id {})".format(e.request_id), file=sys.stderr) return e.exit_code except KeyboardInterrupt: print("\nStopped. A head that was created keeps building; check it with `faceninja.py list`.", file=sys.stderr) return 130 if __name__ == "__main__": sys.exit(main())