| """MPS Forma3D continuation, stable splits, mesh metrics, resumable experiments.""" |
| import argparse |
| import importlib.util |
| import json |
| import math |
| import os |
| from pathlib import Path |
| import random |
| import shutil |
| import subprocess |
| import sys |
| import time |
| from types import SimpleNamespace |
|
|
| sys.path.insert(0,str(Path(__file__).resolve().parents[1])) |
| from common.runtime import ROOT,RUNS,VENDOR,read,write,sha,immutable,exclusive,guard,memory,jsonl,rows,paired_ci,stable_object_splits |
| sys.path.insert(0,str(VENDOR/"forma/src")) |
| from geometry import rotation,rotate_splats,chamfer,mesh_mask,splat_mask,iou,VIEWS |
|
|
| BASE=RUNS/"forma" |
| DATA=BASE/"data" |
| PIN="3469f527c5f6833d654f62653ee92e0f73854b33" |
| VARIANTS=("control","views12","objects1024") |
|
|
| def load_module(name,path): |
| spec=importlib.util.spec_from_file_location(name,path) |
| mod=importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(mod) |
| return mod |
|
|
| def starting_checkpoint(): |
| from huggingface_hub import hf_hub_download |
| return Path(hf_hub_download("Lucasgitguy-co/forma3d-tiny-3","model.pt",revision=PIN)) |
|
|
| def prepare(): |
| import numpy as np |
| import trimesh |
| from image3d_student.data import cache_sample |
| from image3d_student.splats import preview |
| pipeline=load_module("published_pipeline",VENDOR/"forma/recipe/pipeline.py") |
| immutable(DATA/"settings.json",{"recipe":sha(VENDOR/"forma/recipe/pipeline.py"),"objects":1024,"seed":44,"views":12,"version":1}) |
| roots={} |
| for count in (512,1024): |
| root=BASE/f"original-{count}" |
| root.mkdir(parents=True,exist_ok=True) |
| source,helper=pipeline.upstream(root) |
| folder=pipeline.prepare(root,source,helper,SimpleNamespace(objects=count,seed=44)) |
| roots[count]=folder |
| old={} |
| for split in ("train","validation","test"): |
| actual=rows(roots[512]/f"{split}.jsonl") |
| expected=rows(VENDOR/f"forma/splits/{split}.jsonl") |
| if {r["id"] for r in actual}!={r["id"] for r in expected}: |
| raise ValueError("Original 512-object splits did not reproduce") |
| for r in actual: |
| old[r["metadata"]["asset_id"]]=split |
| |
| |
| all_rows=[r for s in ("train","validation","test") for r in rows(roots[1024]/f"{s}.jsonl")] |
| groups={} |
| for r in all_rows: |
| groups.setdefault(r["metadata"]["object_id"],[]).append(r) |
| assignment=stable_object_splits(groups,old) |
| immutable(DATA/"assignments.json",assignment) |
| manifests={"control":[],"views12":[],"objects1024":[],"validation":[],"expanded-validation":[],"test":[],"confirmation":[]} |
| |
| for r in all_rows: |
| uid=r["metadata"]["asset_id"]; split=assignment[r["metadata"]["object_id"]] |
| copied=dict(r) |
| for key in ("image","target"): |
| dest=DATA/r[key]; dest.parent.mkdir(parents=True,exist_ok=True) |
| if not dest.exists(): os.link(roots[1024]/r[key],dest) |
| if uid in old and split=="train": manifests["control"].append(copied) |
| if split=="validation": |
| manifests["expanded-validation"].append(copied) |
| if uid in old: manifests["validation"].append(copied) |
| if split in ("test","confirmation"): manifests[split].append(copied) |
| helper=load_module("prepare_mesh",VENDOR/"forma/prepare_hf.py") |
| fetch=helper.Downloader(BASE/"downloads",8,3072,4) |
| by_asset={r["metadata"]["asset_id"]:r for r in all_rows} |
| for index,(uid,r) in enumerate(sorted(by_asset.items())): |
| meta=r["metadata"]; split=assignment[meta["object_id"]] |
| reference=DATA/"reference"/f"{uid}.npz" |
| augmented=DATA/"augmentation"/f"{uid}.json" |
| if reference.exists() and (split!="train" or augmented.exists()): |
| if split=="train": |
| aug=read(augmented); manifests["objects1024"].extend(aug) |
| if uid in old: manifests["views12"].extend(aug) |
| continue |
| raw=fetch.get(meta["repo"],meta["revision"],meta["path"]) |
| decoded=BASE/"decoded.glb" |
| subprocess.run(["node",str(VENDOR/"forma/recipe/decode.mjs"),str(raw),str(decoded)],check=True,capture_output=True) |
| splats,transform=helper.mesh_splats(decoded,4096,16,int(meta["sha256"][:8],16)^44) |
| scene=trimesh.load_scene(decoded,process=False) |
| parts=[] |
| for node in scene.graph.nodes_geometry: |
| matrix,name=scene.graph[node]; mesh=scene.geometry[name].copy() |
| if isinstance(mesh,trimesh.Trimesh) and len(mesh.faces): |
| mesh.apply_transform(matrix); mesh.apply_transform(np.asarray(transform)); parts.append(mesh) |
| mesh=trimesh.util.concatenate(parts) |
| samples,_=trimesh.sample.sample_surface(mesh,4096,seed=44) |
| reference.parent.mkdir(exist_ok=True) |
| np.savez_compressed(reference,vertices=mesh.vertices.astype(np.float32),faces=mesh.faces,samples=samples.astype(np.float32)) |
| if split=="train": |
| aug=[] |
| for j,(pitch,yaw) in enumerate((p,y) for p in (-20,0,20) for y in (0,90,180,270)): |
| mat=rotation(yaw,pitch) |
| |
| scale=min(1.,.49/float(np.abs(splats.xyz@mat.T).max())) |
| rotated=rotate_splats(splats,mat,scale) |
| sid=f"abo-{uid}-aug{j}" |
| stage=BASE/"augmentation-stage" |
| if stage.exists(): shutil.rmtree(stage) |
| stage.mkdir() |
| preview(rotated,size=128).save(stage/"input.png") |
| item=cache_sample(stage,sid,stage/"input.png",rotated,16,source=meta.get("source","mesh-proxy:codepawl/abo-glb"), |
| extra={**meta,"input_rotation":mat.tolist(),"framing_scale":scale}) |
| for key in ("image","target"): |
| (stage/item[key]).replace(DATA/item[key]) |
| aug.append(item) |
| write(augmented,aug) |
| manifests["objects1024"].extend(aug) |
| if uid in old: manifests["views12"].extend(aug) |
| decoded.unlink(missing_ok=True) |
| print(f"Prepared geometry/views {index+1}/{len(by_asset)}",flush=True) |
| for name,rs in manifests.items(): jsonl(DATA/f"{name}.jsonl",rs) |
| write(DATA/"ready.json",{k:len(v) for k,v in manifests.items()}) |
|
|
| def device(): |
| import torch |
| from image3d_student.engine import device_for |
| d=device_for("mps") |
| torch.set_num_threads(4) |
| if hasattr(torch.mps,"set_per_process_memory_fraction"): |
| torch.mps.set_per_process_memory_fraction(.65) |
| return d |
|
|
| def checkpoint_model(path,dev): |
| import torch |
| from image3d_student.model import Student |
| ck=torch.load(path,map_location="cpu",weights_only=True) |
| if ck["config"]!={"grid":16,"width":64}: |
| raise ValueError("Expected unchanged tiny architecture") |
| model=Student(**ck["config"]).to(dev) |
| model.load_state_dict(ck["model"]) |
| return model |
|
|
| def evaluate(model,manifest,threshold,out): |
| import numpy as np |
| import torch |
| from image3d_student.data import CachedDataset |
| from image3d_student.model import distillation_loss |
| from image3d_student.splats import Splats |
| dataset=CachedDataset(manifest) |
| model.eval(); predictions=[] |
| dev=next(model.parameters()).device |
| for idx,row in enumerate(dataset.rows): |
| image,target=dataset[idx] |
| with torch.no_grad(): |
| start=time.monotonic() |
| pred=model(image[None].to(dev)) |
| loss,terms=distillation_loss(pred,{k:v[None].to(dev) for k,v in target.items()},16) |
| pred={k:v[0].cpu().numpy() for k,v in pred.items()} |
| latency=time.monotonic()-start |
| occ=1/(1+np.exp(-np.clip(pred["occ"][:,0],-50,50)))>=threshold |
| truth=target["occ"].numpy()[:,0]>.5 |
| splats=Splats(**{k:pred[k][occ] for k in ("xyz","rgb","scale","quat","opacity")}) if occ.any() else None |
| meta=row["metadata"] |
| with np.load(DATA/"reference"/f"{meta['asset_id']}.npz",allow_pickle=False) as ref: |
| mat=np.asarray(meta.get("input_rotation",rotation(np.rad2deg(meta["yaw_radians"])))) |
| scale=meta.get("framing_scale",1.) |
| vertices=ref["vertices"]@mat.T*scale; samples=ref["samples"]@mat.T*scale |
| silhouettes=[] |
| for yaw,pitch in VIEWS: |
| view=rotation(yaw,pitch) |
| cache=DATA/"masks"/f"{row['id']}-{yaw}-{pitch}.npy" |
| if cache.exists(): mask=np.load(cache,allow_pickle=False) |
| else: |
| mask=mesh_mask(vertices,ref["faces"],view) |
| cache.parent.mkdir(exist_ok=True); np.save(cache,mask) |
| silhouettes.append(iou(splat_mask(splats,view),mask)) |
| predictions.append({"id":row["id"],"group":meta["object_id"],"loss":float(loss), |
| "occupancy_iou":iou(occ,truth),"silhouette_iou":float(np.mean(silhouettes)), |
| "chamfer":chamfer(pred["xyz"][occ],samples),"empty":not bool(occ.any()),"latency_seconds":latency, |
| **{k:float(v) for k,v in terms.items()}}) |
| summary={k:float(np.mean([r[k] for r in predictions])) for k in ("loss","occupancy_iou","silhouette_iou","chamfer","latency_seconds")} |
| summary.update(threshold=threshold,examples=len(predictions),empty_outputs=sum(r["empty"] for r in predictions),manifest_sha256=sha(manifest),**memory()) |
| jsonl(out.with_suffix(".jsonl"),predictions); write(out,summary) |
| return summary |
|
|
| def eligible(base,candidate): |
| return candidate["chamfer"]<=base["chamfer"] and candidate["occupancy_iou"]>=base["occupancy_iou"] and candidate["silhouette_iou"]>base["silhouette_iou"] |
|
|
| def benchmark(batch,seconds): |
| import torch |
| from torch.utils.data import default_collate |
| from image3d_student.data import CachedDataset |
| from image3d_student.model import distillation_loss |
| dev=device(); model=checkpoint_model(starting_checkpoint(),dev) |
| dataset=CachedDataset(DATA/"control.jsonl") |
| images,target=default_collate([dataset[i%len(dataset)] for i in range(batch)]) |
| images=images.to(dev); target={k:v.to(dev) for k,v in target.items()} |
| opt=torch.optim.AdamW(model.parameters(),lr=5e-5,weight_decay=1e-4) |
| start=None; count=0 |
| while start is None or time.monotonic()-start<seconds: |
| opt.zero_grad(set_to_none=True) |
| loss,_=distillation_loss(model(images),target,16) |
| if not torch.isfinite(loss): raise RuntimeError("Nonfinite benchmark loss") |
| loss.backward(); torch.nn.utils.clip_grad_norm_(model.parameters(),1.,error_if_nonfinite=True); opt.step() |
| torch.mps.synchronize() |
| if start is None: start=time.monotonic() |
| else: count+=1 |
| if count%10==0: guard(torch.mps.driver_allocated_memory()) |
| elapsed=time.monotonic()-start |
| write(BASE/f"benchmark-{batch}.json",{"batch":batch,"steps":count,"seconds":elapsed,"samples_per_second":count*batch/elapsed, |
| "steps_per_second":count/elapsed,"mps_driver_bytes":torch.mps.driver_allocated_memory(),**guard(torch.mps.driver_allocated_memory())}) |
|
|
| def audit(): |
| import torch |
| import numpy as np |
| from image3d_student.data import CachedDataset |
| from image3d_student.model import distillation_loss |
| from image3d_student.engine import evaluate as historical_evaluate |
| from torch.utils.data import DataLoader |
| v3=starting_checkpoint() |
| dataset=CachedDataset(DATA/"test.jsonl") |
| for label,path in (("base_v2",BASE/"original-512/upstream/model.pt"),("candidate_v3",v3)): |
| model=checkpoint_model(path,device()) |
| score=historical_evaluate(model,DataLoader(dataset,batch_size=1),device()) |
| published=read(VENDOR/"forma/evaluation.json")[label]["metrics"] |
| if any(not math.isclose(score[k],published[k],rel_tol=1e-4,abs_tol=1e-5) for k in published): |
| raise RuntimeError(f"{label} historical baseline did not reproduce") |
| write(BASE/f"{label}-historical.json",score) |
| image,target=dataset[0] |
| a=checkpoint_model(v3,"cpu"); b=checkpoint_model(v3,device()) |
| gradients=[]; losses=[] |
| for model in (a,b): |
| dev=next(model.parameters()).device |
| loss,_=distillation_loss(model(image[None].to(dev)),{k:v[None].to(dev) for k,v in target.items()},16) |
| loss.backward(); losses.append(float(loss)) |
| gradients.append([p.grad.detach().cpu().numpy() for p in model.parameters()]) |
| parity=np.allclose(losses[0],losses[1],rtol=1e-3,atol=1e-4) and all(np.allclose(x,y,rtol=1e-3,atol=1e-4) for x,y in zip(*gradients)) |
| write(BASE/"parity.json",{"passed":bool(parity),"losses":losses}) |
| if not parity: raise RuntimeError("Forma CPU/MPS numeric parity failed") |
|
|
| def save(path,model,opt,meta,rng): |
| import torch |
| |
| def cpu(value): |
| if torch.is_tensor(value): return value.detach().cpu() |
| if isinstance(value,dict): return {k:cpu(v) for k,v in value.items()} |
| if isinstance(value,list): return [cpu(v) for v in value] |
| return value |
| ck={"format":1,"config":model.config,"model":cpu(model.state_dict()),"optimizer":cpu(opt.state_dict()), |
| "step":meta["step"],"metadata":meta,"python_rng":rng.getstate()} |
| torch.save(ck,path.with_suffix(".tmp")); path.with_suffix(".tmp").replace(path) |
|
|
| def train(variant,seed,batch): |
| import torch |
| from torch.utils.data import default_collate |
| from image3d_student.data import CachedDataset |
| from image3d_student.model import distillation_loss |
| dev=device(); folder=BASE/"experiments"/f"{variant}-seed{seed}" |
| manifest=DATA/f"{variant}.jsonl" |
| immutable(folder/"settings.json",{"variant":variant,"seed":seed,"initial_batch":batch,"steps":6000,"lr":5e-5, |
| "starting_sha256":sha(starting_checkpoint()),"manifest_sha256":sha(manifest),"script":sha(__file__)}) |
| if (folder/"complete.json").exists(): return |
| dataset=CachedDataset(manifest); model=checkpoint_model(starting_checkpoint(),dev) |
| optimizer=torch.optim.AdamW(model.parameters(),lr=5e-5,weight_decay=1e-4) |
| rng=random.Random(seed); torch.manual_seed(seed) |
| meta={"step":0,"best_step":0,"best_silhouette":-1.,"batch":batch,"history":[],"contains_synthetic":False} |
| if (folder/"last.pt").exists(): |
| ck=torch.load(folder/"last.pt",map_location="cpu",weights_only=True) |
| model.load_state_dict(ck["model"]); optimizer.load_state_dict(ck["optimizer"]) |
| meta=ck["metadata"]; rng.setstate(ck["python_rng"]) |
| baseline=read(BASE/"baseline-validation.json") |
| started=time.monotonic() |
| try: |
| while meta["step"]<6000: |
| |
| if meta["step"] and meta["step"]%1000==0 and not any(h["step"]==meta["step"] for h in meta["history"]): |
| metric=evaluate(model,DATA/"validation.jsonl",.5,folder/f"validation-{meta['step']}.json") |
| ok=eligible(baseline,metric) |
| meta["history"].append({"step":meta["step"],"eligible":ok,**metric}) |
| if ok and metric["silhouette_iou"]>meta["best_silhouette"]: |
| meta.update(best_step=meta["step"],best_silhouette=metric["silhouette_iou"]) |
| save(folder/"best.pt",model,optimizer,meta,rng) |
| save(folder/"last.pt",model,optimizer,meta,rng) |
| model.train() |
| images,target=default_collate([dataset[rng.randrange(len(dataset))] for _ in range(meta["batch"])]) |
| try: |
| guard(torch.mps.driver_allocated_memory()) |
| optimizer.zero_grad(set_to_none=True) |
| loss,_=distillation_loss(model(images.to(dev)),{k:v.to(dev) for k,v in target.items()},16) |
| if not torch.isfinite(loss): raise RuntimeError("Nonfinite loss") |
| loss.backward(); torch.nn.utils.clip_grad_norm_(model.parameters(),1.,error_if_nonfinite=True); optimizer.step() |
| except (MemoryError,RuntimeError) as exc: |
| if not isinstance(exc,MemoryError) and "out of memory" not in str(exc).lower(): raise |
| if meta["batch"]==1: raise |
| meta["batch"]=max(1,meta["batch"]//2) |
| optimizer.zero_grad(set_to_none=True); torch.mps.empty_cache() |
| save(folder/"last.pt",model,optimizer,meta,rng) |
| print(f"Memory pressure: reducing batch to {meta['batch']}",flush=True) |
| continue |
| meta["step"]+=1 |
| if meta["step"]%50==0: |
| print(f"{variant} seed={seed} step={meta['step']}/6000 loss={float(loss):.5f} batch={meta['batch']}",flush=True) |
| save(folder/"last.pt",model,optimizer,meta,rng) |
| |
| if not any(h["step"]==6000 for h in meta["history"]): |
| metric=evaluate(model,DATA/"validation.jsonl",.5,folder/"validation-6000.json") |
| ok=eligible(baseline,metric); meta["history"].append({"step":6000,"eligible":ok,**metric}) |
| if ok and metric["silhouette_iou"]>meta["best_silhouette"]: |
| meta.update(best_step=6000,best_silhouette=metric["silhouette_iou"]) |
| save(folder/"best.pt",model,optimizer,meta,rng) |
| finally: |
| save(folder/"last.pt",model,optimizer,meta,rng) |
| write(folder/"complete.json",{**meta,"seconds_this_invocation":time.monotonic()-started,**memory()}) |
|
|
| def threshold(path,folder): |
| model=checkpoint_model(path,device()); scores=[] |
| for t in (.3,.4,.5,.6): |
| out=folder/f"threshold-{t}.json" |
| if not out.exists(): evaluate(model,DATA/"validation.jsonl",t,out) |
| scores.append(read(out)) |
| baseline=read(BASE/"baseline-validation.json") |
| valid=[s for s in scores if s["chamfer"]<=baseline["chamfer"] and s["occupancy_iou"]>=baseline["occupancy_iou"]] |
| chosen=max(valid,key=lambda s:s["silhouette_iou"]) if valid else next(s for s in scores if s["threshold"]==.5) |
| write(folder/"threshold-selection.json",chosen) |
| return chosen |
|
|
| def gallery(path,out,threshold_value): |
| import numpy as np |
| import torch |
| from PIL import Image,ImageDraw |
| from image3d_student.data import CachedDataset |
| from image3d_student.splats import Splats,preview,write_ply,read_ply |
| model=checkpoint_model(path,device()); model.eval() |
| dataset=CachedDataset(DATA/"confirmation.jsonl") |
| |
| result=rows(out/"confirmation.jsonl") |
| ranked=sorted(result,key=lambda r:r["silhouette_iou"]) |
| chosen={r["id"] for r in ranked[:2]+ranked[-2:]+ranked[len(ranked)//2:len(ranked)//2+2]} |
| for i,row in enumerate(dataset.rows): |
| if row["id"] not in chosen: continue |
| with torch.no_grad(): pred=model(dataset[i][0][None].to(device())) |
| keep=pred["occ"][0,:,0].sigmoid()>=threshold_value |
| if not bool(keep.any()): continue |
| splats=Splats(**{k:pred[k][0,keep].cpu().numpy() for k in ("xyz","rgb","scale","quat","opacity")}) |
| ply=out/(row["id"]+".ply"); write_ply(ply,splats); read_ply(ply) |
| with np.load(DATA/"reference"/f"{row['metadata']['asset_id']}.npz",allow_pickle=False) as ref: |
| vertices=ref["vertices"]@rotation(np.rad2deg(row["metadata"]["yaw_radians"])).T |
| frames=[] |
| for yaw in range(0,360,30): |
| left=preview(splats,size=192,yaw=np.deg2rad(yaw)) |
| right=Image.fromarray(mesh_mask(vertices,ref["faces"],rotation(yaw),192).astype(np.uint8)*255).convert("RGB") |
| frame=Image.new("RGB",(384,216)); frame.paste(left,(0,24)); frame.paste(right,(192,24)) |
| d=ImageDraw.Draw(frame); d.text((4,4),"Prediction",fill="white"); d.text((196,4),"Reference silhouette",fill="white") |
| frames.append(frame) |
| frames[0].save(out/(row["id"]+".gif"),save_all=True,append_images=frames[1:],duration=100,loop=0) |
|
|
| def worker(action,*args,optional=False): |
| result=subprocess.run([sys.executable,str(Path(__file__).resolve()),action,*map(str,args)]) |
| if result.returncode and not optional: raise RuntimeError(f"{action} failed; rerun to resume") |
| return result.returncode==0 |
|
|
| def run_all(seconds): |
| immutable(BASE/"run-config.json",{"script":sha(__file__),"geometry":sha(Path(__file__).with_name("geometry.py")), |
| "common":sha(ROOT/"common/runtime.py"),"lock":sha(ROOT/"locks/forma.txt"),"benchmark_seconds":seconds}) |
| if not (DATA/"ready.json").exists(): worker("prepare") |
| if not (BASE/"parity.json").exists() or not read(BASE/"parity.json")["passed"]: worker("audit") |
| benchmarks=[] |
| for batch in (4,8,16): |
| path=BASE/f"benchmark-{batch}.json" |
| if not path.exists(): worker("benchmark","--batch",batch,"--seconds",seconds,optional=True) |
| if path.exists(): benchmarks.append(read(path)) |
| if not benchmarks: raise RuntimeError("No batch size passed the memory benchmark") |
| batch=max(benchmarks,key=lambda x:x["samples_per_second"])["batch"] |
| write(BASE/"batch-selection.json",{"batch":batch}) |
| if not (BASE/"baseline-validation.json").exists(): worker("baseline") |
| for variant in VARIANTS: worker("train","--variant",variant,"--seed",44,"--batch",batch) |
| candidates=[] |
| |
| worker("threshold","--checkpoint",starting_checkpoint(),"--out",BASE/"threshold-baseline") |
| for variant in VARIANTS: |
| folder=BASE/"experiments"/f"{variant}-seed44" |
| meta=read(folder/"complete.json") |
| if meta["best_step"]: |
| worker("threshold","--checkpoint",folder/"best.pt","--out",folder) |
| candidates.append((read(folder/"threshold-selection.json")["silhouette_iou"],variant)) |
| selection={"variant":None,"checkpoint":str(starting_checkpoint()),"threshold":read(BASE/"threshold-baseline/threshold-selection.json")["threshold"]} |
| baseline_quality=read(BASE/"threshold-baseline/threshold-selection.json")["silhouette_iou"] |
| if candidates and max(candidates)[0]>baseline_quality: |
| _,variant=max(candidates) |
| worker("train","--variant",variant,"--seed",45,"--batch",batch) |
| repeat=BASE/"experiments"/f"{variant}-seed45" |
| if read(repeat/"complete.json")["best_step"]: |
| worker("threshold","--checkpoint",repeat/"best.pt","--out",repeat) |
| if read(repeat/"threshold-selection.json")["silhouette_iou"]>baseline_quality: |
| folder=BASE/"experiments"/f"{variant}-seed44" |
| selection={"variant":variant,"checkpoint":str(folder/"best.pt"),"threshold":read(folder/"threshold-selection.json")["threshold"],"repeat_checkpoint":str(repeat/"best.pt"),"repeat_threshold":read(repeat/"threshold-selection.json")["threshold"]} |
| immutable(BASE/"selection.json",selection) |
| worker("final") |
|
|
| def final(): |
| selected=read(BASE/"selection.json") |
| baseline_dir=BASE/"confirmation-baseline"; baseline_dir.mkdir(exist_ok=True) |
| evaluate(checkpoint_model(starting_checkpoint(),device()),DATA/"confirmation.jsonl",.5,baseline_dir/"confirmation.json") |
| before=rows(baseline_dir/"confirmation.jsonl") |
| report={"selection":selected,"runs":{},"baseline":read(baseline_dir/"confirmation.json"),"batch":read(BASE/"batch-selection.json")} |
| choices=[("selected",selected["checkpoint"],selected["threshold"])] |
| if selected.get("repeat_checkpoint"): choices.append(("repeat",selected["repeat_checkpoint"],selected["repeat_threshold"])) |
| for name,path,t in choices: |
| out=BASE/f"confirmation-{name}"; out.mkdir(exist_ok=True) |
| score=evaluate(checkpoint_model(path,device()),DATA/"confirmation.jsonl",t,out/"confirmation.json") |
| if t!=.5: evaluate(checkpoint_model(path,device()),DATA/"confirmation.jsonl",.5,out/"historical-threshold.json") |
| after=rows(out/"confirmation.jsonl") |
| report["runs"][name]={"score":score,"checkpoint_sha256":sha(path),"ci":{m:paired_ci(before,after,m) for m in ("silhouette_iou","occupancy_iou","chamfer")}} |
| if name=="selected": gallery(path,out,t) |
| report["limitations"]="ABO Gaussian-proxy input only. Chamfer is center-to-mesh-surface-sample distance; silhouettes use orthographic alpha>=0.5 vs triangle rasterization at 96px. Not photorealistic or natural-photo evaluation." |
| write(BASE/"report.json",report) |
| (BASE/"MODEL_CARD.md").write_text("# Local Forma3D experiment\n\n"+report["limitations"]+"\n\nSee report.json and confirmation-selected/*.gif. Existing image3d CLI accepts the selected format=1 checkpoint. No publication performed.\n") |
| print(f"Finished. Report: {BASE / 'report.json'}",flush=True) |
|
|
| def main(): |
| p=argparse.ArgumentParser(description=__doc__) |
| p.add_argument("action",choices=["all","prepare","audit","benchmark","baseline","train","threshold","final"]) |
| p.add_argument("--seconds",type=int,default=900); p.add_argument("--batch",type=int,default=4) |
| p.add_argument("--variant",choices=VARIANTS,default="control"); p.add_argument("--seed",type=int,default=44) |
| p.add_argument("--checkpoint",type=Path); p.add_argument("--out",type=Path,default=BASE/"threshold-baseline") |
| a=p.parse_args(); BASE.mkdir(parents=True,exist_ok=True) |
| if min(a.seconds,a.batch)<1: p.error("seconds and batch must be positive") |
| if a.action=="all": |
| with exclusive(): run_all(a.seconds) |
| elif a.action=="prepare": prepare() |
| elif a.action=="audit": audit() |
| elif a.action=="benchmark": benchmark(a.batch,a.seconds) |
| elif a.action=="baseline": evaluate(checkpoint_model(starting_checkpoint(),device()),DATA/"validation.jsonl",.5,BASE/"baseline-validation.json") |
| elif a.action=="train": train(a.variant,a.seed,a.batch) |
| elif a.action=="threshold": threshold(a.checkpoint,a.out) |
| else: final() |
|
|
| if __name__=="__main__": main() |
|
|
