from __future__ import annotations import ast import csv import json from pathlib import Path import numpy as np ROOT = Path(__file__).resolve().parents[1] SOURCE = ROOT / "src/drl_pinball/eval/infer_train.py" def _load_helpers(*names: str): tree = ast.parse(SOURCE.read_text()) selected = [node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names] namespace = { "Path": Path, "Dict": dict, "Any": object, "List": list, "Tuple": tuple, "np": np, "json": json, "csv": csv, "hashlib": __import__("hashlib"), "CaseSpec": object, "shutil": __import__("shutil"), "NUM_STEPS": 360, "TAIL_WINDOW": 180, "_POLICY_SENSOR_UNITS": "legacy-policy-v1", "REFERENCE_TABLE_ATOL": 1e-6, "REPRO_REWARD_ATOL": 0.02, "REPRO_COMPONENT_ATOL": 0.02, "REPRO_DTW_ATOL": 0.02, "REPRO_ACTION_MEAN_ATOL": 0.03, "OUT_BASE": ROOT / "src/drl_pinball/eval/output/train", } exec(compile(ast.Module(body=selected, type_ignores=[]), str(SOURCE), "exec"), namespace) return [namespace[name] for name in names] def test_vecnormalize_is_frozen_for_canonical_inference(): (freeze,) = _load_helpers("_freeze_vecnormalize") class FakeVecNormalize: training = True norm_reward = True vec = freeze(FakeVecNormalize()) assert vec.training is False assert vec.norm_reward is False def test_seed_signal_serialization_contains_only_compact_series(tmp_path): (save_signals,) = _load_helpers("_save_seed_signals") values = np.ones(4, dtype=np.float32) result = { "seed": "43", "sensors": np.ones((4, 6)), "forces": np.ones((4, 6)), "actions": np.ones((4, 3)), "rewards": values, "r_cd_series": values, "r_cl_series": values, "r_sim_series": values, "sim_raw_series": values, } path = save_signals(tmp_path, result) with np.load(path) as artifact: assert set(artifact.files) == {"sensors", "forces", "actions", "rewards", "r_cd", "r_cl", "r_sim", "sim_raw"} assert not ({"ux", "uy", "vorticity"} & set(artifact.files)) def test_evaluator_contract_is_canonical_and_gpu_imports_are_lazy(): source = SOURCE.read_text() assert "scene_manifest" not in source assert "provenance" not in source assert "from drl_pinball.case_registry import" in source assert "NUM_STEPS = 360" in source assert "TAIL_WINDOW = 180" in source assert "deterministic=True" in source assert "vmin=-0.001, vmax=0.001" in source tree = ast.parse(source) top_imports = [node for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom))] assert all("pycuda" not in ast.unparse(node) for node in top_imports) def test_prepare_output_preserves_existing_baseline(tmp_path): (prepare,) = _load_helpers("_prepare_output") existing = tmp_path / "kar_re100" existing.mkdir() sentinel = existing / "metrics.json" sentinel.write_text("baseline") try: prepare("kar_re100", False, tmp_path) except FileExistsError as exc: assert "--overwrite" in str(exc) else: raise AssertionError("existing baseline must be protected") assert prepare("kar_re100", True, tmp_path) == existing assert not sentinel.exists() def test_illusion_bundle_uses_verified_calibration_harmonics(tmp_path): load_json, sha256, schema, adapt, lineage, resolve = _load_helpers( "_load_json", "_sha256", "_calibration_schema", "_adapt_native_illusion_products", "_normalizer_lineage", "_resolve_seed_artifacts") lineage.__globals__["CASE_IDS"] = ("ill_1L",) resolve.__globals__.update( _sha256=sha256, _load_json=load_json, _calibration_schema=schema, _adapt_native_illusion_products=adapt, _normalizer_lineage=lineage) class Case: case_id = "ill_1L" seeds = (43,) scene_type = "illusion" si = 1200 config_path = tmp_path / "config.json" calibration_path = tmp_path / "calibrations/ill_1L/calibration.json" def model_dir(self, seed): return tmp_path / f"ill_1L_seed{seed}/models" case = Case() run = case.model_dir(43).parent (run / "models").mkdir(parents=True) case.calibration_path.parent.mkdir(parents=True) case.calibration_path.write_text(json.dumps({ "schema_version": "drl-pinball-calibration-v2", "sensor_units": "celeris-area-time-average", })) (run / "models/best_model.zip").write_bytes(b"model") (run / "vec_normalize.pkl").write_bytes(b"normalizer") np.save(run / "target.npy", np.ones((150, 6), dtype=np.float32)) np.save(case.calibration_path.parent / "target.npy", np.ones((150, 6), dtype=np.float32)) (run / "calibration.json").write_text(json.dumps({"SENSOR_CC": 78.0})) try: resolve(case, 43) except FileNotFoundError as exc: assert "target_harmonics.json" in str(exc) assert "registry calibration bundle" in str(exc) else: raise AssertionError("missing calibration harmonics must fail") harmonics_path = case.calibration_path.parent / "target_harmonics.json" harmonics = [ {"dc": float(i + 1), "amps": [float(i + 2)], "freqs": [0.1], "phases": [0.2]} for i in range(8) ] harmonics_path.write_text(json.dumps(harmonics)) np.save(case.calibration_path.parent / "target.npy", np.zeros((150, 6), dtype=np.float32)) try: resolve(case, 43) except ValueError as exc: assert "target mismatch" in str(exc) assert "SHA256" in str(exc) else: raise AssertionError("mismatched calibration target must fail") np.save(case.calibration_path.parent / "target.npy", np.ones((150, 6), dtype=np.float32)) bundle = resolve(case, 43) assert bundle["target_path"] == (run / "target.npy").resolve() assert bundle["harmonics_path"] == harmonics_path.resolve() assert np.array_equal(bundle["target_states"], np.full((150, 6), 78.0, dtype=np.float32)) adapted = bundle["target_harmonics"] assert adapted[0]["dc"] == 78.0 and adapted[0]["amps"] == [156.0] assert adapted[0]["freqs"] == [0.1] and adapted[0]["phases"] == [0.2] assert adapted[5]["dc"] == 468.0 assert adapted[6] == harmonics[6] and adapted[7] == harmonics[7] metadata = bundle["illusion_native_to_legacy_adaptation"] assert metadata["applied"] is True and metadata["factor"] == 78.0 assert metadata["source_target_sha256"] == metadata["registry_target_sha256"] def test_normalizer_lineage_selects_d075_best_and_others_final(tmp_path): (lineage,) = _load_helpers("_normalizer_lineage") lineage.__globals__["CASE_IDS"] = ("kar_d075", "kar_re60") d075_path, d075_reason = lineage("kar_d075", tmp_path) other_path, other_reason = lineage("kar_re60", tmp_path) assert d075_path == tmp_path / "best_vecnormalize.pkl" assert "A/B" in d075_reason assert other_path == tmp_path / "vec_normalize.pkl" assert "final compatibility alias" in other_reason try: lineage("unknown", tmp_path) except ValueError as exc: assert "No retained-artifact normalizer lineage" in str(exc) else: raise AssertionError("unregistered lineage must fail closed") def test_action_means_use_tail_window_only(): (build_metrics,) = _load_helpers("_build_metrics") actions = np.vstack((np.full((180, 3), 10.0), np.array([[1.0, 2.0, 3.0]] * 180))) best = { "seed": "45", "avg_reward": 0.9, "r_cd": 0.8, "r_cl": 0.7, "r_sim": 0.6, "sim_raw": 0.5, "actions": actions, } metrics = build_metrics("kar_re100", best) assert (metrics["aF_mean"], metrics["aT_mean"], metrics["aB_mean"]) == (1.0, 2.0, 3.0) assert metrics["action_mean_window"] == 180 def test_metrics_only_returns_before_signals_and_vorticity_paths(): source = SOURCE.read_text() tree = ast.parse(source) evaluate = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "evaluate_case") metrics_guard = next( node for node in evaluate.body if isinstance(node, ast.If) and ast.unparse(node.test) == "metrics_only" ) assert isinstance(metrics_guard.body[-1], ast.Return) guard_line = metrics_guard.lineno save_seed_line = next(node.lineno for node in ast.walk(evaluate) if isinstance(node, ast.Call) and ast.unparse(node.func) == "_save_seed_signals") replay_lines = [node.lineno for node in ast.walk(evaluate) if isinstance(node, ast.Call) and ast.unparse(node.func) in {"_save_vorticity", "_generate_target_vorticity"}] assert save_seed_line < guard_line save_parent = next(node for node in ast.walk(evaluate) if isinstance(node, ast.If) and any(getattr(child, "lineno", -1) == save_seed_line for child in ast.walk(node))) assert ast.unparse(save_parent.test) == "not metrics_only" assert replay_lines and all(line > guard_line for line in replay_lines) def _configure_validation(validate, load_json, tmp_path, metrics, csv_values, json_values): metrics_dir = tmp_path / "out/kar_re100" metrics_dir.mkdir(parents=True) (metrics_dir / "metrics.json").write_text(json.dumps(metrics)) fields = [ "case_id", "selected_seed", "eval_reward_mean", "eval_r_cd_mean", "eval_r_cl_mean", "eval_r_sim_mean", "dtw_similarity", "action_front_mean", "action_top_mean", "action_bottom_mean", ] csv_path = tmp_path / "latest.csv" csv_path.write_text( ",".join(fields) + "\n" + ",".join(str(csv_values[field]) for field in fields) + "\n" ) json_path = tmp_path / "latest.json" json_path.write_text(json.dumps({ "scope": {"eval_steps": 360, "action_tail_steps": 180}, "evaluation": [json_values], })) validate.__globals__.update( OUT_BASE=tmp_path / "out", REFERENCE_CSV=csv_path, REFERENCE_JSON=json_path, _load_json=load_json, ) def test_validation_accepts_documented_fresh_re100_deltas_and_reports_them(tmp_path): load_json, validate = _load_helpers("_load_json", "validate_outputs") reference = { "case_id": "kar_re100", "selected_seed": 45, "eval_reward_mean": .931258, "eval_r_cd_mean": .981787, "eval_r_cl_mean": .980641, "eval_r_sim_mean": .856324, "dtw_similarity": .918458, "action_front_mean": .015262, "action_top_mean": -.239834, "action_bottom_mean": .226212, } fresh = { "best_seed": "45", "reward_mean": .937275, "r_cd_mean": .980295, "r_cl_mean": .988632, "r_sim_mean": .866492, "dtw_sim_v5": .923246, "aF_mean": .033458, "aT_mean": -.243172, "aB_mean": .214681, } _configure_validation(validate, load_json, tmp_path, fresh, reference, reference) reports, failures = validate(("kar_re100",)) assert failures == [] assert len(reports) == 8 assert all("delta=" in report and report.endswith("PASS") for report in reports) def test_validation_keeps_strict_tables_and_fails_reproduction_gate(tmp_path): load_json, validate = _load_helpers("_load_json", "validate_outputs") reference = { "case_id": "kar_re100", "selected_seed": 45, "eval_reward_mean": 1.0, "eval_r_cd_mean": 2.0, "eval_r_cl_mean": 3.0, "eval_r_sim_mean": 4.0, "dtw_similarity": 5.0, "action_front_mean": 0.1, "action_top_mean": 0.2, "action_bottom_mean": 0.3, } csv_reference = dict(reference, eval_reward_mean=1.000002) fresh = { "best_seed": 44, "reward_mean": 1.021, "r_cd_mean": 2.0, "r_cl_mean": 3.0, "r_sim_mean": 4.0, "dtw_sim_v5": 5.0, "aF_mean": 0.1, "aT_mean": 0.2, "aB_mean": 0.3, } _configure_validation(validate, load_json, tmp_path, fresh, csv_reference, reference) reports, failures = validate(("kar_re100",)) assert reports == [] assert any("CSV/JSON disagree" in failure and "strict atol=1e-06" in failure for failure in failures) csv_reference["eval_reward_mean"] = reference["eval_reward_mean"] _configure_validation(validate, load_json, tmp_path / "second", fresh, csv_reference, reference) reports, failures = validate(("kar_re100",)) assert any(report.endswith("FAIL") and "reward_mean" in report for report in reports) assert any("best_seed=44" in failure for failure in failures) assert any("exceeds reproduction atol=0.020" in failure for failure in failures)