diff --git a/src/drl_pinball/legacy_test/acquire.py b/src/drl_pinball/legacy_test/acquire.py index aa586a2..8198234 100644 --- a/src/drl_pinball/legacy_test/acquire.py +++ b/src/drl_pinball/legacy_test/acquire.py @@ -5,7 +5,10 @@ import argparse import hashlib import json import os +import platform import shutil +import subprocess +import sys from collections import deque from pathlib import Path from typing import Any, Callable @@ -21,17 +24,26 @@ from drl_pinball.acquisition import ( write_phase_cycle_artifacts, ) from .cases import get_case +from SR_analysis.configs import get_scene +from SR_analysis.stage_3_validate import SymbolicPolicy, load_formula_pair +from SR_analysis.utils.data_contracts import CAUSAL_ALIGNMENT from .metrics import erase_reward_terms, frozen_reference_comparison, reward_terms, sha256_file from .core.dtw_metrics import gen_target_states_at from .runtime import erase_policy_observation, load_policy_norm, policy_observation, reset_runtime, run_historical_interval CASE_NAME, ROLE = "karman_re100", "controlled" # compatibility defaults PERIODIC_CASES = ("karman_re50", "karman_re100", "karman_re200", "karman_re400", - "illusion_075L", "illusion_1L", "illusion_15L") + "illusion_075L", "illusion_1L", "illusion_15L", + "karman_re25", "karman_re70", "karman_re150", "karman_re300", + "illusion_05L", "illusion_06L", "illusion_08L", "illusion_12L", "illusion_2L") SUPPORTED_CASES = PERIODIC_CASES + ("steady", "vortex_lamb", "vortex_taylor", "erase") -PERIODIC_ROLES = ("controlled", "target", "zero") +PERIODIC_ROLES = ("controlled", "target", "zero", "sr") STEADY_ROLES = ("target", "constant", "zero") ROLES = PERIODIC_ROLES +SR_VARIANTS = ("k_front0", "k_rear0", "k_rear1", "i_front0", "i_front1", "i_rear0", "i_rear1") +SR_VARIANT_CASE = {**{name: "karman_re100" for name in SR_VARIANTS[:3]}, + **{name: "illusion_15L" for name in SR_VARIANTS[3:]}} +SR_ROLE_DIR = {name: f"sr_{name}" for name in SR_VARIANTS} WARMUP_INTERVALS, COLLECT_BOUNDARIES = 480, 160 NX, U0, SI, CYCLE_LENGTH = 1280, 0.01, 800, 30 # compatibility defaults MIN_FREE_BYTES = 4 * 1024**3 @@ -47,6 +59,18 @@ def _atomic_json(path: Path, payload: dict[str, Any]) -> None: os.replace(temporary, path) +def _execution_provenance(repo_root: Path): + try: + git_sha = subprocess.run(["git", "-C", str(repo_root), "rev-parse", "HEAD"], check=True, capture_output=True, text=True).stdout.strip() + git_dirty = bool(subprocess.run(["git", "-C", str(repo_root), "status", "--porcelain"], check=True, capture_output=True, text=True).stdout) + except (OSError, subprocess.CalledProcessError): + git_sha, git_dirty = None, None + return {"git_sha": git_sha, "git_dirty": git_dirty, + "environment": {"python": sys.version, "platform": platform.platform(), "numpy": np.__version__, + "conda_env": os.environ.get("CONDA_DEFAULT_ENV"), + "cuda_visible_devices": os.environ.get("CUDA_VISIBLE_DEVICES")}} + + def _identity(path: Path) -> dict[str, Any]: return {"path": str(path.resolve()), "sha256": sha256_file(str(path)), "bytes": path.stat().st_size} @@ -103,6 +127,7 @@ def _role_semantics(role: str) -> str: "target": "live builder-generated target trajectory; actions, rewards, and policy normalization unavailable", "zero": "physical-zero/uncontrolled trajectory using counter-bias normalized action; frozen norm used only for native reward evaluation", "constant": "established steady constant control at case.action_bias*U0 from normalized action zero", + "sr": "causal symbolic-regression policy using hash-bound article formulas", }[role] @@ -118,7 +143,53 @@ def _controlled_reference(role, bundle, columns, conv_len): def _case_roles(case): if case.name not in SUPPORTED_CASES: raise ValueError(f"unsupported Legacy acquisition case {case.name!r}") - return STEADY_ROLES if case.scene == "steady" else PERIODIC_ROLES + if case.scene == "steady": return STEADY_ROLES + if case.scene in ("vortex", "erase"): return ("controlled", "target", "zero") + return PERIODIC_ROLES if case.model is not None else ("target", "zero", "sr") + + +def _sr_formula_paths(repo_root: Path, case, variant: str | None = None): + if case.scene not in ("karman", "illusion") or case.sr_scene is None: + raise ValueError(f"case {case.name!r} has no SR deployment contract") + if variant is not None and SR_VARIANT_CASE.get(variant) != case.name: + raise ValueError(f"SR variant {variant!r} is restricted to {SR_VARIANT_CASE.get(variant)!r}") + family = case.scene + canonical = repo_root / "src/SR_analysis/results/runs" / f"article-refit-{family}-topology-a-20260718/formulas" + paths = {"front": canonical / "joint_front.json", "rear": canonical / "joint_rear_shared_upper.json"} + if variant is not None: + head = "front" if "front" in variant else "rear" + term = variant[-1] + paths[head] = repo_root / "src/SR_analysis/results/runs/article-ablation-formulas-v2-20260718/formulas" / f"{family}_{head}__delete_t{term}.json" + return paths + + +def _assert_sr_contract(case, cfg, data): + expected = {"scene_id": case.scene, "re_code": case.re_code, + "sample_interval": case.sample_interval, "action_scale": case.action_scale, + "action_bias": tuple(case.action_bias), "n_objects_env": 7 if case.scene == "karman" else 6, + "obs_slice": (2, 14) if case.scene == "karman" else (0, 12)} + for key, value in expected.items(): + actual = tuple(cfg[key]) if key in ("action_bias", "obs_slice") else cfg[key] + if actual != value: + raise ValueError(f"SR/Legacy contract mismatch for {key}: {actual!r} != {value!r}") + if case.scene == "illusion" and float(cfg["target_radius"]) != float(case.target_radius_l): + raise ValueError("SR/Legacy target geometry contract mismatch") + config = data["config"] + if (int(config["sample_interval"]), int(config["n_obj_total"])) != (case.sample_interval, expected["n_objects_env"]): + raise ValueError("SR/Legacy built runtime contract mismatch") + + +def _resolve_sr(repo_root: Path, case, variant: str | None): + cfg = get_scene(case.sr_scene) + paths = _sr_formula_paths(repo_root, case, variant) + pair = load_formula_pair(paths["front"], paths["rear"]) + harmonics_path = (repo_root / "src/SR_analysis/data/illusion" / case.sr_scene / "target_harmonics.json" + if case.scene == "illusion" else None) + harmonics = None + if harmonics_path is not None: + with harmonics_path.open(encoding="utf-8") as stream: harmonics = json.load(stream) + return {"cfg": cfg, "pair": pair, "formula_paths": paths, "variant": variant, + "harmonics": harmonics, "harmonics_path": harmonics_path} def _resolve_bundle( @@ -129,7 +200,7 @@ def _resolve_bundle( _case_roles(case) frozen = repo_root / "src" / "SR_analysis" / "data" / case.scene / (case.reference_case or case.name) needs_model = case.model is not None and role == "controlled" - needs_frozen_policy = (case.model is not None and role != "target" and + needs_frozen_policy = (case.model is not None and role not in ("target", "sr") and case.scene != "erase" and not (case.scene == "vortex" and role == "zero")) norm_path = frozen / "norm.json" if needs_frozen_policy else None @@ -248,7 +319,7 @@ def _observation_slices(case, role, boundary_obs): def _collect_role(role, data, model, ff, policy_norm, scratch: Path, - capture_field: Callable = _capture_boundary_field, case_name: str = CASE_NAME): + capture_field: Callable = _capture_boundary_field, case_name: str = CASE_NAME, sr_policy=None): case = get_case(case_name) if role not in _case_roles(case): raise ValueError(f"unknown acquisition role {role!r} for {case.name}") @@ -268,11 +339,17 @@ def _collect_role(role, data, model, ff, policy_norm, scratch: Path, field_buffer = None total = WARMUP_INTERVALS + COLLECT_BOUNDARIES harmonics = data.get("target_harmonics") + if role == "sr" and sr_policy is None: raise ValueError("sr role requires a symbolic policy") + raw_for_policy = np.asarray(data["norm"]["save_states"], np.float64)[-1].copy() if role == "sr" else None steady = case.scene == "steady" for interval in range(total): if role == "controlled": action, _ = model.predict(obs, deterministic=True) action, command = _command(case, action, expected_objects, ff.DATA_TYPE) + elif role == "sr": + omega, _, _ = sr_policy.action(raw_for_policy, interval) + action = (np.asarray(omega, np.float64) / U0 - np.asarray(case.action_bias)) / case.action_scale + action, command = _command(case, action, expected_objects, ff.DATA_TYPE) elif role == "zero": action = -np.asarray(case.action_bias, np.float32) / np.float32(case.action_scale) action, command = _command(case, action, expected_objects, ff.DATA_TYPE) @@ -286,6 +363,8 @@ def _collect_role(role, data, model, ff, policy_norm, scratch: Path, action = np.full(3, np.nan, np.float32); command = np.zeros(expected_objects, ff.DATA_TYPE) _run_interval(ff, si, command) boundary_obs = np.asarray(ff.obs, np.float32).copy() + if role == "sr": + sr_policy.observe(np.asarray(command[-3:], np.float64)); raw_for_policy = boundary_obs[:12] if case.scene == "illusion" else boundary_obs[2:14] sensors, forces = _observation_slices(case, role, boundary_obs) if role == "target": effective = np.full(3, np.nan, np.float32) @@ -297,7 +376,7 @@ def _collect_role(role, data, model, ff, policy_norm, scratch: Path, terms = {name: np.nan for name in ("reward", "reward_cd", "reward_cl", "native_legacy_dtw")} else: terms = reward_terms(case, data["target_states"], harmonics, np.asarray(fifo), policy_norm["force_norm_fact"], interval) - if not steady: + if not steady and role != "sr": target_force = gen_target_states_at(interval + 1, harmonics)[:2] if harmonics is not None else None obs = policy_observation(raw, policy_norm, target_force=target_force) effective = np.asarray(ff.current_effective_action(), np.float32)[-3:].copy() @@ -378,6 +457,17 @@ PHASE_FIELD_KEYS = { } +def _sr_metadata(sr): + if sr is None: return None + pair = sr["pair"] + files = {name: _identity(path) for name, path in sr["formula_paths"].items()} + return {"variant": sr["variant"], "formula_ids": {name: doc["artifact_id"] for name, doc in (("front", pair.front), ("rear", pair.rear))}, + "formula_files": files, "pair_hash": pair.pair_hash, + "deployment_expression_hashes": {"front": pair.front["deployment_expression_hash"], "rear": pair.rear["deployment_expression_hash"]}, + "alignment": CAUSAL_ALIGNMENT, "initialization": "both action lags seeded from cfg.policy_init_action; first action uses current raw post-FIFO state; observe after interval", + "config": _canonical_json_value(sr["cfg"]), "harmonics_file": _optional_file_provenance(sr["harmonics_path"])} + + def _finalize(role_dir: Path, scratch: Path, rows, field_buffer, data, bundle, storage, role="controlled"): case = bundle["case"]; columns = _columns(rows) phase_meta = {} @@ -446,6 +536,7 @@ def _finalize(role_dir: Path, scratch: Path, rows, field_buffer, data, bundle, s "field_contract": {"quantity": "q/RHO_ref", "RHO_ref": 1, "source": "public current_step_velocity_field q/U0 multiplied by field_config.velocity U0"}, "field_count": 1 if case.scene == "steady" else 8, "nonperiodic": case.scene == "steady", "role_semantics": _role_semantics(role), + "sr_deployment": _sr_metadata(bundle.get("sr")) if role == "sr" else None, "timeline_contract": "V5 lattice_step is an absolute solver count; Legacy physical_time is derived only from relative Legacy lattice_step", "source_provenance": { "model_file": _identity(bundle["model_path"]) if role == "controlled" else None, @@ -461,6 +552,7 @@ def _finalize(role_dir: Path, scratch: Path, rows, field_buffer, data, bundle, s ("builder_recomputed" if role != "target" else "not_applicable")), "scratch_cleanup": {"complete": not scratch.exists(), "path": str(scratch)}, "resolved_optane_path": str(storage["resolved_output_root"]), "phase_variables": list(phase_values), + "execution_provenance": _execution_provenance(Path(__file__).resolve().parents[3]), **phase_meta}) @@ -484,8 +576,10 @@ def _validate_staged_role(role_dir: Path, case_name: str = CASE_NAME) -> None: def _acquire_standard(role, *, case_name=CASE_NAME, output_root=None, overwrite=False, device_id=0, repo_root=None, storage_validator=validate_output_storage, - runtime_factory=None, finalizer=_finalize, **unused): + runtime_factory=None, finalizer=_finalize, variant=None, **unused): case = get_case(case_name) + if variant is not None and role != "sr": raise ValueError("--variant is valid only with role sr") + if variant is not None and variant not in SR_VARIANTS: raise ValueError(f"unknown SR variant {variant!r}") if role not in _case_roles(case): raise ValueError(f"unknown acquisition role {role!r} for {case_name}") repo = Path(repo_root) if repo_root is not None else Path(__file__).resolve().parents[3] storage = storage_validator(repo_mapping=default_reproduction_mapping(repo), @@ -494,7 +588,9 @@ def _acquire_standard(role, *, case_name=CASE_NAME, output_root=None, overwrite= bundle = _resolve_bundle(repo) else: bundle = _resolve_bundle(repo, case_name, role) - prepared = prepare_role_output(storage, "legacy", case_name, role, overwrite=overwrite) + sr = _resolve_sr(repo, case, variant) if role == "sr" else None + output_role = SR_ROLE_DIR[variant] if variant is not None else role + prepared = prepare_role_output(storage, "legacy", case_name, output_role, overwrite=overwrite) scratch = create_scratch(prepared["scratch_root"]) ff = None try: @@ -502,9 +598,14 @@ def _acquire_standard(role, *, case_name=CASE_NAME, output_root=None, overwrite= data, model, ff = _default_runtime(bundle["case"], bundle, device_id, role) else: data, model, ff = runtime_factory(bundle["case"], bundle, device_id, role) + if sr is not None: + _assert_sr_contract(case, sr["cfg"], data) + if sr["harmonics"] is not None: data["target_harmonics"] = sr["harmonics"] + bundle["sr"] = sr policy_norm = (None if role == "target" else - (data["norm"] if bundle["case"].scene == "steady" else bundle["norm"])) - rows, fields = _collect_role(role, data, model, ff, policy_norm, scratch, case_name=case_name) + (data["norm"] if role == "sr" or bundle["case"].scene == "steady" else bundle["norm"])) + policy = SymbolicPolicy(sr["pair"], sr["cfg"], sr["harmonics"]) if sr is not None else None + rows, fields = _collect_role(role, data, model, ff, policy_norm, scratch, case_name=case_name, sr_policy=policy) finalizer(prepared["role_dir"], scratch, rows, fields, data, bundle, storage, role) _validate_staged_role(prepared["role_dir"], case_name) return publish_role_output(prepared) @@ -792,26 +893,43 @@ def _acquire_special(role,case_name,output_root,overwrite,device_id,repo_root,st finally: if ff is not None and hasattr(ff,"close"): ff.close() -def acquire_role(role, *, case_name=CASE_NAME, output_root=None, overwrite=False, device_id=0, repo_root=None, storage_validator=validate_output_storage, runtime_factory=None, finalizer=_finalize, vortex_y_offset_l0=0): +def acquire_role(role, *, case_name=CASE_NAME, output_root=None, overwrite=False, device_id=0, repo_root=None, storage_validator=validate_output_storage, runtime_factory=None, finalizer=_finalize, vortex_y_offset_l0=0, variant=None): case=get_case(case_name) + if variant is not None and role != "sr": raise ValueError("variant is valid only for the sr role") + if variant is not None and variant not in SR_VARIANTS: raise ValueError(f"unknown SR variant {variant!r}") if role not in _case_roles(case): raise ValueError(f"unknown acquisition role {role!r} for {case_name}") if case.scene in ("vortex","erase"): return _acquire_special(role,case_name,output_root,overwrite,device_id,repo_root,storage_validator,runtime_factory,vortex_y_offset_l0) - return _acquire_standard(role,case_name=case_name,output_root=output_root,overwrite=overwrite,device_id=device_id,repo_root=repo_root,storage_validator=storage_validator,runtime_factory=runtime_factory,finalizer=finalizer) + return _acquire_standard(role,case_name=case_name,output_root=output_root,overwrite=overwrite,device_id=device_id,repo_root=repo_root,storage_validator=storage_validator,runtime_factory=runtime_factory,finalizer=finalizer,variant=variant) def acquire_controlled(**kwargs): return acquire_role("controlled", **kwargs) +def campaign_matrix(): + training = PERIODIC_CASES[:7] + generalization = PERIODIC_CASES[7:] + rows = [(name, "sr", None) for name in training] + rows.extend((name, role, None) for name in generalization for role in ("target", "zero", "sr")) + rows.extend((case, "sr", variant) for variant, case in SR_VARIANT_CASE.items()) + return rows + + def main() -> int: parser = argparse.ArgumentParser(description="Acquire Legacy full-matrix evidence") - parser.add_argument("case", choices=SUPPORTED_CASES) - parser.add_argument("--role", choices=ROLES + ("constant",), required=True) + parser.add_argument("case", choices=SUPPORTED_CASES, nargs="?") + parser.add_argument("--role", choices=ROLES + ("constant",)) + parser.add_argument("--variant", choices=SR_VARIANTS) + parser.add_argument("--list-campaign", action="store_true") parser.add_argument("--device-id", type=int, default=0) parser.add_argument("--vortex-y-offset-l0", type=int, default=0) parser.add_argument("--output-root", type=Path); parser.add_argument("--overwrite", action="store_true") args = parser.parse_args() - path = acquire_role(args.role, case_name=args.case, output_root=args.output_root, overwrite=args.overwrite, device_id=args.device_id, vortex_y_offset_l0=args.vortex_y_offset_l0) + if args.list_campaign: + for case, role, variant in campaign_matrix(): print(case, role, variant or "canonical") + print(f"count={len(campaign_matrix())}"); return 0 + if args.case is None or args.role is None: parser.error("case and --role are required unless --list-campaign is used") + path = acquire_role(args.role, case_name=args.case, output_root=args.output_root, overwrite=args.overwrite, device_id=args.device_id, vortex_y_offset_l0=args.vortex_y_offset_l0, variant=args.variant) print(path); return 0 diff --git a/src/drl_pinball/legacy_test/cases.py b/src/drl_pinball/legacy_test/cases.py index 29eb89c..01a62ef 100644 --- a/src/drl_pinball/legacy_test/cases.py +++ b/src/drl_pinball/legacy_test/cases.py @@ -16,12 +16,19 @@ class Case: target_radius_l: float | None = None vortex_type: str | None = None reference_case: str | None = None + sr_scene: str | None = None CASES = { - **{f"karman_re{re}": Case(f"karman_re{re}", "karman", f"d1a3o12_re{re}", re_code=float(re)) for re in (50, 100, 200, 400)}, - "illusion_075L": Case("illusion_075L", "illusion", "d1a3o14_250525_imit_075L_2U_400S", sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.75, reference_case="illusion_0.75L"), - "illusion_1L": Case("illusion_1L", "illusion", "d1a3o14_250525_imit_1L_2U_600S", sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.0), - "illusion_15L": Case("illusion_15L", "illusion", "d1a3o14_250525_imit_15L_2U", action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.5, reference_case="illusion_1.5L"), + **{f"karman_re{re}": Case(f"karman_re{re}", "karman", f"d1a3o12_re{re}", re_code=float(re), sr_scene=f"karman_re{re}") for re in (50, 100, 200, 400)}, + "illusion_075L": Case("illusion_075L", "illusion", "d1a3o14_250525_imit_075L_2U_400S", sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.75, reference_case="illusion_0.75L", sr_scene="illusion_0.75L"), + "illusion_1L": Case("illusion_1L", "illusion", "d1a3o14_250525_imit_1L_2U_600S", sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.0, sr_scene="illusion_1L"), + "illusion_15L": Case("illusion_15L", "illusion", "d1a3o14_250525_imit_15L_2U", action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.5, reference_case="illusion_1.5L", sr_scene="illusion_1.5L"), + **{f"karman_re{re}": Case(f"karman_re{re}", "karman", None, re_code=float(re), sr_scene=f"karman_re{re}") for re in (25, 70, 150, 300)}, + "illusion_05L": Case("illusion_05L", "illusion", None, sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.5, reference_case="illusion_0.5L", sr_scene="illusion_0.5L"), + "illusion_06L": Case("illusion_06L", "illusion", None, sample_interval=400, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.6, reference_case="illusion_0.6L", sr_scene="illusion_0.6L"), + "illusion_08L": Case("illusion_08L", "illusion", None, sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=0.8, reference_case="illusion_0.8L", sr_scene="illusion_0.8L"), + "illusion_12L": Case("illusion_12L", "illusion", None, sample_interval=600, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=1.2, reference_case="illusion_1.2L", sr_scene="illusion_1.2L"), + "illusion_2L": Case("illusion_2L", "illusion", None, sample_interval=800, action_bias=(0.0, -2.0, 2.0), conv_len=36, target_radius_l=2.0, sr_scene="illusion_2L"), "vortex_lamb": Case("vortex_lamb", "vortex", "vortex_lamb", steps=150, action_scale=4.0, vortex_type="lamb"), "vortex_taylor": Case("vortex_taylor", "vortex", "vortex_taylor", steps=150, action_scale=4.0, vortex_type="taylor"), "erase": Case("erase", "erase", "d1a3o12_250729_250326_erase", sample_interval=600, action_scale=8.0, action_bias=(0.0, -8.0, 8.0), conv_len=36), diff --git a/src/drl_pinball/legacy_test/tests/test_acquire_matrix.py b/src/drl_pinball/legacy_test/tests/test_acquire_matrix.py index 91917a0..cf2b9f1 100644 --- a/src/drl_pinball/legacy_test/tests/test_acquire_matrix.py +++ b/src/drl_pinball/legacy_test/tests/test_acquire_matrix.py @@ -7,12 +7,15 @@ import src.drl_pinball.legacy_test.acquire as acquire def test_allowlist_roles_and_case_sample_intervals(): assert acquire.SUPPORTED_CASES == ( "karman_re50", "karman_re100", "karman_re200", "karman_re400", - "illusion_075L", "illusion_1L", "illusion_15L", "steady", "vortex_lamb", "vortex_taylor", "erase", + "illusion_075L", "illusion_1L", "illusion_15L", + "karman_re25", "karman_re70", "karman_re150", "karman_re300", + "illusion_05L", "illusion_06L", "illusion_08L", "illusion_12L", "illusion_2L", + "steady", "vortex_lamb", "vortex_taylor", "erase", ) assert [acquire.get_case(name).sample_interval for name in ("karman_re50", "illusion_075L", "illusion_1L", "illusion_15L", "steady")] == [800, 400, 600, 800, 800] assert acquire._case_roles(acquire.get_case("steady")) == ("target", "constant", "zero") - assert acquire._case_roles(acquire.get_case("illusion_075L")) == ("controlled", "target", "zero") + assert acquire._case_roles(acquire.get_case("illusion_075L")) == ("controlled", "target", "zero", "sr") def test_scene_observation_slices_are_exact(): diff --git a/src/drl_pinball/legacy_test/tests/test_acquire_roles.py b/src/drl_pinball/legacy_test/tests/test_acquire_roles.py index e9c1240..0dbbe89 100644 --- a/src/drl_pinball/legacy_test/tests/test_acquire_roles.py +++ b/src/drl_pinball/legacy_test/tests/test_acquire_roles.py @@ -230,3 +230,4 @@ def test_role_failure_transaction_preserves_existing(monkeypatch, tmp_path, role def test_cli_exposes_all_roles(): source = Path(acquire.__file__).read_text(encoding="utf-8") assert 'choices=ROLES' in source + assert '--variant' in source and '--list-campaign' in source diff --git a/src/drl_pinball/legacy_test/tests/test_acquire_sr.py b/src/drl_pinball/legacy_test/tests/test_acquire_sr.py new file mode 100644 index 0000000..9139d0a --- /dev/null +++ b/src/drl_pinball/legacy_test/tests/test_acquire_sr.py @@ -0,0 +1,90 @@ +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pytest + +import src.drl_pinball.legacy_test.acquire as acquire + + +def test_generalization_registry_has_no_ppo_and_exact_sr_mapping(): + expected = { + "karman_re25": ("karman_re25", 25, 800), "karman_re70": ("karman_re70", 70, 800), + "karman_re150": ("karman_re150", 150, 800), "karman_re300": ("karman_re300", 300, 800), + "illusion_05L": ("illusion_0.5L", 100, 400), "illusion_06L": ("illusion_0.6L", 100, 400), + "illusion_08L": ("illusion_0.8L", 100, 600), "illusion_12L": ("illusion_1.2L", 100, 600), + "illusion_2L": ("illusion_2L", 100, 800), + } + for name, (scene, re_code, si) in expected.items(): + case = acquire.get_case(name) + assert case.model is None and case.sr_scene == scene + assert (case.re_code, case.sample_interval) == (re_code, si) + assert acquire._case_roles(case) == ("target", "zero", "sr") + with pytest.raises(ValueError, match="unknown acquisition role"): + acquire.acquire_role("controlled", case_name=name) + + +def test_formula_authority_and_variant_mapping(): + root = Path(acquire.__file__).resolve().parents[3] + canonical = acquire._sr_formula_paths(root, acquire.get_case("karman_re70")) + assert "article-refit-karman-topology-a-20260718" in str(canonical["front"]) + pair = acquire._sr_formula_paths(root, acquire.get_case("karman_re100"), "k_rear1") + assert pair["front"] == canonical["front"] + assert pair["rear"].name == "karman_rear__delete_t1.json" + illusion = acquire._sr_formula_paths(root, acquire.get_case("illusion_15L"), "i_front0") + assert illusion["front"].name == "illusion_front__delete_t0.json" + assert "article-refit-illusion-topology-a-20260718" in str(illusion["rear"]) + with pytest.raises(ValueError, match="restricted"): + acquire._sr_formula_paths(root, acquire.get_case("karman_re50"), "k_front0") + + +def test_campaign_matrix_is_exact_41_unique_output_roles(): + matrix = acquire.campaign_matrix() + assert len(matrix) == 41 + keys = [(case, acquire.SR_ROLE_DIR[variant] if variant else role) for case, role, variant in matrix] + assert len(keys) == len(set(keys)) + assert sum(variant is not None for _, _, variant in matrix) == 7 + assert sum(acquire.get_case(case).model is None for case, _, _ in matrix) == 27 + + +def test_sr_symbolic_timing_conversion_and_historical_reset(monkeypatch, tmp_path): + case = acquire.get_case("karman_re100") + class Policy: + def __init__(self): self.events = [] + def action(self, raw, step): + self.events.append(("action", np.asarray(raw).copy(), step)) + return np.array([.01, -.02, .03]), None, None + def observe(self, omega): self.events.append(("observe", np.asarray(omega).copy())) + class Flow: + DATA_TYPE = np.float32 + def __init__(self): self.action=np.ones(7,np.float32); self.obs=np.arange(14,dtype=np.float32); self.commands=[] + def restore_ddf(self): pass + def apply_ddf(self): pass + def run(self, steps, command): + assert np.array_equal(self.action, np.zeros(7,np.float32)) + self.commands.append(command.copy()); self.action=command.copy(); self.obs += 1 + def current_effective_action(self): return self.action + data={"config":{"nx":1280,"ny":1,"sample_interval":800,"n_obj_total":7}, + "norm":{"save_states":np.vstack((np.zeros(12),np.arange(12))).astype(np.float32)}, + "target_states":np.zeros((30,6),np.float32)} + monkeypatch.setattr(acquire,"WARMUP_INTERVALS",0); monkeypatch.setattr(acquire,"COLLECT_BOUNDARIES",1) + monkeypatch.setattr(acquire,"reward_terms",lambda *a:{"reward":1.,"reward_cd":1.,"reward_cl":1.,"native_legacy_dtw":1.}) + policy=Policy(); flow=Flow() + rows,_=acquire._collect_role("sr",data,None,flow,{"force_norm_fact":1.},tmp_path, + capture_field=lambda *a:{"ux":np.zeros((1,1280)),"uy":np.zeros((1,1280))},sr_policy=policy) + assert policy.events[0][0] == "action" and np.array_equal(policy.events[0][1], np.arange(12)) + assert policy.events[1][0] == "observe" + assert np.allclose(flow.commands[0][-3:], [.01,-.02,.03]) + expected=(np.array([1.,-2.,3.])-np.asarray(case.action_bias))/case.action_scale + assert np.allclose(rows[0]["action_normalized"],expected) + + +def test_sr_metadata_binds_formula_and_harmonics(tmp_path): + front=tmp_path/"front.json"; rear=tmp_path/"rear.json"; harmonics=tmp_path/"harmonics.json" + front.write_text("front"); rear.write_text("rear"); harmonics.write_text("[]") + pair=SimpleNamespace(pair_hash="p",front={"artifact_id":"f","deployment_expression_hash":"df"},rear={"artifact_id":"r","deployment_expression_hash":"dr"}) + meta=acquire._sr_metadata({"pair":pair,"formula_paths":{"front":front,"rear":rear},"variant":None, + "cfg":{"policy_init_action":(0,-4,4)},"harmonics_path":harmonics}) + assert meta["pair_hash"] == "p" and meta["alignment"] == "causal_post_state_to_next_action" + assert all(len(item["sha256"]) == 64 for item in meta["formula_files"].values()) + assert meta["harmonics_file"]["file_identity"]["sha256"] diff --git a/src/drl_pinball/legacy_test/tests/test_contracts.py b/src/drl_pinball/legacy_test/tests/test_contracts.py index 8e3717a..b9af0e9 100644 --- a/src/drl_pinball/legacy_test/tests/test_contracts.py +++ b/src/drl_pinball/legacy_test/tests/test_contracts.py @@ -18,7 +18,7 @@ from src.drl_pinball.legacy_test.runtime import ( ) def test_config_matrix(): - assert set(CASES)=={"karman_re50","karman_re100","karman_re200","karman_re400","illusion_075L","illusion_1L","illusion_15L","vortex_lamb","vortex_taylor","erase","steady"} + assert set(CASES)=={"karman_re50","karman_re100","karman_re200","karman_re400","karman_re25","karman_re70","karman_re150","karman_re300","illusion_075L","illusion_1L","illusion_15L","illusion_05L","illusion_06L","illusion_08L","illusion_12L","illusion_2L","vortex_lamb","vortex_taylor","erase","steady"} assert [CASES[f"karman_re{x}"].re_code for x in (50,100,200,400)] == [50,100,200,400] assert [CASES[x].sample_interval for x in ("illusion_075L","illusion_1L","illusion_15L")] == [400,600,800] assert get_case("erase").model == "d1a3o12_250729_250326_erase"