terraflow.climate_impact

Climate-impact pipeline orchestrator introduced in v0.5.0 (#138e). Glues terraflow.temporal.compute_per_station_aggregations to terraflow.climate.ClimateInterpolator and writes the sibling climate_features.parquet artifact alongside the historical features.parquet.

The orchestrator is auto-invoked by terraflow.pipeline.run_pipeline when the loaded config declares both climate.temporal_aggregations and climate.scenarios (since v0.5.0; #138f). It is also exposed as a public function for callers that want to drive the climate-impact path without going through the full pipeline.

Overview

• load_timeseries_csv(path) — parse and validate the long-format station daily CSV. Required columns: station_id, lat, lon, date, temperature_c, precipitation_mm.
• run_climate_impact_features(cfg, run_dir, cells_df) — orchestrate the full path: compute per-station aggregations, kriging-interpolate each <rule>__<scenario> column to cell centroids, write the artifact. Returns the absolute path to climate_features.parquet.

Artifact contract

Column	Type	Meaning
cell_id	int	Stable within a run; matches features.parquet.cell_id
<rule_label>__<scenario_name>	float	One column per TemporalAggregation × Scenario pair

Merge with features.parquet on cell_id for a complete cell-level panel.

Quick example

import pandas as pd
from terraflow.pipeline import run_pipeline

df = run_pipeline("examples/demo_config_climate_impact.yml")
run_dir = df.attrs["run_dir"]

features = pd.read_parquet(f"{run_dir}/features.parquet")
climate  = pd.read_parquet(f"{run_dir}/climate_features.parquet")
panel    = features.merge(climate, on="cell_id")

# panel now has historical suitability + every <rule>__<scenario> column.

API Reference

climate_impact

Climate-impact pipeline orchestration (flagship #138e).

Glues together the schema (#138a), aggregation engine (#138b), CMIP6 ingest (#138c), and hazard module (#138d) into one entry point that produces scenario-tagged cell-level columns from a long-format station time-series CSV.

The function lives in its own module so the historical single-period run_pipeline flow stays unchanged. Downstream tools merge the new climate_features.parquet artifact with the existing features.parquet on cell_id.

Usage::

from terraflow.climate_impact import run_climate_impact_features
from terraflow.config import load_config

cfg = load_config("config.yml")
out_path = run_climate_impact_features(cfg, run_dir, cells_df)

load_timeseries_csv(path)

Load and validate a long-format station time-series CSV.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the CSV file. Must contain columns station_id, lat, lon, date, temperature_c, precipitation_mm.	required

Returns:

Type	Description
pd.DataFrame:	Validated DataFrame with date parsed as datetime64[ns].

Source code in terraflow/climate_impact.py

def load_timeseries_csv(path: str | Path) -> pd.DataFrame:
    """Load and validate a long-format station time-series CSV.

    Parameters
    ----------
    path:
        Path to the CSV file. Must contain columns ``station_id``, ``lat``,
        ``lon``, ``date``, ``temperature_c``, ``precipitation_mm``.

    Returns
    -------
    pd.DataFrame:
        Validated DataFrame with ``date`` parsed as ``datetime64[ns]``.
    """
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"climate timeseries CSV not found: {path}")
    df = pd.read_csv(path)
    _validate_timeseries_schema(df)
    df[DATE_COL] = pd.to_datetime(df[DATE_COL])
    df["lat"] = df["lat"].astype(float)
    df["lon"] = df["lon"].astype(float)
    df[TEMPERATURE_COL] = df[TEMPERATURE_COL].astype(float)
    df[PRECIPITATION_COL] = df[PRECIPITATION_COL].astype(float)
    return df

run_climate_impact_features(cfg, run_dir, cells_df)

Compute scenario × rule cell-level derived columns and write the artifact.

Parameters:

Name	Type	Description	Default
cfg	PipelineConfig	Validated pipeline configuration with non-empty climate.temporal_aggregations and climate.scenarios.	required
run_dir	Path	Run directory under <output_dir>/runs/<run_fingerprint>/.	required
cells_df	DataFrame	Cell centroid DataFrame with at minimum cell_id, lat, lon.	required

Returns:

Name	Type	Description
Path	Path	Absolute path to the written climate_features.parquet.

Raises:

Type	Description
ValueError:	If the config lacks timeseries_csv or the CSV schema is invalid.
FileNotFoundError:	If the timeseries CSV does not exist.

Source code in terraflow/climate_impact.py

def run_climate_impact_features(
    cfg: PipelineConfig,
    run_dir: Path,
    cells_df: pd.DataFrame,
) -> Path:
    """Compute scenario × rule cell-level derived columns and write the artifact.

    Parameters
    ----------
    cfg:
        Validated pipeline configuration with non-empty
        ``climate.temporal_aggregations`` and ``climate.scenarios``.
    run_dir:
        Run directory under ``<output_dir>/runs/<run_fingerprint>/``.
    cells_df:
        Cell centroid DataFrame with at minimum ``cell_id``, ``lat``, ``lon``.

    Returns
    -------
    Path:
        Absolute path to the written ``climate_features.parquet``.

    Raises
    ------
    ValueError:
        If the config lacks ``timeseries_csv`` or the CSV schema is invalid.
    FileNotFoundError:
        If the timeseries CSV does not exist.
    """
    if not cfg.climate.temporal_aggregations or not cfg.climate.scenarios:
        raise ValueError(
            "run_climate_impact_features called with empty "
            "temporal_aggregations or scenarios — nothing to compute."
        )
    if not cfg.climate.timeseries_csv:
        raise ValueError(
            "run_climate_impact_features requires climate.timeseries_csv "
            "(long-format station daily data)."
        )

    logger.info(
        f"Climate-impact features: {len(cfg.climate.temporal_aggregations)} rules "
        f"× {len(cfg.climate.scenarios)} scenarios → "
        f"{len(cfg.climate.temporal_aggregations) * len(cfg.climate.scenarios)} derived columns"
    )

    timeseries_df = load_timeseries_csv(cfg.climate.timeseries_csv)
    stations_df = _stations_with_aggregations(timeseries_df, cfg)
    if stations_df.empty:
        raise RuntimeError(
            "no per-station aggregations produced; "
            "check that the timeseries CSV covers the configured scenario windows."
        )

    derived_columns = [
        c for c in stations_df.columns if c not in (STATION_ID_COL, "lat", "lon")
    ]

    cell_lats = cells_df["lat"].to_numpy()
    cell_lons = cells_df["lon"].to_numpy()

    result = pd.DataFrame({"cell_id": cells_df["cell_id"].to_numpy()})
    for column in derived_columns:
        result[column] = _interpolate_column_to_cells(
            stations_df, column, cell_lats, cell_lons, cfg
        ).to_numpy()

    out_path = run_dir / "climate_features.parquet"
    result.to_parquet(out_path, index=False)
    logger.info(
        f"Wrote climate_features.parquet ({len(derived_columns)} columns) → {out_path}"
    )
    return out_path