Skip to content

Contracts API

Contracts

Defines the "data contracts": the schemas defining the precise shape of each data source and its semantics.

It also owns the thin configuration layer that sits beside those schemas: the CV fold config, and the class_target/import_class pair that turns a class into a _target_ string and back. Both are model-agnostic and need nothing heavier than pydantic and PyYAML.

Dependency Isolation

This package is designed to be extremely lightweight. It defines the shape of the data using Patito and Polars, but it does not contain any ML-specific logic or heavy dependencies like MLflow. This ensures that any component in the system (e.g., a data ingestion script or a dashboard) can import these schemas without bringing in the entire ML stack.

Key Data Contracts

  • PowerTimeSeries: Half-hourly power observations (MW or MVA) per time_series_id, as received from NGED.
  • TimeSeriesMetadata: Substation and customer meter metadata, including lat/lon, H3 index, and asset type (primary substation, GSP, BSP, solar PV, wind, BESS, etc.).
  • Nwp: ECMWF ENS NWP weather data in physical units (Float32), on disk and in memory alike. The on-disk copy is rounded to a 13-bit significand and laid out for compression and row-group pruning by delta_store.nwp.
  • AllFeatures: The final joined dataset passed to ML models. Primary key is (time_series_id, power_fcst_init_time, valid_time[, ensemble_member]). Includes NWP weather variables, power lag/rolling features and datetime features. time_series_type is the one metadata column it can carry, and only when a feature set asks for it.
  • PowerForecast: ML model output schema. power_fcst is in MW (active power) or MVA (apparent power), with the unit given per time_series_id in TimeSeriesMetadata. A planned change will normalise it to [−1, +1] for NGED to multiply by a capacity — see Forecast Building Blocks. Includes power_fcst_model_name, power_fcst_model_version, power_fcst_init_time, nwp_init_time, valid_time, time_series_id, and ensemble_member.

Design Principles

  • The contract is the authoritative account of what the data means. It says what the data should be, not what some current code path happens to produce. So when code and contract disagree, the code is the first suspect: a null the contract forbids usually means an upstream join kept a row it should have dropped, or a caller passed input it should have rejected. Widening a field to | None, or relaxing a range, so that a failing validate() passes buries that defect in the one place the rest of the system trusts. Fix the code instead, and change the contract only when you can say what the data now means and why that meaning is right. Get the change agreed before making it, including a widening that looks like a formality — every reader of contracts is relying on it to still mean what it said yesterday.
  • Column naming: Prefer snake_case, except for acronyms or SI units. Capitalise "DER" (distributed energy resource) and use uppercase for "MW" (megawatts).
  • Semantic checks: Range validation should be generous — the aim is to catch physically impossible values (e.g., 1 GW from a 1 MW solar farm), not possible-but-unlikely values.
  • Datetime ranges: Timestamps on the columns where external data enters — PowerTimeSeries.time, Nwp.init_time and Nwp.valid_time — are bounded to [MIN_PLAUSIBLE_DATETIME, MAX_PLAUSIBLE_DATETIME] (2000-01-01 to 2100-01-01, inclusive), which rejects a corrupt feed or an epoch-unit mix-up without ever excluding a real reading. The check lives in each model's validate override via check_datetime_bounds, because Patito silently ignores ge/le on a datetime field — it derives its bounds checks from the JSON schema's minimum/maximum, which JSON Schema defines for numbers only. Columns on our own output schemas (PowerForecast, EffectiveCapacity, AllFeatures) have not opted in: they are computed from already-bounded inputs rather than received from outside.
  • Degrade, don't abort, at an ingestion boundary: validate() stays strict everywhere — it is also used as a hard assertion in tests and R&D code, where a raise-on-violation contract must not silently change. But a single malformed row from an external feed should not abort ingestion of every other well-formed row in the batch, so PowerTimeSeries.drop_implausible_rows() filters out rows with an out-of-range or minute-misaligned time before validate() runs, returning the survivors plus a count of what was dropped. Only the NGED JSON ingestion path (nged_data.read_nged_json) calls it; the duplicate/sortedness checks in validate() are never relaxed, because those indicate a bug in our own pipeline rather than malformed external data.
  • No lookahead bias: AllFeatures carries power_fcst_init_time (when we make the forecast) as a distinct field from nwp_init_time (when the NWP model ran). Power lag features are nullified by nullify_leaky_lags() when the lag is shorter than or equal to the forecast lead time.

contracts.common

Shared building blocks for the Patito schemas.

The canonical UTC dtype, the plausible-datetime bounds and the checks that enforce them, and the delivery quantile levels.

Attributes

UTC_DATETIME_DTYPE = pl.Datetime(time_unit='us', time_zone='UTC') module-attribute

MIN_PLAUSIBLE_DATETIME = datetime(2000, 1, 1, tzinfo=UTC) module-attribute

The earliest timestamp a bounded datetime column may carry (inclusive).

A column is bounded when its model's validate passes it to :func:check_datetime_bounds; the constant says nothing about columns that have not opted in.

NGED telemetry cannot predate the instrumentation that produced it, and the ECMWF archive we ingest begins later still, so no legitimate row is older than this. The bound is also deliberately far later than 1847: Europe/London ran on local mean time at UTC−0:01:15 until then, so a pre-1848 timestamp produces a sub-minute UTC offset and a nonsensical value for every local-time feature. Enforcing this bound is what guarantees the local-time features in ml_core never see a sub-minute UTC offset.

MAX_PLAUSIBLE_DATETIME = datetime(2100, 1, 1, tzinfo=UTC) module-attribute

The latest timestamp a bounded datetime column may carry (inclusive).

This is a fixed date rather than an offset from the current time, so validation never depends on the wall clock: a frame that validated when it was written still validates when it is read back years later, and tests need no clock control. A fixed far-future bound still catches an epoch-unit mix-up, where Unix milliseconds read as seconds land tens of thousands of years in the future — the plausible failure on any path that converts a numeric timestamp rather than parsing an ISO-8601 string. It deliberately does not catch a small clock skew that ships tomorrow's data as today's; that is a monitoring concern, not a contract one.

DELIVERY_QUANTILES = (0.01, 0.02, 0.05, 0.1, 0.2, 0.35, 0.5, 0.65, 0.8, 0.9, 0.95, 0.98, 0.99) module-attribute

The thirteen quantile levels agreed with NGED for the delivery tables.

Deliberately tail-heavy: NGED is far more interested in the tails than the shoulders. This tuple is the single source of truth for every quantile-indexed artefact — the pinball-loss metric_param labels today, and the percentile columns of the delivery-table representations (Representations 2 and 3) when those land in v0.5.

Functions:

check_datetime_bounds(dataframe, column, *more_columns)

Raise ValueError if any timestamp lies outside the plausible-datetime range.

Call this from a Patito model's validate override, after super().validate(). It exists because Patito silently ignores ge/le on a datetime field: Patito derives its bounds checks from the Pydantic JSON schema's minimum/maximum keywords, which JSON Schema defines for numbers only, so a datetime field's Ge/Le metadata never reaches the JSON schema and no check is ever generated. (ge/le on a numeric field works normally, which is why PowerTimeSeries.power can state its bounds on the field itself.)

Parameters:

Name Type Description Default
dataframe DataFrame

An already-validated frame. Every named column must be a datetime column.

 required
column str

Name of a datetime column to bound.

 required
*more_columns str

Names of any further datetime columns to bound.

 ()

Raises:

Type Description
ValueError

If any value is before :data:MIN_PLAUSIBLE_DATETIME or after :data:MAX_PLAUSIBLE_DATETIME. Nulls are ignored — absence is not malformedness — and an empty frame always passes.
Source code in packages/contracts/src/contracts/common.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def check_datetime_bounds(dataframe: pl.DataFrame, column: str, *more_columns: str) -> None:
    """Raise ``ValueError`` if any timestamp lies outside the plausible-datetime range.

    Call this from a Patito model's ``validate`` override, after ``super().validate()``. It exists
    because Patito **silently ignores** ``ge``/``le`` on a datetime field: Patito derives its bounds
    checks from the Pydantic JSON schema's ``minimum``/``maximum`` keywords, which JSON Schema
    defines for numbers only, so a datetime field's ``Ge``/``Le`` metadata never reaches the JSON
    schema and no check is ever generated. (``ge``/``le`` on a *numeric* field works normally, which
    is why ``PowerTimeSeries.power`` can state its bounds on the field itself.)

    Args:
        dataframe: An already-validated frame. Every named column must be a datetime column.
        column: Name of a datetime column to bound.
        *more_columns: Names of any further datetime columns to bound.

    Raises:
        ValueError: If any value is before :data:`MIN_PLAUSIBLE_DATETIME` or after
            :data:`MAX_PLAUSIBLE_DATETIME`. Nulls are ignored — absence is not malformedness — and
            an empty frame always passes.
    """
    columns = (column, *more_columns)
    extremes = dataframe.select(
        *(pl.col(name).min().alias(f"min_{name}") for name in columns),
        *(pl.col(name).max().alias(f"max_{name}") for name in columns),
    ).row(0, named=True)
    for name in columns:
        _raise_if_outside_plausible_range(name, extremes[f"min_{name}"], extremes[f"max_{name}"])

split_by_datetime_plausibility(dataframe, column)

Partition dataframe into (plausible, implausible) rows by column.

A row is implausible when its column value is before :data:MIN_PLAUSIBLE_DATETIME or after :data:MAX_PLAUSIBLE_DATETIME — the same bounds :func:check_datetime_bounds enforces. Nulls are always plausible (absence is not malformedness).

