Spaces:
Paused
Paused
File size: 3,578 Bytes
d4d5551 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """Validity gate: rejection of implausible scenarios + weight-cap policy."""
from __future__ import annotations
from sensorflow.nextgen import counterfactual as cf
from sensorflow.nextgen import validity as v
from sensorflow.nextgen.models import TransformationStep, ValidityReport
def _validate(scenario):
b = cf.load_bundle(scenario.scenario_id)
return v.validate_scenario(
scenario.scenario_id, b["sequence"], b["actors"],
scenario.environment, scenario.provenance.seed,
source_overlap_pairs=b["source_overlap_pairs"],
source_features=b["source_features"])
def test_gate_rejects_planted_physically_implausible_scenario():
s = cf.generate_counterfactuals(
[TransformationStep(kind="actors.teleport", params={"jump_m": 30.0})],
seed=7, n_scenarios=1, frames_per_sequence=40)[0]
rep = _validate(s)
assert rep.accepted is False
assert rep.evaluation_weight == 0.0
assert any("acceleration" in r or "mismatch" in r for r in rep.reasons)
phys = next(c for c in rep.checks if c["check"] == "physical_plausibility")
assert phys["passed"] is False
def test_gate_accepts_plausible_transformation_with_full_scores():
s = cf.generate_counterfactuals(
[TransformationStep(kind="actors.occluded_emergence")],
seed=7, n_scenarios=1, frames_per_sequence=40)[0]
rep = _validate(s)
assert rep.accepted is True
assert rep.evaluation_weight == 1.0
assert 0.8 <= rep.simulation_fidelity_score <= 1.0
assert 0.5 <= rep.counterfactual_validity <= 1.0
assert {c["check"] for c in rep.checks} == {
"physical_plausibility", "temporal_consistency", "sensor_consistency",
"identity_trajectory_consistency", "distribution_similarity"}
def test_per_scenario_weight_policy():
assert v.weight_policy(False, 0.95) == (0.0, False) # rejected -> 0
assert v.weight_policy(True, 0.95) == (1.0, False) # high fidelity
w, capped = v.weight_policy(True, 0.65) # low fidelity
assert capped is True and w == v.LOW_FIDELITY_WEIGHT_CAP
def _report(sid, accepted, fidelity):
w, capped = v.weight_policy(accepted, fidelity)
return ValidityReport(scenario_id=sid, accepted=accepted,
simulation_fidelity_score=fidelity,
counterfactual_validity=fidelity,
realism_confidence=fidelity,
evaluation_weight=w, weight_capped=capped)
def test_low_fidelity_scenarios_cannot_dominate_suite():
# 10 low-fidelity accepted + 2 high-fidelity: uncapped, low fidelity
# would carry 3.0 of 5.0 total weight (60%). The suite policy must cap
# the low-fidelity share at LOW_FIDELITY_SUITE_SHARE.
reports = ([_report(f"low-{i}", True, 0.6) for i in range(10)]
+ [_report(f"high-{i}", True, 0.95) for i in range(2)]
+ [_report("rejected", False, 0.9)])
out = v.apply_suite_weight_policy(reports)
assert out["scaled_down"] is True
assert out["low_fidelity_share"] <= v.LOW_FIDELITY_SUITE_SHARE + 1e-9
assert out["weights"]["rejected"] == 0.0
assert out["weights"]["high-0"] == 1.0
assert out["weights"]["low-0"] < v.LOW_FIDELITY_WEIGHT_CAP
def test_suite_policy_leaves_high_fidelity_majorities_alone():
reports = ([_report(f"high-{i}", True, 0.9) for i in range(8)]
+ [_report("low-0", True, 0.7)])
out = v.apply_suite_weight_policy(reports)
assert out["scaled_down"] is False
assert out["weights"]["low-0"] == v.LOW_FIDELITY_WEIGHT_CAP
|