terraflow.temporal

Multi-temporal climate aggregation engine introduced in v0.5.0 (#138b). Computes per-station aggregations of a long-format daily climate time-series and fans them out across configured scenarios. The output is the cell-input that terraflow.climate_impact interpolates to the suitability grid.

Overview

• filter_scenario(df, scenario) — slice a long-format station time-series to a Scenario.period year window.
• aggregate_per_station(df, rule) — dispatch a TemporalAggregation to its kind-specific implementation. Returns one scalar per station.
• compute_per_station_aggregations(df, rules, scenarios) — outer product. Returns a DataFrame indexed by station_id with columns <rule_label>__<scenario_name> (one per rule × scenario pair).

Supported kind values (declared on ClimateConfig.temporal_aggregations):

kind	Required fields	Notes
annual_mean	—	Long-run mean of temperature_c.
seasonal_mean	months: [1..12]	Mean over selected calendar months.
growing_degree_days	base_temp_c: float	Cumulative sum of max(0, T - base) across every row in the scenario window — divide by year count for annualised GDD. Implementation in terraflow.hazard.
frost_days	threshold_c: float	Window-cumulative count of days at-or-below threshold. WMO ETCCDI FD0 (when threshold = 0) is per-year — divide by year count for the canonical annual value.
heat_stress_days	threshold_c: float	Window-cumulative count of days at-or-above threshold. WMO ETCCDI TX35 (when threshold = 35) is per-year — divide by year count for the canonical annual value.
precip_percentile	percentile: 0..100	Nth percentile of daily precipitation.
spei	timescale_months: int > 0	Simplified Thornthwaite SPEI at the final month of the input window. Implementation in terraflow.hazard.

Stations with no rows after a scenario filter resolve to NaN so downstream kriging LOOCV sees missing values cleanly rather than silently dropping stations.

Quick example

import pandas as pd
from terraflow.temporal import compute_per_station_aggregations
from terraflow.config import TemporalAggregation, Scenario

df = pd.read_csv("data/demo_timeseries.csv", parse_dates=["date"])
rules = [
    TemporalAggregation(kind="annual_mean"),
    TemporalAggregation(kind="growing_degree_days", base_temp_c=10.0),
]
scenarios = [
    Scenario(name="historical", period=[1991, 2020]),
    Scenario(name="ssp245",     period=[2041, 2070]),
]
panel = compute_per_station_aggregations(df, rules, scenarios)
panel.head()
#         annual_mean__historical  annual_mean__ssp245  ...
# KS01                       12.4                15.1

API Reference

temporal

Multi-temporal climate aggregation engine (flagship #138b).

Takes a long-format station time-series DataFrame and produces one scalar per station for each TemporalAggregation rule. Downstream callers (the climate-impact pipeline path landing in #138e) interpolate those per-station scalars onto cell centroids via the existing ClimateInterpolator.

Expected input shape::

station_id   lat       lon         date         temperature_c   precipitation_mm
KS-001       38.95    -100.50      2010-01-01   -2.3            0.0
KS-001       38.95    -100.50      2010-01-02   -1.1            1.4
...

Each aggregate_per_station call returns a pandas.Series indexed by station_id with scalar float values. Stations with no data after filtering yield NaN rather than raising — this lets downstream kriging LOOCV handle missing stations cleanly.

Hazard-specific kinds (growing_degree_days, frost_days, heat_stress_days, spei) dispatch into :mod:terraflow.hazard.

aggregate_per_station(df, rule)

Apply one TemporalAggregation rule to a station time-series.

Parameters:

Name	Type	Description	Default
df	DataFrame	Long-format station time-series. Must contain station_id and date columns plus whatever variable column the rule needs.	required
rule	TemporalAggregation	A validated :class:terraflow.config.TemporalAggregation instance.	required

Returns:

Type	Description
pd.Series:	Indexed by unique station_id. Float scalar per station. NaN when a station has no rows after filtering. Series name is set to a human-readable column identifier matching the rule's kind and parameters.

Source code in terraflow/temporal.py

def aggregate_per_station(
    df: pd.DataFrame,
    rule: TemporalAggregation,
) -> pd.Series:
    """Apply one ``TemporalAggregation`` rule to a station time-series.

    Parameters
    ----------
    df:
        Long-format station time-series. Must contain ``station_id`` and
        ``date`` columns plus whatever variable column the rule needs.
    rule:
        A validated :class:`terraflow.config.TemporalAggregation` instance.

    Returns
    -------
    pd.Series:
        Indexed by unique ``station_id``. Float scalar per station. NaN
        when a station has no rows after filtering. Series ``name`` is set
        to a human-readable column identifier matching the rule's kind
        and parameters.
    """
    if STATION_ID_COL not in df.columns:
        raise ValueError(
            f"input DataFrame missing required column {STATION_ID_COL!r}; "
            f"got columns: {sorted(df.columns)}"
        )

    if rule.kind == "growing_degree_days":
        from .hazard import growing_degree_days

        assert rule.base_temp_c is not None
        result = growing_degree_days(df, base_temp_c=rule.base_temp_c)
        return result.rename(f"growing_degree_days__base{rule.base_temp_c:g}")

    if rule.kind == "frost_days":
        from .hazard import frost_days

        assert rule.threshold_c is not None
        result = frost_days(df, threshold_c=rule.threshold_c)
        return result.rename(f"frost_days__t{rule.threshold_c:g}")

    if rule.kind == "heat_stress_days":
        from .hazard import heat_stress_days

        assert rule.threshold_c is not None
        result = heat_stress_days(df, threshold_c=rule.threshold_c)
        return result.rename(f"heat_stress_days__t{rule.threshold_c:g}")

    if rule.kind == "spei":
        from .hazard import spei

        assert rule.timescale_months is not None
        result = spei(df, timescale_months=rule.timescale_months)
        return result.rename(f"spei__ts{rule.timescale_months}")

    if rule.kind == "annual_mean":
        return _aggregate_annual_mean(df).rename("annual_mean")

    if rule.kind == "seasonal_mean":
        # months is required for seasonal_mean per #138a model_validator.
        assert rule.months is not None, "seasonal_mean.months should be validated"
        result = _aggregate_seasonal_mean(df, months=rule.months)
        months_label = "_".join(str(m) for m in rule.months)
        return result.rename(f"seasonal_mean__months_{months_label}")

    if rule.kind == "precip_percentile":
        assert rule.percentile is not None
        result = _aggregate_precip_percentile(df, percentile=rule.percentile)
        # Integer-valued percentiles render as p50 / p95 / p100 for
        # readability; fractional percentiles use the full str() so values
        # differing in the 6th+ significant digit (e.g. 99.99999 vs 100.0)
        # do NOT collide on the same label. ``:g`` was rounding both to
        # ``p100``, which would have silently overwritten one column in
        # ``compute_per_station_aggregations``.
        pct = rule.percentile
        if pct == int(pct):
            pct_label = str(int(pct))
        else:
            pct_label = str(pct).replace(".", "p")
        return result.rename(f"precip_percentile__p{pct_label}")

    raise ValueError(f"unsupported temporal_aggregation kind: {rule.kind!r}")

compute_per_station_aggregations(df, rules, scenarios)

Outer-product of rules × scenarios → per-station, per-(rule,scenario) scalars.

Result columns are <rule_label>__<scenario_name>. The index is the union of station_ids observed in df. Hazard-kind rules (handled in

138d) trigger :class:NotImplementedError from the dispatcher.

Returns an empty DataFrame when rules or scenarios is empty — this matches the v0.4.x single-period behaviour.

Source code in terraflow/temporal.py

def compute_per_station_aggregations(
    df: pd.DataFrame,
    rules: list[TemporalAggregation],
    scenarios: list[Scenario],
) -> pd.DataFrame:
    """Outer-product of *rules* × *scenarios* → per-station, per-(rule,scenario) scalars.

    Result columns are ``<rule_label>__<scenario_name>``. The index is the
    union of station_ids observed in *df*. Hazard-kind rules (handled in
    #138d) trigger :class:`NotImplementedError` from the dispatcher.

    Returns an empty DataFrame when *rules* or *scenarios* is empty —
    this matches the v0.4.x single-period behaviour.
    """
    if not rules or not scenarios:
        return pd.DataFrame()

    columns: dict[str, pd.Series] = {}
    all_station_ids = pd.Index(
        df[STATION_ID_COL].dropna().unique(), name=STATION_ID_COL
    )

    for scenario in scenarios:
        scenario_df = filter_scenario(df, scenario)
        for rule in rules:
            series = aggregate_per_station(scenario_df, rule)
            column_label = f"{series.name}__{scenario.name}"
            if column_label in columns:
                # Two rules generated the same column label — surface it
                # rather than silently overwriting the earlier series.
                raise ValueError(
                    f"duplicate generated column label {column_label!r}; "
                    "two TemporalAggregation rules produced the same name "
                    "(check for near-identical percentile values or repeated "
                    "rule kinds)."
                )
            # Re-index against the union of stations so each column has the
            # same index. Missing stations end up NaN.
            columns[column_label] = series.reindex(all_station_ids)

    out = pd.DataFrame(columns, index=all_station_ids)
    out.index.name = STATION_ID_COL
    return out

filter_scenario(df, scenario)

Return the subset of df whose date year falls in the scenario.

The scenario period is closed-inclusive: [year_min, year_max].

Parameters:

Name	Type	Description	Default
df	DataFrame	Long-format station time-series. Must contain a date column parsable as datetime64[ns].	required
scenario	Scenario	A validated :class:terraflow.config.Scenario instance.	required

Returns:

Type	Description
pd.DataFrame:	Filtered rows. Empty DataFrames are returned unchanged when no rows match the scenario window.

Source code in terraflow/temporal.py

def filter_scenario(df: pd.DataFrame, scenario: Scenario) -> pd.DataFrame:
    """Return the subset of *df* whose ``date`` year falls in the scenario.

    The scenario period is closed-inclusive: ``[year_min, year_max]``.

    Parameters
    ----------
    df:
        Long-format station time-series. Must contain a ``date`` column
        parsable as ``datetime64[ns]``.
    scenario:
        A validated :class:`terraflow.config.Scenario` instance.

    Returns
    -------
    pd.DataFrame:
        Filtered rows. Empty DataFrames are returned unchanged when no
        rows match the scenario window.
    """
    if DATE_COL not in df.columns:
        raise ValueError(
            f"input DataFrame missing required column {DATE_COL!r}; "
            f"got columns: {sorted(df.columns)}"
        )
    dates = pd.to_datetime(df[DATE_COL])
    year_min, year_max = scenario.period
    mask = (dates.dt.year >= year_min) & (dates.dt.year <= year_max)
    return df.loc[mask].copy()