Use this at an ingestion boundary to drop-and-report malformed external rows instead of aborting the whole batch; use :func:check_datetime_bounds where a hard assertion is appropriate instead (e.g. inside a Patito model's validate).

Parameters:

Name Type Description Default
dataframe DataFrame

Any frame; column must be a datetime column.

 required
column str

Name of the datetime column to test.

 required

Returns:

Type Description
tuple[DataFrame, DataFrame]

(plausible, implausible), each keeping dataframe's row order and schema.
Source code in packages/contracts/src/contracts/common.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def split_by_datetime_plausibility(
    dataframe: pl.DataFrame, column: str
) -> tuple[pl.DataFrame, pl.DataFrame]:
    """Partition ``dataframe`` into ``(plausible, implausible)`` rows by ``column``.

    A row is implausible when its ``column`` value is before :data:`MIN_PLAUSIBLE_DATETIME` or
    after :data:`MAX_PLAUSIBLE_DATETIME` — the same bounds :func:`check_datetime_bounds` enforces.
    Nulls are always plausible (absence is not malformedness).

    Use this at an ingestion boundary to drop-and-report malformed external rows instead of
    aborting the whole batch; use :func:`check_datetime_bounds` where a hard assertion is
    appropriate instead (e.g. inside a Patito model's ``validate``).

    Args:
        dataframe: Any frame; ``column`` must be a datetime column.
        column: Name of the datetime column to test.

    Returns:
        ``(plausible, implausible)``, each keeping ``dataframe``'s row order and schema.
    """
    is_implausible = pl.col(column).is_not_null() & (
        (pl.col(column) < MIN_PLAUSIBLE_DATETIME) | (pl.col(column) > MAX_PLAUSIBLE_DATETIME)
    )
    return dataframe.filter(~is_implausible), dataframe.filter(is_implausible)

quantile_label(quantile)

Return the canonical p{level} label for a quantile, e.g. 0.05 → "p5".

The label format matches the percentile column names agreed with NGED for the delivery tables (see DELIVERY_QUANTILES).

Source code in packages/contracts/src/contracts/common.py 
151
152
153
154
155
156
157
def quantile_label(quantile: float) -> str:
    """Return the canonical ``p{level}`` label for a quantile, e.g. ``0.05`` → ``"p5"``.

    The label format matches the percentile column names agreed with NGED for the delivery
    tables (see ``DELIVERY_QUANTILES``).
    """
    return f"p{round(quantile * 100)}"

validate_schema(model, df)

Validate a Polars DataFrame's or LazyFrame's schema against a Patito model.

Raises DataFrameValidationError on failure. On LazyFrames this materializes no data; it just calls collect_schema().

Source code in packages/contracts/src/contracts/common.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def validate_schema(model: type[pt.Model], df: pl.DataFrame | pl.LazyFrame) -> None:
    """Validate a Polars DataFrame's or LazyFrame's schema against a Patito model.

    Raises `DataFrameValidationError` on failure. On LazyFrames this materializes no data; it just
    calls `collect_schema()`.
    """
    # Get actual schema
    actual_schema = dict(df.collect_schema()) if isinstance(df, pl.LazyFrame) else dict(df.schema)

    # Check for missing columns
    missing_cols = set(model.dtypes.keys()) - set(actual_schema.keys())
    if missing_cols:
        error = ErrorWrapper(
            MissingColumnsError(f"Missing columns: {missing_cols}"), loc=tuple(missing_cols)
        )
        raise DataFrameValidationError([error], model)

    # Check for dtype mismatches
    errors = []
    for col, expected_dtype in model.dtypes.items():
        if col in actual_schema:
            actual_dtype = actual_schema[col]
            if actual_dtype != expected_dtype:
                errors.append(
                    ErrorWrapper(
                        ColumnDTypeError(
                            f"Column '{col}' expected {expected_dtype}, got {actual_dtype}"
                        ),
                        loc=(col,),
                    )
                )

    # Raise the native Patito exception if errors were found
    if errors:
        raise DataFrameValidationError(errors, model)

contracts.power_schemas

Data schemas for the NGED substation forecast project.

Attributes

LIST_OF_TIME_SERIES_TYPES = ('BESS', 'Biofuel', 'CHP', 'Data Centre', 'Disaggregated Demand', 'Energy from Waste', 'EV Charging', 'Geothermal', 'Hydro', 'Hydrogen Electrolysis', 'Industrial Demand', 'Mixed (Demand)', 'Mixed (Generation)', 'Other (Demand)', 'Other (Generation)', 'Other (Storage)', 'Peaking Plant', 'PV', 'Rail', 'Raw Flow', 'Synchronous Condenser', 'Wind') module-attribute

All time-series type values used in NGED data.

Types present in the V1 trial area: BESS, Biofuel, Disaggregated Demand, Other (Generation), PV, Raw Flow, Wind.

Notes:

  • BESS: Battery energy storage system.
  • Disaggregated Demand: In the trial area, exclusively associated with "Primary" substations. All "Primary" substations in the trial area have their TimeSeriesType set to "Disaggregated Demand". Indicates that NGED have already removed metered generation connected to that primary.
  • Raw Flow: Used for BSP and GSP substations.

FoldId = str module-attribute

Fold identifier for PowerForecast.fold_id.

A CV fold's id is a short label defined in conf/cv/default.yaml (e.g. "mid_2025_to_mid_2026"); fold identity is config-driven, never hard-coded here. "live" is the reserved sentinel for a production forecast that belongs to no CV fold.

Classes

DropImplausibleRowsResult

Bases: NamedTuple

Result of PowerTimeSeries.drop_implausible_rows.

Source code in packages/contracts/src/contracts/power_schemas.py 
20
21
22
23
24
class DropImplausibleRowsResult(NamedTuple):
    """Result of ``PowerTimeSeries.drop_implausible_rows``."""

    survivors: pl.DataFrame
    n_dropped: int
Attributes
survivors instance-attribute
n_dropped instance-attribute

PowerTimeSeries

Bases: Model

Half-hourly power observations (MW or MVA), one row per (time_series_id, time).

Source code in packages/contracts/src/contracts/power_schemas.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
class PowerTimeSeries(pt.Model):
    """Half-hourly power observations (MW or MVA), one row per (time_series_id, time)."""

    time_series_id: int = _get_time_series_id_dtype()

    time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "End time of the 30-minute observation period (all NGED data is already half-hourly)."
            f" Must fall between {MIN_PLAUSIBLE_DATETIME:%Y-%m-%d} and"
            f" {MAX_PLAUSIBLE_DATETIME:%Y-%m-%d} (enforced by `validate`, not by the field, because"
            " Patito ignores `ge`/`le` on datetime fields — see `check_datetime_bounds`)."
        ),
    )

    power: float = pt.Field(
        dtype=pl.Float32,
        ge=-1000,
        le=1000,
        description=(
            "Average power (MW or MVA) over the preceding 30-minute period. Unit defined in "
            "TimeSeriesMetadata."
            " Sign convention depends on `substation_type` in `TimeSeriesMetadata`. At a"
            " substation (`BSP`, `GSP`, `Primary`), positive means power flowing towards"
            " end-users and negative means excess generation flowing back into the grid. At a"
            " customer meter (`EHV Customer`, `HV Customer`), positive means the customer is"
            " sending power to NGED's grid and negative means the customer is drawing power from"
            " it. Those five values are the whole enum, so every series falls into exactly one"
            " case."
        ),
    )

    @classmethod
    def validate(  # ty: ignore[invalid-method-override]
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:
        """Validate the given dataframe, ensuring time is plausible, at :00 or :30, and unique."""
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        # Validate time falls in the plausible range (Patito ignores `ge`/`le` on datetime fields)
        check_datetime_bounds(validated_df, "time")

        # Validate time is at :00 or :30
        minutes = validated_df["time"].dt.minute()
        if not minutes.is_in([0, 30]).all():
            raise ValueError("time must be at the top or bottom of the hour (minute 00 or 30).")

        # Validate uniqueness of (time_series_id, time)
        if validated_df.select(["time_series_id", "time"]).is_duplicated().any():
            raise ValueError("Duplicate entries found for (time_series_id, time).")

        # Validate the time_series_id column is sorted
        if not validated_df["time_series_id"].is_sorted():
            raise ValueError("time_series_id is not sorted!")

        # Validate the time column is sorted (within each time_series_id group)
        if (
            not validated_df.group_by("time_series_id")
            .agg(pl.col("time").diff().min() > 0)["time"]
            .all()
        ):
            raise ValueError("the `time` column is not sorted!")

        return validated_df

    @classmethod
    def drop_implausible_rows(cls, dataframe: pl.DataFrame) -> DropImplausibleRowsResult:
        """Drop rows with a malformed ``time``, returning ``(survivors, n_dropped)``.

        A row is dropped when its ``time`` lies outside the plausible datetime range, is null (the
        schema declares ``time`` non-nullable, so a null this early is already malformed), or does
        not fall on the top or bottom of the hour (minute 00 or 30). All three indicate a
        malformed upstream reading — not a bug in our own pipeline — so under [inherent
        stability](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/)
        an ingestion boundary should degrade the batch rather than abort it entirely.

        This exists alongside ``validate``, which stays strict and raises on the same two rules:
        ``validate`` is also used as a hard assertion in tests and R&D code, where a
        raise-on-violation contract must not silently change. Call this method BEFORE
        ``validate``, and only at a boundary that receives data from outside our system (e.g.
        NGED's raw JSON feed) — the uniqueness and sortedness checks in ``validate`` are NOT
        relaxed here, because those indicate a bug in OUR pipeline, not malformed external data,
        and should keep raising.

        Args:
            dataframe: An already-cast frame with a ``time`` column; need not yet be validated.

        Returns:
            ``(survivors, n_dropped)``. ``survivors`` keeps ``dataframe``'s row order.
        """
        survivors, _ = split_by_datetime_plausibility(dataframe, "time")
        # `dt.minute()` is null for a null `time` and `.filter()` drops a row on a null predicate,
        # so this also drops the null `time`s the non-nullable schema forbids.
        survivors = survivors.filter(pl.col("time").dt.minute().is_in([0, 30]))
        # Counted as a height difference rather than by summing the rejected partitions: whatever
        # a filter does with a null predicate, every row that left is counted exactly once.
        return DropImplausibleRowsResult(survivors, dataframe.height - survivors.height)

    # Define it as a ClassVar so Patito/Pydantic knows it's not a data field
    columns_to_sort_by: ClassVar[tuple[str, str]] = ("time_series_id", "time")
Attributes
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
time = pt.Field(dtype=UTC_DATETIME_DTYPE, description=f'End time of the 30-minute observation period (all NGED data is already half-hourly). Must fall between {MIN_PLAUSIBLE_DATETIME} and {MAX_PLAUSIBLE_DATETIME} (enforced by `validate`, not by the field, because Patito ignores `ge`/`le` on datetime fields — see `check_datetime_bounds`).') class-attribute instance-attribute
power = pt.Field(dtype=(pl.Float32), ge=(-1000), le=1000, description="Average power (MW or MVA) over the preceding 30-minute period. Unit defined in TimeSeriesMetadata. Sign convention depends on `substation_type` in `TimeSeriesMetadata`. At a substation (`BSP`, `GSP`, `Primary`), positive means power flowing towards end-users and negative means excess generation flowing back into the grid. At a customer meter (`EHV Customer`, `HV Customer`), positive means the customer is sending power to NGED's grid and negative means the customer is drawing power from it. Those five values are the whole enum, so every series falls into exactly one case.") class-attribute instance-attribute
columns_to_sort_by = ('time_series_id', 'time') class-attribute
Methods:
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the given dataframe, ensuring time is plausible, at :00 or :30, and unique.

Source code in packages/contracts/src/contracts/power_schemas.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@classmethod
def validate(  # ty: ignore[invalid-method-override]
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:
    """Validate the given dataframe, ensuring time is plausible, at :00 or :30, and unique."""
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    # Validate time falls in the plausible range (Patito ignores `ge`/`le` on datetime fields)
    check_datetime_bounds(validated_df, "time")

    # Validate time is at :00 or :30
    minutes = validated_df["time"].dt.minute()
    if not minutes.is_in([0, 30]).all():
        raise ValueError("time must be at the top or bottom of the hour (minute 00 or 30).")

    # Validate uniqueness of (time_series_id, time)
    if validated_df.select(["time_series_id", "time"]).is_duplicated().any():
        raise ValueError("Duplicate entries found for (time_series_id, time).")

    # Validate the time_series_id column is sorted
    if not validated_df["time_series_id"].is_sorted():
        raise ValueError("time_series_id is not sorted!")

    # Validate the time column is sorted (within each time_series_id group)
    if (
        not validated_df.group_by("time_series_id")
        .agg(pl.col("time").diff().min() > 0)["time"]
        .all()
    ):
        raise ValueError("the `time` column is not sorted!")

    return validated_df
drop_implausible_rows(dataframe) classmethod

Drop rows with a malformed time, returning (survivors, n_dropped).

A row is dropped when its time lies outside the plausible datetime range, is null (the schema declares time non-nullable, so a null this early is already malformed), or does not fall on the top or bottom of the hour (minute 00 or 30). All three indicate a malformed upstream reading — not a bug in our own pipeline — so under inherent stability an ingestion boundary should degrade the batch rather than abort it entirely.

This exists alongside validate, which stays strict and raises on the same two rules: validate is also used as a hard assertion in tests and R&D code, where a raise-on-violation contract must not silently change. Call this method BEFORE validate, and only at a boundary that receives data from outside our system (e.g. NGED's raw JSON feed) — the uniqueness and sortedness checks in validate are NOT relaxed here, because those indicate a bug in OUR pipeline, not malformed external data, and should keep raising.

Parameters:

Name Type Description Default
dataframe DataFrame

An already-cast frame with a time column; need not yet be validated.

 required

Returns:

Type Description
DropImplausibleRowsResult

(survivors, n_dropped). survivors keeps dataframe's row order.
Source code in packages/contracts/src/contracts/power_schemas.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@classmethod
def drop_implausible_rows(cls, dataframe: pl.DataFrame) -> DropImplausibleRowsResult:
    """Drop rows with a malformed ``time``, returning ``(survivors, n_dropped)``.

    A row is dropped when its ``time`` lies outside the plausible datetime range, is null (the
    schema declares ``time`` non-nullable, so a null this early is already malformed), or does
    not fall on the top or bottom of the hour (minute 00 or 30). All three indicate a
    malformed upstream reading — not a bug in our own pipeline — so under [inherent
    stability](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/)
    an ingestion boundary should degrade the batch rather than abort it entirely.

    This exists alongside ``validate``, which stays strict and raises on the same two rules:
    ``validate`` is also used as a hard assertion in tests and R&D code, where a
    raise-on-violation contract must not silently change. Call this method BEFORE
    ``validate``, and only at a boundary that receives data from outside our system (e.g.
    NGED's raw JSON feed) — the uniqueness and sortedness checks in ``validate`` are NOT
    relaxed here, because those indicate a bug in OUR pipeline, not malformed external data,
    and should keep raising.

    Args:
        dataframe: An already-cast frame with a ``time`` column; need not yet be validated.

    Returns:
        ``(survivors, n_dropped)``. ``survivors`` keeps ``dataframe``'s row order.
    """
    survivors, _ = split_by_datetime_plausibility(dataframe, "time")
    # `dt.minute()` is null for a null `time` and `.filter()` drops a row on a null predicate,
    # so this also drops the null `time`s the non-nullable schema forbids.
    survivors = survivors.filter(pl.col("time").dt.minute().is_in([0, 30]))
    # Counted as a height difference rather than by summing the rejected partitions: whatever
    # a filter does with a null predicate, every row that left is counted exactly once.
    return DropImplausibleRowsResult(survivors, dataframe.height - survivors.height)

TimeSeriesMetadata

Bases: Model

One row per substation or asset: its name, location, H3 index, and substation type.

Source code in packages/contracts/src/contracts/power_schemas.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class TimeSeriesMetadata(pt.Model):
    """One row per substation or asset: its name, location, H3 index, and substation type."""

    time_series_id: int = _get_time_series_id_dtype(unique=True)

    time_series_name: str = pt.Field(
        dtype=pl.String,
        description="Human-readable name for the substation or asset.",
        examples=[
            "ALFORD 33 11kV S STN",
            "BAMBERS FARM WIND GENERATION MABLETHORPE 33kV S ST",
            "Leverton Solar Park",
        ],
    )

    time_series_type: str = pt.Field(
        dtype=pl.Enum(LIST_OF_TIME_SERIES_TYPES),
        description=(
            "Asset category (e.g. ‘PV’, ‘Wind’, ‘Disaggregated Demand’). See "
            "LIST_OF_TIME_SERIES_TYPES."
        ),
    )

    units: str = pt.Field(
        dtype=pl.Enum(["MW", "MVA"]),
        description=(
            "Power unit for this time series: ‘MW’ (active power) or ‘MVA’ (apparent power)."
        ),
    )

    licence_area: str = pt.Field(
        dtype=pl.Enum(["EMids"]),
        description="NGED licence area (for the trail area, this is always ‘EMids’).",
    )

    substation_number: int = pt.Field(
        dtype=pl.Int32,
        gt=0,
        lt=1_000_000,
        description=(
            "Perhaps surprisingly, each customer meter in the NGED trial area has its own "
            "substation_number (not one per physical substation)."
        ),
    )

    substation_type: str = pt.Field(
        dtype=pl.Enum(["BSP", "EHV Customer", "GSP", "HV Customer", "Primary"]),
        description=(
            "Substation voltage level / role: BSP, EHV Customer, GSP, HV Customer, or Primary."
            " HV = high voltage. EHV = extra high voltage."
        ),
    )

    latitude: float = pt.Field(
        dtype=pl.Float32,
        ge=49,
        le=61,  # UK latitude range
        description=(
            "Latitude in decimal degrees. For customer time series, gives the location of the "
            "substation, not the customer's site."
        ),
    )

    longitude: float = pt.Field(
        dtype=pl.Float32,
        ge=-9,
        le=2,  # UK longitude range
        description=(
            "Longitude in decimal degrees. For customer time series, gives the location of the "
            "substation, not the customer's site."
        ),
    )

    information: str | None = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        description="Free-text NGED notes field; always null in the V1 trial area.",
    )

    area_wkt: str | None = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        # Maps to the nested Area.WKT field in the JSON data.
        description=(
            "WKT polygon for the asset’s area. In the trial, only Primary substations have this."
            " NGED don’t have polygons for customer sites (though they hope to add that in future)."
            " For customer sites, where present, refers to the area covered by the generator"
            " itself."
        ),
    )

    area_center_lat: float | None = pt.Field(
        dtype=pl.Float32,
        allow_missing=True,
        description=(
            "Centroid latitude of the area polygon. For customer sites, the area, where present, "
            "refers to the area covered by the generator itself."
        ),
    )

    area_center_lon: float | None = pt.Field(
        dtype=pl.Float32,
        allow_missing=True,
        description=(
            "Centroid longitude of the area polygon. For customer sites, the area, where present, "
            "refers to the area covered by the generator itself."
        ),
    )

    h3_res_5: int = pt.Field(
        dtype=pl.UInt64,
        description="H3 discrete spatial index at resolution 5.",
    )
Attributes
time_series_id = _get_time_series_id_dtype(unique=True) class-attribute instance-attribute
time_series_name = pt.Field(dtype=(pl.String), description='Human-readable name for the substation or asset.', examples=['ALFORD 33 11kV S STN', 'BAMBERS FARM WIND GENERATION MABLETHORPE 33kV S ST', 'Leverton Solar Park']) class-attribute instance-attribute
time_series_type = pt.Field(dtype=(pl.Enum(LIST_OF_TIME_SERIES_TYPES)), description='Asset category (e.g. ‘PV’, ‘Wind’, ‘Disaggregated Demand’). See LIST_OF_TIME_SERIES_TYPES.') class-attribute instance-attribute
units = pt.Field(dtype=(pl.Enum(['MW', 'MVA'])), description='Power unit for this time series: ‘MW’ (active power) or ‘MVA’ (apparent power).') class-attribute instance-attribute
licence_area = pt.Field(dtype=(pl.Enum(['EMids'])), description='NGED licence area (for the trail area, this is always ‘EMids’).') class-attribute instance-attribute
substation_number = pt.Field(dtype=(pl.Int32), gt=0, lt=1000000, description='Perhaps surprisingly, each customer meter in the NGED trial area has its own substation_number (not one per physical substation).') class-attribute instance-attribute
substation_type = pt.Field(dtype=(pl.Enum(['BSP', 'EHV Customer', 'GSP', 'HV Customer', 'Primary'])), description='Substation voltage level / role: BSP, EHV Customer, GSP, HV Customer, or Primary. HV = high voltage. EHV = extra high voltage.') class-attribute instance-attribute
latitude = pt.Field(dtype=(pl.Float32), ge=49, le=61, description="Latitude in decimal degrees. For customer time series, gives the location of the substation, not the customer's site.") class-attribute instance-attribute
longitude = pt.Field(dtype=(pl.Float32), ge=(-9), le=2, description="Longitude in decimal degrees. For customer time series, gives the location of the substation, not the customer's site.") class-attribute instance-attribute
information = pt.Field(dtype=(pl.String), allow_missing=True, description='Free-text NGED notes field; always null in the V1 trial area.') class-attribute instance-attribute
area_wkt = pt.Field(dtype=(pl.String), allow_missing=True, description='WKT polygon for the asset’s area. In the trial, only Primary substations have this. NGED don’t have polygons for customer sites (though they hope to add that in future). For customer sites, where present, refers to the area covered by the generator itself.') class-attribute instance-attribute
area_center_lat = pt.Field(dtype=(pl.Float32), allow_missing=True, description='Centroid latitude of the area polygon. For customer sites, the area, where present, refers to the area covered by the generator itself.') class-attribute instance-attribute
area_center_lon = pt.Field(dtype=(pl.Float32), allow_missing=True, description='Centroid longitude of the area polygon. For customer sites, the area, where present, refers to the area covered by the generator itself.') class-attribute instance-attribute
h3_res_5 = pt.Field(dtype=(pl.UInt64), description='H3 discrete spatial index at resolution 5.') class-attribute instance-attribute

PowerForecast

Bases: Model

Forecast data schema for deterministic ensemble forecasts.

Internal vs delivered schema (Milestone 1 report Table 1, p.28): the columns experiment_name, fold_id, and ml_flow_experiment_id are INTERNAL-ONLY — they exist on this schema and the internal power_forecasts Delta table to support cross-validation and the leaderboard, but they are NOT part of the power_forecast table delivered to NGED.

Source code in packages/contracts/src/contracts/power_schemas.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
class PowerForecast(pt.Model):
    """Forecast data schema for deterministic ensemble forecasts.

    Internal vs delivered schema (Milestone 1 report Table 1, p.28): the columns
    ``experiment_name``, ``fold_id``, and ``ml_flow_experiment_id`` are INTERNAL-ONLY —
    they exist on this schema and the internal ``power_forecasts`` Delta table to support
    cross-validation and the leaderboard, but they are NOT part of the ``power_forecast``
    table delivered to NGED.
    """

    valid_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        constraints=pl.col("valid_time") > pl.col("power_fcst_init_time"),
        description=(
            "The target time this forecast is valid for. Constrained to be strictly after"
            " power_fcst_init_time: a row targeting a valid time at or before its own"
            " initialisation is an undeliverable hindcast row — at power_fcst_init_time that"
            " valid time is already observed. The live service only forecasts strictly future"
            " valid times and bulk-mode feature engineering drops hindcast rows at source, so"
            " a constraint violation here indicates a pipeline regression."
        ),
    )

    time_series_id: int = _get_time_series_id_dtype()

    ensemble_member: int = pt.Field(
        dtype=pl.Int8, description="Ensemble member index. 0 is the control NWP ensemble member."
    )

    ml_flow_experiment_id: int | None = pt.Field(
        dtype=pl.Int32,
        allow_missing=True,
        description=(
            "MLflow experiment ID; links to the MLflow experiment that produced this forecast."
        ),
    )

    nwp_init_time: datetime | None = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description=(
            "The datetime that the underlying weather forecast was initialised. "
            "Null for models that do not use NWP (e.g. persistence baselines)."
        ),
    )

    power_fcst_model_name: str = pt.Field(
        dtype=pl.String,  # String, not Categorical — see experiment_name below.
        description=(
            "Identifier for our ML-based power forecasting model. Model-family identity set by the "
            "BaseForecaster subclass (MODEL_NAME)."
        ),
    )

    # String (not Categorical): experiment_name/fold_id are the Delta partition columns and delta-rs
    # stores dictionary-encoded columns as String anyway; String keeps them cast-free and lets
    # predicate pushdown work. See the "Delta Lake dictionary-encoded columns" section of the
    # `polars-patito-gotchas` skill.
    experiment_name: str = pt.Field(
        dtype=pl.String,
        description=(
            "Per-experiment key identifying the experiment that produced this forecast."
            " Distinct from `power_fcst_model_name`, which is the model-family identity"
            " (`MODEL_NAME`); do not overload that with experiment identity."
            " Forecasts are partitioned in Delta by (experiment_name, fold_id)."
            " INTERNAL-ONLY: projected out of the `power_forecast` table delivered to NGED."
        ),
    )

    power_fcst_model_version: int = pt.Field(
        dtype=pl.Int16,
        description=(
            "Model version integer, bumped with each breaking change to the model implementation."
        ),
    )

    power_fcst_init_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "The datetime that the power forecast was initialised. This might be called `t0` in "
            "some other OCF projects."
        ),
    )

    power_fcst: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "The power forecast itself in units of MW (active power) or MVA (apparent power)."
            " The unit is defined in the `TimeSeriesMetadata` for this `time_series_id`."
            " Sign convention depends on `substation_type` in `TimeSeriesMetadata`. At a"
            " substation (`BSP`, `GSP`, `Primary`), positive means power flowing towards"
            " end-users and negative means excess generation flowing back into the grid. At a"
            " customer meter (`EHV Customer`, `HV Customer`), positive means the customer is"
            " sending power to NGED's grid and negative means the customer is drawing power from"
            " it. Those five values are the whole enum, so every series falls into exactly one"
            " case."
            " Rows read back from the internal `power_forecasts` Delta table carry reduced"
            " precision: values are rounded to a 13-bit significand at write time"
            " (max relative error 2^-13 ≈ 1.2e-4, far below forecast error) to aid compression;"
            " see `delta_store.power_forecasts`."
            # PLANNED: We intend to change `power_fcst` to a normalised value in the range
            # [-1, +1] (which NGED multiplies by a capacity to recover MW/MVA), per the
            # delivery-contract design agreed with NGED in the Milestone 1 report. The switch is
            # planned for v0.5, using the static P99 `effective_capacity` estimate that already
            # exists — the same scalar the `metrics` pipeline already divides by for NMAE — so it
            # no longer needs to wait for a time-varying capacity estimate.
        ),
    )

    fold_id: FoldId = pt.Field(
        dtype=pl.String,  # String, not Categorical — see experiment_name above.
        description=(
            "Identifies the source of this forecast row.  "
            "For cross-validation runs, the value is the fold's label from conf/cv/default.yaml "
            "(e.g. 'mid_2025_to_mid_2026').  "
            "'live' means a production forecast with no associated CV fold.  "
            "All forecasts — CV and live — live in the same Delta table; "
            "filter on this column to select the population you need."
        ),
    )

    PRIMARY_KEY: ClassVar[tuple[str, ...]] = (
        "time_series_id",
        "power_fcst_init_time",
        "valid_time",
        "ensemble_member",
    )
    """At most one forecast per series, per init time, per target time, per ensemble member."""

    @classmethod
    def validate(  # ty: ignore[invalid-method-override]
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:
        """Validate the given dataframe, ensuring the primary key is unique.

        A duplicated primary key means either a join fanned out on the way here or the same rows
        were written twice, and both corrupt the metrics computed from them. It is our own bug
        rather than the outside world misbehaving, so this raises rather than degrading — see
        <https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/>.
        """
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        # `n_unique`, not `is_duplicated().any()`: the two are equivalent here (every primary-key
        # column is non-nullable) but `is_duplicated` materialises a per-row mask, costing ~5x the
        # peak memory on a predict-sized frame.
        pk_cols = list(cls.PRIMARY_KEY)
        if validated_df.select(pk_cols).n_unique() != validated_df.height:
            raise ValueError(
                f"Duplicate entries found for primary key columns: {pk_cols}. "
                "Either an upstream join fanned out or these rows were written twice."
            )

        return validated_df
Attributes
valid_time = pt.Field(dtype=UTC_DATETIME_DTYPE, constraints=(pl.col('valid_time') > pl.col('power_fcst_init_time')), description='The target time this forecast is valid for. Constrained to be strictly after power_fcst_init_time: a row targeting a valid time at or before its own initialisation is an undeliverable hindcast row — at power_fcst_init_time that valid time is already observed. The live service only forecasts strictly future valid times and bulk-mode feature engineering drops hindcast rows at source, so a constraint violation here indicates a pipeline regression.') class-attribute instance-attribute
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
ensemble_member = pt.Field(dtype=(pl.Int8), description='Ensemble member index. 0 is the control NWP ensemble member.') class-attribute instance-attribute
ml_flow_experiment_id = pt.Field(dtype=(pl.Int32), allow_missing=True, description='MLflow experiment ID; links to the MLflow experiment that produced this forecast.') class-attribute instance-attribute
nwp_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='The datetime that the underlying weather forecast was initialised. Null for models that do not use NWP (e.g. persistence baselines).') class-attribute instance-attribute
power_fcst_model_name = pt.Field(dtype=(pl.String), description='Identifier for our ML-based power forecasting model. Model-family identity set by the BaseForecaster subclass (MODEL_NAME).') class-attribute instance-attribute
experiment_name = pt.Field(dtype=(pl.String), description='Per-experiment key identifying the experiment that produced this forecast. Distinct from `power_fcst_model_name`, which is the model-family identity (`MODEL_NAME`); do not overload that with experiment identity. Forecasts are partitioned in Delta by (experiment_name, fold_id). INTERNAL-ONLY: projected out of the `power_forecast` table delivered to NGED.') class-attribute instance-attribute
power_fcst_model_version = pt.Field(dtype=(pl.Int16), description='Model version integer, bumped with each breaking change to the model implementation.') class-attribute instance-attribute
power_fcst_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description='The datetime that the power forecast was initialised. This might be called `t0` in some other OCF projects.') class-attribute instance-attribute
power_fcst = pt.Field(dtype=(pl.Float32), description="The power forecast itself in units of MW (active power) or MVA (apparent power). The unit is defined in the `TimeSeriesMetadata` for this `time_series_id`. Sign convention depends on `substation_type` in `TimeSeriesMetadata`. At a substation (`BSP`, `GSP`, `Primary`), positive means power flowing towards end-users and negative means excess generation flowing back into the grid. At a customer meter (`EHV Customer`, `HV Customer`), positive means the customer is sending power to NGED's grid and negative means the customer is drawing power from it. Those five values are the whole enum, so every series falls into exactly one case. Rows read back from the internal `power_forecasts` Delta table carry reduced precision: values are rounded to a 13-bit significand at write time (max relative error 2^-13 ≈ 1.2e-4, far below forecast error) to aid compression; see `delta_store.power_forecasts`.") class-attribute instance-attribute
fold_id = pt.Field(dtype=(pl.String), description="Identifies the source of this forecast row. For cross-validation runs, the value is the fold's label from conf/cv/default.yaml (e.g. 'mid_2025_to_mid_2026'). 'live' means a production forecast with no associated CV fold. All forecasts — CV and live — live in the same Delta table; filter on this column to select the population you need.") class-attribute instance-attribute
PRIMARY_KEY = ('time_series_id', 'power_fcst_init_time', 'valid_time', 'ensemble_member') class-attribute

At most one forecast per series, per init time, per target time, per ensemble member.

Methods:
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the given dataframe, ensuring the primary key is unique.

A duplicated primary key means either a join fanned out on the way here or the same rows were written twice, and both corrupt the metrics computed from them. It is our own bug rather than the outside world misbehaving, so this raises rather than degrading — see https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/.

