Eager cuda placement at startup; remove self-test

app.py

@@ -34,7 +34,7 @@ TURBO_REPO = os.environ.get("H3_TURBO_REPO", "lightx2v/Minimax-h3-Turbo")
 TURBO_FILE = os.environ.get("H3_TURBO_FILE", "minimax_h3_ref2v_turbo_4step_v0.1_bf16.safetensors")
 GPU_SIZE = os.environ.get("H3_GPU_SIZE", "xlarge").strip() or None
 ATTENTION = os.environ.get("H3_ATTENTION", "_native_cudnn").lower()
-PLACEMENT = os.environ.get("H3_PLACEMENT", "lazy").lower()
+PLACEMENT = os.environ.get("H3_PLACEMENT", "eager").lower()  # eager: place on cuda at startup (ZeroGPU best practice)
 
 FPS, FRAMES_PER_CHUNK, LATENTS_PER_CHUNK = 24, 17, 5
 CANVAS_MULTIPLE = 32
@@ -101,10 +101,17 @@ def load_pipeline() -> str | None:
         pipe.transformer_ref.set_attention_backend(ATTENTION)
         try:
             pipe.load_lora_weights(hf_hub_download(TURBO_REPO, TURBO_FILE), adapter_name="turbo")
+            pipe.set_adapters(["turbo"], [1.0])
             TURBO_READY = True
             print(f"[bgswap] turbo LoRA ready ({TURBO_REPO})", flush=True)
         except Exception as error:
             print(f"[bgswap] turbo unavailable ({type(error).__name__}: {error})", flush=True)
+        # Place the whole pipeline (incl. the turbo LoRA) on cuda at startup. ZeroGPU's CUDA emulation handles
+        # `.to("cuda")` outside the GPU function and startup placement is much faster than deferring it to the first
+        # request - so the first generation is no longer penalised by a cold transfer. H3_PLACEMENT=lazy opts out.
+        if PLACEMENT != "lazy":
+            pipe.to("cuda")
+            print("[bgswap] pipeline placed on cuda at startup", flush=True)
         PIPE, MANAGER = pipe, manager
         print(f"[bgswap] generator ready in {time.time() - started:.0f}s", flush=True)
     except Exception as error:
@@ -332,7 +339,10 @@ def get_duration(request, steps, seed, **_):
     denoise = int(steps) * (STEP_LINEAR * sequence + STEP_QUADRATIC * sequence ** 2) * SAFETY
     encode = 5 + reference_rows(references, num_frames) * 1e-3
     share = (height * width * num_frames) / DEFAULT_CANVAS_PIXELS
-    total = (PLACEMENT_ALLOWANCE + encode + SOURCE_ENCODE_PER_DEFAULT_CANVAS * share + denoise
+    # Only budget the cuda-transfer allowance when placement is deferred to the GPU call; eager placement pays it
+    # once at startup, off the clock, so the per-request reservation (and queue priority) improves.
+    placement = PLACEMENT_ALLOWANCE if PLACEMENT == "lazy" else 0
+    total = (placement + encode + SOURCE_ENCODE_PER_DEFAULT_CANVAS * share + denoise
              + DECODE_BASE + DECODE_PER_DEFAULT_CANVAS * share + 10)
     duration = max(MIN_GPU_DURATION, min(MAX_GPU_DURATION, int(total)))
     print(f"[bgswap] S={sequence} -> reserving {duration}s ({denoise:.0f}s denoise at {steps} steps)", flush=True)
@@ -535,30 +545,8 @@ demo = (gr.Workflow(graph=os.path.join(_HERE, "workflow.json"), bind={"generate_
 
 
 # ----------------------------------------------------------------------------------------------------------- self-test
-def _selftest():
-    """Local, no network: derive a mask + reference from the prototype's existing matte/green and check shapes."""
-    base = "/home/ubuntu/ltx-h3-bgswap"
-    green = read_video(f"{base}/green.mp4")
-    matte = read_video(f"{base}/matte.mp4")
-    n = floor_frames(min(len(green), len(matte)))
-    assert n % FRAMES_PER_CHUNK == LATENTS_PER_CHUNK and n <= len(green), n
-    hard = background_mask(matte[:n], 0.0)
-    soft = background_mask(matte[:n], 0.2)
-    assert hard.shape == (n, green.shape[1], green.shape[2]), hard.shape
-    assert set(np.unique(hard)) <= {0.0, 1.0}, "hard mask must be binary"
-    assert {round(v, 3) for v in np.unique(soft).tolist()} == {0.2, 1.0}, "relight sets the subject interior to 0.2, bg to 1"
-    ref = gray_reference(green[0], matte[0])
-    assert Image.open(ref).size == (green.shape[2], green.shape[1])
-    print(f"selftest OK: n={n}, hard repaint {hard.mean()*100:.1f}%, ref {ref}")
-
-
 if __name__ == "__main__":
-    import sys
-
-    if "--selftest" in sys.argv:
-        _selftest()
-    else:
-        if demo is None:
-            raise SystemExit("gr.Workflow requires gradio >= 6.25")
-        load_pipeline()
-        demo.launch(show_error=True, allowed_paths=[OUTPUT_DIR, os.path.join(_HERE, "examples")])
+    if demo is None:
+        raise SystemExit("gr.Workflow requires gradio >= 6.25")
+    load_pipeline()
+    demo.launch(show_error=True, allowed_paths=[OUTPUT_DIR, os.path.join(_HERE, "examples")])