terraflow.validation

Spatial-block cross-validation (Roberts et al. 2017) on the suitability label grid. Invoked from the CLI as terraflow validate -c config.yml; results are appended to report.json under the validation key. For spatial-autocorrelation diagnostics on score residuals, call esda.Moran directly on features.parquet; for inter-rater agreement against a reference label set, call sklearn.metrics.cohen_kappa_score.

API Reference

validation

Model validation module — spatial block cross-validation.

Users wanting spatial autocorrelation diagnostics (e.g. Moran's I) should call esda.Moran directly on features.parquet. Users wanting inter-rater agreement (e.g. Cohen's κ) should call sklearn.metrics.cohen_kappa_score directly. TerraFlow does not maintain wrappers around either, since neither earns Methods-section citations.

run_validation(config_path)

Run model validation and append a validation block to report.json.

Loads the TerraFlow config, locates the most recent pipeline run directory (the one with the latest features.parquet), and computes spatial block CV. Results are written atomically to the existing report.json under the "validation" key.

Parameters:

Name	Type	Description	Default
config_path	Path	Path to a TerraFlow YAML config file that includes a validation: section.	required

Returns:

Name	Type	Description
Path	Path	Absolute path to the updated report.json.

Raises:

Type	Description
ValueError:	If the config has no validation: section.
FileNotFoundError:	If no pipeline run directory containing features.parquet is found.

Source code in terraflow/validation.py

def run_validation(config_path: Path) -> Path:
    """Run model validation and append a validation block to report.json.

    Loads the TerraFlow config, locates the most recent pipeline run directory
    (the one with the latest ``features.parquet``), and computes spatial block
    CV. Results are written atomically to the existing ``report.json`` under
    the ``"validation"`` key.

    Parameters
    ----------
    config_path:
        Path to a TerraFlow YAML config file that includes a ``validation:``
        section.

    Returns
    -------
    Path:
        Absolute path to the updated ``report.json``.

    Raises
    ------
    ValueError:
        If the config has no ``validation:`` section.
    FileNotFoundError:
        If no pipeline run directory containing ``features.parquet`` is found.
    """
    data = load_config_dict(config_path)
    cfg = build_config(data)

    if cfg.validation is None:
        raise ValueError(
            "Config file has no 'validation:' section. "
            "Add a validation: block with optional n_blocks_side and buffer_deg "
            "fields. See TerraFlow documentation for details."
        )

    val_cfg = cfg.validation

    run_dir = resolve_run_dir(config_path)
    features_path = run_dir / "features.parquet"
    if not features_path.exists():
        raise FileNotFoundError(
            f"No pipeline run found at {run_dir}. "
            "Run `terraflow run -c config.yml` before running validation."
        )

    logger.info(f"Running validation on {run_dir}")

    # Load features
    df = pd.read_parquet(features_path)
    lats = df["lat"].values
    lons = df["lon"].values
    labels = df["label"].values

    # Spatial block CV
    fold_accs = _spatial_block_cv(
        lats,
        lons,
        labels,
        n_blocks_side=val_cfg.n_blocks_side,
        buffer_deg=val_cfg.buffer_deg,
    )
    mean_fold_accuracy: Optional[float] = (
        float(np.mean(fold_accs)) if fold_accs else None
    )

    # Read existing report.json and append validation block
    report_path = run_dir / "report.json"
    if report_path.exists():
        with report_path.open("r", encoding="utf-8") as fh:
            report: Dict[str, Any] = json.load(fh)
    else:
        report = {}

    report["validation"] = {
        "method": "spatial_block_cv",
        "citation": "Roberts et al. 2017, Ecography",
        "n_blocks_side": val_cfg.n_blocks_side,
        "buffer_deg": val_cfg.buffer_deg,
        "n_folds": len(fold_accs),
        "mean_fold_accuracy": mean_fold_accuracy,
        "kriging_loocv_rmse": report.get("kriging_loocv"),
        "note": (
            "model has no free parameters; fold accuracy reflects spatial "
            "label consistency, not fit generalization"
        ),
    }

    _atomic_write_json(report_path, report)
    logger.info(f"Validation block written to {report_path}")

    return report_path