Source code in packages/contracts/src/contracts/power_schemas.py 
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
@classmethod
def validate(  # ty: ignore[invalid-method-override]
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:
    """Validate the given dataframe, ensuring the primary key is unique.

    A duplicated primary key means either a join fanned out on the way here or the same rows
    were written twice, and both corrupt the metrics computed from them. It is our own bug
    rather than the outside world misbehaving, so this raises rather than degrading — see
    <https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/>.
    """
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    # `n_unique`, not `is_duplicated().any()`: the two are equivalent here (every primary-key
    # column is non-nullable) but `is_duplicated` materialises a per-row mask, costing ~5x the
    # peak memory on a predict-sized frame.
    pk_cols = list(cls.PRIMARY_KEY)
    if validated_df.select(pk_cols).n_unique() != validated_df.height:
        raise ValueError(
            f"Duplicate entries found for primary key columns: {pk_cols}. "
            "Either an upstream join fanned out or these rows were written twice."
        )

    return validated_df

EffectiveCapacity

Bases: Model

Effective capacity of each time series at each half-hourly timestep.

Delivered to NGED as effective_capacity Delta table (Table 4 in the Milestone 1 report). This table is backward-looking only — it does not cover the forecast period.

v0.1 implementation: one row per time_series_id, time set to the end of the available observation history, effective_capacity_mw = P99 of abs(power) over the full observed history. This is a static scalar per series.

Planned upgrade (v0.7): replace the P99 scalar with a time-varying capacity estimate (see https://openclimatefix.github.io/nged-substation-forecast/techniques/convex-optimisation/ and https://openclimatefix.github.io/nged-substation-forecast/techniques/differentiable-physics/ for the candidate estimation methods), giving one row per (time_series_id, time) half-hourly timestep. This schema is unchanged; the effective_capacity asset body changes and the metrics pipeline swaps its time_series_id-only NMAE-denominator join for a temporal as-of join. Do not pre-densify the v0.1 scalar into one row per half-hour — densifying a constant buys nothing, and the as-of join handles sparse capacity rows naturally.

Source code in packages/contracts/src/contracts/power_schemas.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
class EffectiveCapacity(pt.Model):
    """Effective capacity of each time series at each half-hourly timestep.

    Delivered to NGED as ``effective_capacity`` Delta table (Table 4 in the Milestone 1 report).
    This table is backward-looking only — it does not cover the forecast period.

    **v0.1 implementation:** one row per ``time_series_id``, ``time`` set to the end of
    the available observation history, ``effective_capacity_mw`` = P99 of ``abs(power)`` over
    the full observed history. This is a static scalar per series.

    **Planned upgrade (v0.7):** replace the P99 scalar with a time-varying capacity estimate
    (see
    <https://openclimatefix.github.io/nged-substation-forecast/techniques/convex-optimisation/> and
    <https://openclimatefix.github.io/nged-substation-forecast/techniques/differentiable-physics/>
    for the candidate estimation methods),
    giving one row per ``(time_series_id, time)`` half-hourly timestep. This schema is unchanged;
    the ``effective_capacity`` asset body changes and the ``metrics`` pipeline swaps its
    ``time_series_id``-only NMAE-denominator join for a temporal as-of join. Do **not** pre-densify
    the v0.1 scalar into one row per half-hour — densifying a constant buys nothing, and the as-of
    join handles sparse capacity rows naturally.
    """

    time_series_id: int = _get_time_series_id_dtype()

    time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "The half-hourly timestep this capacity estimate applies to. "
            "In v0.1, this is the end of the available observation history for that series."
        ),
    )

    effective_capacity_mw: float = pt.Field(
        dtype=pl.Float32,
        gt=0,
        description=(
            "OCF's estimate of the effective capacity (MW) of this asset at this timestep. "
            "For generators: absorbs PV panel degradation, partial inverter trips, etc., "
            "but ignores ANM curtailment — a wind farm ANM-capped at 5 MW with 10 MW physical "
            "capability has effective_capacity_mw = 10. "
            "For substations: the 99th percentile of observed load over a rolling time window, "
            "under normal running arrangement only. 'Switched' power (Table 5) should be "
            "added or subtracted when a switching event is in effect."
        ),
    )
Attributes
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
time = pt.Field(dtype=UTC_DATETIME_DTYPE, description='The half-hourly timestep this capacity estimate applies to. In v0.1, this is the end of the available observation history for that series.') class-attribute instance-attribute
effective_capacity_mw = pt.Field(dtype=(pl.Float32), gt=0, description="OCF's estimate of the effective capacity (MW) of this asset at this timestep. For generators: absorbs PV panel degradation, partial inverter trips, etc., but ignores ANM curtailment — a wind farm ANM-capped at 5 MW with 10 MW physical capability has effective_capacity_mw = 10. For substations: the 99th percentile of observed load over a rolling time window, under normal running arrangement only. 'Switched' power (Table 5) should be added or subtracted when a switching event is in effect.") class-attribute instance-attribute

Functions:

contracts.weather_schemas

Contracts for numerical weather prediction data.

The Nwp frame as stored and read, plus the run-completeness and data-quality reports that assess an ingested ECMWF ENS run.

Attributes

WeatherFeature = Literal['temperature_2m', 'dew_point_temperature_2m', 'wind_speed_10m', 'wind_direction_10m', 'wind_speed_100m', 'wind_direction_100m', 'pressure_surface', 'pressure_reduced_to_mean_sea_level', 'geopotential_height_500hpa', 'downward_long_wave_radiation_flux_surface', 'downward_short_wave_radiation_flux_surface', 'precipitation_surface', 'categorical_precipitation_type_surface'] module-attribute

ECMWF_ENS_ENSEMBLE_MEMBERS = frozenset(range(51)) module-attribute

The ensemble members every ECMWF IFS ENS run carries: the control member plus 50 perturbed members, indexed 0-50 by Dynamical.org (matching Nwp.ensemble_member, which is 0-based).

The count is a fixed property of the ECMWF IFS ENS configuration, corroborated by our own data: the 2026-07-14 incident described in https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/ lost a forecast step for "50 of 51 ensemble members", and the per-partition row arithmetic in https://openclimatefix.github.io/nged-substation-forecast/architecture/performance/ is 1671 H3 cells x 51 ensemble members x 85 native steps.

ECMWF_ENS_LEAD_TIME_HOURS = tuple(range(0, 145, 3)) + tuple(range(150, 361, 6)) module-attribute

The 85 native forecast steps of the ECMWF IFS ENS 15-day 0.25-degree dataset, in hours after init_time: 3-hourly out to 144 h (49 steps, lead-0 inclusive), then 6-hourly out to 360 h (36 steps) — i.e. init_time to init_time + 15 days inclusive.

Lead-0 is included: it is a real step, distinguished only by the de-accumulated variables being legitimately null there (see Nwp.deaccumulated_var_names).

ECMWF_ENS_H3_RESOLUTION = 5 module-attribute

The H3 spatial resolution the ecmwf_ens ingest pipeline grids ECMWF ENS data to, and the resolution power_time_series_and_metadata computes for every time series so the two line up on one grid. Every NWP model shares this one resolution today; per-model resolution (so a future NWP source can use a different one) is tracked in https://github.com/openclimatefix/nged-substation-forecast/issues/114.

Classes

NwpModelId

Bases: StrEnum

The NWP models we ingest, used as the nwp_model_id column's vocabulary.

Source code in packages/contracts/src/contracts/weather_schemas.py 
40
41
42
43
class NwpModelId(StrEnum):
    """The NWP models we ingest, used as the `nwp_model_id` column's vocabulary."""

    ECMWF_ENS_0_25_degree = auto()
Attributes
ECMWF_ENS_0_25_degree = auto() class-attribute instance-attribute

NwpVariableWhollyMissing

Bases: ValueError

A de-accumulated NWP variable is null in every slice beyond lead-0 of a run.

The field is wholesale absent rather than locally corrupt, so Nwp.validate rejects it: an all-null weather column would otherwise train and serve as a silently-degraded input for the full 15-day horizon, which is worse than falling back on the previous run.

A distinct exception type rather than a bare ValueError because the ecmwf_ens asset retries it. An upstream publication still in progress can present this way — a variable's chunks read as fill-value null until the worker writing them commits — so waiting is a better first response than failing the partition. Note the limit of that: it only reaches this check when the unwritten variables are the de-accumulated ones. The nine instantaneous variables are non-nullable, so a frame missing one of those is rejected by base Patito validation first, with no retry — as is an all-null categorical_precipitation_type_surface, which is nullable but carries its own historical invariant. See https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Source code in packages/contracts/src/contracts/weather_schemas.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class NwpVariableWhollyMissing(ValueError):
    """A de-accumulated NWP variable is null in *every* slice beyond lead-0 of a run.

    The field is wholesale absent rather than locally corrupt, so `Nwp.validate` rejects it: an
    all-null weather column would otherwise train and serve as a silently-degraded input for the
    full 15-day horizon, which is worse than falling back on the previous run.

    A distinct exception type rather than a bare `ValueError` because the `ecmwf_ens` asset
    retries it. An upstream publication still in progress can present this way — a variable's
    chunks read as fill-value null until the worker writing them commits — so waiting is a better
    first response than failing the partition. Note the limit of that: it only reaches this check
    when the unwritten variables are the de-accumulated ones. The nine instantaneous variables are
    non-nullable, so a frame missing one of those is rejected by base Patito validation first, with
    no retry — as is an all-null `categorical_precipitation_type_surface`, which is nullable but
    carries its own historical invariant. See
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """

Nwp

Bases: Model

Weather data schema for NWP forecasts.

Gridded ECMWF ENS ensemble weather, one row per (nwp_model_id, init_time, valid_time, ensemble_member, h3_index).

Stored on disk as plain Float32, rounded to a significand-bit budget by delta_store.nwp.write_nwp — see https://openclimatefix.github.io/nged-substation-forecast/architecture/overview/ for the physical format and measured numbers.

validate is the fatal ingest gate; assess_nwp_quality reports the tolerated nulls in the de-accumulated variables (the known upstream ECMWF ENS corruption) and assess_nwp_run_completeness reports a run that is missing whole members, steps or cells — both as non-fatal checks, because both are the upstream provider misbehaving rather than a contract violation. Which patterns are fatal versus tolerated, and why, is documented at https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Source code in packages/contracts/src/contracts/weather_schemas.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
class Nwp(pt.Model):
    """Weather data schema for NWP forecasts.

    Gridded ECMWF ENS ensemble weather, one row per
    (nwp_model_id, init_time, valid_time, ensemble_member, h3_index).

    Stored on disk as plain Float32, rounded to a significand-bit budget by
    `delta_store.nwp.write_nwp` — see
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/overview/> for the
    physical format and measured numbers.

    `validate` is the fatal ingest gate; `assess_nwp_quality` reports the tolerated nulls in the
    de-accumulated variables (the known upstream ECMWF ENS corruption) and
    `assess_nwp_run_completeness` reports a run that is missing whole members, steps or cells —
    both as non-fatal checks, because both are the upstream provider misbehaving rather than a
    contract violation. Which patterns are fatal versus tolerated, and why, is documented at
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """

    # `dtype=pl.String` (not `Enum`, which Delta cannot store) means an out-of-vocabulary value is
    # no longer physically inexpressible the way it was under `Enum` — a plain `str` column can
    # hold any string at construction time, with no error. The `constraints=` check below is now
    # the *only* defence, and it fires at `Nwp.validate()`, not at construction: a frame built and
    # never validated can carry an unrecognised `nwp_model_id` all the way through. Every current
    # caller validates, so this is latent rather than live today.
    nwp_model_id: str = pt.Field(
        dtype=pl.String,
        constraints=pl.col("nwp_model_id").is_in([model.name for model in NwpModelId]),
        description="Which NWP model produced this row (e.g. 'ECMWF_ENS_0_25_degree').",
    )

    init_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description="When the NWP model run was initialised.",
    )

    valid_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "The time for which this NWP value is valid. Most variables (temperature, wind, "
            "pressure, geopotential) are instantaneous — they describe conditions at this moment. "
            "Precipitation and radiation (downward_long_wave_radiation_flux_surface, "
            "downward_short_wave_radiation_flux_surface, precipitation_surface) are period-ending "
            "rates: each value represents the average rate over the period that ends at valid_time "
            "(i.e. the preceding forecast step interval). Dynamical.org de-accumulates these from "
            "ECMWF's raw cumulative fields before we receive them. Resampling must honour that "
            "distinction — the step interval is 3 h out to lead 144 h and 6 h beyond it, so "
            "treating a period-ending value as instantaneous shifts it by up to 3 h. Conventions "
            "for every variable, and what the shift costs: "
            "<https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>"
        ),
    )

    ensemble_member: int = pt.Field(
        dtype=pl.Int8,
        description="Ensemble member index (0-based).",
    )

    # `Int64`, not `UInt64`, because Delta has no unsigned integer types: the column is stored as
    # `int64` whatever we declare, and declaring `UInt64` only bought a cast in `scan_delta` that
    # stopped Polars pushing `h3_index` filters into the Parquet scan.
    #
    # Signed is safe for the whole H3 index space, not just the values we happen to store. H3
    # reserves bit 63 of its 64-bit index as always zero — it is not a value bit in any index mode
    # (cell, directed edge, vertex) at any resolution — so every H3 index is below 2**63, which is
    # exactly the range a signed `Int64` covers. A cell index sets mode bits 59-62 to 1 and so tops
    # out around 2**60, well inside that. Sampling every resolution 0-15 agrees: the largest index
    # seen is ~6.5e17 against `Int64`'s ~9.2e18 ceiling, and bit 63 is never set.
    #
    # The guarantee does not depend on the resolution, so raising or lowering
    # `ECMWF_ENS_H3_RESOLUTION`, or giving a future NWP model its own resolution (issue 114),
    # cannot overflow this column. Resolution lives in bits 52-55 and leaves bit 63 alone.
    h3_index: int = pt.Field(
        dtype=pl.Int64,
        description=(
            "H3 cell index, at `ECMWF_ENS_H3_RESOLUTION` (every NWP model shares it today)."
        ),
    )

    temperature_2m: float = pt.Field(
        dtype=pl.Float32,
        description="Air temperature at 2 m above ground. Unit: degrees C.",
        ge=-100,
        le=100,
    )

    dew_point_temperature_2m: float = pt.Field(
        dtype=pl.Float32,
        description="Dew-point temperature at 2 m. Unit: degrees C.",
        ge=-100,
        le=100,
    )

    wind_speed_10m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind speed at 10 m. Unit: meters per second. This is the magnitude of the cell's"
            " *vector* mean wind, not the mean of its grid points' scalar speeds. See"
            " <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>"
        ),
        ge=0,
        le=200,  # Gemini says the highest non-tornadic surface wind speed recorded was 113 m/s
    )

    wind_direction_10m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind direction at 10 m. The angle where the wind is coming from. Degrees."
            " 0° is North; 90° is East. This is a *circular* quantity: 359° and 1° are two degrees"
            " apart, so interpolating, averaging, differencing or taking a quantile of it as if it"
            " were an ordinary number is wrong wherever the values straddle North. Convert to"
            " components first. See"
            " <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>"
        ),
        ge=0,
        le=360,
    )

    wind_speed_100m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind speed at 100 m. Unit: meters per second. This is a vector mean, with the same"
            " caveat as `wind_speed_10m`. It is the height that matters most for wind generation,"
            " because a turbine power curve responds to the scalar speed at each grid point. See"
            " <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>"
        ),
        ge=0,
        le=200,  # Gemini says the highest non-tornadic surface wind speed recorded was 113 m/s
    )

    wind_direction_100m: float = pt.Field(
        dtype=pl.Float32,
        description=(
            "Wind direction at 100 m. The angle where the wind is coming from. Degrees. 0° is "
            "North; 90° is East. This is circular, with the same handling requirement as "
            "`wind_direction_10m`."
        ),
        ge=0,
        le=360,
    )

    pressure_surface: float = pt.Field(
        dtype=pl.Float32,
        description="Surface pressure. Unit: Pa.",
        ge=0,
        le=200_000,  # Max in 2 years of ECMWF ENS = 105_760
    )

    pressure_reduced_to_mean_sea_level: float = pt.Field(
        dtype=pl.Float32,
        description="Mean sea-level pressure. Unit: Pa.",
        ge=0,
        le=200_000,  # Max in 2 years of ECMWF ENS = 105_727
    )

    geopotential_height_500hpa: float = pt.Field(
        dtype=pl.Float32,
        description="Geopotential height of the 500 hPa pressure surface. Unit: m.",
        ge=0,
        le=10_000,  # Max in 2 years of ECMWF ENS = 6_030
    )

    # Precipitation and radiation variables are null for the first forecast step (lead time 0) in
    # ECMWF ENS. Also note that, whilst these variables accumulate over forecast steps in ECMWF's
    # raw forecasts, we get ECMWF ENS from Dynamical.org, and Dynamical.org de-accumulates these
    # values before we receive them. So these are true _rates_.
    downward_long_wave_radiation_flux_surface: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "Downward long-wave radiation flux at surface. Note that this variable is all-null for "
            "lead time 0. Unit: W m-2."
        ),
        ge=0,
        le=1500,  # Max in 2 years of ECMWF ENS = 445
    )

    downward_short_wave_radiation_flux_surface: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "Downward short-wave (solar) radiation flux at surface. Note that this variable is"
            " all-null for lead time 0. Unit: W m-2."
        ),
        ge=0,
        le=1500,  # Max in 2 years of ECMWF ENS = 892
    )

    precipitation_surface: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "Total precipitation rate at surface, de-accumulated by Dynamical. Note that this "
            "variable is all-null for lead time 0. Unit: kg m-2 s-1."
        ),
        ge=0,
        le=0.01,  # Max in 2 years of ECMWF ENS = 0.006
    )

    categorical_precipitation_type_surface: int | None = pt.Field(
        dtype=pl.Int16,
        ge=0,
        le=255,
        description=(
            "This field is always NaN for init_times on and before 2024-11-12, and populated from"
            " the 2024-11-13 00Z run onwards (confirmed by direct inspection of the source data)."
            " Derived from ECMWF's `ptype` field. See https://codes.ecmwf.int/grib/param-db/260015"
            " 0=No precipitation; 1=Rain; 2=Thunderstorm; 3=Freezing rain; 4=Mixed/ice;"
            " 5=Snow; 6=Wet snow; 7=Mixture of rain and snow; 8=Ice pellets; 9=Graupel;"
            " 10=Hail; 11=Drizzle; 12=Freezing drizzle; 13=Hail (less than 5 mm);"
            " 14=Hail (greater than or equal to 5 mm);"
            " 15-191=Reserved; 192-254=Reserved for local use; 255=Missing"
        ),
    )

    # ClassVars: excluded from Patito/Pydantic model fields so they're not treated as data columns.
    categorical_var_names: ClassVar[frozenset[str]] = frozenset(
        {"categorical_precipitation_type_surface"}
    )

    deaccumulated_var_names: ClassVar[frozenset[str]] = frozenset(
        {
            "precipitation_surface",
            "downward_short_wave_radiation_flux_surface",
            "downward_long_wave_radiation_flux_surface",
        }
    )
    """The variables Dynamical.org de-accumulates from ECMWF's cumulative source fields to rates.

    All three are legitimately null at lead-0, and they share a de-accumulation step whose known
    upstream corruption leaves further nulls beyond it. Those are tolerated at ingest and reported
    by :func:`assess_nwp_quality`; only a variable null in *every* slice beyond lead-0 is fatal.
    Which corruption patterns arrive, what the H3 aggregation absorbs before they reach a validated
    frame, and why the survivors are tolerated:
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#nulls-in-the-de-accumulated-variables-tolerated>.
    """

    # Columns that aren't NWP variables:
    _non_var_column_names: ClassVar[frozenset[str]] = frozenset(
        {"nwp_model_id", "init_time", "valid_time", "ensemble_member", "h3_index"}
    )

    @classmethod
    def all_weather_var_names(cls) -> frozenset[str]:
        """All meteorological variable field names (continuous + categorical)."""
        return frozenset(cls.model_fields) - cls._non_var_column_names

    @classmethod
    def continuous_var_names(cls) -> frozenset[str]:
        """Meteorological variable field names suitable for linear interpolation."""
        return cls.all_weather_var_names() - cls.categorical_var_names

    @classmethod
    def validate(
        cls,
        dataframe: pl.DataFrame,
        columns: Sequence[str] | None = None,
        allow_missing_columns: bool = False,
        allow_superfluous_columns: bool = False,
        drop_superfluous_columns: bool = False,
    ) -> pt.DataFrame[Self]:  # ty:ignore[invalid-method-override]
        """Validate the frame.

        Bounds `init_time`/`valid_time` to the plausible datetime range, rejects a de-accumulated
        variable that is wholly missing (smaller null patterns are tolerated), and enforces both
        uniqueness and the ptype-introduction invariant.
        """
        validated_df = super().validate(
            dataframe=dataframe,
            columns=columns,
            allow_missing_columns=allow_missing_columns,
            allow_superfluous_columns=allow_superfluous_columns,
            drop_superfluous_columns=drop_superfluous_columns,
        )

        # Patito ignores `ge`/`le` on datetime fields, so the range check lives here.
        check_datetime_bounds(validated_df, "init_time", "valid_time")
        cls._check_no_wholly_missing_deaccumulated_variable(validated_df)
        cls._check_unique(validated_df)
        cls._check_variables_that_were_introduced_after_start_of_dataset(validated_df)
        return validated_df

    @classmethod
    def _check_no_wholly_missing_deaccumulated_variable(cls, dataframe: pt.DataFrame[Self]) -> None:
        """Reject a de-accumulated variable that carries no weather at all.

        That is, one that is null in *every* (ensemble_member, valid_time) slice beyond lead-0 of
        a run.

        Every smaller null pattern is *tolerated* and reported by :func:`assess_nwp_quality`
        instead, so this is a cliff rather than a slope: a run one slice short of empty lands with
        a warning. There is no tunable fraction — the test is that *nothing* survives.

        Why an absent column is the one case worth discarding a run over:
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#a-wholly-missing-variable-and-instantaneous-nulls-fatal>.
        Why every smaller pattern is not — the slice arithmetic, and the interpolation argument
        that makes a tolerated slice survivable:
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#nulls-in-the-de-accumulated-variables-tolerated>.

        Two things a caller must not assume:

        - The judgement is made per `init_time`, so a run whose column is empty is caught even
          inside a frame holding other, healthy runs — but by the same token, a frame filtered down
          to nothing *but* a wholly-null slice is indistinguishable from an empty column and does
          raise, even though that same slice was deliberately landed when the whole run was
          validated. Latent rather than live today: the only production caller validates one whole
          run, and reads go through `scan_delta`/`set_model`, which do not validate.
        - Raising is not the end of the partition. :class:`NwpVariableWhollyMissing` is a distinct
          type because the `ecmwf_ens` asset retries it rather than failing outright.
        """
        slices_per_run = (
            dataframe.filter(pl.col("valid_time") > pl.col("init_time"))
            .group_by("init_time")
            .agg(n_slices=pl.struct("ensemble_member", "valid_time").n_unique())
        )
        wholly_missing = (
            _deaccumulated_null_breakdown(dataframe)
            .filter(pl.col("n_null") == pl.col("n_total"))
            .group_by("variable", "init_time")
            .agg(n_whole_null_slices=pl.len())
            # Per `init_time`: comparing a variable's wholly-null slice count against the frame's
            # *total* would let a wholly-missing run hide behind a healthy one in the same frame.
            .join(slices_per_run, on="init_time", how="inner")
            .filter(pl.col("n_whole_null_slices") == pl.col("n_slices"))
        )
        if wholly_missing.height:
            offenders = sorted(
                (row["variable"], str(row["init_time"]), row["n_slices"])
                for row in wholly_missing.iter_rows(named=True)
            )
            raise NwpVariableWhollyMissing(
                "De-accumulated NWP variable(s) null in every (ensemble_member, valid_time) slice "
                "beyond lead-0 of their run — the field is wholesale absent, not the "
                f"locally-corrupt scatter we tolerate. Offending (variable, init_time, n_slices): "
                f"{offenders}"
            )

    @classmethod
    def _check_unique(cls, dataframe: pt.DataFrame[Self]) -> None:
        if (
            dataframe.select(
                ["nwp_model_id", "init_time", "valid_time", "ensemble_member", "h3_index"]
            )
            .is_duplicated()
            .any()
        ):
            raise ValueError(
                "Duplicate entries found for "
                "(nwp_model_id, init_time, valid_time, ensemble_member, h3_index)."
            )

    @classmethod
    def _check_variables_that_were_introduced_after_start_of_dataset(
        cls, dataframe: pt.DataFrame[Self]
    ) -> None:
        """Check the introduction date of `categorical_precipitation_type_surface`.

        The column is all-null when init_time <= 2024-11-12, and is never null afterwards.

        Confirmed by direct inspection of the source data: the 2024-11-13 00Z run is the first
        run with `ptype` populated (0% null across all lead times), while 2024-11-12 and earlier
        are 100% null.
        """
        threshold_date = datetime(2024, 11, 12, tzinfo=UTC)

        # Partition the dataframe based on the threshold date
        partition_col = "is_before_or_on_threshold"
        partitioned_df = dataframe.with_columns(
            (pl.col("init_time") <= threshold_date).alias(partition_col)
        )
        partitions = partitioned_df.partition_by(partition_col, as_dict=True)

        empty_df = pl.DataFrame(schema=dataframe.schema)
        before_or_on = partitions.get((True,), empty_df)
        after = partitions.get((False,), empty_df)

        # Check before or on threshold
        if not before_or_on["categorical_precipitation_type_surface"].is_null().all():
            raise ValueError(
                "categorical_precipitation_type_surface must be all null for "
                "init_time <= 2024-11-12"
            )

        # Check after threshold
        if after["categorical_precipitation_type_surface"].is_null().any():
            raise ValueError(
                "categorical_precipitation_type_surface must not be null for init_time > 2024-11-12"
            )

    @classmethod
    def scan_delta(
        cls,
        path: str | Path | None = None,
        storage_options: ObjectStoreOptions | None = None,
    ) -> pt.LazyFrame[Self]:
        """Lazily scan the NWP Delta table, typed and cast to this contract's dtypes.

        The table stores physical-unit `Float32` directly (see `delta_store.nwp.write_nwp`),
        so no rescale step is needed.

        Args:
            path: Path or URI of the ``nwp`` Delta table; defaults to
                ``get_settings().nwp_data_path`` (resolved lazily so importing this module needs
                no ``.env``).
            storage_options: delta-rs object-store options (credentials/endpoint) for a remote
                ``path``; defaults to ``get_settings().storage_options`` (empty for a local
                ``path``).
        """
        settings = get_settings()
        if path is None:
            path = settings.nwp_data_path
        if storage_options is None:
            storage_options = settings.storage_options
        return (
            pt.LazyFrame.from_existing(
                pl.scan_delta(path, storage_options=typeddict_to_dict(storage_options))
            )
            .set_model(cls)
            .cast()
        )
