diff --git a/predicators/code_sim_learning/observation_track.py b/predicators/code_sim_learning/observation_track.py index 9952056b3..7f0ae8dfa 100644 --- a/predicators/code_sim_learning/observation_track.py +++ b/predicators/code_sim_learning/observation_track.py @@ -355,15 +355,40 @@ def topple_onsets(series_by_id: Mapping[int, Sequence[Tuple[float, float]]], def propagation_intervals(onsets: Dict[int, float]) -> Dict[int, float]: """Each onset relative to the earliest one, in seconds. - The first domino to fall defines the origin and contributes a zero - that carries no information, so it is dropped: what friction sets is - how fast the cascade travels *down the row*, not when the push - happened. + NOTHING is dropped, including the domino that defines the origin. + That domino contributes a zero, which carries no information when + both streams agree on which fell first -- and everything when they + do not. + + This used to drop every entry at the earliest time, which is a + different thing from dropping the origin whenever two dominoes tie. + They tie routinely: the sim samples one state per action, 83.3 ms, + while the track runs at 60 fps, so the pushed domino and the one + beside it land on the SAME sim step and are 5x resolvable on camera. + On run_20260819_104757 the twin toppled all five cleanly, yet + domino_3 and domino_4 both came back at 0.0833 s, both were dropped, + and the track -- which had them 350 ms apart -- kept domino_4. It + then had no counterpart and drew the full missing-cascade penalty, + reporting that the twin's chain never reached a domino it had in + fact laid flat. + + Dropping exactly ONE would need the two streams to agree on WHICH, + and neither can see the other from here: a local tie-break on id + picks the wrong one as easily as the right one, and picking wrong + reproduces the same false penalty on the other domino. Keeping every + entry needs no agreement. Where both streams do agree on the origin + it costs one residual that is identically zero. """ - if len(onsets) < 2: + # `not onsets`, NOT `len < 2`. Keeping the origin (above) while dropping a + # single-onset stream is inconsistent, and the inconsistency costs a + # residual: one stream collapses to {} while the other keeps every entry + # including its origin, so a domino BOTH streams watched fall has no + # counterpart and draws the missing-cascade penalty. That is 5 penalties + # where the pre-origin-keeping code had 4. + if not onsets: return {} first = min(onsets.values()) - return {obj_id: t - first for obj_id, t in onsets.items() if t > first} + return {obj_id: t - first for obj_id, t in onsets.items()} def sim_topple_series( @@ -432,9 +457,24 @@ def settled_xy_before_cascade( confirm_deg=confirm_deg, onset_deg=onset_deg, min_persist=min_persist) + # NOTHING when nothing topples. There is then no moment at which the two + # streams are known to describe the same arrangement, and the previous + # fallback -- states[0] -- is not merely arbitrary but reliably WRONG for + # a take that starts at the push: states[0] is the pre-prologue layout + # while the track shows the post-prologue one, so the dominoes the arm + # PLACES are compared against positions they have not occupied since + # before the episode began. On run_20260819_152448 that produced 2 of 5 + # matched, 63 times, with the unmatched set exactly the three placed + # dominoes and the matched pair exactly the two the arm never touches -- + # the signature of anchoring on the episode's start. + # + # Refusing is the honest answer. The caller contributes no residuals for + # that trajectory, which is right: it has no cascade to score anyway. + if not onsets: + return {} # Onset times are state INDICES here, since the series was built with a # step of one: only the ordering matters for choosing a state. - cut = int(min(onsets.values())) if onsets else 0 + cut = int(min(onsets.values())) settled = states[max(0, min(cut, len(states) - 1))] return { obj.name: (float(settled.get(obj, "x")), float(settled.get(obj, "y"))) @@ -445,8 +485,9 @@ def settled_xy_before_cascade( def interval_residuals(sim_intervals: Dict[int, float], obs_intervals: Dict[int, float], - missing_penalty_s: float) -> List[float]: - """``sim - obs`` per domino, in seconds, over the union of both. + missing_penalty_s: float, + scale_s: float = 1.0) -> List[float]: + """``sim - obs`` per domino over the union of both, divided by scale. A domino present in one side only is the strongest evidence the track carries -- the cascade completed under one friction and @@ -454,15 +495,24 @@ def interval_residuals(sim_intervals: Dict[int, float], ``missing_penalty_s`` rather than skipped. Skipping it would make a friction that stops the cascade early look *better* than one that reproduces it, because it would simply have fewer terms. + + ``scale_s`` divides both the differences and the penalty, turning + seconds into a fraction of whatever the caller considers the natural + span -- the units every consumer of a residual already assumes. It + defaults to 1.0, which leaves the raw seconds this returned before + the divisor existed. See + :func:`rollout_objective._interval_scale` for what is passed and + why it is read off the track rather than the rollout. """ + denom = scale_s if scale_s > 0.0 else 1.0 residuals: List[float] = [] for obj_id in sorted(set(sim_intervals) | set(obs_intervals)): sim_t = sim_intervals.get(obj_id) obs_t = obs_intervals.get(obj_id) if sim_t is None or obs_t is None: - residuals.append(missing_penalty_s) + residuals.append(missing_penalty_s / denom) continue - residuals.append(sim_t - obs_t) + residuals.append((sim_t - obs_t) / denom) return residuals diff --git a/predicators/code_sim_learning/rollout_objective.py b/predicators/code_sim_learning/rollout_objective.py index 79ee1f627..a564b34d2 100644 --- a/predicators/code_sim_learning/rollout_objective.py +++ b/predicators/code_sim_learning/rollout_objective.py @@ -39,8 +39,25 @@ # sweep evaluates the objective dozens of times and an episode's track is a # multi-megabyte JSON. Also what stops the post-processing wait below being # re-entered on every evaluation. Cleared by ``reset_track_cache``, which the -# tests use; a run only ever reads one manifest. -_TRACK_CACHE: Dict[str, List[Any]] = {} +# tests use. +# +# The KEY carries every config value the cached tracks depend on, and the +# frame transform is applied on the way out rather than before storing. +# +# Keying on the path alone was wrong twice over. The frame transform used to +# be applied before storing, so whichever caller loaded first fixed the frame +# for every later one, and a single load with the transform unset left every +# subsequent evaluation matching base-frame track positions against +# world-frame twin states -- invisible, because it degrades the id matching +# rather than raising, and on run_20260819_133802 it held matching at 2 of 5 +# dominoes for 99 evaluations. An earlier version of this comment then claimed +# the cache holds "the FILE's contents, never anything derived from a config", +# which is FALSE: load_tracks takes fallback_fps and wait_s, and for a track +# without per-frame timestamps every sample time is index/fallback_fps, baked +# straight into angles_deg. Same first-loader-poisons-the-process class, one +# field over. +_TrackKey = Tuple[str, float, float] +_TRACK_CACHE: Dict[_TrackKey, List[Any]] = {} def reset_track_cache() -> None: @@ -96,6 +113,7 @@ def compute_rollout_residuals( latent_init: Any = None, scaling: Optional[ResidualScaling] = None, config: Optional[SysIdConfig] = None, + episode_count: Optional[int] = None, ) -> np.ndarray: """Rollout residuals (predicted - observed, scaled) as a flat vector. @@ -103,11 +121,16 @@ def compute_rollout_residuals( prediction pipeline, same iteration order — the sim is deterministic, so the same theta yields the same vector, as finite-difference Jacobians require). + + ``episode_count`` is how many trajectories the CALLER holds, when + that differs from how many were passed here: see + :func:`_iter_rollout_residual_terms`. """ return np.asarray(list( _iter_rollout_residual_terms(base_env, trajectories, params, residual_features, physical_names, rules, - latent_init, scaling, config)), + latent_init, scaling, config, + episode_count)), dtype=float) @@ -143,6 +166,7 @@ def _iter_rollout_residual_terms( latent_init: Any, scaling: Optional[ResidualScaling] = None, config: Optional[SysIdConfig] = None, + episode_count: Optional[int] = None, ) -> Iterator[float]: """Yield per-feature residuals for the joint forward model. @@ -178,10 +202,28 @@ def _iter_rollout_residual_terms( loaded = _load_scored_track(config) if config.score_observed_only else None tracks: List[Any] = loaded or [] score_intervals = bool(tracks) + # One track per trajectory means each trajectory is a whole episode. + # Otherwise rest-point segmentation split ONE episode into several, and + # the track covers all of them at once. + # + # COUNT THE CALLER'S LIST, NOT THIS CALL'S. Pairing is a property of the + # whole trajectory set, and per_trajectory_rms scores segments ONE AT A + # TIME: with a single track loaded, len(tracks) == len(trajectories) is + # then 1 == 1 for every segment, and each fragment gets treated as a + # whole episode and anchored on itself. Segments before the cascade have + # nothing to anchor on, so they yielded NO residuals -- which + # per_trajectory_rms scored as a perfect RMS of 0, and the trimmer then + # kept them and dropped the one segment that held the cascade + # (run_20260820_123606: best RMS ['0', '1.497'] against a 0.1 bar, so + # the only evidence in the run was discarded as unexplainable). + n_trajectories = (episode_count + if episode_count is not None else len(trajectories)) + paired_tracks = len(tracks) == n_trajectories # Once per objective evaluation, and once per EPISODE rather than per # scored segment: the positions only line up at the episode's start. - id_maps = (_episode_id_maps(tracks, trajectories, config) + id_maps = (_episode_id_maps(tracks, trajectories, config, paired_tracks) if score_intervals else []) + episode_rollouts: List[List[State]] = [] physical = {n: params[n] for n in physical_names if n in params} rules_list = list(rules) latent_mode = bool(rules_list) and has_latent_rules(rules_list) @@ -231,9 +273,17 @@ def _run_rules_post_step(env: Any, sim_state: State, i: int) -> None: # are the twin's own simulation and including them would let the # defect this flag exists to fix outvote the real evidence by # thousands of terms to a handful. - yield from _interval_residual_terms( - sim_states, _track_for(tracks, traj_index, len(trajectories)), - id_maps[traj_index], config, summary_w) + if paired_tracks: + # One track per trajectory: each trajectory IS an episode, so + # scoring it on its own already is per-episode. + yield from _interval_residual_terms(sim_states, states, + tracks[traj_index], + id_maps[traj_index], + config, summary_w) + else: + # Segments of ONE episode. Held, not scored: see the episode + # -level yield after this loop. + episode_rollouts.append(sim_states) continue endpoint_residuals: List[float] = [] for i, sim_state in enumerate(sim_states): @@ -269,6 +319,11 @@ def _run_rules_post_step(env: Any, sim_state: State, i: int) -> None: yield from _onset_residuals([states[0]] + sim_states, states, residual_features, config.settle_tol, summary_w) + if score_intervals and not paired_tracks and episode_rollouts: + yield from _episode_interval_terms(episode_rollouts, + [s for s, _ in trajectories], + tracks[-1], id_maps[0], config, + summary_w) def _load_scored_track(config: SysIdConfig) -> Optional[Any]: @@ -288,9 +343,12 @@ def _load_scored_track(config: SysIdConfig) -> Optional[Any]: "track path is set; falling back to per-step scoring, which " "under open-loop execution scores the twin against itself.") return None - cached = _TRACK_CACHE.get(config.track_path) + key: _TrackKey = (config.track_path, float(config.track_fallback_fps), + float(config.track_wait_s)) + cached = _TRACK_CACHE.get(key) if cached is not None: - return cached + # Transform on the way OUT, never on the way in: see _TRACK_CACHE. + return [_track_in_world_frame(t, config) for t in cached] try: tracks = load_tracks(config.track_path, config.track_fallback_fps, config.track_wait_s) @@ -306,9 +364,8 @@ def _load_scored_track(config: SysIdConfig) -> Optional[Any]: "or none has been post-processed yet); falling back to per-step " "scoring.", config.track_path) return None - tracks = [_track_in_world_frame(t, config) for t in tracks] - _TRACK_CACHE[config.track_path] = tracks - return tracks + _TRACK_CACHE[key] = tracks + return [_track_in_world_frame(t, config) for t in tracks] def _track_in_world_frame(track: Any, config: SysIdConfig) -> Any: @@ -366,7 +423,8 @@ def _track_for(tracks: List[Any], index: int, n_trajectories: int) -> Any: def _episode_id_maps(tracks: List[Any], trajectories: List[RolloutTrajectory], - config: SysIdConfig) -> List[Dict[str, int]]: + config: SysIdConfig, + paired: bool) -> List[Dict[str, int]]: """Match track ids to object names once per EPISODE, not per segment. The match is positional, and the one moment the two position sets @@ -378,12 +436,17 @@ def _episode_id_maps(tracks: List[Any], trajectories: List[RolloutTrajectory], tolerance, so a per-segment match silently drops those objects and the intervals they carry. - When the counts pair one-to-one, each trajectory is its own episode - and anchors on its own initial state. Otherwise segmentation has - split something -- ``_track_for`` scores every trajectory against - the most recent track -- and the anchor is the FIRST trajectory's - initial state, which is where the episode began, before the plan - moved anything. + When ``paired``, each trajectory is its own episode and anchors on + its own initial state. Otherwise segmentation has split something -- + ``_track_for`` scores every trajectory against the most recent track + -- and the anchor is the FIRST trajectory's initial state, which is + where the episode began, before the plan moved anything. + + ``paired`` is DECIDED BY THE CALLER and passed in, never re-derived + from ``len(tracks) == len(trajectories)`` here: that comparison is a + coincidence when the caller is scoring one segment at a time, and + reading it as pairing anchors a fragment on itself. See + :func:`_iter_rollout_residual_terms`. """ # pylint: disable-next=import-outside-toplevel from predicators.code_sim_learning.observation_track import \ @@ -402,15 +465,26 @@ def _map_for(states: List[State], track: Any) -> Dict[str, int]: min_persist=config.onset_min_persist) track_xy = track.pre_cascade_xy or track.first_xy name_to_id = match_ids_by_xy(twin_xy, track_xy) - if not name_to_id: - logging.warning( - "falling back to matching track ids by object name, which " - "assumes the initialization boxes were drawn in the env's " - "own domino order") - name_to_id = track_name_to_id(states[0], prefix) - return name_to_id + if name_to_id: + return name_to_id + # Names are the fallback for a track that carries NO POSITIONS, which + # is what track_name_to_id documents itself as. They are not a + # fallback for a track that has positions the twin could not be + # anchored against: the ids are box-drawing order, and on + # run_20260819_152448 the true mapping was a permutation + # (domino_3 -> id 4, domino_4 -> id 3), so matching by name would have + # attributed each domino's topple to another one -- silently, and + # with a full set of confident-looking residuals. An empty mapping + # and no residuals is the better failure. + if track_xy: + return {} + logging.warning( + "the track carries no positions, so track ids are matched by " + "object name, which assumes the initialization boxes were drawn " + "in the env's own domino order") + return track_name_to_id(states[0], prefix) - if len(tracks) == len(trajectories): + if paired: return [ _map_for(list(states), tracks[i]) for i, (states, _) in enumerate(trajectories) @@ -422,8 +496,132 @@ def _map_for(states: List[State], track: Any) -> Dict[str, int]: return [shared] * len(trajectories) -def _interval_residual_terms(sim_states: List[State], track: Any, - name_to_id: Dict[str, int], config: SysIdConfig, +def _interval_scale(obs_intervals: Dict[int, float], penalty: float) -> float: + """Divisor putting interval residuals in the units everything expects. + + Every consumer downstream of the objective assumes residuals are + DIMENSIONLESS -- a fraction of typical motion -- because that is what + :func:`compute_residual_scaling` makes the per-step ones: each linear + feature over its observed span, each angle over pi. The interval + branch never went through it. Its residuals are SECONDS, and they + were handed to the same consumers anyway. + + The trim threshold is where that bites. It is + ``trim_rms_factor * noise_sigma`` = 0.1, meaning "10% of typical + motion", and applying it to seconds demands the twin reproduce a + cascade to within 0.1 s / sqrt(summary_weight) ~ 45 ms at the best + grid point before the fit will look at the data at all. Nothing real + passes that, so under interval scoring the trimmer dropped every + segment holding a cascade -- and, before the sibling fix in this + module, kept the ones holding none. + + The scale is the OBSERVED propagation span: the camera's own measure + of how long this cascade took, so a residual of 1.0 means "out by + the whole cascade". Read off the track and never off the rollout, + because a theta-dependent divisor is an objective a fit can game -- + stalling the chain would stretch the sim's span and shrink its own + residuals. + + Falls back to ``penalty`` when the track saw fewer than two onsets: + there is no observed span then, and the residuals that remain are + one-sided penalties, which that choice puts at 1.0 apiece. + """ + span = max(obs_intervals.values()) if obs_intervals else 0.0 + if span > 0.0: + return span + return penalty if penalty > 0.0 else 1.0 + + +def _episode_interval_terms(rollouts: List[List[State]], + recorded: List[List[State]], track: Any, + name_to_id: Dict[str, int], config: SysIdConfig, + summary_w: float) -> Iterator[float]: + """Yield ONE set of propagation-interval residuals for a whole episode. + + Rest-point segmentation is a ROLLOUT device -- multiple shooting, to + stop early divergence compounding across an entire manipulation. It + is not a statement about how the evidence divides. The track covers + the whole episode, so comparing against it is inherently an + episode-level operation, and doing it per segment created three + problems that each needed their own guard: + + * the same observed intervals were compared once per segment, so a + cascade watched by one camera was counted as many times as the + episode happened to be cut; + * segments with no cascade in them drew a missing-cascade penalty + for every observed interval, which needed a skip gate -- and that + gate could not consult the track, the only theta-independent + witness, because the track spans the whole episode while a segment + is a sub-range of it; + * when every segment skipped, the objective was exactly 0 and flat + in theta, which the fit reads as "this parameter does not matter". + run_20260819_163114 refused all five parameters that way. + + Scoring once removes all three by construction rather than by guard. + + The rollouts are concatenated with a per-segment time offset. They + are NOT one continuous simulation -- each is re-anchored at rest with + velocities zeroed, which is the whole point of the segmentation -- but + onset detection only needs each domino's fall to lie within one + segment, and a cascade runs in well under a second while segments are + cut at quiescence. A cascade straddling a boundary would be + misreported, and that is the assumption this rests on. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.observation_track import \ + interval_residuals, propagation_intervals, sim_topple_series, \ + topple_onsets + if not rollouts or not name_to_id: + return + step_s = CFG.pybullet_sim_steps_per_action / 240.0 + + def _across(segments: List[List[State]]) -> Dict[int, Any]: + """One (seconds, fall_deg) series per domino, spanning the episode.""" + joined: Dict[int, List[Any]] = {} + offset = 0.0 + for states in segments: + for obj_id, series in sim_topple_series(states, step_s, + name_to_id).items(): + joined.setdefault(obj_id, []).extend( + (t + offset, a) for t, a in series) + offset += len(states) * step_s + return joined + + def _onsets(series: Any) -> Dict[int, float]: + """Both sides detected identically, which is the point.""" + return topple_onsets(series, + confirm_deg=config.onset_confirm_deg, + onset_deg=config.onset_deg, + min_persist=config.onset_min_persist) + + sim_intervals = propagation_intervals(_onsets(_across(rollouts))) + obs_series = { + obj_id: series + for obj_id, series in track.angles_deg.items() + if obj_id in set(name_to_id.values()) + } + obs_intervals = propagation_intervals(_onsets(obs_series)) + if not sim_intervals and not obs_intervals: + # Neither the twin nor the camera saw a cascade anywhere in the + # episode. There is nothing to disagree about, and a penalty would + # claim otherwise. + logging.warning( + "no cascade in either the rollout or the track for this episode, " + "so it contributes no interval residuals. Nothing was measured; " + "this is NOT evidence that the parameters do not matter.") + return + total_steps = sum(len(s) for s in rollouts) + penalty = max(track.duration_s, step_s * total_steps) + del recorded # kept in the signature for symmetry with the per-segment path + scale = _interval_scale(obs_intervals, penalty) + for res in interval_residuals(sim_intervals, obs_intervals, penalty, + scale): + yield summary_w * res + + +def _interval_residual_terms(sim_states: List[State], recorded: List[State], + track: Any, name_to_id: Dict[str, int], + config: SysIdConfig, summary_w: float) -> Iterator[float]: """Yield (sim - observed) propagation intervals, in seconds. @@ -443,13 +641,34 @@ def _interval_residual_terms(sim_states: List[State], track: Any, # the episode's initial state is contemporaneous with the track's first # frame; see that function. if not name_to_id: - logging.warning( - "no object name starts with %r, so nothing in the rollout maps " - "onto the track's domino ids; this trajectory contributes no " - "interval residuals.", config.track_object_prefix) + # SAY WHICH. An empty mapping used to mean one thing -- no object + # carries the prefix -- and this warning still said so after I made + # it mean a second: the anchor could not be established, because + # nothing topples in the states the matcher was given. On + # run_20260819_163114 the message fired 61 times claiming no object + # starts with "domino_" while the twin held domino_0 through + # domino_4 and cascaded all five to 90 degrees. A log that + # misattributes its own cause is worse than a quiet one: it sends + # whoever reads it to the wrong place. + named = sorted(o.name for o in sim_states[0] + if o.name.startswith(config.track_object_prefix)) + if named: + logging.warning( + "no domino topples anywhere in this trajectory, so there is " + "no moment at which the twin and the track can be compared: " + "positions are matched just BEFORE the cascade, and there is " + "no cascade here. %s went unmatched and this trajectory " + "contributes no interval residuals. This is NOT evidence " + "that the parameters do not matter -- nothing was measured.", + named) + else: + logging.warning( + "no object name starts with %r, so nothing in the rollout " + "maps onto the track's domino ids; this trajectory " + "contributes no interval residuals.", + config.track_object_prefix) return step_s = CFG.pybullet_sim_steps_per_action / 240.0 - sim_series = sim_topple_series(sim_states, step_s, name_to_id) def _onsets(series: Any) -> Dict[int, float]: """Both sides detected identically, which is the point.""" @@ -458,13 +677,62 @@ def _onsets(series: Any) -> Dict[int, float]: onset_deg=config.onset_deg, min_persist=config.onset_min_persist) - sim_intervals = propagation_intervals(_onsets(sim_series)) + sim_series = sim_topple_series(sim_states, step_s, name_to_id) + # A segment with no cascade in it scores NOTHING, rather than a penalty + # per observed interval. Rest-point segmentation splits an episode into + # several scored trajectories while the track covers the whole episode, so + # a segment that is pick-and-place has no onsets to offer and every + # observed interval reads as a cascade the sim failed to reproduce. On + # run_20260819_104757 that put 21-24 WHOLE penalties into every evaluation + # -- the reported SSEs were integer multiples of one penalty -- burying + # four real residuals of 0.000-0.267 s under a penalty mass of 1.96e+05. + # Being near-constant in theta it also made every physical parameter read + # as flat, which is what the agent then declined on. + # + # BOTH sides must be empty, and requiring both is the whole subtlety. + # Skipping on the rollout alone would be actively harmful: a theta that + # STALLS the cascade also produces no onsets, so its segment would be + # skipped and score zero -- making a friction that breaks the chain look + # BETTER than one that reproduces it, the exact inversion + # interval_residuals penalises a one-sided domino to prevent. The recorded + # states are the twin's own baseline simulation under open-loop, so a + # segment where THEY cascade is a real cascade segment and keeps its + # penalties however badly this theta does. Gating on the recorded states + # alone is wrong in the other direction: a caller may hand in a degenerate + # recorded trajectory (test_the_objective_prefers_the_cascade_that_matches + # _the_track passes two identical states) while the rollout carries the + # cascade being scored. + sim_onsets = _onsets(sim_series) + if (len(sim_onsets) < 2 and + len(_onsets(sim_topple_series(recorded, step_s, name_to_id))) < 2): + # Silent ONLY when the mapping is whole. A skip means "no cascade + # here", and that reading depends on having named every domino: with + # a partial mapping the same emptiness can equally mean "the dominoes + # that fell are the ones I could not identify", and returning nothing + # then reports a flat objective built on a measurement that never + # happened. On run_20260819_133802 a stale cached track held the + # matching at 2 of 5 for 99 evaluations, every segment fell under the + # two onsets an interval needs, and the fit read the resulting zeros + # as "insensitive to friction" -- a decision made on data the + # objective had quietly declined to score. + if len(name_to_id) < len(track.angles_deg): + logging.warning( + "segment scored nothing: only %d of the track's %d domino(s) " + "could be matched, and the matched ones show no cascade. This " + "is NOT evidence that the parameters do not matter -- the " + "objective could not measure them here.", len(name_to_id), + len(track.angles_deg)) + return + + sim_intervals = propagation_intervals(sim_onsets) obs_intervals = propagation_intervals(_onsets(track.angles_deg)) # A cascade that fails to propagate in one of the two is the strongest # evidence there is, so the stand-in is the track's own span rather than # a small number: it must cost more than any real disagreement. penalty = max(track.duration_s, step_s * len(sim_states)) - for res in interval_residuals(sim_intervals, obs_intervals, penalty): + scale = _interval_scale(obs_intervals, penalty) + for res in interval_residuals(sim_intervals, obs_intervals, penalty, + scale): yield summary_w * res @@ -597,11 +865,33 @@ def per_trajectory_rms( ``scaling`` the RMS is in the scored features' native units (meters / radians per residual); with it, a dimensionless fraction of typical motion. + + NO RESIDUALS IS NOT A PERFECT FIT. An empty residual vector means + nothing was measured -- under interval scoring, a segment that holds + no cascade has nothing the track can be compared against -- and this + used to return 0.0 for it, the best score obtainable. The trimmer in + :func:`physical_sysid.fit_params_rollout_trimmed` keeps whatever + scores below its bar, so a vacuous segment was kept as ideal + evidence while the segment carrying the real cascade was dropped for + scoring above it. Infinity is the honest answer: a trajectory + nothing could be measured on must never win a comparison against one + that was actually scored. + + ``trajectories`` are scored ONE AT A TIME, so each call gets a list + of length 1; the full count goes down as ``episode_count`` so the + objective can still tell an episode from a segment of one. """ out: List[float] = [] for traj in trajectories: - res = compute_rollout_residuals(base_env, [traj], params, - residual_features, physical_names, - rules, latent_init, scaling, config) - out.append(float(np.sqrt(np.mean(res**2))) if res.size else 0.0) + res = compute_rollout_residuals(base_env, [traj], + params, + residual_features, + physical_names, + rules, + latent_init, + scaling, + config, + episode_count=len(trajectories)) + out.append( + float(np.sqrt(np.mean(res**2))) if res.size else float("inf")) return out diff --git a/predicators/pybullet_helpers/real_robot_executor.py b/predicators/pybullet_helpers/real_robot_executor.py index 972f9ed42..24580adf7 100644 --- a/predicators/pybullet_helpers/real_robot_executor.py +++ b/predicators/pybullet_helpers/real_robot_executor.py @@ -694,8 +694,15 @@ def _max_position_divergence(predicted: State, def _snapshot_perception(recorder: Any) -> MarkerlessSnapshotPerception: """The scene look that a snapshot rebuild uses instead of a live one.""" - serials = recorder.serials - serial = CFG.real_robot_snapshot_camera or (serials[0] if serials else "") + # str(), because a launcher config's "30264679" arrives as an INT: + # utils.string_to_python_object parses an all-digit serial as a number on + # the way in from the command line, and the recorder reports serials as + # strings. Comparing the two rejected a camera that was in the list, and + # said so in a message that showed the wanted serial unquoted beside a + # list of quoted ones -- which is the only visible clue. + serials = [str(s) for s in recorder.serials] + wanted = CFG.real_robot_snapshot_camera + serial = str(wanted) if wanted else (serials[0] if serials else "") if not serial: raise ValueError( "real_robot_snapshot_rebuild needs a camera to fit the scene " diff --git a/scripts/configs/predicatorv3/exp_domino_real_replay.yaml b/scripts/configs/predicatorv3/exp_domino_real_replay.yaml new file mode 100644 index 000000000..0f3093b46 --- /dev/null +++ b/scripts/configs/predicatorv3/exp_domino_real_replay.yaml @@ -0,0 +1,65 @@ +# Replay: the domino friction fit, scored against an ALREADY-RECORDED track. +# Usage: +# python scripts/local/launch_simp.py -c predicatorv3/exp_domino_real_replay.yaml +# +# Same experiment as exp_domino_real.yaml, with the two slow halves removed: +# the arm does not move and the cameras do not look. Everything downstream of +# the track -- id matching, interval residuals, the parameter sweep, and the +# agent's decision about what to declare -- runs exactly as it does live. +# +# WHY THIS EXISTS. One live episode costs a scene reset, ~110 s of arm motion +# and ~3 min of markerless post-processing, and it produces the same track +# every time the layout is the same. The objective is what has been changing, +# not the data, so iterating on the objective against a known-good recording +# is the loop that matters. run_20260820_123606 is that recording: four +# dominoes cascaded 3 -> 2 -> 1 -> 0, the twin reproduced all four, and the +# ids match 5 of 5 offline. +# +# WHAT STILL RUNS. The twin simulates the fixed plan, which is where the +# recorded trajectory comes from -- that half is pure PyBullet and was never +# the slow part. The arm is dry, so the plan's motion is a no-op at the +# hardware boundary, and perception is the captured scene file rather than a +# camera. The trajectory this produces is the same computation the live run +# performed, because under open-loop nothing corrects the twin mid-episode. +# +# WHAT THIS CANNOT TELL YOU. Whether the real world would have cascaded +# differently at a different friction. The track is fixed, so the replay +# answers "what does the fit do with this evidence", never "is the evidence +# right". Re-record when the scene or the plan changes. +--- +includes: + - exp_domino_real.yaml +ENVS: + domino_real: + FLAGS: + # -- the arm and the cameras, both off --------------------------------- + # Dry: no arm is built and arm calls are no-ops, so the plan is + # simulated and then dropped at the hardware boundary. The executor is + # still attached (real_robot_execute stays True) so the same shipping + # and batching path runs -- it just ships into nothing. + real_robot_dry: True + # "scene_file" replays domino_real_scene: cameraless, and it reports the + # captured layout, which is what the twin has to start from for its + # trajectory to match the recorded run's. + real_robot_perception: "scene_file" + # No takes, so no ZED session, no SVOs and no markerless pipeline. This + # also stops tracks.json being rewritten, which is what makes the frozen + # manifest below safe to point at. + real_robot_record_episodes: False + real_robot_process_takes: False + # Nothing to reset between episodes when nothing moved, and nothing to + # draw boxes on when no camera looked. Both of these BLOCK on a human + # (a terminal prompt and an OpenCV drag window), which would defeat the + # point of a replay. + real_robot_human_reset: False + real_robot_pick_boxes_at_start: False + real_robot_snapshot_rebuild: False + # -- the evidence ------------------------------------------------------ + # The frozen copy, NOT logs/zed_tracks/tracks.json: that file is + # rewritten by every live run, so a replay pointed at it would silently + # start scoring whatever was recorded most recently. + code_sim_learning_rollout_track_path: "logs/zed_tracks/replay_20260820_124013.json" + # The track is already on disk and complete, so there is nothing to wait + # for. Left long enough to be a real error rather than a hang if the + # manifest ever points somewhere wrong. + code_sim_learning_track_wait_s: 30 diff --git a/tests/code_sim_learning/test_observation_track.py b/tests/code_sim_learning/test_observation_track.py index 32bf16673..94d2e1b0f 100644 --- a/tests/code_sim_learning/test_observation_track.py +++ b/tests/code_sim_learning/test_observation_track.py @@ -133,15 +133,59 @@ def test_missing_frames_do_not_break_the_detector(): # -- intervals --------------------------------------------------------------- def test_intervals_are_relative_to_the_first_onset(): - """What friction sets is how fast the cascade travels down the row, so the - first onset is an origin and contributes nothing.""" + """What friction sets is how fast the cascade travels down the row, so + every onset is measured from the earliest. + + The origin is KEPT, at zero. It carries no information while both + streams agree on which domino fell first, and everything when they + do not -- see + test_a_tie_at_the_earliest_onset_keeps_both_dominoes. + """ intervals = propagation_intervals({0: 10.0, 1: 10.2, 2: 10.5}) - assert 0 not in intervals + assert intervals[0] == pytest_approx(0.0) assert intervals[1] == pytest_approx(0.2) assert intervals[2] == pytest_approx(0.5) +def test_a_tie_at_the_earliest_onset_keeps_both_dominoes(): + """The sim samples one state per action, 83.3 ms, while the track runs at + 60 fps -- so the pushed domino and the one beside it land on the SAME sim + step and are 5x resolvable on camera. + + Dropping every entry at the earliest time would delete BOTH, and the + track keeps the second, which then has no counterpart and draws the + full missing-cascade penalty -- reporting that the twin's chain never + reached a domino it had in fact laid flat. Measured on + run_20260819_104757: domino_3 and domino_4 both at 0.0833 s in the + twin, 350 ms apart on camera. + """ + tied = propagation_intervals({0: 0.0833, 1: 0.0833, 2: 0.3333}) + + assert set(tied) == {0, 1, 2}, "a tie must not delete both dominoes" + assert tied[0] == pytest_approx(0.0) + assert tied[1] == pytest_approx(0.0) + assert tied[2] == pytest_approx(0.25) + + +def test_streams_that_disagree_on_the_origin_still_compare(): + """Neither stream can see the other from inside propagation_intervals, so + dropping exactly one would need an agreement they cannot reach. + + Keeping every entry needs no agreement: where the two disagree about + which domino fell first, both sides still carry both dominoes and + the comparison yields real differences instead of two penalties. + """ + # The sim has 0 first; the track resolves 1 as first instead. + sim = propagation_intervals({0: 1.00, 1: 1.00, 2: 1.30}) + obs = propagation_intervals({0: 1.35, 1: 1.00, 2: 1.30}) + + assert set(sim) == set(obs), \ + "every domino must have a counterpart, whoever the origin is" + residuals = interval_residuals(sim, obs, missing_penalty_s=999.0) + assert 999.0 not in residuals, "no term may fall back to the penalty" + + def test_intervals_are_invariant_to_a_clock_offset(): """This is why alignment can be an event rather than a clock reading: a constant offset between the track's clock and the robot's cancels.""" @@ -157,9 +201,18 @@ def test_intervals_are_invariant_to_a_clock_offset(): assert got[key] == pytest_approx(value, abs=1e-6) -def test_one_onset_yields_no_intervals(): - """A cascade of one has nothing to say about propagation.""" - assert propagation_intervals({0: 1.0}) == {} +def test_one_onset_yields_its_origin_not_nothing(): + """A cascade of one still says WHICH domino fell, and the other stream + needs a counterpart for it. + + Collapsing to {} while the origin is kept for longer cascades is an + inconsistency that costs a residual: one stream returns nothing + while the other keeps every entry including its origin, so a domino + BOTH streams watched fall has no counterpart and draws the missing- + cascade penalty. + """ + assert propagation_intervals({0: 1.0}) == {0: 0.0} + assert propagation_intervals({}) == {} def test_a_cascade_that_stalls_on_one_side_is_penalised_not_skipped(): @@ -187,10 +240,10 @@ def test_sim_series_converts_roll_to_degrees(): assert series[0][1][0] == pytest_approx(0.0833) -def _domino_state(positions): - """A state with dominoes at the given (x, y).""" +def _domino_state(positions, roll=0.0): + """A state with dominoes at the given (x, y), optionally toppling.""" return State({ - Object(name, _DOMINO): [x, y, 0, 0, 0, 0, 0, 0] + Object(name, _DOMINO): [x, y, 0, 0, roll, 0, 0, 0] for name, (x, y) in positions.items() }) @@ -372,20 +425,30 @@ def test_ids_are_matched_once_per_episode_not_per_segment(caplog): n_frames=20, source="test", first_xy={ - 0: start["domino_0"], - 1: start["domino_1"], - 2: start["domino_2"] + 0: after_place["domino_0"], + 1: after_place["domino_1"], + 2: after_place["domino_2"] }) - # Two segments of one episode: the second begins after the place. + # Two segments of one episode: the second begins after the place and + # carries the cascade, which is where the anchor comes from. A track of a + # push-only take shows the PLACED row, not where the episode began. + toppling = [ + _domino_state(after_place, roll=math.radians(a)) + for a in _fall(steps=20) + ] trajectories = [([_domino_state(start)], []), - ([_domino_state(after_place)], [])] + ([_domino_state(after_place)] + toppling, [])] with caplog.at_level("WARNING"): - maps = _episode_id_maps([track], trajectories, SysIdConfig.from_cfg()) + maps = _episode_id_maps([track], + trajectories, + SysIdConfig.from_cfg(), + paired=False) expected = {"domino_0": 0, "domino_1": 1, "domino_2": 2} assert maps == [expected, expected], \ - "every segment of an episode maps by where the episode STARTED" + "every segment of an episode shares ONE mapping, taken from the " \ + "arrangement the cascade actually ran along" assert "could not match" not in caplog.text @@ -441,7 +504,10 @@ def _state(positions, roll): # One episode, split into two scored segments by the place. segments = [(states[:10], []), (states[10:], [])] - maps = _episode_id_maps([track], segments, SysIdConfig.from_cfg()) + maps = _episode_id_maps([track], + segments, + SysIdConfig.from_cfg(), + paired=False) expected = {"domino_0": 2, "domino_1": 0, "domino_2": 1} assert maps == [expected, expected], \ @@ -449,6 +515,82 @@ def _state(positions, roll): "episode began" +def test_no_cascade_means_no_anchor_rather_than_the_episode_start(): + """The fallback this replaces was not arbitrary, it was reliably WRONG. + + settled_xy_before_cascade used to return states[0] when nothing + toppled. For a take that starts at the push that is the pre-prologue + layout, while the track shows the post-prologue one -- so the dominoes + the arm PLACES get compared against positions they have not occupied + since before the episode began, and only the ones it never touches + still match. + + Measured on run_20260819_152448: 2 of 5 matched, 63 times, with the + unmatched set exactly the three placed dominoes. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.observation_track import \ + settled_xy_before_cascade + placed = { + "domino_0": (0.60, 1.30), + "domino_1": (0.70, 1.30), + "domino_2": (0.80, 1.30), + } + still = [_domino_state(placed)] * 8 + toppling = still + [ + _domino_state(placed, roll=math.radians(a)) for a in _fall(steps=20) + ] + + assert settled_xy_before_cascade(still, "domino_") == {}, \ + "no cascade means there is no moment the two streams are known " \ + "to share, so no anchor" + assert set(settled_xy_before_cascade(toppling, "domino_")) == set(placed) + + +def test_names_are_not_a_fallback_for_a_track_that_has_positions( + tmp_path, monkeypatch): + """track_name_to_id is for a track carrying NO positions. + + Using it when the twin merely could not be anchored is worse than + scoring nothing: the ids are box-drawing order, and on + run_20260819_152448 the true mapping was a permutation (domino_3 -> + id 4, domino_4 -> id 3), so a name match would have attributed each + domino's topple to another one -- silently, with a full set of + confident-looking residuals. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.config import SysIdConfig + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.observation_track import \ + ObservationTrack + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import \ + _episode_id_maps + del monkeypatch, tmp_path + utils.reset_config({"code_sim_learning_track_object_prefix": "domino_"}) + placed = {"domino_0": (0.60, 1.30), "domino_1": (0.70, 1.30)} + # Nothing topples, so there is no anchor -- but the track HAS positions. + trajectories = [([_domino_state(placed)] * 4, [])] + track = ObservationTrack(angles_deg={ + 0: _series(*_fall()), + 1: _series(*_fall()) + }, + n_frames=20, + source="test", + first_xy={ + 0: (0.60, 1.30), + 1: (0.70, 1.30) + }) + + maps = _episode_id_maps([track], + trajectories, + SysIdConfig.from_cfg(), + paired=True) + + assert maps == [{}], \ + "a positioned track must not be matched by name as a consolation" + + def test_paired_tracks_still_anchor_on_their_own_episode(): """One track per trajectory means no segmentation happened, so each trajectory is its own episode and anchors on its own initial state -- it @@ -486,10 +628,26 @@ def _track(first_xy): 1: layout_b["domino_0"] }), ] - trajectories = [([_domino_state(layout_a)], []), - ([_domino_state(layout_b)], [])] - maps = _episode_id_maps(tracks, trajectories, SysIdConfig.from_cfg()) + def _episode(layout): + """One episode: the layout, then its cascade. + + A cascade is required now: with nothing toppling there is no + moment the two streams are known to share, and + settled_xy_before_cascade refuses rather than anchoring on a + state that may be the wrong one. + """ + return [_domino_state(layout)] + [ + _domino_state(layout, roll=math.radians(a)) + for a in _fall(steps=20) + ] + + trajectories = [(_episode(layout_a), []), (_episode(layout_b), [])] + + maps = _episode_id_maps(tracks, + trajectories, + SysIdConfig.from_cfg(), + paired=True) assert maps == [{ "domino_0": 0, @@ -1014,6 +1172,412 @@ def _fake_rollout_states(_env, _init, _actions, physical, **_kwargs): assert sse_wrong > 10 * max(sse_true, 1e-9) +def test_the_track_cache_does_not_fix_the_frame_for_the_whole_process( + tmp_path): + """What is cached must be the FILE, never a config-dependent derivative. + + The frame transform used to be applied before storing, with the path + alone as the key, so whichever caller loaded first fixed the frame + for every later one. A single load with the transform unset then + left every subsequent evaluation matching base-frame track positions + against world-frame twin states -- which does not raise, it silently + degrades the id matching. On run_20260819_133802 that held matching + at 2 of 5 dominoes for 99 evaluations. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.config import SysIdConfig + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import \ + _load_scored_track, reset_track_cache + + # Its own fixture, with CENTRES: the frame transform moves positions, and + # the shared _cascade_track carries angles only. + frames = [{ + "index": + t, + "timestamp_ns": + int(t * (1e9 / 60.0)), + "dominoes": [{ + "id": i, + "fall_deg": 0.0, + "center_base_m": [0.55, -0.15 + 0.1 * i, 0.0] + } for i in range(4)], + } for t in range(5)] + track_path = _write_track(tmp_path, frames) + + def _loaded(yaw, xy): + """The track as a caller with this frame config would see it.""" + utils.reset_config({ + "code_sim_learning_rollout_score_observed_only": True, + "code_sim_learning_rollout_track_path": track_path, + "code_sim_learning_track_frame_yaw": yaw, + "code_sim_learning_track_frame_xy": xy, + }) + return _load_scored_track(SysIdConfig.from_cfg()) + + reset_track_cache() + # First loader has NO transform -- the case that used to poison the cache. + untransformed = _loaded(0.0, (0.0, 0.0))[0].first_xy + # A later caller asks for the quarter turn the domino env actually uses. + transformed = _loaded(1.5707963267948966, (0.75, 0.72))[0].first_xy + + assert untransformed != transformed, \ + "the second caller inherited the first caller's frame" + for obj_id, (x, y) in untransformed.items(): + want = (0.75 - y, 0.72 + x) + got = transformed[obj_id] + assert got[0] == pytest_approx(want[0], abs=1e-6) + assert got[1] == pytest_approx(want[1], abs=1e-6) + + +def test_the_cache_key_carries_the_fps_the_timings_depend_on(tmp_path): + """A track without per-frame timestamps has every sample time computed as + index/fallback_fps, baked straight into angles_deg. + + Keying the cache on the path alone therefore let the first loader + fix the TIMEBASE for the whole process too -- the same first-loader- + poisons-everyone bug as the frame transform, one field over. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.config import SysIdConfig + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import \ + _load_scored_track, reset_track_cache + + # No timestamp_ns anywhere, so the fps is what sets the times. + frames = [{ + "index": + t, + "dominoes": [{ + "id": 0, + "fall_deg": min(max((t - 4) / 8.0, 0.0), 1.0) * 90.0 + }], + } for t in range(40)] + track_path = _write_track(tmp_path, frames) + + def _at(fps): + """The track as a caller asking for this fps would see it.""" + utils.reset_config({ + "code_sim_learning_rollout_score_observed_only": True, + "code_sim_learning_rollout_track_path": track_path, + "code_sim_learning_track_fallback_fps": fps, + }) + return _load_scored_track(SysIdConfig.from_cfg())[0].angles_deg[0] + + reset_track_cache() + slow = _at(30.0) + fast = _at(120.0) + + assert slow[-1][0] == pytest_approx(fast[-1][0] * 4.0, abs=1e-6), \ + "the second caller inherited the first caller's timebase" + + +def test_a_partial_mapping_does_not_score_zero_in_silence( + tmp_path, monkeypatch, caplog): + """A skip means "no cascade here", which is only readable when every domino + could be named. + + With a partial mapping the same emptiness can mean "the dominoes + that fell are the ones I could not identify". Returning nothing then + reports a flat objective built on a measurement that never happened + -- which the fit read as "insensitive to friction". + """ + track_path = _cascade_track(tmp_path, [0, 12, 20, 24]) + still = _cascade_states([10_000] * 4)[:40] + + with caplog.at_level("WARNING"): + sse = _segment_sse(still, still, track_path, monkeypatch) + + assert sse == 0.0 + # The warning says "could be matched", so the obvious negative assertion + # -- "could not be matched" not in text -- is VACUOUSLY true and would + # pass with the guard deleted. Match the string the code actually logs. + assert "could be matched" not in caplog.text, \ + "a WHOLE mapping with no cascade is a legitimate silent skip" + + +def _cascade_track(tmp_path, onsets): + """A track whose dominoes fall at the given frame indices.""" + frames = [] + for t in range(200): + recs = [{ + "id": i, + "fall_deg": min(max((t - o) / 8.0, 0.0), 1.0) * 90.0 + } for i, o in enumerate(onsets)] + frames.append({ + "index": t, + "timestamp_ns": int(t * (1e9 / 60.0)), + "dominoes": recs + }) + return _write_track(tmp_path, frames) + + +def _segment_sse(sim_states, recorded, track_path, monkeypatch): + """What one scored segment contributes, through the real objective.""" + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning import rollout_objective + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import \ + compute_rollout_sse, reset_track_cache + monkeypatch.setattr(rollout_objective, "rollout_states", + lambda *_a, **_k: sim_states) + utils.reset_config({ + "code_sim_learning_rollout_score_observed_only": True, + "code_sim_learning_rollout_track_path": track_path, + }) + reset_track_cache() + return compute_rollout_sse(None, [(recorded, [None])], {"friction": 0.5}, + {}, ["friction"]) + + +def test_one_episode_yields_one_set_of_residuals_not_one_per_segment( + tmp_path, monkeypatch): + """The track covers the WHOLE episode, so comparing against it is an + episode-level operation. + + Scoring per segment compared the same observed intervals once per + segment -- a cascade watched by one camera counted as many times as + the episode happened to be cut. Segmentation is a rollout device + (multiple shooting, to bound divergence), not a statement about how + the evidence divides. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning import rollout_objective + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import \ + compute_rollout_residuals, reset_track_cache + track_path = _cascade_track(tmp_path, [0, 12, 20, 24]) + cascade = _cascade_states([0, 2, 4, 6]) + monkeypatch.setattr(rollout_objective, "rollout_states", + lambda *_a, **_k: cascade) + utils.reset_config({ + "code_sim_learning_rollout_score_observed_only": True, + "code_sim_learning_rollout_track_path": track_path, + }) + + def _terms(n_segments): + """Residual count for one episode cut into n pieces.""" + reset_track_cache() + return len( + compute_rollout_residuals(None, [(cascade, [None])] * n_segments, + {"friction": 0.5}, {}, ["friction"])) + + one = _terms(1) + three = _terms(3) + + assert one > 0, "the episode must score something" + assert three == one, \ + "cutting the episode into more segments must not multiply the " \ + "evidence -- the same observed intervals were being counted once " \ + "per segment" + + +def test_a_segment_with_no_cascade_scores_nothing_instead_of_penalties( + tmp_path, monkeypatch): + """Segmentation splits an episode; the track covers all of it. + + A pick-and-place segment has no onsets to offer, so every observed + interval used to read as a cascade the sim had failed to reproduce + and drew the full missing-cascade penalty. On run_20260819_104757 + that put 21-24 WHOLE penalties into every evaluation -- the SSEs + came back as integer multiples of one -- which made every physical + parameter read as flat and is what the agent then declined on. + """ + track_path = _cascade_track(tmp_path, [0, 12, 20, 24]) + still = _cascade_states([10_000] * 4)[:40] + + assert _segment_sse(still, still, track_path, monkeypatch) == 0.0, \ + "a segment with no cascade on either side must contribute nothing" + + +def test_a_theta_that_stalls_the_cascade_is_still_penalised( + tmp_path, monkeypatch): + """The trap in that skip, and why the recorded states are consulted too. + + A candidate that BREAKS the chain also produces no onsets. Skipping + on the rollout alone would score it zero -- making a friction that + stops the cascade look better than one that reproduces it, the + inversion interval_residuals penalises a one-sided domino to + prevent. + """ + track_path = _cascade_track(tmp_path, [0, 12, 20, 24]) + stalled = _cascade_states([10_000] * 4)[:40] + # The recorded states DO cascade, so this is a real cascade segment. + recorded = _cascade_states([0, 2, 4, 6]) + + assert _segment_sse(stalled, recorded, track_path, monkeypatch) > 0.0, \ + "a stalled cascade in a real cascade segment must still be penalised" + + +def test_the_objective_scores_a_cascade_the_same_however_slow_it_was( + tmp_path, monkeypatch): + """The scaling has to reach the objective, not just exist beside it. + + Two episodes disagreeing by the SAME FRACTION of their own cascade + must score identically: one cascade takes twice as long as the other + and the twin is wrong by twice as much, so the twin is equally wrong + in both. In raw seconds the slow one scores 4x the fast one purely + for having taken longer, which is what let a slow cascade sit above + the trim bar while an identically-wrong fast one passed. + """ + fast_dir = tmp_path / "fast" + slow_dir = tmp_path / "slow" + fast_dir.mkdir() + slow_dir.mkdir() + fast = _segment_sse(_cascade_states([0, 3, 6, 9]), + _cascade_states([0, 3, 6, 9]), + _cascade_track(fast_dir, [0, 10, 20, 30]), monkeypatch) + slow = _segment_sse(_cascade_states([0, 6, 12, 18]), + _cascade_states([0, 6, 12, 18]), + _cascade_track(slow_dir, [0, 20, 40, 60]), monkeypatch) + + assert fast > 0.0, "the twin disagrees with the track, so this must score" + assert slow == pytest.approx(fast, rel=1e-6), \ + "the same proportional disagreement must cost the same; scoring in " \ + "raw seconds charges the slower cascade 4x for its duration alone" + + +def test_the_episode_path_scales_its_residuals_too(tmp_path, monkeypatch): + """The two interval paths are wired separately, so both need proving. + + One track against one trajectory scores through + ``_interval_residual_terms``; one track against the several segments + an episode was cut into scores through ``_episode_interval_terms``. + A fix applied to one and not the other is invisible in a test that + only drives the first. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning import rollout_objective + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import \ + compute_rollout_sse, reset_track_cache + + def _episode_sse(sim_states, track_path): + """One episode cut into two segments, so the counts cannot pair.""" + monkeypatch.setattr(rollout_objective, "rollout_states", + lambda *_a, **_k: sim_states) + utils.reset_config({ + "code_sim_learning_rollout_score_observed_only": True, + "code_sim_learning_rollout_track_path": track_path, + }) + reset_track_cache() + return compute_rollout_sse(None, [(sim_states, [None])] * 2, + {"friction": 0.5}, {}, ["friction"]) + + fast_dir = tmp_path / "fast" + slow_dir = tmp_path / "slow" + fast_dir.mkdir() + slow_dir.mkdir() + fast = _episode_sse(_cascade_states([0, 3, 6, 9]), + _cascade_track(fast_dir, [0, 10, 20, 30])) + slow = _episode_sse(_cascade_states([0, 6, 12, 18]), + _cascade_track(slow_dir, [0, 20, 40, 60])) + + assert fast > 0.0, "the twin disagrees with the track, so this must score" + assert slow == pytest.approx(fast, rel=1e-6), \ + "the episode path must scale by the observed span as well" + + +def test_interval_residuals_are_a_fraction_of_the_observed_span(): + """Seconds are the wrong units for every consumer of a residual. + + The trim threshold is ``trim_rms_factor * noise_sigma`` = 0.1, + meaning "10% of typical motion" because per-step residuals go + through compute_residual_scaling. Interval residuals never did, so + the same bar demanded the twin reproduce a cascade to ~45 ms before + the fit would look at it, and every cascade-bearing segment was + dropped as unexplainable. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.rollout_objective import _interval_scale + + # A real cascade: the span is the last domino's interval. + obs = {3: 0.0, 2: 0.7669, 1: 0.9001, 0: 1.0668} + sim = {3: 0.0, 2: 0.0833, 1: 0.4167, 0: 0.5833} + scale = _interval_scale(obs, penalty=25.62) + + assert scale == pytest.approx(1.0668), \ + "the scale is the observed propagation span, read off the track" + scaled = interval_residuals(sim, obs, 25.62, scale) + raw = interval_residuals(sim, obs, 25.62) + + assert scaled == pytest.approx([r / 1.0668 for r in raw]) + assert max(abs(r) for r in scaled) < 1.0, \ + "a disagreement smaller than the whole cascade must score under 1.0" + # The divisor must not come from the rollout: a theta that stalls the + # chain would stretch the sim's span and shrink its own residuals. + assert _interval_scale({}, penalty=25.62) == pytest.approx(25.62), \ + "with no observed span the penalty is the scale, so a one-sided " \ + "domino costs 1.0 rather than an unbounded number of seconds" + + +def test_a_trajectory_with_no_residuals_scores_infinite_not_zero(monkeypatch): + """Nothing measured must never outrank something measured. + + ``per_trajectory_rms`` used to turn an empty residual vector into an + RMS of 0.0 -- the best score obtainable. Under interval scoring a + segment holding no cascade yields exactly that empty vector, so on + run_20260820_123606 the trimmer saw best RMS ['0', '1.497'] against + a 0.1 bar, kept the segment that measured nothing and dropped the + one carrying the only cascade in the run. + """ + # pylint: disable-next=import-outside-toplevel + import numpy as np + + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning import rollout_objective + + def _fake(_env, trajectories, *_a, **_k): + """Empty for the first trajectory, two real residuals for the + second.""" + return (np.asarray([], dtype=float) if trajectories[0][0] == "empty" + else np.asarray([0.3, 0.4], dtype=float)) + + monkeypatch.setattr(rollout_objective, "compute_rollout_residuals", _fake) + rms = rollout_objective.per_trajectory_rms(None, [("empty", []), + ("real", [])], {}, {}, + []) + + assert math.isinf(rms[0]), \ + "a trajectory nothing could be measured on must not score 0.0, " \ + "which is the best RMS there is and beats every real measurement" + assert rms[1] == pytest.approx(math.sqrt((0.3**2 + 0.4**2) / 2)) + assert rms[0] > rms[1], \ + "the unmeasured trajectory must rank WORSE than the measured one" + + +def test_per_trajectory_rms_reports_the_callers_own_episode_count(monkeypatch): + """Pairing is a property of the whole set, not of one call's sublist. + + ``per_trajectory_rms`` scores segments one at a time, so every call + reaches the objective with a list of length 1. With a single track + loaded, ``len(tracks) == len(trajectories)`` is then 1 == 1 and each + fragment was treated as a whole episode and anchored on itself -- + which is why segments before the cascade reported all five dominoes + unmatched. The full count has to travel with the call. + """ + # pylint: disable-next=import-outside-toplevel + import numpy as np + + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning import rollout_objective + + seen = [] + + def _fake(_env, trajectories, *_a, **kwargs): + """Record what the objective was told about the caller's list.""" + seen.append((len(trajectories), kwargs.get("episode_count"))) + return np.asarray([1.0], dtype=float) + + monkeypatch.setattr(rollout_objective, "compute_rollout_residuals", _fake) + rollout_objective.per_trajectory_rms(None, [("a", []), ("b", []), + ("c", [])], {}, {}, []) + + assert seen == [(1, 3), (1, 3), (1, 3)], \ + "each call scores one trajectory but must report that the caller " \ + "holds 3, so one track cannot be mistaken for a per-episode pairing" + + def pytest_approx(value, abs=1e-9): # pylint: disable=redefined-builtin """Local approx so the comparisons above read as equations.""" return pytest.approx(value, abs=abs) diff --git a/tests/pybullet_helpers/test_real_robot_executor.py b/tests/pybullet_helpers/test_real_robot_executor.py index afbb1c62e..87ec43159 100644 --- a/tests/pybullet_helpers/test_real_robot_executor.py +++ b/tests/pybullet_helpers/test_real_robot_executor.py @@ -1766,6 +1766,34 @@ def _watched_ensure_boxes(*args, **kwargs): return seen +def test_a_numeric_camera_serial_from_a_config_is_still_recognised(): + """An all-digit serial arrives from a launcher config as an INT. + + utils.string_to_python_object parses "30264679" as a number on the way + in from the command line, while the recorder reports its serials as + strings -- so the membership test rejected a camera that was in the + list, and the message showed the wanted serial unquoted beside a list + of quoted ones, which was the only visible clue. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.pybullet_helpers.real_robot_executor import \ + _snapshot_perception + + class _Rec: + """Only the attribute the lookup reads.""" + serials = ["32294776", "30264679"] + + utils.reset_config({ + "real_robot_snapshot_camera": 30264679, + "real_robot_snapshot_frames": 5, + }) + + perception = _snapshot_perception(_Rec()) + + # pylint: disable-next=protected-access + assert perception._serial == "30264679" + + def test_boxes_are_drawn_at_the_boundary_when_the_take_opens_later( recorder, tmp_path): """After the prologue rearranges the row, before the take opens.