Buckets:
| """cwm3/data/build_kin_causal_v4b.py — hands_on / hand-event label v4b (pre-registered docs/decisions.md 2026-09-16T23:00:36Z, last row). | |
| Imports data/build_kin_causal_v4.py (v4, FROZEN) as a module and runs ITS build (V4.main -> V4.build_route_series -> the v4 causal | |
| ellipse series, hysteresis, persistence, grid events, v2-identity assertions, kinhist hard link) with exactly two overrides, applied | |
| to the imported v4 module only for the duration of the build (context manager `patched`; restored afterwards, so the v4 module keeps | |
| its v4 semantics in the same process — the v4 tests keep passing when both modules are imported): | |
| 1. STATE DISTANCE CONVENTION (the only change in the label definition): | |
| v4 : d = min over wrists of the exact SIGNED distance to the FILLED ellipse (negative inside -> the interior is 'on') | |
| v4b: d_rim = min over wrists of |signed distance to the fitted ellipse| (rim band on both sides of the rim; interior NOT 'on') | |
| The absolute value is taken PER WRIST before the min (the wrist nearest to the rim decides): a wrist deep inside the ellipse | |
| (lap / torso projected behind the wheel in 2-D) can no longer make the state ON. Implemented by calling V4.wheel_signed_distance | |
| once per wrist (the other wrist masked) and taking np.fmin of the absolute values — the v4 distance code is reused, not copied. | |
| hands_on = V4.hyst_hands_v4 on d_rim: ON if d_rim < 0.04, OFF if d_rim > 0.08, otherwise keep; reset after a gap > 10 frames (1 s), | |
| undecided (NaN) after a reset until the first decisive frame. hand_leave / hand_return = V3.confirmed_transitions (>= 1.0 s | |
| persistence before and after, stamp = first frame of the new state); S0_c = raw causal state at the anchor (no lookahead). | |
| 2. OUTPUT: results/targets_kin_causal_routes_v4b/ (label_version "v4b", wheel_ref_rule text, run-json name build_kin_causal_routes_v4b). | |
| UNCHANGED by construction (same code path, same constants, same per-device RNG): the causal per-device RANSAC ellipse fit of the rim | |
| from both wrists over manual moving frames (refit cadence 50 frames, gate >= 200 inliers & >= 90 deg arc, undefined -> NaN, NO fallback | |
| to a point rule), head_turned / head_turn_onset, and ALL input arrays (K_c / K0_c / Kvalid / Kframe / K0valid / row_ok / route_decoded | |
| asserted bit-identical to results/targets_kin_causal_routes_v2 by the v4 build; kinhist_causal_* hard-linked to the v2 files). | |
| Because the ellipse series is deterministic given the pose, ellipse0_c / wheel_ref_defined / route_refs ell__<uid> are expected to be | |
| identical to the v4 files (checked by eval/kin_v4b_build_stats.py and tests/test_kin_causal_v4b.py). In the output files | |
| `dwheel0_c` / `d4__<uid>` hold d_rim (>= 0 where finite). Tests: tests/test_kin_causal_v4b.py. | |
| """ | |
| import contextlib, sys | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "data")); sys.path.insert(0, str(ROOT / "common")) | |
| import build_kin_causal_v4 as V4 # noqa: E402 (v4 machinery, frozen file, imported not copied) | |
| OUT_V4B = ROOT / "results" / "targets_kin_causal_routes_v4b" | |
| LABEL_VERSION = "v4b" | |
| HYST_ON, HYST_OFF, PERSIST_S, GAP_RESET, FPS = V4.HYST_ON, V4.HYST_OFF, V4.PERSIST_S, V4.GAP_RESET, V4.FPS # unchanged: 0.04 / 0.08 / 1.0 s / 10 frames | |
| _V4_DIST_TEXT = "d = min over wrists of the exact SIGNED distance to the filled ellipse (negative inside)" | |
| _V4B_DIST_TEXT = "d_rim = min over wrists of |exact signed distance to the ellipse| (RIM BAND on both sides of the rim; the interior is NOT on; v4b)" | |
| assert _V4_DIST_TEXT in V4.WHEEL_RULE, "v4 rule text changed — update the v4b override" | |
| WHEEL_RULE = V4.WHEEL_RULE.replace(_V4_DIST_TEXT, _V4B_DIST_TEXT) | |
| RUN_NAME_V4, RUN_NAME_V4B = "build_kin_causal_routes_v4", "build_kin_causal_routes_v4b" | |
| _V4_SIGNED_DISTANCE = V4.wheel_signed_distance # the ORIGINAL v4 function, bound at import (inside `patched` V4.wheel_signed_distance is v4b's) | |
| # ------------------------------------------------------------------------------------------------------------------ v4b primitive | |
| def wheel_rim_distance(q, ell): | |
| """d_rim[N] = min over wrists of |signed distance of the wrist to the ellipse ell[N,5]| (NaN where the ellipse or both wrists are undefined). | |
| Uses the original V4.wheel_signed_distance (bound at import as _V4_SIGNED_DISTANCE) per wrist, the other wrist masked to NaN (np.fmin | |
| ignores it), so the exact Eberly distance of v4 is reused unchanged; only the sign convention differs: |.| per wrist, then the min.""" | |
| qL = np.array(q, np.float32, copy=True); qL[:, 2:4] = np.nan | |
| qR = np.array(q, np.float32, copy=True); qR[:, 0:2] = np.nan | |
| return np.fmin(np.abs(_V4_SIGNED_DISTANCE(qL, ell)), np.abs(_V4_SIGNED_DISTANCE(qR, ell))).astype(np.float32) | |
| def hyst_hands_v4b(d_rim): | |
| """Unchanged v4 hysteresis (ON < 0.04, OFF > 0.08, reset after > 1 s gap) applied to d_rim.""" | |
| return V4.hyst_hands_v4(d_rim) | |
| def run_name(name): | |
| """Run-json / input-listing name of the v4 build -> v4b.""" | |
| return name.replace(RUN_NAME_V4, RUN_NAME_V4B) | |
| def patched(): | |
| """Apply the two v4b overrides to the imported v4 module for the duration of the block; restore v4 afterwards (also on error).""" | |
| keys = ("wheel_signed_distance", "OUT_V4", "LABEL_VERSION", "WHEEL_RULE", "write_run_json") | |
| saved = {k: getattr(V4, k) for k in keys} | |
| def _write_run_json(out_dir, name, code_files, *a, **k): | |
| return saved["write_run_json"](out_dir, run_name(name), [Path(__file__)] + list(code_files), *a, **k) | |
| V4.wheel_signed_distance = wheel_rim_distance; V4.OUT_V4 = OUT_V4B; V4.LABEL_VERSION = LABEL_VERSION; V4.WHEEL_RULE = WHEEL_RULE | |
| V4.write_run_json = _write_run_json | |
| try: | |
| yield | |
| finally: | |
| for k, v in saved.items(): setattr(V4, k, v) | |
| # ------------------------------------------------------------------------------------------------------------------ main | |
| def main(argv=None): | |
| argv = sys.argv[1:] if argv is None else list(argv) | |
| print(f"v4b build = v4 build with the RIM-BAND distance convention: d_rim = min over wrists |signed distance to the ellipse| (interior NOT on), " | |
| f"hysteresis ON < {HYST_ON} / OFF > {HYST_OFF}, gap reset {GAP_RESET} frames, persistence {PERSIST_S} s (unchanged); default out {OUT_V4B}", flush=True) | |
| with patched(): | |
| V4.main(argv) | |
| print("KIN_CAUSAL_ROUTES_V4B_DONE", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 6.53 kB
- Xet hash:
- 51638cdd3e186014cbe3ebdf2a50ef9d7d95e0d61d220da88984daf87e4904e1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.