Attributes
nwp_model_id = pt.Field(dtype=(pl.String), constraints=(pl.col('nwp_model_id').is_in([(model.name) for model in NwpModelId])), description="Which NWP model produced this row (e.g. 'ECMWF_ENS_0_25_degree').") class-attribute instance-attribute
init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description='When the NWP model run was initialised.') class-attribute instance-attribute
valid_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description="The time for which this NWP value is valid. Most variables (temperature, wind, pressure, geopotential) are instantaneous — they describe conditions at this moment. Precipitation and radiation (downward_long_wave_radiation_flux_surface, downward_short_wave_radiation_flux_surface, precipitation_surface) are period-ending rates: each value represents the average rate over the period that ends at valid_time (i.e. the preceding forecast step interval). Dynamical.org de-accumulates these from ECMWF's raw cumulative fields before we receive them. Resampling must honour that distinction — the step interval is 3 h out to lead 144 h and 6 h beyond it, so treating a period-ending value as instantaneous shifts it by up to 3 h. Conventions for every variable, and what the shift costs: <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>") class-attribute instance-attribute
ensemble_member = pt.Field(dtype=(pl.Int8), description='Ensemble member index (0-based).') class-attribute instance-attribute
h3_index = pt.Field(dtype=(pl.Int64), description='H3 cell index, at `ECMWF_ENS_H3_RESOLUTION` (every NWP model shares it today).') class-attribute instance-attribute
temperature_2m = pt.Field(dtype=(pl.Float32), description='Air temperature at 2 m above ground. Unit: degrees C.', ge=(-100), le=100) class-attribute instance-attribute
dew_point_temperature_2m = pt.Field(dtype=(pl.Float32), description='Dew-point temperature at 2 m. Unit: degrees C.', ge=(-100), le=100) class-attribute instance-attribute
wind_speed_10m = pt.Field(dtype=(pl.Float32), description="Wind speed at 10 m. Unit: meters per second. This is the magnitude of the cell's *vector* mean wind, not the mean of its grid points' scalar speeds. See <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>", ge=0, le=200) class-attribute instance-attribute
wind_direction_10m = pt.Field(dtype=(pl.Float32), description='Wind direction at 10 m. The angle where the wind is coming from. Degrees. 0° is North; 90° is East. This is a *circular* quantity: 359° and 1° are two degrees apart, so interpolating, averaging, differencing or taking a quantile of it as if it were an ordinary number is wrong wherever the values straddle North. Convert to components first. See <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/>', ge=0, le=360) class-attribute instance-attribute
wind_speed_100m = pt.Field(dtype=(pl.Float32), description='Wind speed at 100 m. Unit: meters per second. This is a vector mean, with the same caveat as `wind_speed_10m`. It is the height that matters most for wind generation, because a turbine power curve responds to the scalar speed at each grid point. See <https://openclimatefix.github.io/nged-substation-forecast/architecture/nwp-variable-conventions/#wind-is-stored-as-speed-and-direction-and-why>', ge=0, le=200) class-attribute instance-attribute
wind_direction_100m = pt.Field(dtype=(pl.Float32), description='Wind direction at 100 m. The angle where the wind is coming from. Degrees. 0° is North; 90° is East. This is circular, with the same handling requirement as `wind_direction_10m`.', ge=0, le=360) class-attribute instance-attribute
pressure_surface = pt.Field(dtype=(pl.Float32), description='Surface pressure. Unit: Pa.', ge=0, le=200000) class-attribute instance-attribute
pressure_reduced_to_mean_sea_level = pt.Field(dtype=(pl.Float32), description='Mean sea-level pressure. Unit: Pa.', ge=0, le=200000) class-attribute instance-attribute
geopotential_height_500hpa = pt.Field(dtype=(pl.Float32), description='Geopotential height of the 500 hPa pressure surface. Unit: m.', ge=0, le=10000) class-attribute instance-attribute
downward_long_wave_radiation_flux_surface = pt.Field(dtype=(pl.Float32), description='Downward long-wave radiation flux at surface. Note that this variable is all-null for lead time 0. Unit: W m-2.', ge=0, le=1500) class-attribute instance-attribute
downward_short_wave_radiation_flux_surface = pt.Field(dtype=(pl.Float32), description='Downward short-wave (solar) radiation flux at surface. Note that this variable is all-null for lead time 0. Unit: W m-2.', ge=0, le=1500) class-attribute instance-attribute
precipitation_surface = pt.Field(dtype=(pl.Float32), description='Total precipitation rate at surface, de-accumulated by Dynamical. Note that this variable is all-null for lead time 0. Unit: kg m-2 s-1.', ge=0, le=0.01) class-attribute instance-attribute
categorical_precipitation_type_surface = pt.Field(dtype=(pl.Int16), ge=0, le=255, description="This field is always NaN for init_times on and before 2024-11-12, and populated from the 2024-11-13 00Z run onwards (confirmed by direct inspection of the source data). Derived from ECMWF's `ptype` field. See https://codes.ecmwf.int/grib/param-db/260015 0=No precipitation; 1=Rain; 2=Thunderstorm; 3=Freezing rain; 4=Mixed/ice; 5=Snow; 6=Wet snow; 7=Mixture of rain and snow; 8=Ice pellets; 9=Graupel; 10=Hail; 11=Drizzle; 12=Freezing drizzle; 13=Hail (less than 5 mm); 14=Hail (greater than or equal to 5 mm); 15-191=Reserved; 192-254=Reserved for local use; 255=Missing") class-attribute instance-attribute
categorical_var_names = frozenset({'categorical_precipitation_type_surface'}) class-attribute
deaccumulated_var_names = frozenset({'precipitation_surface', 'downward_short_wave_radiation_flux_surface', 'downward_long_wave_radiation_flux_surface'}) class-attribute

The variables Dynamical.org de-accumulates from ECMWF's cumulative source fields to rates.

All three are legitimately null at lead-0, and they share a de-accumulation step whose known upstream corruption leaves further nulls beyond it. Those are tolerated at ingest and reported by :func:assess_nwp_quality; only a variable null in every slice beyond lead-0 is fatal. Which corruption patterns arrive, what the H3 aggregation absorbs before they reach a validated frame, and why the survivors are tolerated: https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/#nulls-in-the-de-accumulated-variables-tolerated.

Methods:
all_weather_var_names() classmethod

All meteorological variable field names (continuous + categorical).

Source code in packages/contracts/src/contracts/weather_schemas.py 
336
337
338
339
@classmethod
def all_weather_var_names(cls) -> frozenset[str]:
    """All meteorological variable field names (continuous + categorical)."""
    return frozenset(cls.model_fields) - cls._non_var_column_names
continuous_var_names() classmethod

Meteorological variable field names suitable for linear interpolation.

Source code in packages/contracts/src/contracts/weather_schemas.py 
341
342
343
344
@classmethod
def continuous_var_names(cls) -> frozenset[str]:
    """Meteorological variable field names suitable for linear interpolation."""
    return cls.all_weather_var_names() - cls.categorical_var_names
validate(dataframe, columns=None, allow_missing_columns=False, allow_superfluous_columns=False, drop_superfluous_columns=False) classmethod

Validate the frame.

Bounds init_time/valid_time to the plausible datetime range, rejects a de-accumulated variable that is wholly missing (smaller null patterns are tolerated), and enforces both uniqueness and the ptype-introduction invariant.

Source code in packages/contracts/src/contracts/weather_schemas.py 
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
@classmethod
def validate(
    cls,
    dataframe: pl.DataFrame,
    columns: Sequence[str] | None = None,
    allow_missing_columns: bool = False,
    allow_superfluous_columns: bool = False,
    drop_superfluous_columns: bool = False,
) -> pt.DataFrame[Self]:  # ty:ignore[invalid-method-override]
    """Validate the frame.

    Bounds `init_time`/`valid_time` to the plausible datetime range, rejects a de-accumulated
    variable that is wholly missing (smaller null patterns are tolerated), and enforces both
    uniqueness and the ptype-introduction invariant.
    """
    validated_df = super().validate(
        dataframe=dataframe,
        columns=columns,
        allow_missing_columns=allow_missing_columns,
        allow_superfluous_columns=allow_superfluous_columns,
        drop_superfluous_columns=drop_superfluous_columns,
    )

    # Patito ignores `ge`/`le` on datetime fields, so the range check lives here.
    check_datetime_bounds(validated_df, "init_time", "valid_time")
    cls._check_no_wholly_missing_deaccumulated_variable(validated_df)
    cls._check_unique(validated_df)
    cls._check_variables_that_were_introduced_after_start_of_dataset(validated_df)
    return validated_df
scan_delta(path=None, storage_options=None) classmethod

Lazily scan the NWP Delta table, typed and cast to this contract's dtypes.

The table stores physical-unit Float32 directly (see delta_store.nwp.write_nwp), so no rescale step is needed.

Parameters:

Name Type Description Default
path str | Path | None

Path or URI of the nwp Delta table; defaults to get_settings().nwp_data_path (resolved lazily so importing this module needs no .env).

 None
storage_options ObjectStoreOptions | None

delta-rs object-store options (credentials/endpoint) for a remote path; defaults to get_settings().storage_options (empty for a local path).

 None
Source code in packages/contracts/src/contracts/weather_schemas.py 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
@classmethod
def scan_delta(
    cls,
    path: str | Path | None = None,
    storage_options: ObjectStoreOptions | None = None,
) -> pt.LazyFrame[Self]:
    """Lazily scan the NWP Delta table, typed and cast to this contract's dtypes.

    The table stores physical-unit `Float32` directly (see `delta_store.nwp.write_nwp`),
    so no rescale step is needed.

    Args:
        path: Path or URI of the ``nwp`` Delta table; defaults to
            ``get_settings().nwp_data_path`` (resolved lazily so importing this module needs
            no ``.env``).
        storage_options: delta-rs object-store options (credentials/endpoint) for a remote
            ``path``; defaults to ``get_settings().storage_options`` (empty for a local
            ``path``).
    """
    settings = get_settings()
    if path is None:
        path = settings.nwp_data_path
    if storage_options is None:
        storage_options = settings.storage_options
    return (
        pt.LazyFrame.from_existing(
            pl.scan_delta(path, storage_options=typeddict_to_dict(storage_options))
        )
        .set_model(cls)
        .cast()
    )

NwpQualityReport dataclass

Non-fatal data-quality summary for one NWP run.

Carries the tolerated nulls that the de-accumulated variables still hold after the H3 spatial aggregation, beyond lead-0. Usually that means whole (ensemble_member, valid_time) slices that arrived empty, plus the cells where upstream scatter happened to take out every contributing grid point; the two are counted separately because they warrant different responses, but neither fails the run. Only a variable that is null in every slice is fatal, and :meth:Nwp.validate rejects that before this runs.

Read this as "how much did we lose", not as "how corrupt was the feed". The aggregation absorbs most per-pixel upstream corruption before it reaches a cell, so this is a poor proxy for the upstream null rate. That rate is measured where it lives, on the raw grid, by :class:dynamical_data.ecmwf_ens.upstream_nulls.UpstreamNullRate; the ecmwf_ens asset publishes both on one check, and they are not comparable as rates.

Source code in packages/contracts/src/contracts/weather_schemas.py 
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
@dataclass(frozen=True)
class NwpQualityReport:
    """Non-fatal data-quality summary for one NWP run.

    Carries the *tolerated* nulls that the de-accumulated variables still hold after the H3 spatial
    aggregation, beyond lead-0. Usually that means whole (ensemble_member, valid_time) slices that
    arrived empty, plus the cells where upstream scatter happened to take out every contributing
    grid point; the two are counted separately because they warrant different responses, but
    neither fails the run. Only a variable that is null in *every* slice is fatal, and
    :meth:`Nwp.validate` rejects that before this runs.

    Read this as "how much did we lose", not as "how corrupt was the feed". The aggregation absorbs
    most per-pixel upstream corruption before it reaches a cell, so this is a poor proxy for the
    upstream null rate. That rate is measured where it lives, on the raw grid, by
    :class:`dynamical_data.ecmwf_ens.upstream_nulls.UpstreamNullRate`; the ``ecmwf_ens`` asset
    publishes both on one check, and they are not comparable as rates.
    """

    affected: pl.DataFrame
    """One row per affected (variable, init_time, ensemble_member, valid_time) slice, with
    ``n_null`` and ``n_total`` cell counts. Empty when the run is clean."""

    @property
    def n_null_cells(self) -> int:
        """Total null cells across all affected slices."""
        return int(self.affected["n_null"].sum()) if self.affected.height else 0

    @property
    def n_affected_slices(self) -> int:
        """Number of (variable, member, valid_time) slices carrying at least one null."""
        return self.affected.height

    @property
    def n_whole_null_slices(self) -> int:
        """Affected slices that arrived *entirely* null.

        The field is missing altogether for that one (variable, member, valid_time).

        Worth watching separately from the scattered case below: a rising count is the shape a
        partial upstream publication takes, and it is the number this report is best at, because a
        wholly-null slice reaches the cells intact however they are aggregated.
        """
        if not self.affected.height:
            return 0
        return int((self.affected["n_null"] == self.affected["n_total"]).sum())

    @property
    def n_scattered_slices(self) -> int:
        """Affected slices carrying only *some* null cells.

        Expect this to be small. Scattered upstream corruption is mostly absorbed by the H3
        aggregation, so a slice reaches this count only where the scatter took out every grid
        point of some cell. It is *not* a measure of the upstream per-pixel null rate.
        """
        return self.n_affected_slices - self.n_whole_null_slices

    @property
    def is_healthy(self) -> bool:
        """True when the run has no unexpected nulls."""
        return self.affected.height == 0

    @property
    def affected_variables(self) -> tuple[str, ...]:
        """The de-accumulated variables carrying unexpected nulls, sorted."""
        if not self.affected.height:
            return ()
        return tuple(sorted(self.affected["variable"].unique().to_list()))
Attributes
affected instance-attribute

One row per affected (variable, init_time, ensemble_member, valid_time) slice, with n_null and n_total cell counts. Empty when the run is clean.

n_null_cells property

Total null cells across all affected slices.

n_affected_slices property

Number of (variable, member, valid_time) slices carrying at least one null.

n_whole_null_slices property

Affected slices that arrived entirely null.

The field is missing altogether for that one (variable, member, valid_time).

Worth watching separately from the scattered case below: a rising count is the shape a partial upstream publication takes, and it is the number this report is best at, because a wholly-null slice reaches the cells intact however they are aggregated.

n_scattered_slices property

Affected slices carrying only some null cells.

Expect this to be small. Scattered upstream corruption is mostly absorbed by the H3 aggregation, so a slice reaches this count only where the scatter took out every grid point of some cell. It is not a measure of the upstream per-pixel null rate.

