scripts/render_real2sim_demo.py

#!/usr/bin/env python
"""S9 real2sim visual deliverable: real video + observed-vs-simulated silhouettes, honestly.

**What this renders, per object.** Two silhouettes per frame, reprojected onto the
real camera image via :func:`fpgm.viz.overlays.project_silhouette_hull` (the same
convex-hull re-projection ``fpgm.datagen.object_poses.compute_silhouette_iou`` and
``write_object_poses_debug_video`` already use -- see this script's own
docstrings below for why nothing here writes a second projector):

* **observed** -- S6's measured pose track (``poses.npz``'s ``T_world_obj``),
  amber.
* **simulated** -- the MuJoCo posterior's highest-weight particle's rollout
  (:func:`fpgm.physics.simulate.MujocoSimulator.simulate_batched`), magenta.

Plus a small per-frame HUD: frame index, whether S6 actually scored this frame
as an observation, the current observed-vs-simulated translation error in mm,
and (for an object whose simulation is anchored above empty space -- see
below) an explicit, computed-not-hardcoded note of why.

**This is a visualization script, not a new identification run.** It reuses
every piece of ``fpgm.physics`` exactly as ``scripts/run_physics_identification.py``
does (:func:`~fpgm.physics.scene.build_observed_track`,
:func:`~fpgm.physics.scene.build_sim_spec`,
:class:`~fpgm.physics.simulate.MujocoSimulator`,
:func:`~fpgm.physics.likelihood.log_likelihood`,
:func:`~fpgm.physics.inference.accumulate`) and touches no file under
``src/fpgm/physics/``. The one deliberate simplification: the material prior
always comes from :meth:`~fpgm.physics.priors.VlmPriorProposer.fallback_verdict`
(the documented "unknown material" default) rather than a live VLM call --
this script's job is showing what the *simulator* currently produces from a
measured pose track, not re-running the VLM subprocess, and the VLM verdict
only widens/narrows the *prior* particles are drawn from, not the rollout
dynamics themselves. (Confirmed against the demo episode's own prior physics
runs, ``outputs/physics_multibody/.../brick/physics.json`` /
``.../drawer/physics.json``: both were already run this same way, i.e. this is
not a special-cased shortcut invented for this script.)

**If the simulated silhouette falls away from the observed one, most frames,
that is not a bug in this script.** ``fpgm.physics.scene.build_sim_spec`` always
anchors a body at the episode's FIRST PNP frame -- if that frame is not actually
resting near the static-scene heightfield (e.g. the object has not been placed
yet in this phase of the episode), a free body free-falls before any contact is
possible, exactly as it should. See ``fpgm.physics.scene``'s own module
docstring (the heightfield section, and "Scenes are multi-body...") and
:func:`~fpgm.physics.scene.build_sim_spec`'s own
``anchor_height_above_heightfield_m`` diagnostic, which this script reads and
prints into both the HUD and ``summary.json`` rather than re-deriving the
number itself. (Historical note: an earlier version of this module used a
single FITTED PLANE as the scene's only static surface, which was frequently
the wrong surface entirely -- see ``fpgm.physics.scene``'s module docstring for
the measured bug that motivated replacing it with an observed-depth
heightfield.)

Usage::

    PYTHONPATH=src python scripts/render_real2sim_demo.py

    # non-default episode/objects:
    PYTHONPATH=src python scripts/render_real2sim_demo.py \\
        --uuid <uuid> --camera <serial> --labels drawer brick --particles 64
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "src"))

import numpy as np  # noqa: E402
from scipy.spatial.transform import Rotation  # noqa: E402

from fpgm.config_datagen import DatagenProfile  # noqa: E402
from fpgm.data.droid_raw import read_mp4_properties  # noqa: E402
from fpgm.datagen.robot_buffers import (  # noqa: E402
    load_extrinsics_candidates,
    read_native_camera_intrinsics,
)
from fpgm.geometry.camera import Camera  # noqa: E402
from fpgm.geometry.transforms import transform_points  # noqa: E402
from fpgm.physics import inference, likelihood, materials, priors, scene, simulate  # noqa: E402
from fpgm.physics.types import (  # noqa: E402
    PRISMATIC_PARAMS,
    RIGID_PARAMS,
    EpisodeLogLik,
    ParamSpace,
    PhysicsError,
    SimResult,
)
from fpgm.utils.io import atomic_write, ensure_dir  # noqa: E402
from fpgm.utils.logging import get_logger, setup_logging  # noqa: E402
from fpgm.utils.timing import StepTimer  # noqa: E402
from fpgm.viz.overlays import (  # noqa: E402
    VideoWriter,
    draw_hud,
    draw_silhouette,
    project_silhouette_hull,
    read_frames_bgr,
)

logger = get_logger("render_real2sim_demo")

_DEFAULT_UUID = "AUTOLab+0d4edc83+2023-10-21-19h-07m-04s"
_DEFAULT_CAMERA = "22008760"
_DEFAULT_LABELS = ("drawer", "brick")
_DEFAULT_N_PARTICLES = 64
_DEFAULT_SEED = 0
_DEFAULT_OUT_DIR = REPO_ROOT / "outputs" / "real2sim_demo"

# BGR (OpenCV convention), chosen for contrast against both the amber DROID
# tabletop and each other -- not ``fpgm.viz.overlays.color_for``'s indexed
# palette, since these two colours mean "observed" / "simulated" for every
# object, not "which object" (color_for(i) already means the latter, one per
# label, in S6's own object_poses_debug.mp4 -- reusing it here would collide
# meanings).
_OBSERVED_COLOR_BGR = (0, 210, 255)  # amber
_SIMULATED_COLOR_BGR = (230, 60, 220)  # magenta

#: Heightfield-surface clearance above which fpgm.physics.scene.build_sim_spec itself
#: already flags "anchor frame is above the observed heightfield surface" as a
#: limitation (see that function's own docstring) -- reused verbatim here rather than a
#: second threshold, so this script's "known cause" note fires exactly when
#: build_sim_spec's own diagnostic does.
_ANCHOR_ABOVE_SUPPORT_GATE_M = 0.05


# --------------------------------------------------------------------------- #
# Geometry loading -- observed silhouette needs the SAME canonical points
# fpgm.physics.scene loaded to build the convex-hull collision mesh, at their
# raw (un-rescaled) MESH-vs-OBSERVED_SURFACE convention.
# --------------------------------------------------------------------------- #


def _load_canonical_points(geometry_path: Path, label: str) -> np.ndarray:
    """``(N, 3)`` points in the object's own canonical/local frame.

    Mirrors ``fpgm.physics.scene``'s own private ``_canonical_geometry_points``
    dispatch (MESH -> ``.glb`` vertices via trimesh, OBSERVED_SURFACE -> a
    ``.npz``'s ``points_canonical`` key) rather than importing that
    underscore-prefixed name across a module boundary this task is not
    allowed to modify. Five lines, keyed only on file suffix, matching
    ``fpgm.datagen.geometry_ops.write_object_mesh_glb`` /
    ``write_object_point_cloud_npz`` -- the two writers this reads back.
    """
    if geometry_path.suffix == ".glb":
        import trimesh

        mesh = trimesh.load(str(geometry_path), force="mesh", process=False)
        return np.asarray(mesh.vertices, dtype=np.float64)
    if geometry_path.suffix == ".npz":
        with np.load(geometry_path) as npz:
            return np.asarray(npz["points_canonical"], dtype=np.float64)
    raise PhysicsError(f"{label}: unrecognised geometry file {geometry_path}")


def _load_camera_for_episode(
    profile: DatagenProfile,
    uuid: str,
    camera_serial: str,
    video_width: int,
    video_height: int,
) -> Camera:
    """The same :class:`Camera` S6 built for this episode/camera.

    Native scene-flow intrinsics rescaled to the mp4's own resolution, paired
    with whichever extrinsic candidate S2's own alignment gate chose (recorded
    in ``robot_buffers/meta.json``'s ``chosen_extrinsics``). Mirrors
    ``EpisodePipeline._run_s6``'s "camera + extrinsics: mirror S2's own
    construction" block (``fpgm/datagen/pipeline.py``) -- duplicated rather
    than imported because that block lives inline in a private pipeline
    method, not a reusable function; two call sites in that same file
    (``_run_s6``, the S8 export path) already duplicate it themselves, so this
    is a third occurrence of an established pattern, not a new one.
    """
    master_dir = profile.paths.master_dir(uuid, camera_serial)
    trajectory_path = profile.paths.episode_dir(uuid) / "trajectory.h5"
    camera_intrinsics_native = read_native_camera_intrinsics(
        profile.paths.flows_h5(uuid), camera_serial
    )
    extrinsics_candidates = load_extrinsics_candidates(
        profile.paths.cameras_json(uuid), trajectory_path, camera_serial
    )
    s2_meta = json.loads((master_dir / "robot_buffers" / "meta.json").read_text())
    chosen_name = s2_meta["payload"]["chosen_extrinsics"]
    chosen = next(c for c in extrinsics_candidates if c.name == chosen_name)
    return Camera(camera_intrinsics_native.scaled(video_width, video_height), chosen.world_to_cam)


# --------------------------------------------------------------------------- #
# MuJoCo (w, x, y, z) quaternion -> rotation matrix, for rendering the
# simulated rollout's silhouette (fpgm.physics.likelihood only ever needed the
# inverse direction, matrix -> quat, for its residuals).
# --------------------------------------------------------------------------- #


def _quat_wxyz_to_matrix(quat_wxyz: np.ndarray) -> np.ndarray:
    """``(..., 4)`` MuJoCo-order unit quaternions -> ``(..., 3, 3)`` rotation matrices.

    ``fpgm.physics.likelihood`` implements the opposite direction
    (``_matrix_to_quat_wxyz``) as local numpy specifically to avoid
    ``scipy.spatial.transform``'s ``(x, y, z, w)`` reordering *and* a measured
    ill-conditioning in its own geodesic-angle computation near 0/pi radians
    (see that module's docstring). Neither concern applies to a plain
    quaternion-to-matrix conversion (no angle extraction, no subtraction of
    near-equal floats), so this direction uses ``scipy`` directly rather than
    hand-rolling a second Shepperd's-method implementation for one call site.
    """
    quat_wxyz = np.asarray(quat_wxyz, dtype=np.float64)
    quat_xyzw = quat_wxyz[..., [1, 2, 3, 0]]
    flat = Rotation.from_quat(quat_xyzw.reshape(-1, 4)).as_matrix()
    return flat.reshape(*quat_wxyz.shape[:-1], 3, 3)


def _divergence_frame(poses_row: np.ndarray) -> int | None:
    """First frame this rollout has no pose (NaN), or ``None`` if it never diverged.

    ``fpgm.physics.worker_mujoco`` fills ``poses`` with NaN up front and only
    overwrites a frame once ``mj_step`` for it succeeds (see that module's own
    per-particle loop, which ``break``s out -- without writing that frame's
    pose -- the instant it detects divergence). The first NaN row IS the
    divergence point by that contract; this reads it directly rather than
    re-deriving a divergence test of its own.
    """
    finite_per_frame = np.all(np.isfinite(poses_row), axis=-1)
    if finite_per_frame.all():
        return None
    return int(np.argmax(~finite_per_frame))


def _select_best_particle(
    track: Any,
    sim_result: SimResult,
    prior: Any,
    particles: np.ndarray,
    *,
    timer: StepTimer,
) -> tuple[int, float, EpisodeLogLik, str | None]:
    """Highest-weight particle's index, plus its ESS and a degeneracy note if any.

    "Posterior-mean rollout" per the task brief means: score every particle's
    rollout against the observed track exactly as
    ``fpgm.physics.inference.accumulate`` would for a real identification run
    (one episode, softmax over its log-likelihoods), then take the argmax --
    well-defined even when the posterior is degenerate (ESS -> 1, all mass on
    one particle), which is the measured, expected case here (both objects'
    prior ``physics.json`` runs report ``ess=1.0``).

    Falls back to a "least-diverged" ranking -- more frames survived before
    hitting NaN, ties broken by lower raw translation+rotation residual via
    :func:`~fpgm.physics.likelihood.se3_residuals` (which, unlike
    :func:`~fpgm.physics.likelihood.log_likelihood`, does not force ``-inf``
    on a diverged particle -- it still has real pre-divergence residuals) --
    only in the genuinely degenerate case ``accumulate`` itself refuses to
    report a posterior over: every particle's total log-weight is ``-inf``
    (see that function's own docstring). This is the brief's own explicitly
    anticipated worst case for the brick ("if every brick particle diverges
    ... still render whatever the least-diverged particle did").

    Returns:
        ``(best_idx, ess, episode_loglik, degeneracy_note)``. ``degeneracy_note``
        is ``None`` on the normal path, else the ``PhysicsError`` message
        ``accumulate`` raised.
    """
    loglik = likelihood.log_likelihood(track, sim_result.poses, sim_result.ok, timer=timer)
    episode_ll = EpisodeLogLik(
        uuid=track.uuid,
        camera_serial=track.camera_serial,
        label=track.label,
        loglik=loglik,
        n_obs_frames=track.n_valid,
        n_diverged=int(np.sum(~np.asarray(sim_result.ok, dtype=bool))),
    )
    try:
        posterior = inference.accumulate(prior, particles, [episode_ll], timer=timer)
        return int(np.argmax(posterior.weights)), float(posterior.ess), episode_ll, None
    except PhysicsError as exc:
        trans_res, rot_res = likelihood.se3_residuals(track, sim_result.poses, timer=timer)
        survived = np.array(
            [
                sim_result.poses.shape[1] if (df := _divergence_frame(p)) is None else df
                for p in sim_result.poses
            ]
        )
        residual_sum = np.nansum(trans_res, axis=1) + np.nansum(rot_res, axis=1)
        # np.lexsort's LAST key is primary: rank by most-frames-survived first,
        # lowest residual second.
        order = np.lexsort((residual_sum, -survived))
        return int(order[0]), 0.0, episode_ll, str(exc)


# --------------------------------------------------------------------------- #
# Per-object render
# --------------------------------------------------------------------------- #


def render_object(
    uuid: str,
    camera_serial: str,
    label: str,
    *,
    profile: DatagenProfile,
    out_dir: Path,
    n_particles: int,
    seed: int,
    timer: StepTimer,
) -> dict[str, Any]:
    """Score + simulate ``label``, write ``real2sim_<label>.mp4``, return its summary dict."""
    logger.info("=== %s: build observed track + sim spec ===", label)
    with timer.step("load_track", uuid=uuid, obj_label=label):
        track = scene.build_observed_track(uuid, camera_serial, label, profile=profile, timer=timer)

    scratch_dir = ensure_dir(out_dir / "scratch" / label)
    with timer.step("build_spec", uuid=uuid, obj_label=label):
        spec, diagnostics = scene.build_sim_spec(
            uuid, camera_serial, label, profile=profile, scratch_dir=scratch_dir, timer=timer,
        )
    for note in diagnostics.get("limitations", []):
        print(f"[{label}] {note}")

    body = spec.bodies[0]
    space = ParamSpace(RIGID_PARAMS)
    if body.kind == "prismatic":
        space = space.extended(PRISMATIC_PARAMS)

    # See module docstring: fallback (uninformative) material prior, not a
    # live VLM call -- this script visualizes the simulator, not the VLM.
    verdict = priors.VlmPriorProposer.fallback_verdict(label)
    prior = materials.material_prior(verdict, space)
    rng = np.random.default_rng(seed)
    particles = prior.sample(rng, n_particles)

    simulator = simulate.MujocoSimulator(scratch_dir=scratch_dir)
    with timer.step("simulate", uuid=uuid, obj_label=label, n=n_particles):
        sim_result = simulator.simulate_batched(spec, particles, space, timer=timer)

    n_diverged = int(np.sum(~np.asarray(sim_result.ok, dtype=bool)))
    with timer.step("score", uuid=uuid, obj_label=label, n=n_particles):
        best_idx, ess, episode_ll, degeneracy_note = _select_best_particle(
            track, sim_result, prior, particles, timer=timer
        )
    if degeneracy_note is not None:
        print(
            f"[{label}] every particle's rollout was inconsistent with the observation "
            f"(accumulate() refused a posterior: {degeneracy_note}); rendering the "
            "least-diverged particle instead, ESS is reported as 0."
        )

    best_poses = sim_result.poses[best_idx]  # (T, 7)
    best_ok = bool(sim_result.ok[best_idx])
    div_frame = _divergence_frame(best_poses)
    print(
        f"{label}: n_particles={n_particles} n_diverged={n_diverged} ess={ess:.3f} "
        f"best_idx={best_idx} best_ok={best_ok} divergence_frame={div_frame} "
        f"spread_nats={episode_ll.spread_nats:.3f}"
    )

    # -- translation error, mm, every frame; NaN after the rollout diverges
    # propagates through np.linalg.norm honestly (no fabricated "error") -- #
    obs_pos_m = track.T_world_obj[:, :3, 3]
    sim_pos_m = best_poses[:, :3]
    trans_err_mm = np.linalg.norm(obs_pos_m - sim_pos_m, axis=1) * 1000.0
    scored = track.valid & np.isfinite(trans_err_mm)
    if scored.any():
        err_summary = {
            "mean_mm": float(np.mean(trans_err_mm[scored])),
            "median_mm": float(np.median(trans_err_mm[scored])),
            "max_mm": float(np.max(trans_err_mm[scored])),
            "n_frames_scored": int(scored.sum()),
        }
    else:
        err_summary = {"mean_mm": None, "median_mm": None, "max_mm": None, "n_frames_scored": 0}

    # -- geometry for the two silhouettes (see module-level helpers' docstrings) -- #
    meta_path = profile.paths.master_dir(uuid, camera_serial) / "object_poses" / "meta.json"
    meta = json.loads(meta_path.read_text())
    geometry_path = Path(meta["payload"]["geometry_paths"][label])
    canonical_points = _load_canonical_points(geometry_path, label)
    scaled_points = canonical_points * diagnostics["scale"]

    episode_dir = profile.paths.episode_dir(uuid)
    mp4_path = episode_dir / "recordings" / "MP4" / f"{camera_serial}.mp4"
    mp4_fps, n_video_frames, (video_w, video_h) = read_mp4_properties(mp4_path)
    camera = _load_camera_for_episode(profile, uuid, camera_serial, video_w, video_h)

    finite_frame = np.all(np.isfinite(best_poses), axis=-1)  # (T,)
    sim_rot = np.full((spec.n_frames, 3, 3), np.nan, dtype=np.float64)
    if finite_frame.any():
        sim_rot[finite_frame] = _quat_wxyz_to_matrix(best_poses[finite_frame, 3:7])

    anchor_clear_m = diagnostics.get("anchor_height_above_heightfield_m")
    known_cause = None
    known_cause_hud = None
    if anchor_clear_m is not None and anchor_clear_m > _ANCHOR_ABOVE_SUPPORT_GATE_M:
        known_cause = (
            f"anchored {anchor_clear_m * 1000:.0f}mm above the observed heightfield surface "
            "directly under it at sim start -- free-falls before any contact is possible"
        )
        # Short enough to fit one HUD line at this video's width -- the full
        # sentence above is the one that goes into summary.json; draw_hud has
        # no text-wrapping, and a 110-char line at 1280px runs off-frame.
        known_cause_hud = (
            f"KNOWN CAUSE: anchored {anchor_clear_m * 1000:.0f}mm above heightfield surface "
            "-> free-falls (see summary.json)"
        )

    out_path = out_dir / f"real2sim_{label}.mp4"
    with timer.step("render_video", uuid=uuid, obj_label=label, n=n_video_frames):
        with VideoWriter(out_path, fps=mp4_fps) as writer:
            for t, frame_bgr in enumerate(read_frames_bgr(mp4_path)):
                if t >= n_video_frames:
                    break
                out = frame_bgr

                verts_obs_world = transform_points(track.T_world_obj[t], canonical_points)
                out = draw_silhouette(
                    out, project_silhouette_hull(verts_obs_world, camera), _OBSERVED_COLOR_BGR
                )

                if finite_frame[t]:
                    verts_sim_world = scaled_points @ sim_rot[t].T + best_poses[t, :3]
                    out = draw_silhouette(
                        out, project_silhouette_hull(verts_sim_world, camera), _SIMULATED_COLOR_BGR
                    )

                if np.isfinite(trans_err_mm[t]):
                    err_str = f"{trans_err_mm[t]:.1f} mm"
                else:
                    err_str = "n/a (sim diverged)"
                obs_status = "scored (S6 valid)" if track.valid[t] else "NOT scored (gap/low-vis)"
                hud = [
                    f"{label}  frame {t}/{n_video_frames - 1}",
                    f"observed (amber): {obs_status}",
                    f"simulated (magenta): {'ok' if finite_frame[t] else 'DIVERGED'}"
                    f"  [best of {n_particles}, {n_diverged} diverged]",
                    f"translation error: {err_str}   ESS={ess:.2f}",
                ]
                if known_cause_hud is not None:
                    hud.append(known_cause_hud)
                out = draw_hud(out, hud)
                writer.write(out)
    logger.info("wrote %s", out_path)

    return {
        "label": label,
        "video_path": str(out_path),
        "n_particles": n_particles,
        "n_diverged": n_diverged,
        "ess": ess,
        "degenerate_posterior_note": degeneracy_note,
        "best_particle_idx": best_idx,
        "best_particle_ok": best_ok,
        "divergence_frame": div_frame,
        "translation_error_mm": err_summary,
        "body_kind": body.kind,
        "n_kinematic_bodies": diagnostics["n_kinematic_bodies"],
        "anchor_height_above_heightfield_m": anchor_clear_m,
        "known_cause": known_cause,
        "spread_nats": episode_ll.spread_nats,
        "n_obs_frames": episode_ll.n_obs_frames,
    }


def render_combined_video(
    out_dir: Path, labels: tuple[str, ...], *, timer: StepTimer,
) -> Path | None:
    """Horizontal concat of the per-object videos, frame-by-frame -- cheap: no re-simulation.

    ``None`` if fewer than two labels were rendered (nothing to place side by
    side) or any per-object video is missing (a prior object's render failed)
    rather than raising -- the per-object deliverables are the ones that
    matter; the combined view is a bonus the task brief calls out as
    "if cheap".
    """
    if len(labels) < 2:
        return None
    paths = [out_dir / f"real2sim_{label}.mp4" for label in labels]
    if not all(p.exists() for p in paths):
        logger.warning("render_combined_video: not all per-object videos exist, skipping")
        return None
    fps, _, _ = read_mp4_properties(paths[0])
    combined_path = out_dir / "real2sim_combined.mp4"
    with timer.step("render_combined_video", n=len(labels)):
        with VideoWriter(combined_path, fps=fps) as writer:
            for frames in zip(*(read_frames_bgr(p) for p in paths), strict=True):
                writer.write(np.concatenate(frames, axis=1))
    logger.info("wrote %s", combined_path)
    return combined_path


# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    p.add_argument("--uuid", default=_DEFAULT_UUID)
    p.add_argument("--camera", default=_DEFAULT_CAMERA)
    p.add_argument("--labels", nargs="+", default=list(_DEFAULT_LABELS))
    p.add_argument("--particles", type=int, default=_DEFAULT_N_PARTICLES)
    p.add_argument("--seed", type=int, default=_DEFAULT_SEED)
    p.add_argument("--out", type=Path, default=_DEFAULT_OUT_DIR)
    p.add_argument("--config", type=Path, default=REPO_ROOT / "configs" / "datagen_droid.yaml")
    p.add_argument("--log-level", default="INFO")
    return p.parse_args()


def main() -> int:
    args = parse_args()
    setup_logging(args.log_level)

    out_dir = ensure_dir(args.out)
    profile = DatagenProfile.from_yaml(args.config)
    timer = StepTimer("render_real2sim_demo")

    objects: dict[str, Any] = {}
    for label in args.labels:
        objects[label] = render_object(
            args.uuid, args.camera, label,
            profile=profile, out_dir=out_dir,
            n_particles=args.particles, seed=args.seed, timer=timer,
        )

    combined_path = render_combined_video(out_dir, tuple(args.labels), timer=timer)

    summary = {
        "uuid": args.uuid,
        "camera_serial": args.camera,
        "n_particles": args.particles,
        "seed": args.seed,
        "objects": objects,
        "combined_video_path": str(combined_path) if combined_path is not None else None,
        "timings": timer.summary(),
    }
    summary_path = out_dir / "summary.json"
    atomic_write(summary_path, json.dumps(summary, indent=2).encode("utf-8"))
    print(f"\nwrote {summary_path}")
    for label, obj in objects.items():
        print(
            f"  {label}: n_diverged={obj['n_diverged']}/{obj['n_particles']} "
            f"ess={obj['ess']:.3f} translation_error_mm={obj['translation_error_mm']}"
        )
    print()
    print(timer.report())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())