scripts/phase6_fuse.py

import csv
import glob
import json

import cv2
import numpy as np

ROOT = "/root/bdmc_pipeline"
H, W = 1920, 1080
fx = fy = H / (2 * np.tan(np.radians(72.0 / 2)))
K = np.array([[fx, 0, W / 2], [0, fy, H / 2], [0, 0, 1.0]])
RADIUS_PX = 110

devmaps = np.load(f"{ROOT}/outputs/devmaps_small.npz")["dev"].astype(np.float32)
devmaps_m = (devmaps - 128.0) / 1000.0
defects = list(csv.DictReader(open(f"{ROOT}/outputs/defects.csv")))
wins = {}
for p in sorted(glob.glob(f"{ROOT}/outputs/tracks/win_*.npz")):
    z = np.load(p)
    wins[int(z["start"])] = {"tracks": z["tracks"], "vis": z["vis"].reshape(z["vis"].shape[0], -1)}

tri_meta = {}
for p in sorted(glob.glob(f"{ROOT}/outputs/tracks/triangulated_*.npz")):
    z = np.load(p)
    tri_meta[int(z["start"])] = {"xyz": z["xyz"], "Rs": z["Rs"], "ts": z["ts"],
                                 "rmse_med": float(np.median(z["rmse"]))}

def track_support(fst, dur, cxc, cyc):
    dvs = []
    for s, w in wins.items():
        lo, hi = max(fst, s), min(fst + dur - 1, s + w["tracks"].shape[0] - 1)
        for f in range(lo, hi + 1):
            tr = w["tracks"][f - s]
            v = w["vis"][f - s].astype(bool)
            xi = np.clip(tr[:, 0].round().astype(int), 0, W - 1)
            yi = np.clip(tr[:, 1].round().astype(int), 0, H - 1)
            dist = np.hypot(xi - cxc, yi - cyc)
            near = v & (dist < RADIUS_PX)
            if near.sum() == 0:
                continue
            dvs.append(devmaps_m[f - 1][yi[near] // 4, xi[near] // 4])
    if not dvs:
        return None
    dv = np.concatenate(dvs)
    return {"n_obs": int(len(dv)),
            "support_frac_lt3cm": round(float((dv < -0.03).mean()), 3),
            "median_dev_cm": round(float(np.median(dv)) * 100, 2),
            "p25_dev_cm": round(float(np.percentile(dv, 25)) * 100, 2)}

rows_out = []
for r in defects:
    c = str(r["centroid_px"]).replace("[", "").replace("]", "").replace(",", " ").split()
    cxc, cyc = float(c[0]), float(c[1])
    fst, dur = int(r["first_frame"]), int(r["duration_frames"])
    rec = dict(r)
    ts = track_support(fst, dur, cxc, cyc)
    if ts is None or ts["n_obs"] < 5:
        rec.update({"track_n_obs": ts["n_obs"] if ts else 0, "track_median_dev_cm": None,
                    "track_support_frac": None, "confidence": "no_coverage"})
    else:
        med, sf = ts["median_dev_cm"], ts["support_frac_lt3cm"]
        conf = ("agree" if (med <= -3.0 and sf >= 0.4)
                else "partial" if med <= -1.5 else "disagree")
        rec.update({**ts, "track_support_frac": sf, "track_median_dev_cm": med,
                    "confidence": conf})
    rows_out.append(rec)

cols = ["first_frame", "t_sec", "duration_frames", "duration_s", "peak_dev_cm", "area_px",
        "centroid_px", "n_obs", "median_dev_cm", "p25_dev_cm", "support_frac_lt3cm",
        "confidence"]
with open(f"{ROOT}/outputs/defects_fused.csv", "w", newline="") as fh:
    wr = csv.DictWriter(fh, fieldnames=cols)
    wr.writeheader()
    wr.writerows([{k: r.get(k) for k in cols} for r in rows_out])

tri_files = glob.glob(f"{ROOT}/outputs/tracks/triangulated_*.npz")
if tri_files:
    med_rmse = float(np.median(np.concatenate([np.load(p)["rmse"] for p in tri_files])))
else:
    med_rmse = 0.0

report = {
    "n_defects": len(rows_out),
    "confidence_counts": {c: sum(1 for r in rows_out if r.get("confidence") == c)
                          for c in ["agree", "partial", "disagree", "no_coverage"]},
    "method": "per-defect: gather all CoTracker3 observations passing through defect region during its confirmed lifetime; compare their depth-residual samples against the flag",
    "median_triangulated_rmse_px": round(med_rmse, 2),
    "sigma_proxy_triangulated_cm_at_5m": round(float(med_rmse * 5.0 / fx * 100), 2),
    "depth_tile_sigma_cm": json.load(open(f"{ROOT}/outputs/phase3_summary.json"))["tile_sigma_median_cm"],
    "note": "no GPS/IMU: egomotion estimated visually (Kabsch on depth-anchored correspondences)",
}
json.dump(report, open(f"{ROOT}/outputs/phase6_report.json", "w"), indent=1)
print(json.dumps(report, indent=1))
print("Phase 6 fusion complete", flush=True)