is_healthy property

True when the run has no unexpected nulls.

affected_variables property

The de-accumulated variables carrying unexpected nulls, sorted.

Methods:
__init__(affected)

NwpRunCompletenessReport dataclass

Run-level shape summary for one ingested NWP run.

Answers a single question: is the whole (member x step x cell) grid there?

Deliberately a report rather than an exception. A short run is the upstream provider misbehaving, not a contract violation, and the never-raise-on-absent-input rule says we land what arrived and warn, rather than throwing away an otherwise-good run. The ecmwf_ens asset wraps this into a WARN, non-blocking AssetCheckResult and publishes the counts as materialisation metadata.

Source code in packages/contracts/src/contracts/weather_schemas.py 
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
@dataclass(frozen=True)
class NwpRunCompletenessReport:
    """Run-level shape summary for one ingested NWP run.

    Answers a single question: is the whole (member x step x cell) grid there?

    Deliberately a *report* rather than an exception. A short run is the upstream provider
    misbehaving, not a contract violation, and the
    [never-raise-on-absent-input rule](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/#the-rules)
    says we land what arrived and warn, rather than throwing away an otherwise-good run. The
    `ecmwf_ens` asset wraps this into a WARN, non-blocking `AssetCheckResult` and publishes the
    counts as materialisation metadata.
    """

    init_times: tuple[datetime, ...]
    """Distinct `init_time` values in the frame. A whole run has exactly one."""

    n_rows: int
    """Rows in the frame."""

    expected_n_rows: int
    """`len(expected_ensemble_members) * len(expected_lead_time_hours) * expected_n_h3_cells` —
    the size of the complete grid."""

    n_ensemble_members: int
    """Distinct `ensemble_member` values observed."""

    n_valid_times: int
    """Distinct `valid_time` values observed."""

    n_h3_cells: int
    """Distinct `h3_index` values observed."""

    expected_n_h3_cells: int
    """Distinct `h3_index` values the H3 grid weights say this run should cover."""

    valid_time_min: datetime | None
    """Earliest `valid_time`, or `None` for an empty frame."""

    valid_time_max: datetime | None
    """Latest `valid_time`, or `None` for an empty frame."""

    missing_ensemble_members: tuple[int, ...]
    """Expected members with no rows at all, sorted."""

    unexpected_ensemble_members: tuple[int, ...]
    """Observed members outside the expected set, sorted — the upstream ensemble changed shape."""

    missing_lead_time_hours: tuple[int, ...]
    """Expected forecast steps (hours after `init_time`) with no rows at all, sorted."""

    unexpected_valid_times: tuple[datetime, ...]
    """Observed `valid_time`s that are not on any expected forecast step, sorted."""

    @property
    def is_complete(self) -> bool:
        """True when the run carries the shape we expect.

        That is, when the frame is one run whose member set, forecast-step set, H3 cell *count*
        and row count all match the expectation.

        Cells are compared by count, not by set: the report never receives the expected `h3_index`
        values, only how many there should be. Substituting one cell for another would therefore
        pass. That is not a live gap, because the asset derives the expected count from the very H3
        grid weights the converter joins against — see
        <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
        """
        return (
            len(self.init_times) == 1
            and not self.missing_ensemble_members
            and not self.unexpected_ensemble_members
            and not self.missing_lead_time_hours
            and not self.unexpected_valid_times
            and self.n_h3_cells == self.expected_n_h3_cells
            and self.n_rows == self.expected_n_rows
        )

    @property
    def h3_cell_shortfall(self) -> int:
        """Expected H3 cells minus observed.

        The difference is *net*, so it is negative if the run carries extra cells and zero if one
        cell was swapped for another.
        """
        return self.expected_n_h3_cells - self.n_h3_cells

    def describe(self) -> str:
        """One human-readable sentence naming every gap, for the asset check's description."""
        gaps = list(self._describe_gaps())
        if not gaps:
            return (
                f"Complete run: {self.n_ensemble_members} ensemble members x "
                f"{self.n_valid_times} forecast steps x {self.n_h3_cells} H3 cells "
                f"= {self.n_rows} rows."
            )
        return "Incomplete NWP run — " + "; ".join(gaps) + "."

    def _describe_gaps(self) -> Iterator[str]:
        """Yield one clause per detected gap, in decreasing order of diagnostic value."""
        if len(self.init_times) != 1:
            yield f"expected exactly one init_time, found {len(self.init_times)}"
        if self.missing_ensemble_members:
            yield f"missing ensemble member(s) {_abbreviate(self.missing_ensemble_members)}"
        if self.unexpected_ensemble_members:
            yield f"unexpected ensemble member(s) {_abbreviate(self.unexpected_ensemble_members)}"
        if self.missing_lead_time_hours:
            yield f"missing lead time(s) in hours {_abbreviate(self.missing_lead_time_hours)}"
        if self.unexpected_valid_times:
            yield f"unexpected valid_time(s) {_abbreviate(self.unexpected_valid_times)}"
        if self.n_h3_cells != self.expected_n_h3_cells:
            yield f"{self.n_h3_cells} H3 cells, expected {self.expected_n_h3_cells}"
        if self.n_rows != self.expected_n_rows:
            yield f"{self.n_rows} rows, expected {self.expected_n_rows}"
Attributes
init_times instance-attribute

Distinct init_time values in the frame. A whole run has exactly one.

n_rows instance-attribute

Rows in the frame.

expected_n_rows instance-attribute

len(expected_ensemble_members) * len(expected_lead_time_hours) * expected_n_h3_cells — the size of the complete grid.

n_ensemble_members instance-attribute

Distinct ensemble_member values observed.

n_valid_times instance-attribute

Distinct valid_time values observed.

n_h3_cells instance-attribute

Distinct h3_index values observed.

expected_n_h3_cells instance-attribute

Distinct h3_index values the H3 grid weights say this run should cover.

valid_time_min instance-attribute

Earliest valid_time, or None for an empty frame.

valid_time_max instance-attribute

Latest valid_time, or None for an empty frame.

missing_ensemble_members instance-attribute

Expected members with no rows at all, sorted.

unexpected_ensemble_members instance-attribute

Observed members outside the expected set, sorted — the upstream ensemble changed shape.

missing_lead_time_hours instance-attribute

Expected forecast steps (hours after init_time) with no rows at all, sorted.

unexpected_valid_times instance-attribute

Observed valid_times that are not on any expected forecast step, sorted.

is_complete property

True when the run carries the shape we expect.

That is, when the frame is one run whose member set, forecast-step set, H3 cell count and row count all match the expectation.

Cells are compared by count, not by set: the report never receives the expected h3_index values, only how many there should be. Substituting one cell for another would therefore pass. That is not a live gap, because the asset derives the expected count from the very H3 grid weights the converter joins against — see https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

h3_cell_shortfall property

Expected H3 cells minus observed.

The difference is net, so it is negative if the run carries extra cells and zero if one cell was swapped for another.

Methods:
__init__(init_times, n_rows, expected_n_rows, n_ensemble_members, n_valid_times, n_h3_cells, expected_n_h3_cells, valid_time_min, valid_time_max, missing_ensemble_members, unexpected_ensemble_members, missing_lead_time_hours, unexpected_valid_times)
describe()

One human-readable sentence naming every gap, for the asset check's description.

Source code in packages/contracts/src/contracts/weather_schemas.py 
712
713
714
715
716
717
718
719
720
721
def describe(self) -> str:
    """One human-readable sentence naming every gap, for the asset check's description."""
    gaps = list(self._describe_gaps())
    if not gaps:
        return (
            f"Complete run: {self.n_ensemble_members} ensemble members x "
            f"{self.n_valid_times} forecast steps x {self.n_h3_cells} H3 cells "
            f"= {self.n_rows} rows."
        )
    return "Incomplete NWP run — " + "; ".join(gaps) + "."

Functions:

assess_nwp_quality(dataframe)

Summarise the tolerated-but-noteworthy nulls in a validated NWP run.

Reports the nulls in the de-accumulated variables (precipitation/radiation) beyond lead-0 that :meth:Nwp.validate deliberately tolerates: whole (ensemble_member, valid_time) slices that arrived empty, and the cells where the upstream per-pixel corruption survived the H3 aggregation by taking out every grid point of a cell. Pure and Dagster-free (unit-testable in isolation); the ecmwf_ens asset wraps the result into a WARN AssetCheckResult. See https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/.

Source code in packages/contracts/src/contracts/weather_schemas.py 
612
613
614
615
616
617
618
619
620
621
622
623
def assess_nwp_quality(dataframe: pt.DataFrame[Nwp]) -> NwpQualityReport:
    """Summarise the tolerated-but-noteworthy nulls in a *validated* NWP run.

    Reports the nulls in the de-accumulated variables (precipitation/radiation) beyond lead-0 that
    :meth:`Nwp.validate` deliberately tolerates: whole (ensemble_member, valid_time) slices that
    arrived empty, and the cells where the upstream per-pixel corruption survived the H3
    aggregation by taking out every grid point of a cell. Pure and Dagster-free
    (unit-testable in isolation); the ``ecmwf_ens`` asset wraps the result into a WARN
    ``AssetCheckResult``. See
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/ecmwf-ens-known-issues/>.
    """
    return NwpQualityReport(affected=_deaccumulated_null_breakdown(dataframe))

assess_nwp_run_completeness(dataframe, expected_n_h3_cells, expected_ensemble_members=ECMWF_ENS_ENSEMBLE_MEMBERS, expected_lead_time_hours=ECMWF_ENS_LEAD_TIME_HOURS)

Summarise whether one ingested NWP run covers its full (member x step x cell) grid.

Called from the ecmwf_ens asset, deliberately not from Nwp.validate: validation runs on arbitrary frames (filtered test fixtures, pruned scans, single-member training reads), whereas completeness is a property of one whole ingested run and would be false on every one of those. Pure and Dagster-free, so it is unit-testable in isolation.

Never raises on a validated Nwp frame — not on an empty one, not on a multi-init_time one, not on one whose valid_times are off-grid — so it cannot turn the warning path into a failure path (rule 7 of Inherent Stability). The qualifier matters: the key columns being non-null and present is exactly what Nwp.validate guarantees, and the sole production caller validates before calling.

Row counting is safe here despite Polars' 32-bit row index: this runs on a single in-memory run (at V1 scale 1671 cells x 51 members x 85 steps ~ 7.24M rows), four orders of magnitude below the 2**32 ceiling that the whole ~5.9B-row NWP Delta table would hit.

Parameters:

Name Type Description Default
dataframe DataFrame[Nwp]

One whole ingested NWP run, already through Nwp.validate.

 required
expected_n_h3_cells int

Distinct H3 cells the run should cover — pass h3_grid["h3_index"].n_unique() so the expectation tracks the H3 grid weights rather than a hard-coded number.

 required
expected_ensemble_members frozenset[int]

Members the run should carry. Defaults to the ECMWF ENS ensemble, the only NwpModelId we ingest today.

 ECMWF_ENS_ENSEMBLE_MEMBERS
expected_lead_time_hours tuple[int, ...]

Forecast steps, in hours after init_time, the run should carry. Defaults to the ECMWF ENS native step structure.

 ECMWF_ENS_LEAD_TIME_HOURS
Source code in packages/contracts/src/contracts/weather_schemas.py 
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
def assess_nwp_run_completeness(
    dataframe: pt.DataFrame[Nwp],
    expected_n_h3_cells: int,
    expected_ensemble_members: frozenset[int] = ECMWF_ENS_ENSEMBLE_MEMBERS,
    expected_lead_time_hours: tuple[int, ...] = ECMWF_ENS_LEAD_TIME_HOURS,
) -> NwpRunCompletenessReport:
    """Summarise whether one ingested NWP run covers its full (member x step x cell) grid.

    Called from the `ecmwf_ens` asset, deliberately **not** from `Nwp.validate`: validation runs on
    arbitrary frames (filtered test fixtures, pruned scans, single-member training reads), whereas
    completeness is a property of one whole ingested run and would be false on every one of those.
    Pure and Dagster-free, so it is unit-testable in isolation.

    Never raises on a validated `Nwp` frame — not on an empty one, not on a multi-`init_time` one,
    not on one whose `valid_time`s are off-grid — so it cannot turn the warning path into a failure
    path (rule 7 of
    [Inherent Stability](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/inherent-stability/#the-rules)).
    The qualifier matters: the key columns being non-null and present is exactly what `Nwp.validate`
    guarantees, and the sole production caller validates before calling.

    Row counting is safe here despite Polars' 32-bit row index: this runs on a single in-memory run
    (at V1 scale 1671 cells x 51 members x 85 steps ~ 7.24M rows), four orders of magnitude below
    the 2**32 ceiling that the whole ~5.9B-row NWP Delta table would hit.

    Args:
        dataframe: One whole ingested NWP run, already through `Nwp.validate`.
        expected_n_h3_cells: Distinct H3 cells the run should cover — pass
            `h3_grid["h3_index"].n_unique()` so the expectation tracks the H3 grid weights rather
            than a hard-coded number.
        expected_ensemble_members: Members the run should carry. Defaults to the ECMWF ENS
            ensemble, the only `NwpModelId` we ingest today.
        expected_lead_time_hours: Forecast steps, in hours after `init_time`, the run should carry.
            Defaults to the ECMWF ENS native step structure.
    """
    init_times = tuple(sorted(dataframe["init_time"].unique().to_list()))
    observed_members = set(dataframe["ensemble_member"].unique().to_list())
    observed_valid_times = set(dataframe["valid_time"].unique().to_list())
    missing_lead_times, unexpected_valid_times = _lead_time_gaps(
        init_times, observed_valid_times, expected_lead_time_hours
    )
    return NwpRunCompletenessReport(
        init_times=init_times,
        n_rows=dataframe.height,
        expected_n_rows=(
            len(expected_ensemble_members) * len(expected_lead_time_hours) * expected_n_h3_cells
        ),
        n_ensemble_members=len(observed_members),
        n_valid_times=len(observed_valid_times),
        n_h3_cells=dataframe["h3_index"].n_unique(),
        expected_n_h3_cells=expected_n_h3_cells,
        valid_time_min=min(observed_valid_times, default=None),
        valid_time_max=max(observed_valid_times, default=None),
        missing_ensemble_members=tuple(sorted(expected_ensemble_members - observed_members)),
        unexpected_ensemble_members=tuple(sorted(observed_members - expected_ensemble_members)),
        missing_lead_time_hours=missing_lead_times,
        unexpected_valid_times=unexpected_valid_times,
    )

contracts.geo_schemas

Spatial contracts: the H3-cell-to-NWP-grid weights that drive the spatial NWP aggregation.

Classes

H3GridWeights

Bases: Model

Schema for the pre-computed H3 grid weights.

This contract defines the mapping between H3 hexagons and a regular latitude/longitude grid. It is used to ensure type safety when passing spatial mapping data from generic geospatial utilities (like packages/geo) to dataset-specific ingestion pipelines (like packages/dynamical_data).

Source code in packages/contracts/src/contracts/geo_schemas.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class H3GridWeights(pt.Model):
    """Schema for the pre-computed H3 grid weights.

    This contract defines the mapping between H3 hexagons and a regular latitude/longitude grid. It
    is used to ensure type safety when passing spatial mapping data from generic geospatial
    utilities (like `packages/geo`) to dataset-specific ingestion pipelines (like
    `packages/dynamical_data`).
    """

    h3_index: int = pt.Field(dtype=pl.UInt64, description="H3 cell index.")

    nwp_lat: float = pt.Field(
        dtype=pl.Float32,
        ge=-90,
        le=90,
        description="Latitude of the NWP grid box centre (decimal degrees).",
    )

    nwp_lon: float = pt.Field(
        dtype=pl.Float32,
        ge=-180,
        le=180,
        description="Longitude of the NWP grid box centre (decimal degrees).",
    )

    proportion: float = pt.Field(
        dtype=pl.Float32,
        ge=0,
        le=1,
        description="Fraction of the H3 hexagon that overlaps this NWP grid box (0 to 1).",
    )
Attributes
h3_index = pt.Field(dtype=(pl.UInt64), description='H3 cell index.') class-attribute instance-attribute
nwp_lat = pt.Field(dtype=(pl.Float32), ge=(-90), le=90, description='Latitude of the NWP grid box centre (decimal degrees).') class-attribute instance-attribute
nwp_lon = pt.Field(dtype=(pl.Float32), ge=(-180), le=180, description='Longitude of the NWP grid box centre (decimal degrees).') class-attribute instance-attribute
proportion = pt.Field(dtype=(pl.Float32), ge=0, le=1, description='Fraction of the H3 hexagon that overlaps this NWP grid box (0 to 1).') class-attribute instance-attribute

contracts.ml_schemas

Contracts for the ML pipeline.

The feature vocabulary, the joined AllFeatures frame handed to models, the eligible-time-series population, and the metrics schema.

Attributes

TimeFeature = Literal['local_time_of_day_sin', 'local_time_of_day_cos', 'local_time_of_year_sin', 'local_time_of_year_cos', 'local_day_of_week_sin', 'local_day_of_week_cos', 'local_day_of_week', 'local_utc_offset_minutes'] module-attribute

SafeInputBaseColumn = Literal['time_series_id', 'time_series_type', 'nwp_lead_time_hours', 'ensemble_member', 'power_fcst_init_time', 'nwp_init_time'] module-attribute

HORIZON_SLICES = ('all', 'intraday', 'day_ahead', 'short_medium_range', 'extended_range') module-attribute

Horizon slice labels matching the four forecast ranges from the project report.

Bands are left-closed intervals of lead time (valid_time − power_fcst_init_time), as implemented by ml_core.metrics:

  • "all": aggregates over all horizons and is always computed
  • "intraday": [0 h, 6 h)
  • "day_ahead": [6 h, 36 h)
  • "short_medium_range": [36 h, 168 h) — day 2 to day 7
  • "extended_range": [168 h, ∞) — day 8 onwards (unbounded above; in practice forecasts stop at the 14-day NWP horizon)

METRIC_NAMES = ('mae', 'nmae', 'rmse', 'mbe', 'crps', 'spread_skill_ratio', 'pinball_loss', 'mean_pinball_loss', 'picp', 'interval_width') module-attribute

Metric names currently implemented.

Full definitions, equations, and design rationale: https://openclimatefix.github.io/nged-substation-forecast/techniques/evaluation-metrics/

Deterministic (scored on the per-run ensemble mean):

  • "mae": mean absolute error (MW)
  • "nmae": normalised MAE (dimensionless; normalised by full-history effective capacity)
  • "rmse": root mean squared error (MW)
  • "mbe": mean bias error (MW; positive = over-prediction)

Probabilistic (scored on the ensemble members before the mean collapse):

  • "crps": fair (finite-ensemble-unbiased) continuous ranked probability score (MW). The only metric here that is comparable across models with different ensemble sizes.
  • "spread_skill_ratio": Fortin-corrected RMS ensemble spread ÷ RMSE of the ensemble mean (dimensionless; 1.0 = well-calibrated, < 1 = underdispersed/overconfident).
  • "pinball_loss": quantile loss (MW) at the quantile named by metric_param.
  • "mean_pinball_loss": unweighted mean of "pinball_loss" over the thirteen DELIVERY_QUANTILES (MW). Tail-heavy by construction, matching NGED's priorities.
  • "picp": prediction-interval coverage probability of the band named by metric_param (dimensionless fraction). The calibrated reference for empirical quantiles from a finite ensemble sits below the nominal coverage — e.g. ≈ 0.769, not 0.8, for p10_p90 at 51 members.
  • "interval_width": mean width (MW) of the band named by metric_param — the sharpness companion to "picp" (coverage is only impressive if the band is also narrow).

QUANTILE_METRIC_PARAMS = tuple((quantile_label(q)) for q in DELIVERY_QUANTILES) module-attribute

metric_param labels for quantile-indexed metrics (pinball loss), e.g. "p1" … "p99".

Derived from DELIVERY_QUANTILES so that evaluation is always scored at exactly the quantile levels the product delivers.

BAND_METRIC_PARAMS = tuple(f'{quantile_label(q)}_{quantile_label(1 - q)}' for q in DELIVERY_QUANTILES if q < 0.5) module-attribute

metric_param labels for the symmetric prediction-interval bands (PICP, interval width).

One band per symmetric pair of DELIVERY_QUANTILES: "p1_p99", "p2_p98", "p5_p95", "p10_p90", "p20_p80", "p35_p65". Together they trace a coverage curve from the 30% band to the 98% band.

METRIC_PARAMS = ('all', *QUANTILE_METRIC_PARAMS, *BAND_METRIC_PARAMS) module-attribute

Parameter values for parametric metrics.

  • "all": all scalar metrics with no extra parameter dimension (MAE, NMAE, RMSE, MBE, CRPS, spread-skill ratio, mean pinball loss).
  • Quantile labels ("p1" … "p99", from QUANTILE_METRIC_PARAMS): pinball loss.
  • Band labels ("p1_p99" … "p35_p65", from BAND_METRIC_PARAMS): PICP and interval width.

