"""Gradio front end for the Sking image -> Minecraft skin pipeline.
Startup order is load-bearing:
0. import spaces installs ZeroGPU's CUDA emulation shim -- MUST be
before anything touches torch, or `.to("cuda")`
dies with "No CUDA GPUs are available".
1. build + launch Gradio immediately, so the container binds port 7860 well
inside HF's health-check window.
2. bootstrap in a thread downloads ~19 GB and verifies SHA-256 in the
background. Blocking startup on this is what caused
RUNTIME_ERROR with no traceback: the platform killed
the container before it ever became healthy.
"""
# --- 0. `spaces` MUST be imported before torch ------------------------------
try:
import spaces # type: ignore
def gpu(duration: int):
return spaces.GPU(duration=duration)
ZERO_GPU = True
except ImportError: # dedicated-GPU Docker Space, or a local run
def gpu(duration: int):
def identity(fn):
return fn
return identity
ZERO_GPU = False
import base64 # noqa: E402
import os # noqa: E402
import html # noqa: E402
import tempfile # noqa: E402
import threading # noqa: E402
import traceback # noqa: E402
from pathlib import Path # noqa: E402
import gradio as gr # noqa: E402
from PIL import Image # noqa: E402
# `weights` first: it owns HF_HOME, which huggingface_hub.constants freezes at
# import time. sking_pipeline/__init__.py now guarantees this regardless, but
# keep the order explicit so the intent survives a future "tidy up imports".
from sking_pipeline import weights # noqa: E402,F401 (isort: skip)
from sking_pipeline import stage1_flux, stage2_uv # noqa: E402
# ZeroGPU platform caps, measured live + documented:
# * a call RESERVES 1.5x the declared duration against the caller's daily
# quota before anything runs (225s requested for duration=150, etc.);
# * a FREE account may not declare more than 120s per call -- duration=300
# raises "ZeroGPU illegal duration ... larger than the maximum allowed";
# only PRO (1500s/day) or dedicated hardware may exceed it;
# * daily quota: ~60-180s anonymous, 300s signed-in free, 1500s PRO.
# Production parity (entropydrop.com sends n_step=100, guidance=4, 1024px)
# costs ~160s of GPU on the measured free path (~1.48 s/step @1024 + stage2),
# so the single-call PRO path (300s) still owns true 100-step parity. Free gets
# closer by SPLITTING Flux and stage 2 into two 120s GPU calls so Flux keeps
# the whole free envelope for itself (~70 steps @1024).
GPU_SECONDS = int(os.environ.get("SKING_GPU_SECONDS", "120")) # free Flux call
GPU_SECONDS_PROD = int(os.environ.get("SKING_GPU_SECONDS_PROD", "300")) # PRO one-shot
STAGE2_SECONDS = int(os.environ.get("SKING_STAGE2_SECONDS", "60")) # free stage-2 call
# Live zero-a10g measurement 2026-09-14 (free button log):
# 33 steps @1024 → 48.5 s wall ⇒ ~1.48 s/step (fixed overhead absorbed)
# Prior model of 1.25 s/step @768 scaled by (1024/768)^2 = 2.22 s/step was
# ~50% pessimistic and clamped free far below what the GPU actually delivers.
# Recalibrate to measured 1024 rate; scale by area for other sizes. Keep a
# small fixed cost + stage-2 estimate so the clamp still has headroom.
_FLUX_FIXED_S = float(os.environ.get("SKING_FLUX_FIXED_S", "4"))
_FLUX_PER_STEP_S_1024 = float(os.environ.get("SKING_FLUX_PER_STEP_S", "1.50"))
_STAGE2_EST_S = float(os.environ.get("SKING_STAGE2_EST_S", "25"))
# Free path: Flux alone owns the 120s call (stage 2 is a separate call).
# Use 0.90 of the envelope so a slow first-of-session step doesn't OOM the
# reservation; the kill still burns quota.
_BUDGET_FRACTION = float(os.environ.get("SKING_BUDGET_FRACTION", "0.90"))
def _per_step_s(size: int) -> float:
"""Seconds per Flux step at `size`×`size`, calibrated at 1024."""
return _FLUX_PER_STEP_S_1024 * (size / 1024.0) ** 2
def _fit_gpu_budget(
steps: int,
size: int,
budget_s: int,
clamp: bool,
*,
include_stage2: bool = True,
):
"""Optionally clamp `steps` so the run fits the declared GPU duration.
`clamp=False` is the production-parity path: runs exactly what was asked
under the (PRO/paid-only) 300s declaration.
`include_stage2=False` is the free split path: Flux owns the whole 120s
envelope and stage 2 runs as a second @spaces.GPU call, so we do not
subtract stage-2 time from the Flux budget. That is what pushes free from
~33 → ~70 steps @1024.
"""
if not clamp:
return steps, size, ""
reserved = _STAGE2_EST_S if include_stage2 else 0.0
budget = budget_s * _BUDGET_FRACTION - _FLUX_FIXED_S - reserved
per_step = _per_step_s(size)
if per_step <= 0:
return steps, size, ""
max_steps = int(budget / per_step)
if steps <= max_steps:
return steps, size, ""
clamped = max(8, max_steps)
path = "single-call" if include_stage2 else "split (Flux-only envelope)"
note = (
f" · ⚠ steps clamped {steps}→{clamped} for free ZeroGPU {path} "
f"(budget {budget_s}s, ~{per_step:.2f}s/step @ {size}px). "
f"True production parity is 100 steps @1024 on the PRO button."
)
return clamped, size, note
# ---------------------------------------------------------------------------
# background bootstrap
# ---------------------------------------------------------------------------
STATE = {
"stage": "starting",
"log": [],
"ready": False,
"stage1": False,
"boot": None,
"error": None,
"last_error": None,
}
_LOCK = threading.Lock()
def _progress(msg: str) -> None:
if STATE.get("halt_logging"):
return
with _LOCK:
STATE["log"].append(msg)
if len(STATE["log"]) > 200:
del STATE["log"][:100]
print(msg, flush=True)
def _bootstrap() -> None:
"""Download, verify and load the models -- at startup, on the main thread.
Both constraints are hard and they pull in opposite directions:
* ZeroGPU's CUDA emulation only intercepts `.to("cuda")` during *startup*,
before the app begins serving. Loading later -- from a worker thread, or
after `launch()` -- falls through to the real `torch._C._cuda_init`, which
the `spaces` patch refuses: "Low-level CUDA init reached ... emulation mode
did not intercept a CUDA operation".
* So the weights must be resident before `launch()` is called.
Downloads are fast on HF's Xet cache (~900 MB/s), so ~19 GB costs seconds;
the model materialisation dominates. Failure here is recorded, not fatal:
stage 2 still serves.
"""
try:
boot = weights.bootstrap(progress=_progress)
except BaseException: # noqa: BLE001
with _LOCK:
STATE["error"] = traceback.format_exc()
STATE["stage"] = "error"
print(STATE["error"], flush=True)
return
with _LOCK:
STATE["boot"] = boot
# Stage 2 needs nothing but the verified bundle, so publish readiness
# here. A stage-1 failure must not take the working half of the app
# down with it -- that bug left the Render tab reporting "still
# loading" even though all 127 files had verified.
STATE["ready"] = True
if boot.get("stage1_viable") and boot.get("flux"):
try:
_progress("loading FLUX.2 [klein] 4B Base during startup")
stage1_flux.load(boot["flux"], boot["lora"], progress=_progress)
with _LOCK:
STATE["stage1"] = True
_progress("stage 1 ready")
except BaseException: # noqa: BLE001 - stage 2 keeps working
with _LOCK:
STATE["error"] = traceback.format_exc()
print(STATE["error"], flush=True)
else:
_progress(f"stage 1 skipped: {boot.get('stage1_reason')}")
with _LOCK:
STATE["stage"] = "ready"
_progress("READY")
def _bundle():
boot = STATE.get("boot")
if not boot:
raise gr.Error("Models are still downloading — see the status panel.")
return boot
def _require_stage1():
if not STATE["ready"]:
raise gr.Error("Models are still downloading — see the status panel.")
if not STATE["stage1"]:
raise gr.Error(
"Stage 1 (FLUX.2 [klein] 4B) is not available on this hardware. "
"Use the 'Render → Skin' tab, which runs the identical production "
"UV parser on a render you supply."
)
def _record_endpoint_error(where: str) -> None:
"""Capture an endpoint traceback and surface it to the *caller*.
Two channels are needed because both passive ones fail here:
* HF's run-log endpoint is a rolling SSE window saturated by Xet progress
bars, and errorMessage is only a tail slice -- both truncated the real
exception mid-frame, repeatedly.
* ZeroGPU executes @spaces.GPU tasks in a forked worker, so writes to our
module-level STATE never reach the parent process that serves
/status_md. Recording there looked like "nothing was captured".
So: print for the log, store for the panel, and re-raise as gr.Error whose
message is the traceback -- that one always reaches the client.
"""
tb = f"[{where}]\n" + traceback.format_exc()
with _LOCK:
STATE["last_error"] = tb
print(tb, flush=True)
raise gr.Error(tb[-1800:])
def status_md() -> str:
with _LOCK:
stage, err, log = STATE["stage"], STATE["error"], list(STATE["log"])
s1, ready = STATE["stage1"], STATE["ready"]
last = STATE.get("last_error")
icon = {"starting": "⏳", "ready": "✅", "error": "❌"}.get(stage, "•")
tail = "\n".join(f" {line}" for line in log[-14:])
head = (
f"### {icon} Status: `{stage}`\n\n"
f"- stage 1 (Flux + LoRA): **{'available' if s1 else 'not loaded'}**\n"
f"- stage 2 (SKING_DDJ_v104): **{'available' if ready else 'loading'}**\n"
f"- ZeroGPU: **{'yes' if ZERO_GPU else 'no (dedicated/local)'}**\n"
)
if err:
head += f"\n**Startup error**\n\n```\n{err.strip()[-2000:]}\n```\n"
if last:
head += f"\n**Last request error**\n\n```\n{last.strip()[-2500:]}\n```\n"
return head + f"\nstartup log
\n\n```\n{tail}\n```\n