State, Physics, Causality, and Memory Evaluation › Object Permanence and Temporal Consistency

## Deterministic selection rule: the first six held-out trajectories in index order,
## with no resampling and no random draw. SHOW_IDX is fixed at construction time.
N_SHOW = 6
SHOW_IDX = np.arange(N_SHOW)
show_pos = pos_all[test_sl][
    SHOW_IDX
]  # (N_SHOW, N_STEPS + 1, 2) positions in metres
show_mask = mask_all[test_sl][SHOW_IDX]  # 1.0 visible, 0.0 behind the occluder
occ_lo, occ_hi = 0.5 - OCC_HALF, 0.5 + OCC_HALF

import numpy as np

## Wall bounces during occlusion are already produced by the reflecting-wall
## dynamics in simulate(); this evaluation reuses the main held-out trajectories
## used by the probes and freeze-drift diagnostic. The bounces come from the physics,
## not from a post-hoc perturbation, so no trajectory is modified after the fact.


def occluded_runs(vis_row):
    """Return maximal (start, stop) index intervals where visibility is 0."""
    runs, start = [], None
    for i, value in enumerate(vis_row):
        if value < 0.5 and start is None:
            start = i
        elif value >= 0.5 and start is not None:
            runs.append((start, i))
            start = None
    if start is not None:
        runs.append((start, len(vis_row)))
    return runs


def first_usable_run(vis_row, min_start=2):
    """First hidden run with two immediately preceding visible frames."""
    for start, stop in occluded_runs(vis_row):
        if (
            start >= min_start
            and stop > start
            and np.all(vis_row[start - 2 : start] >= 0.5)
        ):
            return start, stop
    return None


## Regression fixtures: earlier visibility somewhere in the history is not enough.
assert first_usable_run(np.array([1, 0, 1, 0, 0, 1])) is None
assert first_usable_run(np.array([1, 1, 0, 0, 1])) == (2, 4)
assert first_usable_run(np.array([0, 0, 0, 0])) is None
assert first_usable_run(np.array([1, 1, 1, 1])) is None
assert occluded_runs(np.array([1, 1, 0, 0])) == [(2, 4)]

K_MAX = 6
pos_te = pos_all[test_sl]
obs_te = obs_all[test_sl]
vel_te = vel_all[test_sl]
mask_te = mask_all[test_sl]

predictor_names = [
    "freeze",
    "constant velocity",
    "privileged constant velocity",
]
err_sum = {name: np.zeros(K_MAX) for name in predictor_names}
n_at_k = np.zeros(K_MAX)

for i in range(pos_te.shape[0]):
    for start, stop in occluded_runs(mask_te[i]):
        if start < 2 or not np.all(mask_te[i, start - 2 : start] >= 0.5):
            continue
        p_last = obs_te[i, start - 1]
        v_hat = (obs_te[i, start - 1] - obs_te[i, start - 2]) / DT
        s_true_last = pos_te[i, start - 1]
        v_true = vel_te[i, start - 1]
        for k in range(1, min(stop - start, K_MAX) + 1):
            j = start - 1 + k
            if j >= pos_te.shape[1]:
                break
            truth = pos_te[i, j]
            preds = {
                "freeze": p_last,
                "constant velocity": p_last + k * DT * v_hat,
                "privileged constant velocity": s_true_last + k * DT * v_true,
            }
            for name, pred in preds.items():
                err_sum[name][k - 1] += np.linalg.norm(pred - truth)
            n_at_k[k - 1] += 1

mean_err = {
    name: np.divide(
        err_sum[name], n_at_k, out=np.full(K_MAX, np.nan), where=n_at_k > 0
    )
    for name in predictor_names
}
## An empty contributing set must be missing, not a fabricated zero error.
_empty_error = np.divide(
    np.zeros(2), np.zeros(2), out=np.full(2, np.nan), where=np.zeros(2) > 0
)
assert np.isnan(_empty_error).all()

## Leading-order prediction of the freeze error: k * dt * mean speed on hidden frames.
## The mean speed is a deterministic statistic of vel_te and mask_te, with no sampling,
## using the same main held-out trajectories as the hidden-horizon predictors.
hidden_frames = mask_te < 0.5
mean_speed_hidden = float(np.linalg.norm(vel_te[hidden_frames], axis=-1).mean())
drift_pred = np.arange(1, K_MAX + 1) * DT * mean_speed_hidden
freeze_measured = mean_err["freeze"]

sweep_widths = [0.02, 0.05, 0.10, 0.20, 0.30]
rng_sw = np.random.default_rng(404)
p_sw, v_sw, a_sw, o_sw = simulate(rng_sw, 40, driven=False)

sweep_rows = []
for w in sweep_widths:
    m_w = visibility(p_sw, occ_half=w)
    f_tot = c_tot = t_tot = 0.0
    n_used = 0
    n_censored = 0
    gap_lengths = []
    for i in range(p_sw.shape[0]):
        run = first_usable_run(m_w[i])
        if run is None:
            continue
        start, stop = run
        k = stop - start
        if k < 1:
            continue
        truth = p_sw[i, start - 1 + k]
        p_last = o_sw[i, start - 1]
        v_hat = (o_sw[i, start - 1] - o_sw[i, start - 2]) / DT
        f_tot += np.linalg.norm(p_last - truth)
        c_tot += np.linalg.norm(p_last + k * DT * v_hat - truth)
        t_tot += np.linalg.norm(
            p_sw[i, start - 1] + k * DT * v_sw[i, start - 1] - truth
        )
        n_used += 1
        n_censored += int(stop == m_w.shape[1])
        gap_lengths.append(k)
    denominator = n_used if n_used else np.nan
    sweep_rows.append(
        (
            w,
            float(np.mean(gap_lengths)) if n_used else np.nan,
            f_tot / denominator,
            c_tot / denominator,
            t_tot / denominator,
            n_used,
            n_censored,
        )
    )