EVALUATION_SCOPES = ('leaderboard', 'production_monitoring', 'ad_hoc') module-attribute

Evaluation scopes that coexist in the forecast_metrics table.

  • "leaderboard": CV-fold leaderboard metrics
  • "production_monitoring": live trailing-window monitoring
  • "ad_hoc": one-off analyses (no MLflow run)

EvalScopeType = Literal['leaderboard', 'ad_hoc'] module-attribute

Type annotation for the evaluation scopes currently accepted by the metrics asset config.

Distinct from EVALUATION_SCOPES (the runtime tuple driving the Polars Enum): that tuple includes "production_monitoring" for future use; this Literal reflects only the scopes the asset actually handles today. Expand to match EVALUATION_SCOPES when Phase 8 lands.

TIME_SERIES_TYPE_SLICES = ('all', *LIST_OF_TIME_SERIES_TYPES) module-attribute

Values for the time_series_type metric slice.

Every time_series_type plus the sentinel "all" for the across-everything aggregate.

Classes

AllFeatures

Bases: Model

Final joined dataset ready for the ML model.

Weather features are kept in their physical units (e.g., degrees Celsius, m/s) to ensure precision during interpolation and feature engineering.

DYNAMIC FEATURES: In addition to the explicitly defined columns below, the pipeline supports dynamically generated features. You can request these in your model config:

  • power_lag_{hours}h: The power value shifted by X hours (e.g., power_lag_24h).
  • temperature_2m_rolling_mean_{hours}h: Rolling average of temperature over X hours (e.g., temperature_2m_rolling_mean_6h).

Note: Dynamic features are not explicitly typed as Patito fields below. This is intentional to allow infinite parameterization (e.g., any lag hour) without the overhead of metaprogramming or defining hundreds of static fields. The pipeline dynamically asserts their presence during feature engineering.

Source code in packages/contracts/src/contracts/ml_schemas.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class AllFeatures(pt.Model):
    """Final joined dataset ready for the ML model.

    Weather features are kept in their physical units (e.g., degrees Celsius, m/s)
    to ensure precision during interpolation and feature engineering.

    DYNAMIC FEATURES:
    In addition to the explicitly defined columns below, the pipeline supports
    dynamically generated features. You can request these in your model config:

    * `power_lag_{hours}h`: The power value shifted by X hours (e.g., `power_lag_24h`).
    * `temperature_2m_rolling_mean_{hours}h`: Rolling average of temperature over X hours (e.g.,
      `temperature_2m_rolling_mean_6h`).

    Note: Dynamic features are not explicitly typed as Patito fields below.
    This is intentional to allow infinite parameterization (e.g., any lag hour)
    without the overhead of metaprogramming or defining hundreds of static fields.
    The pipeline dynamically asserts their presence during feature engineering.
    """

    valid_time: datetime = pt.Field(dtype=UTC_DATETIME_DTYPE)
    time_series_id: int = _get_time_series_id_dtype()
    time_series_type: str = pt.Field(
        dtype=pl.Enum(LIST_OF_TIME_SERIES_TYPES),
        allow_missing=True,
        description=(
            "The substation's category. Only emitted when a feature set requests it, and never "
            "null: the feature pipeline drops any time series with no row in the metadata."
        ),
    )
    ensemble_member: int | None = pt.Field(dtype=pl.UInt8, allow_missing=True)

    power_fcst_init_time: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        description=(
            "When OCF's power forecast model was initialised. This might be called `t0` in other "
            "OCF projects."
        ),
    )

    nwp_init_time: datetime | None = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description="When the NWP run was initialised.",
    )

    power: float | None = pt.Field(
        dtype=pl.Float32,
        description=(
            "The observed power, or null where there is no observation. Live inference feeds a "
            "dense spine that runs past the last observation, and training drops the null rows."
        ),
    )
    nwp_lead_time_hours: float | None = pt.Field(dtype=pl.Float32, allow_missing=True)

    # Weather features
    temperature_2m: float | None = _FEATURE_DTYPE
    dew_point_temperature_2m: float | None = _FEATURE_DTYPE
    wind_speed_10m: float | None = _FEATURE_DTYPE
    wind_direction_10m: float | None = _FEATURE_DTYPE
    wind_speed_100m: float | None = _FEATURE_DTYPE
    wind_direction_100m: float | None = _FEATURE_DTYPE
    pressure_surface: float | None = _FEATURE_DTYPE
    pressure_reduced_to_mean_sea_level: float | None = _FEATURE_DTYPE
    geopotential_height_500hpa: float | None = _FEATURE_DTYPE
    downward_long_wave_radiation_flux_surface: float | None = _FEATURE_DTYPE
    downward_short_wave_radiation_flux_surface: float | None = _FEATURE_DTYPE
    precipitation_surface: float | None = _FEATURE_DTYPE
    categorical_precipitation_type_surface: int | None = _FEATURE_DTYPE

    # Derived weather features
    windchill: float | None = _FEATURE_DTYPE

    # Temporal features. `local` means "in the local timezone", e.g. "Europe/London". We use `local`
    # as the main input feature, because it's the local time that mostly drives demand.
    local_utc_offset_minutes: int | None = pt.Field(
        dtype=pl.Int16,
        allow_missing=True,
        description=(
            "Offset of the local time zone from UTC, in minutes (0 or 60 for GB). Minutes keep "
            "sub-hour zones distinct: India's +5:30 is 330 and Nepal's +5:45 is 345. Int16 spans "
            "the real extremes, -720 (Etc/GMT+12) to +840 (Pacific/Kiritimati)."
        ),
    )
    local_time_of_day_sin: float | None = _FEATURE_DTYPE
    local_time_of_day_cos: float | None = _FEATURE_DTYPE
    local_time_of_year_sin: float | None = _FEATURE_DTYPE
    local_time_of_year_cos: float | None = _FEATURE_DTYPE
    local_day_of_week_sin: float | None = _FEATURE_DTYPE
    local_day_of_week_cos: float | None = _FEATURE_DTYPE
    local_day_of_week: str | None = pt.Field(
        dtype=pl.Enum(
            ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
        )
    )
Attributes
valid_time = pt.Field(dtype=UTC_DATETIME_DTYPE) class-attribute instance-attribute
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
time_series_type = pt.Field(dtype=(pl.Enum(LIST_OF_TIME_SERIES_TYPES)), allow_missing=True, description="The substation's category. Only emitted when a feature set requests it, and never null: the feature pipeline drops any time series with no row in the metadata.") class-attribute instance-attribute
ensemble_member = pt.Field(dtype=(pl.UInt8), allow_missing=True) class-attribute instance-attribute
power_fcst_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, description="When OCF's power forecast model was initialised. This might be called `t0` in other OCF projects.") class-attribute instance-attribute
nwp_init_time = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='When the NWP run was initialised.') class-attribute instance-attribute
power = pt.Field(dtype=(pl.Float32), description='The observed power, or null where there is no observation. Live inference feeds a dense spine that runs past the last observation, and training drops the null rows.') class-attribute instance-attribute
nwp_lead_time_hours = pt.Field(dtype=(pl.Float32), allow_missing=True) class-attribute instance-attribute
temperature_2m = _FEATURE_DTYPE class-attribute instance-attribute
dew_point_temperature_2m = _FEATURE_DTYPE class-attribute instance-attribute
wind_speed_10m = _FEATURE_DTYPE class-attribute instance-attribute
wind_direction_10m = _FEATURE_DTYPE class-attribute instance-attribute
wind_speed_100m = _FEATURE_DTYPE class-attribute instance-attribute
wind_direction_100m = _FEATURE_DTYPE class-attribute instance-attribute
pressure_surface = _FEATURE_DTYPE class-attribute instance-attribute
pressure_reduced_to_mean_sea_level = _FEATURE_DTYPE class-attribute instance-attribute
geopotential_height_500hpa = _FEATURE_DTYPE class-attribute instance-attribute
downward_long_wave_radiation_flux_surface = _FEATURE_DTYPE class-attribute instance-attribute
downward_short_wave_radiation_flux_surface = _FEATURE_DTYPE class-attribute instance-attribute
precipitation_surface = _FEATURE_DTYPE class-attribute instance-attribute
categorical_precipitation_type_surface = _FEATURE_DTYPE class-attribute instance-attribute
windchill = _FEATURE_DTYPE class-attribute instance-attribute
local_utc_offset_minutes = pt.Field(dtype=(pl.Int16), allow_missing=True, description="Offset of the local time zone from UTC, in minutes (0 or 60 for GB). Minutes keep sub-hour zones distinct: India's +5:30 is 330 and Nepal's +5:45 is 345. Int16 spans the real extremes, -720 (Etc/GMT+12) to +840 (Pacific/Kiritimati).") class-attribute instance-attribute
local_time_of_day_sin = _FEATURE_DTYPE class-attribute instance-attribute
local_time_of_day_cos = _FEATURE_DTYPE class-attribute instance-attribute
local_time_of_year_sin = _FEATURE_DTYPE class-attribute instance-attribute
local_time_of_year_cos = _FEATURE_DTYPE class-attribute instance-attribute
local_day_of_week_sin = _FEATURE_DTYPE class-attribute instance-attribute
local_day_of_week_cos = _FEATURE_DTYPE class-attribute instance-attribute
local_day_of_week = pt.Field(dtype=(pl.Enum(['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']))) class-attribute instance-attribute

Metrics

Bases: Model

Evaluation metrics for power forecasts — tall format.

One row per (time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name, metric_param). metric_param encodes the extra parameter dimension for metrics that have one, or "all" for scalar metrics with no extra dimension. Examples:

time_series_id fold_id horizon_slice metric_name metric_param metric_value
1 1 all mae all 5.2
1 1 day_ahead rmse all 7.1
1 1 day_ahead pinball_loss p10 2.1
1 1 day_ahead pinball_loss p50 3.4
1 1 day_ahead mean_pinball_loss all 2.4
1 1 day_ahead picp p10_p90 0.78

Primary key: (time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name, metric_param).

Source code in packages/contracts/src/contracts/ml_schemas.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
class Metrics(pt.Model):
    """Evaluation metrics for power forecasts — tall format.

    One row per `(time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name,
    metric_param)`. `metric_param` encodes the extra parameter dimension for metrics that have
    one, or `"all"` for scalar metrics with no extra dimension. Examples:

    | time_series_id | fold_id | horizon_slice | metric_name       | metric_param | metric_value |
    |----------------|---------|---------------|-------------------|--------------|--------------|
    | 1              | 1       | all           | mae               | all          | 5.2          |
    | 1              | 1       | day_ahead     | rmse              | all          | 7.1          |
    | 1              | 1       | day_ahead     | pinball_loss      | p10          | 2.1          |
    | 1              | 1       | day_ahead     | pinball_loss      | p50          | 3.4          |
    | 1              | 1       | day_ahead     | mean_pinball_loss | all          | 2.4          |
    | 1              | 1       | day_ahead     | picp              | p10_p90      | 0.78         |

    Primary key: `(time_series_id, power_fcst_model_name, fold_id, horizon_slice, metric_name,
    metric_param)`.
    """

    time_series_id: int = _get_time_series_id_dtype()

    # String (not Categorical): fold_id/experiment_name are Delta partition columns and delta-rs
    # stores dictionary-encoded columns as String anyway; String keeps them cast-free and lets
    # predicate pushdown work. See the "Delta Lake dictionary-encoded columns" section of the
    # `polars-patito-gotchas` skill.
    power_fcst_model_name: str = pt.Field(
        dtype=pl.String,
        description="Identifier for the ML-based power forecasting model family.",
    )

    fold_id: str = pt.Field(
        dtype=pl.String,
        description=(
            "CV fold year (e.g. '2022'), or 'live' for production forecasts. Matches "
            "PowerForecast.fold_id."
        ),
    )

    horizon_slice: str = pt.Field(
        dtype=pl.Enum(HORIZON_SLICES),
        description=(
            "'all' aggregates over all forecast horizons.  Other values select the HORIZON_SLICES"
            " bands."
        ),
    )

    metric_name: str = pt.Field(
        dtype=pl.Enum(METRIC_NAMES),
        description="The name of the metric (e.g. 'mae', 'rmse').",
    )

    metric_param: str = pt.Field(
        dtype=pl.Enum(METRIC_PARAMS),
        description=(
            "Extra parameter dimension for parametric metrics.  "
            "'all' for scalar metrics (MAE, NMAE, RMSE, MBE, CRPS, spread-skill ratio, mean"
            " pinball loss).  For pinball loss: a DELIVERY_QUANTILES label ('p1' … 'p99').  "
            "For PICP and interval width: a symmetric band label ('p1_p99' … 'p35_p65')."
        ),
    )

    metric_value: float = pt.Field(dtype=pl.Float32, description="The computed metric value.")

    evaluation_scope: str = pt.Field(
        dtype=pl.Enum(EVALUATION_SCOPES),
        allow_missing=True,
        description=(
            "Which evaluation produced this row, so leaderboard, live-monitoring, and "
            "one-off metrics coexist in one table and stay separable."
        ),
    )
    """The columns from `evaluation_scope` and below are populated by the ``metrics`` Dagster asset.
    They are ``allow_missing`` so that the pure ``compute_metrics()`` helper can emit the core
    metric rows and have the asset enrich them with scope/window provenance before the frame is
    written to the ``forecast_metrics`` Delta table."""

    time_series_type: str = pt.Field(
        dtype=pl.Enum(TIME_SERIES_TYPE_SLICES),
        allow_missing=True,
        description=(
            "The time-series category for this row. Never null: `compute_metrics` raises if any "
            "scored series has no metadata row, because a null would drop that series out of "
            "every per-type aggregate while still counting towards the overall mean."
        ),
    )

    window_start: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description=(
            "Inclusive start of the valid_time window this row covers. For a CV fold this "
            "is the fold's val_start; for monitoring it is the trailing-window start."
        ),
    )

    window_end: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description="End of the valid_time window this row covers (fold val_end or window end).",
    )

    window_label: str = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        description="Human label for the window, e.g. '2025', '24h', '7d', 'full_fold'.",
    )

    computed_at: datetime = pt.Field(
        dtype=UTC_DATETIME_DTYPE,
        allow_missing=True,
        description=(
            "When this metric row was written (provenance; orders the append-only "
            "monitoring series and distinguishes recomputations)."
        ),
    )

    mlflow_run_id: str | None = pt.Field(
        dtype=pl.String,
        allow_missing=True,
        description=(
            "Convenience cross-link to the MLflow run this metric row belongs to. "
            "Null for 'ad_hoc' rows, which have no MLflow run."
        ),
    )

    experiment_name: str = pt.Field(
        dtype=pl.String,  # String (not Categorical) — Delta partition column, see fold_id above.
        allow_missing=True,
        description=(
            "Experiment that produced these forecasts. Delta partition key alongside fold_id. "
            "Populated by the metrics Dagster asset after compute_metrics() returns; "
            "allow_missing so compute_metrics() itself need not produce it."
        ),
    )
Attributes
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute
power_fcst_model_name = pt.Field(dtype=(pl.String), description='Identifier for the ML-based power forecasting model family.') class-attribute instance-attribute
fold_id = pt.Field(dtype=(pl.String), description="CV fold year (e.g. '2022'), or 'live' for production forecasts. Matches PowerForecast.fold_id.") class-attribute instance-attribute
horizon_slice = pt.Field(dtype=(pl.Enum(HORIZON_SLICES)), description="'all' aggregates over all forecast horizons. Other values select the HORIZON_SLICES bands.") class-attribute instance-attribute
metric_name = pt.Field(dtype=(pl.Enum(METRIC_NAMES)), description="The name of the metric (e.g. 'mae', 'rmse').") class-attribute instance-attribute
metric_param = pt.Field(dtype=(pl.Enum(METRIC_PARAMS)), description="Extra parameter dimension for parametric metrics. 'all' for scalar metrics (MAE, NMAE, RMSE, MBE, CRPS, spread-skill ratio, mean pinball loss). For pinball loss: a DELIVERY_QUANTILES label ('p1' … 'p99'). For PICP and interval width: a symmetric band label ('p1_p99' … 'p35_p65').") class-attribute instance-attribute
metric_value = pt.Field(dtype=(pl.Float32), description='The computed metric value.') class-attribute instance-attribute
evaluation_scope = pt.Field(dtype=(pl.Enum(EVALUATION_SCOPES)), allow_missing=True, description='Which evaluation produced this row, so leaderboard, live-monitoring, and one-off metrics coexist in one table and stay separable.') class-attribute instance-attribute

The columns from evaluation_scope and below are populated by the metrics Dagster asset. They are allow_missing so that the pure compute_metrics() helper can emit the core metric rows and have the asset enrich them with scope/window provenance before the frame is written to the forecast_metrics Delta table.

time_series_type = pt.Field(dtype=(pl.Enum(TIME_SERIES_TYPE_SLICES)), allow_missing=True, description='The time-series category for this row. Never null: `compute_metrics` raises if any scored series has no metadata row, because a null would drop that series out of every per-type aggregate while still counting towards the overall mean.') class-attribute instance-attribute
window_start = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description="Inclusive start of the valid_time window this row covers. For a CV fold this is the fold's val_start; for monitoring it is the trailing-window start.") class-attribute instance-attribute
window_end = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='End of the valid_time window this row covers (fold val_end or window end).') class-attribute instance-attribute
window_label = pt.Field(dtype=(pl.String), allow_missing=True, description="Human label for the window, e.g. '2025', '24h', '7d', 'full_fold'.") class-attribute instance-attribute
computed_at = pt.Field(dtype=UTC_DATETIME_DTYPE, allow_missing=True, description='When this metric row was written (provenance; orders the append-only monitoring series and distinguishes recomputations).') class-attribute instance-attribute
mlflow_run_id = pt.Field(dtype=(pl.String), allow_missing=True, description="Convenience cross-link to the MLflow run this metric row belongs to. Null for 'ad_hoc' rows, which have no MLflow run.") class-attribute instance-attribute
experiment_name = pt.Field(dtype=(pl.String), allow_missing=True, description='Experiment that produced these forecasts. Delta partition key alongside fold_id. Populated by the metrics Dagster asset after compute_metrics() returns; allow_missing so compute_metrics() itself need not produce it.') class-attribute instance-attribute

EligibleTimeSeries

Bases: Model

The canonical per-fold population of eligible time_series_ids.

Written by the eligible_time_series Dagster asset (one Delta partition per fold_id) and read by trained_cv_model and cv_power_forecasts. Eligibility is a function of data coverage and the fold dates only — never the model or experiment config — so every experiment trains and scores a fold on the identical population, which is what makes leaderboard comparisons apples-to-apples.

One row per eligible (fold_id, time_series_id).

Source code in packages/contracts/src/contracts/ml_schemas.py 
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
class EligibleTimeSeries(pt.Model):
    """The canonical per-fold population of eligible ``time_series_id``s.

    Written by the ``eligible_time_series`` Dagster asset (one Delta partition per ``fold_id``)
    and read by ``trained_cv_model`` and ``cv_power_forecasts``. Eligibility is a function of
    data coverage and the fold dates **only** — never the model or experiment config — so every
    experiment trains and scores a fold on the identical population, which is what makes
    leaderboard comparisons apples-to-apples.

    One row per eligible ``(fold_id, time_series_id)``.
    """

    fold_id: str = pt.Field(
        dtype=pl.String,
        description=(
            "The CV fold this eligibility row belongs to (e.g. 'mid_2025_to_mid_2026'); the Delta "
            "partition key."
        ),
    )
    time_series_id: int = _get_time_series_id_dtype()
Attributes
fold_id = pt.Field(dtype=(pl.String), description="The CV fold this eligibility row belongs to (e.g. 'mid_2025_to_mid_2026'); the Delta partition key.") class-attribute instance-attribute
time_series_id = _get_time_series_id_dtype() class-attribute instance-attribute

Functions:

contracts.config_schemas

Configuration schemas, and the class-path round-trip that _target_ strings ride on.

Attributes

Classes

CvFoldConfig

Bases: BaseModel

Configuration for a single expanding-window CV fold.

leaderboard distinguishes the epoch-pinned leaderboard folds (the apples-to-apples evaluation protocol) from optional non-leaderboard dev folds such as smoke_test: a leaderboard=False fold runs through the identical pipeline but never feeds the leaderboard.

min_training_months overrides CvConfig.min_training_months for this fold alone (None falls back to the config-level value). A short dev fold sets it to its train length so eligibility does not demand the leaderboard's longer history.

