Powered by AppSignal & Oban Pro

Loop 4: 3D latent, EditScore, VoxHammer or OmniGen2

notebooks/4-latent-to-pixal3d.livemd

Loop 4: 3D latent, EditScore, VoxHammer or OmniGen2

Mix.install([
  {:pythonx, "~> 0.4.9"},
  {:kino_pythonx, "~> 0.1.0"},
  {:kino, "~> 0.19.0"}
])
[project]
name = "weft_loop_notebook"
version = "0.0.0"
requires-python = "==3.11.*"
dependencies = [
  "pillow==11.1.0",
  "numpy==2.2.3",
  "matplotlib==3.10.1",
  "requests==2.32.3"
]

What this loop is

Pixal3D turns an image into a latent and a set of views. Every view is scored against the image that conditioned it. What the scores say decides which repair runs, and there are two: edit the latent with VoxHammer, or edit a view in 2D with OmniGen2 and re-enter Pixal3D.

The router is the point

A shape that is wrong along one axis scores well from the view that hides it. So the decision is not made on one number:

what the scores look like what it means which arm
low mean, low spread across views the geometry is agreed on and the appearance is wrong OmniGen2 on a view, then re-enter
high spread across views the views disagree about the shape VoxHammer on the latent

The threshold is a standard deviation, not a variance, and the first version of this notebook had it wrong. A score bounded in 0..1 has a variance of at most 0.25, so a threshold of 0.15 against variance almost never fires: four views at 0.80, 0.20, 0.75 and 0.15 have a variance of 0.091 and would have gone to the 2D arm. test/loop_test.exs caught that before any run did. The number itself is still uncalibrated: 0.15 separates two scripted cases and nothing has compared it to real EditScore output.

That is the whole reason the views come from sphere_hammersley_sequence rather than from a front view somebody picked. One view cannot separate these two cases, and a hand-picked front view is the case where the axis error hides.

One arm does not work yet, and the notebook says so

VoxHammer's server solves and dispatches its seven-step plan and raises NotImplementedError on every step outside WEFTSPUN_STUB=1. It also takes a mesh rather than Pixal3D's latent, so the arm passes through /extract first, which is a second reason it is not a latent edit today.

The router still selects it when the scores say geometry. It reports that it selected an arm that cannot run rather than quietly falling through to the other one, because a silent fallback would make a geometry failure look like an appearance failure that was repaired.

Setup

import sys, json, base64, statistics, requests
from pathlib import Path

HARNESS = Path(r"C:\weftspun-keypoint\7-service\service-livebook\priv\python")
CORPUS = Path(r"C:\weftspun-keypoint\6-datasource\anny-render-corpus")
sys.path.insert(0, str(HARNESS))
from weft_loop import History, Round, pixi_run, plot, PreconditionFailed
from loop4 import extract, region, repair_latent, require_voxhammer

PIXAL3D = "http://localhost:8002"
VOXHAMMER = "http://localhost:8003"
WORK = Path(r"C:\weftspun-keypoint\.loop4")
WORK.mkdir(exist_ok=True)

SOURCE = r"C:\anny_test\hv_0.png"
INSTRUCTION = "a textured mesh matching the figure in the source image"
NVIEWS = 8
SPREAD_SELECTS_LATENT = 0.15  # standard deviation, not variance: see below

Propose: a latent and its views

def generate(image_path, seed=42):
    payload = {"image": base64.b64encode(Path(image_path).read_bytes()).decode(),
               "seed": seed, "nviews": NVIEWS, "view_resolution": 512}
    response = requests.post(f"{PIXAL3D}/predict", json=payload, timeout=1800)
    response.raise_for_status()
    body = response.json()
    views = []
    for i, view in enumerate(body["views"]):
        path = WORK / f"view_{i}.png"
        path.write_bytes(base64.b64decode(view))
        views.append(str(path))
    (WORK / "state.b64").write_text(body["state"], encoding="ascii")
    return body["state"], views

/extract is not called here. Turning the latent into a mesh and a texture is deferred until a score clears its target, because extraction is the expensive half and a shape nobody has scored is not worth decimating.

Score every view, not the best one

