qlroms.utils.builder

Config-driven qlROM building: one YAML file per model, one CLI.

python -m qlroms.utils.builder configs/ks1d_chaotic.yml [--set rom.K=20]

Three blocks, in the same style as the sibling packages' configs. case names the test case and its window (<case module>/<TEST_CASES key> plus the overrides that case accepts), rom says which quantized-local family to fit and with how many charts and modes, and check says what to measure once it is fitted: a closed-loop forecast, optionally against the global (K = 1) ROM of the same family and rank. Unknown keys fail loudly.

Both the case trajectory and the fitted ROM come from the case's own cached pipeline, so re-running a config only pays for what changed. Nothing here assimilates anything -- qlroms.model.QLModel is where a fitted ROM meets an estimator, and that lives outside this package.

load_config(path, block_keys=None, defaults=None, **overrides)

Read a YAML config, apply dotted overrides, fill defaults, reject unknown keys.

block_keys / defaults default to this module's schema; pass another pair to validate a config of a different shape with the same rules.

Source code in qlroms/utils/builder.py

def load_config(path, block_keys=None, defaults=None, **overrides) -> dict:
    """Read a YAML config, apply dotted overrides, fill defaults, reject unknown keys.

    `block_keys` / `defaults` default to this module's schema; pass another pair to
    validate a config of a different shape with the same rules.
    """


    block_keys = BLOCK_KEYS if block_keys is None else block_keys
    defaults = DEFAULTS if defaults is None else defaults

    with open(path) as fh:
        cfg = yaml.safe_load(fh) or {}
    for dotted, value in overrides.items():
        block, _, key = dotted.partition(".")
        if not key:
            raise ValueError(f"Override {dotted!r} must be '<block>.<key>'.")
        cfg.setdefault(block, {})[key] = value

    unknown = sorted(set(cfg) - set(block_keys) - {"name"})
    if unknown:
        raise ValueError(f"Unknown config block(s) {unknown}; allowed: {sorted(block_keys)} plus 'name'.")
    for block, allowed in block_keys.items():
        cfg[block] = {**defaults[block], **(cfg.get(block) or {})}
        bad = sorted(set(cfg[block]) - set(allowed))
        if bad:
            raise ValueError(f"Unknown {block} key(s) {bad}; allowed: {sorted(allowed)}.")
    return cfg

build_case(cfg, K=None)

Case FOM, fitted qlROM and snapshot windows a resolved config describes.

Returns (fom, rom, Xtrain, Xtest, save_dir). K overrides the config's chart count, which is how the global (K = 1) reference ROM is built from the same charts pipeline. Trajectory and ROM are both cached under qlroms.utils.paths.data_dir.

Source code in qlroms/utils/builder.py

def build_case(cfg: dict, K=None) -> tuple:
    """Case FOM, fitted qlROM and snapshot windows a resolved config describes.

    Returns (fom, rom, Xtrain, Xtest, save_dir). `K` overrides the config's chart
    count, which is how the global (K = 1) reference ROM is built from the same
    charts pipeline. Trajectory and ROM are both cached under
    `qlroms.utils.paths.data_dir`.
    """
    module, case, settings = _resolve_case(cfg["case"]["model"])
    # the case's own windows stand unless this run overrides them
    fom, case_cfg = build_fom(module, case=case, overrides=cfg["case"].get("params") or None)
    i0, Ntrain, Ntest = case_cfg["i0"], case_cfg["Ntrain"], case_cfg["Ntest"]

    full_traj = get_full_trajectory(Ntot=i0 + Ntrain + Ntest, model=fom, i0=i0)
    Xtrain, Xtest = full_traj[:, :Ntrain], full_traj[:, Ntrain:Ntrain + Ntest]
    save_dir = module.get_simulation_path(fom, Ntrain=Ntrain)   # models are keyed on Ntrain only

    kind = cfg["rom"]["kind"]
    if kind not in KINDS:
        raise ValueError(f"Unknown rom.kind {kind!r}; choose from {sorted(KINDS)}.")
    # K and r unset in the config mean "whatever this case is usually built at"
    K = K if K is not None else (cfg["rom"]["K"] if cfg["rom"]["K"] is not None else settings.get("K", 10))
    r = cfg["rom"]["r"] if cfg["rom"]["r"] is not None else settings.get("r", 30)
    seed = int(cfg["rom"].get("seed") or 0)
    # rom.clustering goes straight to fit_clusters: kmeans_method, assign_overlapping,
    # overlap_tolerance, kmeans_n_init, kmeans_max_iter (rom.seed is its random_state)
    clustering = {"random_state": seed, **(cfg["rom"].get("clustering") or {})}

    if kind == "qlopinf":
        # equation-free end to end: the charts and the operators both come from the
        # snapshots, so this never touches the case's Galerkin build
        qlOpinf = _equation_free("qlOpinf")
        rom = qlOpinf.from_snapshots(Xtrain, K=int(K), r=int(r), dt=float(fom.dt),
                                     random_state=seed, clustering_kwargs=clustering)
    else:
        rom = module.build_local_model(Xtrain, fom, r=int(r), K=int(K), save_dir=save_dir,
                                       clustering_kwargs=clustering)
        if kind == "qlesn":
            rom = _fit_esn(rom, Xtrain, {**ESN_DEFAULTS, **(cfg["rom"].get("esn") or {})}, seed)
    return fom, rom, Xtrain, Xtest, save_dir

run_build(config, **overrides)

Fit the qlROM a config describes and score it; return the metrics dict.

With check.compare_global the same family is also fitted at K = 1 and scored on the same window, which is the comparison the ql-ROM claim rests on. With check.save_figs the FOM-vs-ROMs page lands next to the cached model.

Source code in qlroms/utils/builder.py

def run_build(config, **overrides) -> dict:
    """Fit the qlROM a config describes and score it; return the metrics dict.

    With ``check.compare_global`` the same family is also fitted at ``K = 1`` and
    scored on the same window, which is the comparison the ql-ROM claim rests on.
    With ``check.save_figs`` the FOM-vs-ROMs page lands next to the cached model.
    """
    cfg = load_config(config, **overrides) if isinstance(config, (str, os.PathLike)) else config
    check = cfg["check"]
    n_steps, threshold = check["n_steps"], float(check["threshold"])

    fom, rom, Xtrain, Xtest, save_dir = build_case(cfg)
    local, X_local = _score(rom, Xtrain, Xtest, n_steps, threshold)
    charts = getattr(rom, "rom", rom)
    out = {"name": cfg.get("name", "qlroms"), "kind": cfg["rom"]["kind"], "K": int(charts.K),
           "r": int(charts.r), "local": local}

    if check["compare_global"]:
        _, rom_g, _, _, _ = build_case(cfg, K=1)
        out["global"], X_global = _score(rom_g, Xtrain, Xtest, n_steps, threshold)
        if check["save_figs"] and X_local is not None:
            from .diagnosis_plots import plot_fom_vs_roms, save_figures_pdf
            n = min(X_local.shape[1], X_global.shape[1])
            fig = plot_fom_vs_roms(Xtest[:, :n], X_local[:, :n], X_global[:, :n], fom,
                                   K=int(charts.K), r=int(charts.r))
            out["figure"] = save_figures_pdf([fig], save_dir,
                                             f"{cfg['rom']['kind']}_K{charts.K}_r{charts.r}.pdf")
    return out