Source code in packages/contracts/src/contracts/config_schemas.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class CvFoldConfig(BaseModel):
    """Configuration for a single expanding-window CV fold.

    ``leaderboard`` distinguishes the epoch-pinned leaderboard folds (the apples-to-apples
    evaluation protocol) from optional non-leaderboard dev folds such as ``smoke_test``: a
    ``leaderboard=False`` fold runs through the identical pipeline but never feeds the leaderboard.

    ``min_training_months`` overrides ``CvConfig.min_training_months`` for this fold alone (``None``
    falls back to the config-level value). A short dev fold sets it to its train length so
    eligibility does not demand the leaderboard's longer history.
    """

    fold_id: FoldId
    train_start: date
    train_end: date
    val_start: date
    val_end: date
    leaderboard: bool = True
    min_training_months: int | None = Field(default=None, ge=1)
Attributes
fold_id instance-attribute
train_start instance-attribute
train_end instance-attribute
val_start instance-attribute
val_end instance-attribute
leaderboard = True class-attribute instance-attribute
min_training_months = Field(default=None, ge=1) class-attribute instance-attribute

CvConfig

Bases: BaseModel

Configuration for expanding-window cross-validation.

The folds list defines the evaluation protocol shared by all experiments on the leaderboard. All models must be evaluated against the same folds to ensure apples-to-apples comparison.

min_training_months controls which time series are eligible for each fold: a time series is only included if it has at least this many months of data before val_start (and data through val_end).

Source code in packages/contracts/src/contracts/config_schemas.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class CvConfig(BaseModel):
    """Configuration for expanding-window cross-validation.

    The folds list defines the evaluation protocol shared by all experiments on the
    leaderboard. All models must be evaluated against the same folds to ensure
    apples-to-apples comparison.

    min_training_months controls which time series are eligible for each fold: a time
    series is only included if it has at least this many months of data before val_start
    (and data through val_end).
    """

    folds: list[CvFoldConfig]
    min_training_months: int = Field(default=6, ge=1)

    @property
    def fold_ids(self) -> list[str]:
        """The fold ids in declaration order (e.g. ``["2022", "2023", ...]``).

        Used to build the ``cv_experiment_folds`` partitions and to expand experiment
        registration into per-fold partition keys — always read from config, never hard-coded.
        """
        return [fold.fold_id for fold in self.folds]

    @property
    def leaderboard_fold_ids(self) -> list[str]:
        """The fold ids of the leaderboard folds only, in declaration order.

        Non-leaderboard dev folds (e.g. ``smoke_test``) are excluded. Used to expand the
        ``full_cv`` / ``register_only`` run modes and to scope leaderboard metrics.
        """
        return [fold.fold_id for fold in self.folds if fold.leaderboard]

    def get_fold(self, fold_id: str) -> CvFoldConfig:
        """Return the fold with the given ``fold_id``.

        Args:
            fold_id: The fold identifier to look up (e.g. ``"2022"``).

        Raises:
            KeyError: If no fold with that id exists in the config.
        """
        for fold in self.folds:
            if fold.fold_id == fold_id:
                return fold
        raise KeyError(f"No fold with fold_id={fold_id!r}; available folds: {self.fold_ids}")
Attributes
folds instance-attribute
min_training_months = Field(default=6, ge=1) class-attribute instance-attribute
fold_ids property

The fold ids in declaration order (e.g. ["2022", "2023", ...]).

Used to build the cv_experiment_folds partitions and to expand experiment registration into per-fold partition keys — always read from config, never hard-coded.

leaderboard_fold_ids property

The fold ids of the leaderboard folds only, in declaration order.

Non-leaderboard dev folds (e.g. smoke_test) are excluded. Used to expand the full_cv / register_only run modes and to scope leaderboard metrics.

Methods:
get_fold(fold_id)

Return the fold with the given fold_id.

Parameters:

Name Type Description Default
fold_id str

The fold identifier to look up (e.g. "2022").

 required

Raises:

Type Description
KeyError

If no fold with that id exists in the config.
Source code in packages/contracts/src/contracts/config_schemas.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
def get_fold(self, fold_id: str) -> CvFoldConfig:
    """Return the fold with the given ``fold_id``.

    Args:
        fold_id: The fold identifier to look up (e.g. ``"2022"``).

    Raises:
        KeyError: If no fold with that id exists in the config.
    """
    for fold in self.folds:
        if fold.fold_id == fold_id:
            return fold
    raise KeyError(f"No fold with fold_id={fold_id!r}; available folds: {self.fold_ids}")

Functions:

class_target(obj)