def score_view(path):
    out = WORK / (Path(path).stem + ".score.json")
    pixi_run("editscore", [
        str(HARNESS / "weft_score.py"),
        "--source", SOURCE, "--edited", str(path),
        "--instruction", INSTRUCTION, "--out", str(out), "--precision", "nf4",
    ])
    return json.loads(out.read_text())["overall"]


def score_all(views):
    scores = [score_view(v) for v in views]
    return {"scores": scores,
            "mean": statistics.fmean(scores),
            "spread": statistics.pstdev(scores)}

The router, and the two arms

def choose_arm(summary):
    if summary["spread"] >= SPREAD_SELECTS_LATENT:
        return "voxhammer", "the views disagree about the shape"
    return "omnigen2", "the views agree and the appearance is wrong"


def repair_omnigen2(view_path, i):
    out = WORK / f"repaired_{i}.png"
    pixi_run("omnigen2", [
        str(CORPUS / "omnigen2_edit.py"),
        "--image", str(view_path), "--out", str(out),
        "--steps", "40", "--precision", "bf16",
    ])
    return str(out)

The latent arm is loop4.repair_latent, and it does three things in the order the services require: ask /health, call Pixal3D's /extract because VoxHammer takes a mesh rather than the latent, and only then edit. Each of those was wrong here before priv/python/test_loop4.py was written: the mask was a name that did not exist, the latent was handed straight to the editor, and unavailability was read from an HTTP 500 when a stubbed VoxHammer answers 200 with stub: true after marking all seven plan steps done without editing anything.

Run it, with a floor to compare against

baseline = score_view(SOURCE)
history = History(baseline=baseline, control=SOURCE)
state, views = generate(SOURCE)

for i in (1, 2):
    summary = score_all(views)
    arm, why = choose_arm(summary)
    print(f"round {i}: mean {summary['mean']:.3f}  spread {summary['spread']:.4f}  -> {arm} ({why})")
    history.rounds.append(Round(index=i, artifact=str(WORK / f"round_{i}"),
                                score=summary["mean"], delta=summary["mean"] - baseline,
                                seconds=0.0,
                                provenance={"scores": summary["scores"], "arm": arm}))
    if i == 2:
        break
    if arm == "voxhammer":
        repair_latent(VOXHAMMER, PIXAL3D, state, SOURCE, WORK, mask=None)
    else:
        worst = summary["scores"].index(min(summary["scores"]))
        state, views = generate(repair_omnigen2(views[worst], worst))

print(history.table())

The baseline is the source scored against itself, because a number without one is not a measurement, and loop 4 had no floor at all until now.

repair_latent raises rather than returning. Today it raises at /health, because VoxHammer is stubbed. When that is wired it will raise one step later, at the point where a returned USD layer has no path back to the .npz that render_view.py renders. Both are recorded as an unavailable round rather than being swapped for the 2D arm, because that swap would file a geometry failure as an appearance failure that was repaired.

Extract, once the score earns it

glb, layer = extract(PIXAL3D, state, WORK)
print(glb, layer)

The USD layer names the GLB as an asset path rather than referencing it, and the GLB carries pure data: skin weights, animation samplers, morph targets. No runtime modifiers, drivers, constraints or custom extensions, because an export that only looks right when the consumer runs our code is not portable.

Provenance

record = {
    "loop": "4-latent-to-pixal3d",
    "source": SOURCE,
    "generator": {"service": "pixal3d-image-to-textured-mesh", "nviews": NVIEWS,
                  "view_resolution": 512, "seed": 42},
    "cameras": "sphere_hammersley_sequence",
    "scorer": {"base": "Qwen/Qwen3-VL-8B-Instruct",
               "adapter": "EditScore/EditScore-Qwen3-VL-8B-Instruct", "precision": "nf4"},
    "view_scores": summary["scores"],
    "mean": summary["mean"],
    "spread": summary["spread"],
    "arm": arm,
    "arm_reason": why,
    "region": "whole object; there is no segmentation step in this loop yet",
    "voxhammer_state": "stub, edit steps not wired",
}
(WORK / "provenance.json").write_text(json.dumps(record, indent=2), encoding="utf-8")
print("wrote", WORK / "provenance.json")