Spaces:
Paused
Paused
| """Closed-loop behavioral evaluation: planted-scenario sanity + provenance.""" | |
| from __future__ import annotations | |
| from sensorflow.nextgen.causal import causal_replay | |
| from sensorflow.nextgen.closedloop import demo_emergence_scenario, run_closed_loop | |
| from sensorflow.nextgen.models import ( | |
| CAUSAL_BEHAVIORAL, CAUSAL_METRIC_ONLY, DataLabel, | |
| ) | |
| def _run(demo, **kw): | |
| return run_closed_loop(demo["actors"], demo["environment"], | |
| demo["scenario_id"], DataLabel.COUNTERFACTUAL, **kw) | |
| def test_late_detection_shortens_ttc_and_margin(): | |
| """Planted case: the same scenario with the critical pedestrian missed | |
| until late must produce shorter TTC and smaller safety margin.""" | |
| demo = demo_emergence_scenario() | |
| ped = demo["critical_instance_id"] | |
| on_time = _run(demo, seed=0, corrected=True) | |
| late = _run(demo, seed=0, | |
| faults=[{"type": "miss", "instance_id": ped, | |
| "from_s": 0.0, "until_s": 3.2}]) | |
| assert late.metrics.min_ttc_s <= on_time.metrics.min_ttc_s | |
| assert late.metrics.min_separation_m < on_time.metrics.min_separation_m | |
| assert late.metrics.safety_margin_m < on_time.metrics.safety_margin_m | |
| def test_behavioral_metrics_sane_and_deterministic(): | |
| demo = demo_emergence_scenario() | |
| a1 = _run(demo, seed=4) | |
| a2 = _run(demo, seed=4) | |
| assert a1.metrics == a2.metrics # seeded determinism | |
| assert a1.metrics.max_deceleration_mps2 <= 8.0 + 1e-6 | |
| assert a1.metrics.planner_interventions >= 1 # ped triggers braking | |
| assert a1.metrics.min_separation_m is not None | |
| assert a1.open_loop["frame_recall"] is not None # open-loop attached | |
| assert a1.data_label == DataLabel.COUNTERFACTUAL # label carried | |
| def test_corrected_mode_is_perfect_perception(): | |
| demo = demo_emergence_scenario() | |
| corrected = _run(demo, seed=0, corrected=True) | |
| assert corrected.open_loop["frame_recall"] == 1.0 | |
| assert corrected.perception_mode == "corrected" | |
| def test_causal_replay_separates_metric_only_from_behavioral(): | |
| """Planted pair: missed crossing pedestrian is BEHAVIORALLY_CONSEQUENTIAL; | |
| a cosmetic classification flip is METRIC_ONLY.""" | |
| demo = demo_emergence_scenario() | |
| ped = demo["critical_instance_id"] | |
| missed = causal_replay(demo["actors"], demo["environment"], | |
| demo["scenario_id"], DataLabel.COUNTERFACTUAL, | |
| seed=0, | |
| faults=[{"type": "miss", "instance_id": ped, | |
| "from_s": 0.0, "until_s": 4.0}]) | |
| assert missed.verdict == CAUSAL_BEHAVIORAL | |
| assert missed.diffs["collision_actual"] or \ | |
| missed.diffs["min_separation_diff_m"] < -0.5 | |
| answers = [c["answer"] for c in missed.causal_chain] | |
| assert answers == [True, True, True] # full causal chain | |
| cosmetic = causal_replay(demo["actors"], demo["environment"], | |
| demo["scenario_id"], DataLabel.COUNTERFACTUAL, | |
| seed=0, | |
| faults=[{"type": "misclassify", | |
| "instance_id": ped, | |
| "as_class": "cyclist"}]) | |
| assert cosmetic.verdict == CAUSAL_METRIC_ONLY | |
| assert cosmetic.causal_chain[2]["answer"] is False | |
| # the flip IS visible open-loop — that's the point of METRIC_ONLY | |
| assert cosmetic.actual.open_loop["n_misclassified"] > 0 | |
| def test_causal_result_carries_data_label_through_report(): | |
| demo = demo_emergence_scenario() | |
| res = causal_replay(demo["actors"], demo["environment"], | |
| demo["scenario_id"], DataLabel.COUNTERFACTUAL, seed=0) | |
| assert res.data_label == DataLabel.COUNTERFACTUAL | |
| assert res.actual.data_label == DataLabel.COUNTERFACTUAL | |
| assert res.corrected.data_label == DataLabel.COUNTERFACTUAL | |