Return the fully-qualified import path of a class (or of an instance's class).

The inverse of import_class. Together they are how a config file, an MLflow experiment tag and a saved model's meta.json all name a Python class: a module.ClassName string that survives being written to disk and read back in another process.

Parameters:

Name Type Description Default
obj type | object

The class to name, or an instance whose class should be named.

 required

Returns:

Type Description
str

The module.ClassName path, e.g. "xgboost_forecaster.forecaster.XGBoostForecaster".

Raises:

Type Description
ValueError

obj's class is not defined at module level — it is nested inside another class or inside a function. import_class resolves a single attribute lookup on a module, so such a class has no path it could resolve. Failing here, where the class is defined, beats emitting a target string that only breaks when something later tries to load it.
Source code in packages/contracts/src/contracts/config_schemas.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def class_target(obj: type | object) -> str:
    """Return the fully-qualified import path of a class (or of an instance's class).

    The inverse of ``import_class``. Together they are how a config file, an MLflow experiment tag
    and a saved model's ``meta.json`` all name a Python class: a ``module.ClassName`` string that
    survives being written to disk and read back in another process.

    Args:
        obj: The class to name, or an instance whose class should be named.

    Returns:
        The ``module.ClassName`` path, e.g. ``"xgboost_forecaster.forecaster.XGBoostForecaster"``.

    Raises:
        ValueError: ``obj``'s class is not defined at module level — it is nested inside another
            class or inside a function. ``import_class`` resolves a single attribute lookup on a
            module, so such a class has no path it could resolve. Failing here, where the class is
            defined, beats emitting a target string that only breaks when something later tries to
            load it.
    """
    cls = obj if isinstance(obj, type) else type(obj)
    if "." in cls.__qualname__:
        raise ValueError(
            f"{cls.__module__}.{cls.__qualname__} is not defined at module level (it is nested "
            "inside a class or a function), so it has no importable path. Move it to module level "
            "to use it as a _target_."
        )
    return f"{cls.__module__}.{cls.__qualname__}"

import_class(target)

Resolve a fully-qualified class path to the class object.

The inverse of class_target: imports the module and looks the class up on it.

Parameters:

Name Type Description Default
target str

A module.ClassName path, as produced by class_target.

 required

Returns:

Type Description
type

The class named by target.

Raises:

Type Description
ValueError

target names no module path, is relative, its module cannot be imported, the module has no such attribute, or the attribute is not a class. A module that fails on its own imports is reported the same way — the original ImportError is chained as __cause__. Any other failure of the module's body (a SyntaxError, say) propagates unchanged, so a broken module reports its own error rather than hiding behind "cannot import".
Source code in packages/contracts/src/contracts/config_schemas.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def import_class(target: str) -> type:
    """Resolve a fully-qualified class path to the class object.

    The inverse of ``class_target``: imports the module and looks the class up on it.

    Args:
        target: A ``module.ClassName`` path, as produced by ``class_target``.

    Returns:
        The class named by ``target``.

    Raises:
        ValueError: ``target`` names no module path, is relative, its module cannot be imported,
            the module has no such attribute, or the attribute is not a class. A module that
            fails on its *own* imports is reported the same way — the original ``ImportError``
            is chained as ``__cause__``. Any other failure of the module's body (a ``SyntaxError``,
            say) propagates unchanged, so a broken module reports its own error rather than
            hiding behind "cannot import".
    """
    module_path, _, class_name = target.rpartition(".")
    # A leading dot would make import_module read the target as a relative import and demand a
    # package to resolve it against; that is a malformed target, so reject it here rather than
    # letting it surface as the TypeError import_module would raise.
    if not module_path or module_path.startswith("."):
        raise ValueError(f"{target!r} is not a fully-qualified class path (expected 'module.Cls').")
    try:
        module = importlib.import_module(module_path)
    except ImportError as error:
        raise ValueError(f"Cannot import module {module_path!r} of target {target!r}.") from error
    try:
        resolved = getattr(module, class_name)
    except AttributeError as error:
        raise ValueError(f"Module {module_path!r} has no attribute {class_name!r}.") from error
    if not isinstance(resolved, type):
        # TRY004 wants a TypeError, but the isinstance check is on what `target` *resolved to*,
        # not on `target` itself, which is a perfectly well-typed str. What is wrong is its value,
        # so every failure here is one ValueError and a caller needs to catch only that.
        raise ValueError(  # noqa: TRY004
            f"Target {target!r} resolved to {resolved!r}, which is not a class."
        )
    return resolved

load_cv_config(path)

Load and validate the cross-validation config from a YAML file.

The CV folds are the leaderboard's evaluation protocol and must be read from conf/cv/default.yaml (never hard-coded) so every experiment and asset shares one canonical definition.

Parameters:

Name Type Description Default
path Path

Path to the CV config YAML (e.g. conf/cv/default.yaml).

 required

Returns:

Type Description
CvConfig

The validated CvConfig.
Source code in packages/contracts/src/contracts/config_schemas.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def load_cv_config(path: Path) -> CvConfig:
    """Load and validate the cross-validation config from a YAML file.

    The CV folds are the leaderboard's evaluation protocol and must be read from
    ``conf/cv/default.yaml`` (never hard-coded) so every experiment and asset shares one
    canonical definition.

    Args:
        path: Path to the CV config YAML (e.g. ``conf/cv/default.yaml``).

    Returns:
        The validated ``CvConfig``.
    """
    with path.open(encoding="utf-8") as file:
        return CvConfig.model_validate(yaml.safe_load(file))

contracts.settings

Settings: every data path and object-store credential the pipeline reads.

Values are resolved from the environment and the workspace .env.

Attributes

url_adapter = TypeAdapter(AnyHttpUrl) module-attribute

PROJECT_ROOT = _find_project_root(Path(__file__)) module-attribute

The repo root — anchor for the repo-relative defaults below (conf/, data/, .env).

Resolved by :func:_find_project_root; see its docstring for the per-install-mode behaviour and the caveat for wheels installed outside a workspace checkout.

Classes

Settings

Bases: BaseSettings

Configuration settings for the NGED substation forecast project.

Source code in packages/contracts/src/contracts/settings.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
class Settings(BaseSettings):
    """Configuration settings for the NGED substation forecast project."""

    mlflow_tracking_uri: str = Field(
        default="sqlite:///mlflow.db",
        description=(
            "MLflow tracking URI. Centralized here for environment-specific configuration"
            " (e.g., local SQLite for development, remote server for production)."
            " The application entrypoint should read this setting and set the"
            " MLFLOW_TRACKING_URI environment variable accordingly, which MLflow will"
            " automatically pick up."
        ),
    )

    # Credentials for NGED's *source* bucket, which lives in NGED's AWS account. Empty by default,
    # not required: `get_nged_s3_store` is their only consumer, and its only caller is the
    # `power_time_series_and_metadata` ingest asset. Making them required would mean every Delta
    # read, training run and dashboard could not build a `Settings` without third-party credentials
    # it never uses — and since `.env` is gitignored, a fresh clone could not run the test suite at
    # all, which is exactly what "the whole system must be exercisable on one laptop" forbids:
    # https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/design-principles/#6-the-whole-system-must-be-exercisable-on-one-laptop
    #
    # Absence is therefore caught at the point of *use* (`get_nged_s3_store`), which also confines
    # a mis-wired secret to the one schedule that needs it: inference reads our own Delta tables and
    # a baked-in model, so failing it over an ingest credential would stop the forecast.
    nged_s3_bucket_url: str = Field(
        default="",
        description=(
            "NGED source S3 bucket URL. Typically stored in the .env file. Empty unless you need"
            " to ingest from NGED's bucket; `get_nged_s3_store` raises if it is."
        ),
    )
    nged_s3_bucket_access_key: str = Field(
        default="",
        description="Access key for the NGED source S3 bucket. Typically stored in the .env file.",
    )
    nged_s3_bucket_secret: str = Field(
        default="",
        description="Secret key for the NGED source S3 bucket. Typically stored in the .env file.",
    )

    def require_nged_source_credentials(self) -> None:
        """Raise ``ValueError`` unless all three NGED source-bucket credentials are set.

        Call this immediately before doing something that reads NGED's bucket, so the failure names
        the missing configuration instead of surfacing as an opaque auth error from ``obstore``.

        Do *not* call it from process start-up or module import to fail a deployment fast: inference
        needs none of these credentials, so that would stop the forecast over a missing ingest
        secret. See the [AWS runbook](https://openclimatefix.github.io/nged-substation-forecast/live_service/aws/#step-8-store-secrets-in-parameter-store).

        Raises:
            ValueError: Naming exactly which of the three environment variables are unset.
        """
        missing = [
            name
            for name, value in (
                ("NGED_S3_BUCKET_URL", self.nged_s3_bucket_url),
                ("NGED_S3_BUCKET_ACCESS_KEY", self.nged_s3_bucket_access_key),
                ("NGED_S3_BUCKET_SECRET", self.nged_s3_bucket_secret),
            )
            if not value
        ]
        if missing:
            raise ValueError(
                f"NGED source-bucket credentials are not set: {', '.join(missing)}. Reading"
                " NGED's telemetry needs all three. Set them in `.env` (locally) or inject them"
                " from Parameter Store (on AWS). Everything that does not read NGED's own bucket"
                " — training, cross-validation, the dashboards, every Delta read — works without"
                " them."
            )

    def get_nged_s3_store(self) -> obstore.store.S3Store:
        """Returns an initialized obstore.store.S3Store instance for the NGED bucket.

        Raises:
            ValueError: If any of the three source-bucket credentials is unset.
        """
        self.require_nged_source_credentials()
        return obstore.store.S3Store.from_url(
            url=self.nged_s3_bucket_url,
            config={
                "aws_access_key_id": self.nged_s3_bucket_access_key,
                "aws_secret_access_key": self.nged_s3_bucket_secret,
            },
        )

    cv_config_path: Path = Field(
        default=PROJECT_ROOT / "conf" / "cv" / "default.yaml",
        description=(
            "Path to the canonical cross-validation fold definitions. These folds are the"
            " shared leaderboard evaluation protocol; read them from here, never hard-coded."
        ),
    )
    # --- Storage roots -------------------------------------------------------------------
    #
    # data_path_internal and data_path_delivery hold the (S3-capable) data tables; local_
    # artifacts_path holds the always-local production model. Why they are split, and
    # what belongs in each:
    # https://openclimatefix.github.io/nged-substation-forecast/live_service/setup/

    data_path_internal: str = Field(
        default=str(PROJECT_ROOT / "data"),
        description=(
            "Root of OCF's own working data tables (NWP, power, forecast_metrics, …). A local"
            " path by default; may be a remote URI (e.g. 's3://bucket/data') so the data tables"
            " live on S3. Every *_data_path below that is left unset derives from this root,"
            " except the NGED-facing delivery tables, which derive from data_path_delivery"
            " instead."
        ),
    )
    data_path_delivery: str = Field(
        default=str(PROJECT_ROOT / "data"),
        description=(
            "Root of the NGED-facing delivery tables (power_forecasts, effective_capacity, …)."
            " Defaults to the same local path as data_path_internal, so local dev sees one"
            " directory and the split is invisible; on AWS this points at the separate,"
            " NGED-facing bucket."
        ),
    )
    local_artifacts_path: str = Field(
        default=str(PROJECT_ROOT / "data"),
        description=(
            "Root of the always-local artifacts (the production model). Kept separate from"
            " the data-table roots because these back local-filesystem-only libraries and"
            " must stay local even when the data tables live on S3."
        ),
    )

    # --- Object-store credentials for the data tables (used only when a data-path root is remote)
    #
    # All empty by default; unset on AWS (object_store auto-discovers the IAM-role credentials),
    # set only for a dev/MinIO endpoint. The AWS/dev split, and how these differ from the
    # nged_s3_bucket_* source-bucket creds above:
    # https://openclimatefix.github.io/nged-substation-forecast/live_service/setup/

    data_store_endpoint_url: str = Field(
        default="",
        description=(
            "S3-compatible endpoint URL for the data tables (e.g. a MinIO/dev endpoint). Empty on"
            " AWS, where the endpoint is inferred. When set, the store is also allowed to use"
            " plain HTTP (dev endpoints rarely have TLS)."
        ),
    )
    data_store_access_key_id: str = Field(
        default="", description="Access key for the data-table object store; empty on AWS (IAM)."
    )
    data_store_secret_access_key: str = Field(
        default="", description="Secret key for the data-table object store; empty on AWS (IAM)."
    )
    data_store_region: str = Field(
        default="", description="Region for the data-table object store; empty to auto-discover."
    )

    @property
    def storage_options(self) -> ObjectStoreOptions:
        """delta-rs / polars / obstore ``storage_options`` for the managed data tables.

        Empty on AWS — object_store auto-discovers the Fargate task's IAM-role credentials and
        region — and empty for a local data-path root (delta-rs ignores it there). Populated from
        the ``data_store_*`` settings only for a dev/MinIO/S3-compatible endpoint. The ``aws_*``
        keys are the shared object_store aliases understood by delta-rs, polars cloud IO, and
        obstore alike, so one value feeds every IO site. Returned as an ``ObjectStoreOptions``
        ``TypedDict`` so ``ty`` checks each key here, where they are authored; widen it to a plain
        ``dict`` at each IO boundary with ``typeddict_to_dict``.
        """
        options: ObjectStoreOptions = {}
        if self.data_store_endpoint_url:
            options["aws_endpoint_url"] = self.data_store_endpoint_url
            options["aws_allow_http"] = "true"
        if self.data_store_access_key_id:
            options["aws_access_key_id"] = self.data_store_access_key_id
        if self.data_store_secret_access_key:
            options["aws_secret_access_key"] = self.data_store_secret_access_key
        if self.data_store_region:
            options["aws_region"] = self.data_store_region
        return options

    # --- Managed data tables (derive from a root unless explicitly set) --------------------
    #
    # Each defaults to "" — a sentinel meaning "derive from data_path_internal or
    # data_path_delivery in _derive_unset_paths". The derive-from-root convention and per-table
    # overrides: https://openclimatefix.github.io/nged-substation-forecast/live_service/setup/

    nged_data_path: str = ""
    """Directory holding the NGED power_time_series Delta table and metadata parquet."""
    nwp_data_path: str = ""
    """Delta table of NWP weather data."""
    power_forecasts_data_path: str = ""
    """Delta table of power forecasts (partitioned by experiment_name, fold_id).

    An NGED-facing delivery table — derives from data_path_delivery, not data_path_internal.
    """
    forecast_metrics_data_path: str = ""
    """Delta table of forecast evaluation metrics."""
    power_time_series_data_path: str = ""
    """Delta table of half-hourly power observations (under nged_data_path)."""
    metadata_path: str = ""
    """Parquet file of per-series substation metadata (under nged_data_path)."""
    eligible_time_series_data_path: str = Field(
        default="",
        description=(
            "Delta table of the canonical per-fold eligible time_series_id population, written"
            " by the eligible_time_series asset (partitioned by fold_id) and read by"
            " trained_cv_model and cv_power_forecasts so every experiment scores a fold on the"
            " identical, experiment-independent population."
        ),
    )
    effective_capacity_data_path: str = Field(
        default="",
        description=(
            "Delta table of per-series effective capacity (v0.1: full-history P99 of |power|),"
            " written by the effective_capacity asset and read by the metrics asset as the NMAE"
            " denominator. An NGED-facing delivery table — derives from data_path_delivery, not"
            " data_path_internal."
        ),
    )
    h3_grid_weights_path: str = ""
    """Parquet file of fractional H3 cell overlap with the GB boundary."""

    # --- Always-local artifacts (derive from local_artifacts_path unless explicitly set) --

    production_model_path: str = Field(
        default="",
        description=(
            "Directory holding the current production model, written by the promoted_model"
            " asset (ml_core.production_helpers.fetch_model_artifacts) and read by"
            " live_forecasts via a plain BaseForecaster disk load — no MLflow at inference"
            " time. Later COPY'd into the container image at build time (issue #222)."
        ),
    )

    # --- Sentry observability (all optional; an empty DSN disables Sentry entirely) ---

    sentry_dsn: str = Field(
        default="",
        description=(
            "Sentry.io project DSN. Empty (the default) disables Sentry entirely — every Sentry"
            " code path becomes a no-op — so laptops and CI need no Sentry config. Set it in the"
            " .env file to enable error telemetry. Passed explicitly to sentry_sdk.init as the"
            " single source of truth (the SDK would otherwise also auto-read the SENTRY_DSN env"
            " var)."
        ),
    )
    sentry_environment: str = Field(
        default="local",
        description=(
            "Sentry environment tag, the dimension that separates deployments in Sentry's UI,"
            " alerts, and cron monitors. The always-on production box sets 'production'; each"
            " developer overrides the 'local' fallback with '<name>-laptop' (e.g. 'jacks-laptop')"
            " so laptop telemetry is cleanly filterable and the production-scoped missed-check-in"
            " alert never fires for a laptop."
        ),
    )
    sentry_traces_sample_rate: float = Field(
        default=0.0,
        description=(
            "Fraction of transactions sampled for Sentry performance tracing (off by default)."
        ),
    )
    sentry_monitor_forecasts: bool = Field(
        default=False,
        description=(
            "Emit the live_forecasts success heartbeat to Sentry's cron monitor (the"
            " missed-check-in alarm). True only on the always-on production deployment; left"
            " False on laptops so an intermittently-run laptop never registers a monitor"
            " environment that Sentry would then mark as missed."
        ),
    )

    @model_validator(mode="after")
    def _derive_unset_paths(self) -> Self:
        """Fill any unset ("") path from its root, so callers always see a concrete path.

        The default layout lives here and nowhere else. A field set explicitly (e.g. via its
        env var) keeps its value; only the "" sentinels are derived.
        """
        self.nged_data_path = self.nged_data_path or uri_join(self.data_path_internal, "NGED")
        self.nwp_data_path = self.nwp_data_path or uri_join(self.data_path_internal, "NWP")
        self.forecast_metrics_data_path = self.forecast_metrics_data_path or uri_join(
            self.data_path_internal, "forecast_metrics"
        )
        self.eligible_time_series_data_path = self.eligible_time_series_data_path or uri_join(
            self.data_path_internal, "eligible_time_series"
        )
        self.h3_grid_weights_path = self.h3_grid_weights_path or uri_join(
            self.data_path_internal, "h3_grid_weights.parquet"
        )
        # NGED-facing delivery tables derive from data_path_delivery instead, so a new delivery
        # table can't silently land in the internal bucket by inheriting the default derivation.
        # `power_forecast` and `effective_capacity` are two of the five tables in NGED's stable
        # delivery contract; see
        # <https://openclimatefix.github.io/nged-substation-forecast/architecture/forecast-delivery/#securing-it>
        # for the rest and why they live in a separate bucket.
        self.power_forecasts_data_path = self.power_forecasts_data_path or uri_join(
            self.data_path_delivery, "power_forecasts"
        )
        self.effective_capacity_data_path = self.effective_capacity_data_path or uri_join(
            self.data_path_delivery, "effective_capacity"
        )
        # Derived from nged_data_path (itself derived above).
        self.power_time_series_data_path = self.power_time_series_data_path or uri_join(
            self.nged_data_path, "power_time_series.delta"
        )
        self.metadata_path = self.metadata_path or uri_join(self.nged_data_path, "metadata.parquet")
        # Always-local artifacts.
        self.production_model_path = self.production_model_path or uri_join(
            self.local_artifacts_path, "production_model"
        )
        return self

    # Tell Pydantic to override defaults with fields set in the .env file.
    model_config = SettingsConfigDict(
        env_file=PROJECT_ROOT / ".env",
        extra="forbid",
        env_file_encoding="utf-8",
        env_prefix="",
    )

    @field_validator("nged_s3_bucket_url")
    @classmethod
    def validate_url(cls, v: str) -> str:
        """Validate that the S3 bucket URL is a valid URL, if one was given at all.

        Empty means "no NGED ingest configured", which is the default and not an error;
        ``require_nged_source_credentials`` is what rejects it where it matters. A *non-empty*
        value still has to be a real URL — strict about malformed, liberal about missing.
        """
        if v:
            url_adapter.validate_python(v)
        return v
Attributes
mlflow_tracking_uri = Field(default='sqlite:///mlflow.db', description='MLflow tracking URI. Centralized here for environment-specific configuration (e.g., local SQLite for development, remote server for production). The application entrypoint should read this setting and set the MLFLOW_TRACKING_URI environment variable accordingly, which MLflow will automatically pick up.') class-attribute instance-attribute
nged_s3_bucket_url = Field(default='', description="NGED source S3 bucket URL. Typically stored in the .env file. Empty unless you need to ingest from NGED's bucket; `get_nged_s3_store` raises if it is.") class-attribute instance-attribute
nged_s3_bucket_access_key = Field(default='', description='Access key for the NGED source S3 bucket. Typically stored in the .env file.') class-attribute instance-attribute
nged_s3_bucket_secret = Field(default='', description='Secret key for the NGED source S3 bucket. Typically stored in the .env file.') class-attribute instance-attribute
cv_config_path = Field(default=(PROJECT_ROOT / 'conf' / 'cv' / 'default.yaml'), description='Path to the canonical cross-validation fold definitions. These folds are the shared leaderboard evaluation protocol; read them from here, never hard-coded.') class-attribute instance-attribute
data_path_internal = Field(default=(str(PROJECT_ROOT / 'data')), description="Root of OCF's own working data tables (NWP, power, forecast_metrics, …). A local path by default; may be a remote URI (e.g. 's3://bucket/data') so the data tables live on S3. Every *_data_path below that is left unset derives from this root, except the NGED-facing delivery tables, which derive from data_path_delivery instead.") class-attribute instance-attribute
data_path_delivery = Field(default=(str(PROJECT_ROOT / 'data')), description='Root of the NGED-facing delivery tables (power_forecasts, effective_capacity, …). Defaults to the same local path as data_path_internal, so local dev sees one directory and the split is invisible; on AWS this points at the separate, NGED-facing bucket.') class-attribute instance-attribute
local_artifacts_path = Field(default=(str(PROJECT_ROOT / 'data')), description='Root of the always-local artifacts (the production model). Kept separate from the data-table roots because these back local-filesystem-only libraries and must stay local even when the data tables live on S3.') class-attribute instance-attribute
data_store_endpoint_url = Field(default='', description='S3-compatible endpoint URL for the data tables (e.g. a MinIO/dev endpoint). Empty on AWS, where the endpoint is inferred. When set, the store is also allowed to use plain HTTP (dev endpoints rarely have TLS).') class-attribute instance-attribute
data_store_access_key_id = Field(default='', description='Access key for the data-table object store; empty on AWS (IAM).') class-attribute instance-attribute
data_store_secret_access_key = Field(default='', description='Secret key for the data-table object store; empty on AWS (IAM).') class-attribute instance-attribute
data_store_region = Field(default='', description='Region for the data-table object store; empty to auto-discover.') class-attribute instance-attribute
storage_options property

delta-rs / polars / obstore storage_options for the managed data tables.

Empty on AWS — object_store auto-discovers the Fargate task's IAM-role credentials and region — and empty for a local data-path root (delta-rs ignores it there). Populated from the data_store_* settings only for a dev/MinIO/S3-compatible endpoint. The aws_* keys are the shared object_store aliases understood by delta-rs, polars cloud IO, and obstore alike, so one value feeds every IO site. Returned as an ObjectStoreOptions TypedDict so ty checks each key here, where they are authored; widen it to a plain dict at each IO boundary with typeddict_to_dict.

nged_data_path = '' class-attribute instance-attribute

Directory holding the NGED power_time_series Delta table and metadata parquet.

nwp_data_path = '' class-attribute instance-attribute

Delta table of NWP weather data.

power_forecasts_data_path = '' class-attribute instance-attribute

Delta table of power forecasts (partitioned by experiment_name, fold_id).

An NGED-facing delivery table — derives from data_path_delivery, not data_path_internal.

forecast_metrics_data_path = '' class-attribute instance-attribute

Delta table of forecast evaluation metrics.

power_time_series_data_path = '' class-attribute instance-attribute

Delta table of half-hourly power observations (under nged_data_path).

metadata_path = '' class-attribute instance-attribute

Parquet file of per-series substation metadata (under nged_data_path).

eligible_time_series_data_path = Field(default='', description='Delta table of the canonical per-fold eligible time_series_id population, written by the eligible_time_series asset (partitioned by fold_id) and read by trained_cv_model and cv_power_forecasts so every experiment scores a fold on the identical, experiment-independent population.') class-attribute instance-attribute
effective_capacity_data_path = Field(default='', description='Delta table of per-series effective capacity (v0.1: full-history P99 of |power|), written by the effective_capacity asset and read by the metrics asset as the NMAE denominator. An NGED-facing delivery table — derives from data_path_delivery, not data_path_internal.') class-attribute instance-attribute
h3_grid_weights_path = '' class-attribute instance-attribute

Parquet file of fractional H3 cell overlap with the GB boundary.

production_model_path = Field(default='', description="Directory holding the current production model, written by the promoted_model asset (ml_core.production_helpers.fetch_model_artifacts) and read by live_forecasts via a plain BaseForecaster disk load — no MLflow at inference time. Later COPY'd into the container image at build time (issue #222).") class-attribute instance-attribute
sentry_dsn = Field(default='', description='Sentry.io project DSN. Empty (the default) disables Sentry entirely — every Sentry code path becomes a no-op — so laptops and CI need no Sentry config. Set it in the .env file to enable error telemetry. Passed explicitly to sentry_sdk.init as the single source of truth (the SDK would otherwise also auto-read the SENTRY_DSN env var).') class-attribute instance-attribute
sentry_environment = Field(default='local', description="Sentry environment tag, the dimension that separates deployments in Sentry's UI, alerts, and cron monitors. The always-on production box sets 'production'; each developer overrides the 'local' fallback with '<name>-laptop' (e.g. 'jacks-laptop') so laptop telemetry is cleanly filterable and the production-scoped missed-check-in alert never fires for a laptop.") class-attribute instance-attribute
sentry_traces_sample_rate = Field(default=0.0, description='Fraction of transactions sampled for Sentry performance tracing (off by default).') class-attribute instance-attribute
sentry_monitor_forecasts = Field(default=False, description="Emit the live_forecasts success heartbeat to Sentry's cron monitor (the missed-check-in alarm). True only on the always-on production deployment; left False on laptops so an intermittently-run laptop never registers a monitor environment that Sentry would then mark as missed.") class-attribute instance-attribute
model_config = SettingsConfigDict(env_file=(PROJECT_ROOT / '.env'), extra='forbid', env_file_encoding='utf-8', env_prefix='') class-attribute instance-attribute
Methods:
require_nged_source_credentials()

Raise ValueError unless all three NGED source-bucket credentials are set.

Call this immediately before doing something that reads NGED's bucket, so the failure names the missing configuration instead of surfacing as an opaque auth error from obstore.

Do not call it from process start-up or module import to fail a deployment fast: inference needs none of these credentials, so that would stop the forecast over a missing ingest secret. See the AWS runbook.

Raises:

Type Description
ValueError

Naming exactly which of the three environment variables are unset.
Source code in packages/contracts/src/contracts/settings.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def require_nged_source_credentials(self) -> None:
    """Raise ``ValueError`` unless all three NGED source-bucket credentials are set.

    Call this immediately before doing something that reads NGED's bucket, so the failure names
    the missing configuration instead of surfacing as an opaque auth error from ``obstore``.

    Do *not* call it from process start-up or module import to fail a deployment fast: inference
    needs none of these credentials, so that would stop the forecast over a missing ingest
    secret. See the [AWS runbook](https://openclimatefix.github.io/nged-substation-forecast/live_service/aws/#step-8-store-secrets-in-parameter-store).

    Raises:
        ValueError: Naming exactly which of the three environment variables are unset.
    """
    missing = [
        name
        for name, value in (
            ("NGED_S3_BUCKET_URL", self.nged_s3_bucket_url),
            ("NGED_S3_BUCKET_ACCESS_KEY", self.nged_s3_bucket_access_key),
            ("NGED_S3_BUCKET_SECRET", self.nged_s3_bucket_secret),
        )
        if not value
    ]
    if missing:
        raise ValueError(
            f"NGED source-bucket credentials are not set: {', '.join(missing)}. Reading"
            " NGED's telemetry needs all three. Set them in `.env` (locally) or inject them"
            " from Parameter Store (on AWS). Everything that does not read NGED's own bucket"
            " — training, cross-validation, the dashboards, every Delta read — works without"
            " them."
        )
get_nged_s3_store()

Returns an initialized obstore.store.S3Store instance for the NGED bucket.

Raises:

Type Description
ValueError

If any of the three source-bucket credentials is unset.
Source code in packages/contracts/src/contracts/settings.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_nged_s3_store(self) -> obstore.store.S3Store:
    """Returns an initialized obstore.store.S3Store instance for the NGED bucket.

    Raises:
        ValueError: If any of the three source-bucket credentials is unset.
    """
    self.require_nged_source_credentials()
    return obstore.store.S3Store.from_url(
        url=self.nged_s3_bucket_url,
        config={
            "aws_access_key_id": self.nged_s3_bucket_access_key,
            "aws_secret_access_key": self.nged_s3_bucket_secret,
        },
    )
validate_url(v) classmethod

Validate that the S3 bucket URL is a valid URL, if one was given at all.

Empty means "no NGED ingest configured", which is the default and not an error; require_nged_source_credentials is what rejects it where it matters. A non-empty value still has to be a real URL — strict about malformed, liberal about missing.

Source code in packages/contracts/src/contracts/settings.py 
374
375
376
377
378
379
380
381
382
383
384
385
@field_validator("nged_s3_bucket_url")
@classmethod
def validate_url(cls, v: str) -> str:
    """Validate that the S3 bucket URL is a valid URL, if one was given at all.

    Empty means "no NGED ingest configured", which is the default and not an error;
    ``require_nged_source_credentials`` is what rejects it where it matters. A *non-empty*
    value still has to be a real URL — strict about malformed, liberal about missing.
    """
    if v:
        url_adapter.validate_python(v)
    return v

Functions:

get_settings() cached

Return the shared, lazily-constructed Settings singleton.

Prefer this over constructing Settings() at module import time: instantiation reads .env and the environment, so deferring it to first use keeps library modules (e.g. the contracts schemas) importable whatever the environment holds, and lets a test change the environment before the first read.

Cached with lru_cache so every caller shares one instance (matching the previous module-level singletons). Call get_settings.cache_clear() in a test that needs to re-read the environment after changing it.

Source code in packages/contracts/src/contracts/settings.py 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
@lru_cache
def get_settings() -> Settings:
    """Return the shared, lazily-constructed ``Settings`` singleton.

    Prefer this over constructing ``Settings()`` at module import time: instantiation reads
    ``.env`` and the environment, so deferring it to first *use* keeps library modules (e.g. the
    ``contracts`` schemas) importable whatever the environment holds, and lets a test change the
    environment before the first read.

    Cached with ``lru_cache`` so every caller shares one instance (matching the previous
    module-level singletons). Call ``get_settings.cache_clear()`` in a test that needs to
    re-read the environment after changing it.
    """
    return Settings()

contracts.uri

Small URI helpers for paths that may be local filesystem paths or remote URIs.

Settings data-location fields are plain str so they can hold either a local path (/home/.../data/NWP) or a remote URI (s3://bucket/NWP). pathlib.Path mangles the latter (Path("s3://b/a") / "c" drops the scheme), so joins route through here.

The existence/parent helpers below give the asset IO layer a single local-or-remote-aware call for the two things it does around every Delta/parquet write: make sure the parent directory exists (a no-op on object stores, which have no directories) and check whether a table/object is already there. Remote calls go through delta-rs / obstore with the caller's storage_options so the same code path serves both a local data-path root and an s3:// one.

Classes

ObjectStoreOptions

Bases: TypedDict

object_store options for the managed data tables.

These are the shared aws_* aliases understood by delta-rs, Polars, and obstore alike, so one value feeds every IO site.

Authored as a TypedDict (rather than a bare dict[str, str]) so ty checks every key where it is written — see Settings.storage_options. Widen it to the plain dict the IO libraries expect with typeddict_to_dict at each call boundary. Empty on AWS (object_store auto-discovers the IAM-role credentials and region) and for a local data-path root.

Source code in packages/contracts/src/contracts/uri.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class ObjectStoreOptions(TypedDict, total=False):
    """object_store options for the managed data tables.

    These are the shared ``aws_*`` aliases understood by delta-rs, Polars, and obstore alike, so
    one value feeds every IO site.

    Authored as a ``TypedDict`` (rather than a bare ``dict[str, str]``) so ``ty`` checks every key
    where it is written — see ``Settings.storage_options``. Widen it to the plain ``dict`` the IO
    libraries expect with ``typeddict_to_dict`` at each call boundary. Empty on AWS (object_store
    auto-discovers the IAM-role credentials and region) and for a local data-path root.
    """

    aws_endpoint_url: str
    aws_allow_http: str
    aws_access_key_id: str
    aws_secret_access_key: str
    aws_region: str
Attributes
aws_endpoint_url instance-attribute
aws_allow_http instance-attribute
aws_access_key_id instance-attribute
aws_secret_access_key instance-attribute
aws_region instance-attribute

Functions:

is_remote_uri(uri)

Return whether uri carries a URI scheme (e.g. s3://) rather than a local path.

Source code in packages/contracts/src/contracts/uri.py 
46
47
48
def is_remote_uri(uri: str) -> bool:
    """Return whether ``uri`` carries a URI scheme (e.g. ``s3://``) rather than a local path."""
    return _SCHEME_SEP in uri

uri_join(base, *parts)

Join parts onto base with local- or remote-aware semantics.

Local bases join through pathlib (yielding an absolute path string); remote, scheme-bearing bases join posix-style so "s3://bucket/a" + "b" stays "s3://bucket/a/b" instead of being mangled by Path.__truediv__.

Source code in packages/contracts/src/contracts/uri.py 
51
52
53
54
55
56
57
58
59
60
def uri_join(base: str, *parts: str) -> str:
    """Join ``parts`` onto ``base`` with local- or remote-aware semantics.

    Local bases join through ``pathlib`` (yielding an absolute path string); remote,
    scheme-bearing bases join posix-style so ``"s3://bucket/a"`` + ``"b"`` stays
    ``"s3://bucket/a/b"`` instead of being mangled by ``Path.__truediv__``.
    """
    if is_remote_uri(base):
        return posixpath.join(base.rstrip("/"), *parts)
    return str(Path(base).joinpath(*parts))

if_local_path_then_make_parent_dir(uri)

Create the parent directory of a local uri; a no-op for a remote URI.

Local filesystems need a table/file's parent directory to exist before a write; object stores (s3://) have no directories — a write creates the key's prefix implicitly — so there is nothing to do and this returns immediately.

Source code in packages/contracts/src/contracts/uri.py 
63
64
65
66
67
68
69
70
71
72
def if_local_path_then_make_parent_dir(uri: str) -> None:
    """Create the parent directory of a *local* ``uri``; a no-op for a remote URI.

    Local filesystems need a table/file's parent directory to exist before a write; object
    stores (``s3://``) have no directories — a write creates the key's prefix implicitly — so
    there is nothing to do and this returns immediately.
    """
    if is_remote_uri(uri):
        return
    Path(uri).parent.mkdir(parents=True, exist_ok=True)

delta_table_exists(uri, storage_options=None)

Return whether a Delta table already exists at uri (local path or remote URI).

Wraps DeltaTable.is_deltatable, which inspects the _delta_log through delta-rs' object_store and so works identically for a local path and an s3:// URI given the matching storage_options. Replaces Path(uri).exists() at the write-guard sites, which would raise on a remote URI.

Source code in packages/contracts/src/contracts/uri.py 
75
76
77
78
79
80
81
82
83
def delta_table_exists(uri: str, storage_options: ObjectStoreOptions | None = None) -> bool:
    """Return whether a Delta table already exists at ``uri`` (local path or remote URI).

    Wraps ``DeltaTable.is_deltatable``, which inspects the ``_delta_log`` through delta-rs'
    object_store and so works identically for a local path and an ``s3://`` URI given the
    matching ``storage_options``. Replaces ``Path(uri).exists()`` at the write-guard sites,
    which would raise on a remote URI.
    """
    return DeltaTable.is_deltatable(uri, storage_options=typeddict_to_dict(storage_options) or {})

object_exists(uri, storage_options=None)

Return whether a single object/file at uri exists (local file or remote object).

For a local uri this is Path.exists(); for a remote URI it issues an object-store head via obstore, so it works for a plain file (e.g. a .parquet) that is not a Delta table. Use delta_table_exists for Delta tables.

Source code in packages/contracts/src/contracts/uri.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def object_exists(uri: str, storage_options: ObjectStoreOptions | None = None) -> bool:
    """Return whether a single object/file at ``uri`` exists (local file or remote object).

    For a local ``uri`` this is ``Path.exists()``; for a remote URI it issues an object-store
    ``head`` via obstore, so it works for a plain file (e.g. a ``.parquet``) that is not a Delta
    table. Use ``delta_table_exists`` for Delta tables.
    """
    if not is_remote_uri(uri):
        return Path(uri).exists()
    parsed = urlparse(uri)
    store = obstore.store.S3Store(parsed.netloc, config=typeddict_to_dict(storage_options) or {})
    try:
        obstore.head(store, parsed.path.lstrip("/"))
    except FileNotFoundError:
        return False
    return True