code_snapshots/WiSER_public_release_code/wiser/config.py

"""Single-source-of-truth configuration for WiSER CIR.

Every runtime knob lives here. Submodules (`models/detr_head.py`,
`models/backbone.py`, `engine/checkpoint_selector.py`, `engine/orchestrate.py`,
`engine/losses.py`) expose configs that MUST be constructed from these
top-level dataclasses so a single edit here propagates.

Defaults carry PROVISIONAL flags where task9 / histogram evidence has not
yet replaced them (pending Round-2 real-data execution).
"""

from __future__ import annotations

from dataclasses import dataclass, field


@dataclass
class ModelConfig:
    # Query budget (AC-10). Ratified: reports/path_count_histogram.md shows
    # K=6 covers 100% of tiny10 samples; K=8 retained for draft parity +
    # room for rare longer-tail distributions on other scene sets.
    query_budget: int = 8
    # Bounded dB head (AC-3, AC-6). Ratified: reports/path_db_range.md.
    # Tiny10 merged-path 1st percentile = -109.3 dB, 99th = -12.3 dB.
    # Set with ~5 dB headroom each side so saturation occurs only in the
    # extreme tail of the observed distribution.
    db_low: float = -115.0
    db_high: float = -7.5
    # Backbone shape controls (AC-4).
    backbone_channels: int = 256
    backbone_tx_emb_channels: int = 128
    backbone_downsample_stages: int = 2
    backbone_blocks_per_stage: int = 2
    backbone_num_heads: int = 8
    backbone_mlp_ratio: float = 4.0
    # Round 8: default flipped to `trellis2` now that the sparse runtime is
    # validated by `tests/test_cache_invariants.py::test_trellis2_real_batch_cuda_smoke`
    # on CUDA. `dense_fallback` stays as an opt-in via `--backbone-kind dense_fallback`
    # for unit tests and CPU-only smoke work.
    backbone_kind: str = "trellis2"  # or "dense_fallback"
    # Head shape controls (AC-3).
    head_num_decoder_layers: int = 2
    head_num_heads: int = 8
    head_dropout: float = 0.0
    # Round-6 C1 peak_db head architecture switch. Default "flat" preserves
    # pre-R6 behaviour and the existing test suite. "deep" enables
    # DeepBoundedDbHead (LayerNorm + residual block + 2-layer out MLP + tanh
    # + data-driven bias init). Opt in via CLI `--peak-db-head-arch deep`.
    peak_db_head_arch: str = "flat"  # or "deep"
    peak_db_hidden: int = 512
    peak_db_init_mean_db: float = -55.0
    # C2 experiment: deep delay head (see models/detr_head.py::DeepNormalizedDelayHead).
    delay_head_arch: str = "flat"      # or "deep"
    delay_hidden: int = 512
    delay_init_mean_ns: float = 4.0
    # C2-v7 experiment: deep exists head.
    exists_head_arch: str = "flat"     # or "deep"
    exists_hidden: int = 512
    exists_init_prob: float = 0.2


@dataclass
class LossConfig:
    match_exists: float = 1.0
    match_delay_ns: float = 1.0
    match_peak_db: float = 1.0
    no_object_exists: float = 1.0
    # Pair-gap auxiliaries DEFAULT OFF per AC-6.
    enable_pair_gap_losses: bool = False
    pair_gap_delay: float = 0.0
    pair_gap_db: float = 0.0
    delay_huber_beta_ns: float = 0.5
    peak_db_huber_beta: float = 2.0


@dataclass
class CheckpointSelectorConfig:
    peak_db_bias_abs_max: float = 3.0
    zero_path_fpr_max: float = 0.10
    count_acc_min: float = 0.5
    delay_mae_ns_max: float = 1.0
    primary_tie_tol: float = 1e-3


@dataclass
class ScaleUpGateConfig:
    count_acc_min: float = 0.9
    delay_mae_ns_max: float = 0.3
    peak_db_bias_abs_max: float = 3.0


@dataclass
class TargetContractConfig:
    mode: str = "merged_delay_bin"
    delay_max_ns: float = 15.0
    num_bins: int = 128
    max_paths: int = 8
    # Frozen by Codex task2 contract; MUST match MergedPathTargetConfig.floor_db.
    floor_db: float = -250.0
    use_bin_center_delay: bool = True


@dataclass
class ThroughputBenchConfig:
    # Plan Version 12: self-contained AC-5 contract. The cross-model WiSER
    # ratio gate was removed per user directive "跳过对比算法 base line a06-7".
    manifest_path: str = "benchmarks/tiny10_triples_manifest.json"
    warmup_steps: int = 20
    measured_steps: int = 100


@dataclass
class ManifestPaths:
    single_scene: str = "manifests/task_a069_single_scene.json"
    tiny10: str = "benchmarks/tiny10_triples_manifest.json"
    tiny30: str = "manifests/task_a069_tiny30.json"
    tiny50: str = "manifests/task_a069_tiny50.json"
    tiny100: str = "manifests/task_a069_tiny100.json"


@dataclass
class TrainConfig:
    model: ModelConfig = field(default_factory=ModelConfig)
    loss: LossConfig = field(default_factory=LossConfig)
    selector: CheckpointSelectorConfig = field(default_factory=CheckpointSelectorConfig)
    scale_up_gate: ScaleUpGateConfig = field(default_factory=ScaleUpGateConfig)
    target_contract: TargetContractConfig = field(default_factory=TargetContractConfig)
    benchmark: ThroughputBenchConfig = field(default_factory=ThroughputBenchConfig)
    manifests: ManifestPaths = field(default_factory=ManifestPaths)
    radiomap_head_forbidden: bool = True


def reject_radiomap_artifacts(manifest_path: str, weights_name: str | None = None) -> None:
    """Hard guard: refuse radiomap manifests or weights.

    Round 7 fix (Codex R6 finding 2): the guard only inspects the
    BASENAME / STEM of the manifest path, not the full absolute path. The
    branch's own ancestor directory is `.../WiSER/...`, so
    a naive substring match over the full path incorrectly rejected every
    WiSER CIR manifest. We now match patterns that a genuine radiomap artifact
    filename would exhibit (`radiomap_*`, `*_radiomap_*`, `*radiomap_tag*`),
    not any ancestor-directory occurrence of the word.

    Called by dataset loader, manifest builder, benchmark script, and
    training orchestrators so the CIR-only guarantee holds at runtime.
    """
    import os

    basename = os.path.basename(str(manifest_path)).lower()
    stem = basename.rsplit(".", 1)[0] if "." in basename else basename

    if "radiomap" in stem:
        raise RuntimeError(
            f"Refusing radiomap manifest in WiSER CIR (CIR-only branch): {manifest_path}"
        )
    if weights_name is not None and str(weights_name).lower().startswith("radiomap_"):
        raise RuntimeError(
            f"Refusing radiomap weights in WiSER CIR (CIR-only branch): {weights_name}"
        )