neural_codec/codec_dcvc_config.py

"""Single source of truth for DCVC canvas-selection parameters.

Reads the ``codec.dcvc`` block of ``preprocessor_config.json`` (found by walking
up from this file to the model directory). Used by ``dcvc_readiness_gen.py`` and
the readiness pipeline (``process_video_bitcost_readiness.py`` /
``process_video_bitcost_mv_mask_collage.py``) so ALL selection knobs are
controlled from the config file — NOT environment variables.

Full schema (defaults = the ``b50`` benchmark config) lives in
``preprocessor_config.json`` under ``codec.dcvc``.
"""
import json
import os
import functools

# Baseline defaults (the b50 benchmark config). Used only when a key is absent
# from preprocessor_config.json's codec.dcvc, so the file stays authoritative.
_DEFAULTS = {
    # DCVC engine
    "qp": 42, "reset_interval": 64, "intra_period": -1, "max_side": 0,
    # readiness sampling / grouping
    "num_sampled_frames": 256, "grouping_mode": "readiness",
    "readiness_sum_threshold_mode": "auto", "group_size": 32,
    "images_per_group": 4, "patch": 16, "max_pixels": 150000,
    "min_group_frames": 8, "max_group_frames": 128,
    "readiness_coverage_bins": 3, "readiness_delta_ratio": 0.05,
    "bitcost_grid": "sub", "bitcost_pct": 99, "decode_backsearch_max": 16,
    "canvas_format": "jpg",
    # selection tuning knobs (our additions)
    "per_frame_cap_ratio": 1.2,   # spread block budget over more time frames
    "bottom_atten": 0.5,          # attenuate bottom-edge bit-cost (de-bias overlays)
    "bottom_band": 0.10,          # fraction of frame height treated as bottom band
    "threshold_scale": 1.0,       # scale readiness threshold (<1 -> more canvases)
    "random_select": False,       # random patch baseline (control)
    "random_seed": 0,
}


@functools.lru_cache(maxsize=1)
def _load_file():
    d = os.path.dirname(os.path.abspath(__file__))
    for _ in range(4):
        # preprocessor_config.json may sit next to this file (bundled layout) or in
        # a sibling ``processor/`` subdir (multi-component model repo layout).
        for p in (os.path.join(d, "preprocessor_config.json"),
                  os.path.join(d, "processor", "preprocessor_config.json")):
            if os.path.exists(p):
                try:
                    with open(p, encoding="utf-8") as f:
                        return json.load(f).get("codec", {}).get("dcvc", {}) or {}
                except Exception:
                    return {}
        d = os.path.dirname(d)
    return {}


def get(key, cast=None):
    """Return codec.dcvc[key] from preprocessor_config.json, else the b50 default.
    ``cast`` optionally coerces (e.g. float, int, bool)."""
    v = _load_file().get(key, _DEFAULTS.get(key))
    if cast is not None and v is not None:
        if cast is bool and isinstance(v, str):
            return v.strip().lower() in ("1", "true", "yes")
        return cast(v)
    return v