NGED Data API

NGED JSON Data

This package reads NGED's telemetry JSON files from S3 and parses them into the PowerTimeSeries and TimeSeriesMetadata schemas (see contracts). The metadata table and the downloaded-files list are the only files this package owns and writes; the parsed power observations are handed back to the caller, which appends them to the power_time_series Delta table (see Usage below). nged_data.storage's module docstring, below on this page, says which functions read that Delta table and which write the metadata table.

Public surface

Only upsert_metadata is re-exported from the package root (from nged_data import upsert_metadata); the other functions live in nged_data.storage and nged_data.cleaning (from nged_data.storage import list_timeseries_json_files, etc.).

• nged_data.storage.list_timeseries_json_files(store) — lists the timeseries JSON files on NGED's S3 bucket, parsing time_series_id, start_time, and end_time out of each file's path.
• nged_data.storage.read_downloaded_files(...), write_downloaded_files(...) — read and replace the downloaded-files list, which records the bucket listing (path and LastModified) that the ingest last processed in full.
• nged_data.storage.select_files_not_yet_downloaded(file_listing, downloaded_files) — keeps the listed files that the downloaded-files list lacks, so each file is downloaded once.
• nged_data.storage.download_and_parse_files(store, paths_df) — downloads the listed files concurrently, in chunks, and parses them in end_time order, returning a DownloadAndParseResult of metadata (TimeSeriesMetadata), power_time_series (PowerTimeSeries), and n_implausible_power_rows_dropped. When every file's data field was null, the power frame is empty.
• nged_data.storage.select_new_rows(time_series, delta_path, storage_options=None) — filters PowerTimeSeries rows down to those missing from the power_time_series Delta table at delta_path, by an anti-join on (time_series_id, time), so a reading is kept regardless of arrival order.
• nged_data.storage.time_series_coverage(delta_path, storage_options=None) — the earliest and latest observation time on disk for each time_series_id in the power_time_series Delta table.
• nged_data.storage.coverage_from_power(power) — the same earliest and latest time per time_series_id, from any lazy PowerTimeSeries frame.
• nged_data.storage.scan_cleaned_power(delta_path, storage_options=None) — scans the cleaned_power_time_series Delta table, keeping only the rows no cleaning rule flagged. Every reader of observed power except the ingest and its freshness check uses scan_cleaned_power.
• nged_data.cleaning.flag_nged_power(power, metadata) — where cleaning rules go. flag_nged_power returns every raw power row plus a drop_reason column, and the clean_nged_power_data Dagster asset writes the result to the cleaned table. The function's docstring says how to add a rule.
• nged_data.upsert_metadata(new_metadata, metadata_path, storage_options=None) — merges a TimeSeriesMetadata snapshot into the stored metadata Parquet file, keeping the newest values per time_series_id and rewriting the file only if the incoming metadata differs from what is stored.

nged_data.read_nged_json parses one downloaded JSON file into the two schemas. All three of its functions are private, and the two that parse a whole file are called only by download_and_parse_files, so the module appears on the API page below carrying just its ExtractedPowerTimeSeries result type.

Data quality

download_and_parse_files drops rows whose time is malformed — outside the plausible datetime range, null, or not aligned to the top or bottom of the hour — via PowerTimeSeries.drop_implausible_rows, and reports how many as n_implausible_power_rows_dropped. Degrading rather than raising on a malformed reading follows inherent stability: a malformed time originates upstream of our pipeline, at the meter or in the telemetry export, not in our own code. Ingestion therefore keeps the rest of the batch rather than aborting it. No other cleaning happens during ingestion. Cleaning rules run afterwards, over the whole stored table, in nged_data.cleaning.

Usage

This package is used by the power_time_series_and_metadata Dagster asset in src/nged_substation_forecast/defs/assets.py.

nged_data.read_nged_json

Extracts metadata and time series from NGED JSON data.

The parser expects two properties of each JSON file. Metadata fields sit at the top level. A data field holds an array of time series data points.

Attributes

log = logging.getLogger(__name__) module-attribute

Classes

NoReadingsInFile

Bases: Exception

Raised by _extract_power_time_series when a file's data field holds no readings.

NGED's meter reported nothing for the period the file covers.

Source code in packages/nged_data/src/nged_data/read_nged_json.py

class NoReadingsInFile(Exception):
    """Raised by `_extract_power_time_series` when a file's `data` field holds no readings.

    NGED's meter reported nothing for the period the file covers.
    """

ExtractedPowerTimeSeries

Bases: NamedTuple

Result of parsing PowerTimeSeries rows out of one NGED JSON file.

n_dropped counts rows dropped by PowerTimeSeries.drop_implausible_rows for a malformed time — see that method's docstring for why ingestion degrades rather than raising.

Source code in packages/nged_data/src/nged_data/read_nged_json.py

class ExtractedPowerTimeSeries(NamedTuple):
    """Result of parsing ``PowerTimeSeries`` rows out of one NGED JSON file.

    ``n_dropped`` counts rows dropped by ``PowerTimeSeries.drop_implausible_rows`` for a
    malformed ``time`` — see that method's docstring for why ingestion degrades rather than
    raising.
    """

    dataframe: pt.DataFrame[PowerTimeSeries]
    n_dropped: int

Attributes

dataframe instance-attribute

n_dropped instance-attribute

nged_data.storage

Listing, downloading, and parsing NGED's telemetry JSON from S3.

Each file yields TimeSeriesMetadata describing one series and PowerTimeSeries power observations from it. The metadata is upserted here, into a Parquet metadata table (see upsert_metadata). The power observations are not. This module never writes PowerTimeSeries rows to disk. It returns them to the caller, and the caller appends them to the power_time_series Delta table. select_new_rows reads that same Delta table, to drop rows already on disk, but does not write to it.

The module downloads each NGED file once. The downloaded-files list (read_downloaded_files, write_downloaded_files) records the bucket listing that the ingest last processed in full, and select_files_not_yet_downloaded keeps only the listed files that the downloaded-files list lacks.

Attributes

log = logging.getLogger(__name__) module-attribute

Classes

NgedFileParseError

Bases: ValueError

Raised when an NGED file is malformed or breaks the TimeSeriesMetadata contract.

Parsing is deterministic, so retrying the download cannot fix the file.

Source code in packages/nged_data/src/nged_data/storage.py

class NgedFileParseError(ValueError):
    """Raised when an NGED file is malformed or breaks the `TimeSeriesMetadata` contract.

    Parsing is deterministic, so retrying the download cannot fix the file.
    """

DownloadedFilesError

Bases: Exception

Raised when the stored downloaded-files list cannot be read or fails validation.

The ingest must stop on this error. Reading a damaged list as an empty one would download every file in NGED's bucket.

Source code in packages/nged_data/src/nged_data/storage.py

class DownloadedFilesError(Exception):
    """Raised when the stored downloaded-files list cannot be read or fails validation.

    The ingest must stop on this error. Reading a damaged list as an empty one would download
    every file in NGED's bucket.
    """

DownloadAndParseResult

Bases: NamedTuple

Result of download_and_parse_files.

n_implausible_power_rows_dropped sums ExtractedPowerTimeSeries.n_dropped across every file in the batch — see PowerTimeSeries.drop_implausible_rows for what gets dropped and why.

Source code in packages/nged_data/src/nged_data/storage.py

class DownloadAndParseResult(NamedTuple):
    """Result of ``download_and_parse_files``.

    ``n_implausible_power_rows_dropped`` sums ``ExtractedPowerTimeSeries.n_dropped`` across every
    file in the batch — see ``PowerTimeSeries.drop_implausible_rows`` for what gets dropped and
    why.
    """

    metadata: pt.DataFrame[TimeSeriesMetadata]
    power_time_series: pt.DataFrame[PowerTimeSeries]
    n_implausible_power_rows_dropped: int

Attributes

metadata instance-attribute

power_time_series instance-attribute

n_implausible_power_rows_dropped instance-attribute

TimeSeriesCoverage

Bases: Model

Per-series observation-time span of the power_time_series Delta table.

first_time/last_time are the earliest/latest observation time for each time_series_id. This frame is a transient intermediate, never persisted. Two callers read the frame: the freshness asset check reads last_time to detect staleness, and cross-validation (CV) fold-eligibility (eligible_time_series_ids) reads both first_time and last_time. A CV fold is one train/test split of the history, and a series is eligible for a fold only if its data covers that split.

The freshness check reads this on-disk recency rather than the asset's materialisation timestamp. A materialisation-freshness policy would miss the failure the check exists to catch. When NGED's telemetry stalls, the ingest asset keeps materialising successfully on schedule, and writes nothing. The materialisation looks fresh; the newest observation on disk does not. Full reasoning: https://openclimatefix.github.io/nged-substation-forecast/architecture/production-deployment/#warn-on-stale-power-data-with-a-dagster-asset-check.

Source code in packages/nged_data/src/nged_data/storage.py

class TimeSeriesCoverage(pt.Model):
    """Per-series observation-time span of the ``power_time_series`` Delta table.

    ``first_time``/``last_time`` are the earliest/latest observation ``time`` for each
    ``time_series_id``. This frame is a transient intermediate, never persisted. Two callers
    read the frame: the freshness asset check reads ``last_time`` to detect staleness, and
    cross-validation (CV) fold-eligibility (``eligible_time_series_ids``) reads both
    ``first_time`` and ``last_time``. A CV fold is one train/test split of the history, and a
    series is eligible for a fold only if its data covers that split.

    The freshness check reads this on-disk recency rather than the asset's materialisation
    timestamp. A materialisation-freshness policy would miss the failure the check exists to
    catch. When NGED's telemetry stalls, the ingest asset keeps materialising successfully on
    schedule, and writes nothing. The materialisation looks fresh; the newest observation on disk
    does not. Full reasoning:
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/production-deployment/#warn-on-stale-power-data-with-a-dagster-asset-check>.
    """

    time_series_id: int = _get_time_series_id_dtype(unique=True)
    first_time: int = pt.Field(dtype=PowerTimeSeries.dtypes["time"])
    last_time: int = pt.Field(dtype=PowerTimeSeries.dtypes["time"])

Attributes

time_series_id = _get_time_series_id_dtype(unique=True) class-attribute instance-attribute

first_time = pt.Field(dtype=PowerTimeSeries.dtypes['time']) class-attribute instance-attribute

last_time = pt.Field(dtype=PowerTimeSeries.dtypes['time']) class-attribute instance-attribute

UpsertMetadataStats

Bases: TypedDict

What the TimeSeriesMetadata upsert did, published as Dagster output metadata.

Source code in packages/nged_data/src/nged_data/storage.py

class UpsertMetadataStats(TypedDict, total=False):
    """What the ``TimeSeriesMetadata`` upsert did, published as Dagster output metadata."""

    metadata_n_new_TimeSeriesIDs: int
    metadata_n_updated_TimeSeriesIDs: int
    metadata_updated_TimeSeriesIDs: Sequence[int]
    metadata_upsert_failed: str
    """Set by the asset when the whole upsert raised, so the power write went ahead without it.

    Read this field's presence as "the metadata table is stale until each affected series
    publishes its next file", not as a power-ingest failure — the power write is unaffected. See
    [Degraded input data](https://openclimatefix.github.io/nged-substation-forecast/live_service/operations/#degraded-input-data-nwp-feed-down-or-telemetry-stalled),
    under "Reading a failed metadata table upsert", for the operational read of this field and what
    a stale metadata table costs while the failure persists.
    """

Attributes

metadata_n_new_TimeSeriesIDs instance-attribute

metadata_n_updated_TimeSeriesIDs instance-attribute

metadata_updated_TimeSeriesIDs instance-attribute

metadata_upsert_failed instance-attribute

Set by the asset when the whole upsert raised, so the power write went ahead without it.

Read this field's presence as "the metadata table is stale until each affected series publishes its next file", not as a power-ingest failure — the power write is unaffected. See Degraded input data, under "Reading a failed metadata table upsert", for the operational read of this field and what a stale metadata table costs while the failure persists.

Functions:

list_timeseries_json_files(store)

List all the timeseries JSON files in NGED's S3 bucket.

Each path encodes start_time, end_time, and time_series_id, and is assumed to be of the form:

timeseries/1774512000000_1774533600000/TimeSeries_23_20260326T080000Z_20260326T140000Z.json

_process_file_listing parses those three fields back out, and its aligned comment traces the regex against this same key.

A key that does not match yields null captures and fails _ProcessedFileListing.validate. That validation failure aborts the listing for the whole bucket rather than skipping the one object. Aborting the whole listing is deliberately stricter than the null-data handling in download_and_parse_files, which logs the offending file and carries on. The difference is where the fault lies. A malformed reading originates upstream of our pipeline, at the meter or in the telemetry export. A key we cannot parse means NGED's naming convention has changed. Every start_time, end_time, and time_series_id this function returns is then suspect — including the values parsed from the keys that still match.

Source code in packages/nged_data/src/nged_data/storage.py

def list_timeseries_json_files(
    store: obstore.store.S3Store,
) -> pt.DataFrame[_ProcessedFileListing]:
    """List all the timeseries JSON files in NGED's S3 bucket.

    Each path encodes `start_time`, `end_time`, and `time_series_id`, and is assumed to be of the
    form:

        timeseries/1774512000000_1774533600000/TimeSeries_23_20260326T080000Z_20260326T140000Z.json

    `_process_file_listing` parses those three fields back out, and its aligned comment traces the
    regex against this same key.

    A key that does not match yields null captures and fails `_ProcessedFileListing.validate`. That
    validation failure aborts the listing for the whole bucket rather than skipping the one object.
    Aborting the whole listing is deliberately stricter than the null-`data` handling in
    `download_and_parse_files`, which logs the offending file and carries on. The difference is
    where the fault lies. A malformed reading originates upstream of our pipeline, at the meter or
    in the telemetry export. A key we cannot parse means NGED's naming convention has changed. Every
    `start_time`, `end_time`, and `time_series_id` this function returns is then suspect — including
    the values parsed from the keys that still match.
    """
    raw_file_listing: list[_RawFileListItem] = []
    total_objects = 0
    for chunk in store.list(prefix="timeseries"):
        # `list()` returns the file listing in chunks of `chunk_size=50` items per chunk.
        for object_meta in chunk:
            total_objects += 1
            if object_meta["path"].endswith(".json"):
                raw_file_listing.append(
                    _RawFileListItem(
                        path=object_meta["path"],
                        filesize_bytes=object_meta["size"],
                        last_modified=object_meta["last_modified"],
                    ),
                )
    log.info(f"JSON files on NGED's S3: {len(raw_file_listing)} out of {total_objects=}")
    return _process_file_listing(raw_file_listing)

read_downloaded_files(list_of_downloaded_files_path, power_table_path, metadata_path, storage_options=None)

Read the bucket listing that the ingest last processed in full.

The list describes files loaded into the power_time_series Delta table and the metadata parquet. If an operator moves the power table or the metadata parquet aside to rebuild that table, the list would record files as loaded whose readings the rebuilt table lacks. The list therefore counts as empty, and the next run downloads every file, whenever the list file, the power table, or the metadata parquet does not exist.

Parameters:

Name	Type	Description	Default
list_of_downloaded_files_path	str	Local path or remote URI of the list's parquet file.	required
power_table_path	str	Local path or remote URI of the power_time_series Delta table.	required
metadata_path	str	Local path or remote URI of the metadata parquet file.	required
storage_options	ObjectStoreOptions | None	Object-store credentials/endpoint for remote paths; None/empty for local paths.	None

Returns:

Type	Description
DataFrame[_DownloadedFiles]	The stored path and last_modified of each file, or an empty frame of the same type.

Raises:

Type	Description
DownloadedFilesError	if the list file exists but is damaged or fails validation. A transient object-store error is not wrapped, so the caller's retry guard can retry it.

Source code in packages/nged_data/src/nged_data/storage.py

def read_downloaded_files(
    list_of_downloaded_files_path: str,
    power_table_path: str,
    metadata_path: str,
    storage_options: ObjectStoreOptions | None = None,
) -> pt.DataFrame[_DownloadedFiles]:
    """Read the bucket listing that the ingest last processed in full.

    The list describes files loaded into the `power_time_series` Delta table and the metadata
    parquet. If an operator moves the power table or the metadata parquet aside to rebuild that
    table, the list would record files as loaded whose readings the rebuilt table lacks. The list
    therefore counts as empty, and the next run downloads every file, whenever the list file, the
    power table, or the metadata parquet does not exist.

    Args:
        list_of_downloaded_files_path: Local path or remote URI of the list's parquet file.
        power_table_path: Local path or remote URI of the `power_time_series` Delta table.
        metadata_path: Local path or remote URI of the metadata parquet file.
        storage_options: Object-store credentials/endpoint for remote paths; ``None``/empty for
            local paths.

    Returns:
        The stored `path` and `last_modified` of each file, or an empty frame of the same type.

    Raises:
        DownloadedFilesError: if the list file exists but is damaged or fails validation. A
            transient object-store error is not wrapped, so the caller's retry guard can retry it.
    """
    if not (
        object_exists(list_of_downloaded_files_path, storage_options)
        and delta_table_exists(power_table_path, storage_options)
        and object_exists(metadata_path, storage_options)
    ):
        log.info(
            f"No usable downloaded-files list at {list_of_downloaded_files_path};"
            " using an empty one."
        )
        return _empty_downloaded_files()
    # A local parquet file can be torn or truncated. An object-store write replaces the object in
    # one request, so a remote file can be off-contract but never torn. A Polars error reading a
    # remote file is therefore a transient object-store error, which the caller's retry guard should
    # retry.
    unreadable_errors: tuple[type[Exception], ...] = (pt.exceptions.DataFrameValidationError,)
    if not is_remote_uri(list_of_downloaded_files_path):
        unreadable_errors += (pl.exceptions.PolarsError, OSError)
    try:
        stored = pl.read_parquet(
            list_of_downloaded_files_path, storage_options=typeddict_to_dict(storage_options)
        )
        return pt.DataFrame(_DownloadedFiles.validate(stored)).set_model(_DownloadedFiles)
    except unreadable_errors as exc:
        raise DownloadedFilesError(
            f"Could not read the downloaded-files list at {list_of_downloaded_files_path}."
            " Delete the file to download every file in NGED's bucket again."
        ) from exc

write_downloaded_files(list_of_downloaded_files_path, listing, storage_options=None)

Replace the downloaded-files list with listing's path and last_modified columns.

A local write goes through a temporary file and Path.replace, so a killed process cannot leave a torn list. A torn list would stop the ingest, whereas a torn metadata parquet does not. An object-store write replaces the object in one request.

Parameters:

Name	Type	Description	Default
list_of_downloaded_files_path	str	Local path or remote URI of the list's parquet file.	required
listing	DataFrame[_ProcessedFileListing]	The whole bucket listing that the run processed.	required
storage_options	ObjectStoreOptions | None	Object-store credentials/endpoint for a remote path; None/empty for a local path.	None

Source code in packages/nged_data/src/nged_data/storage.py

def write_downloaded_files(
    list_of_downloaded_files_path: str,
    listing: pt.DataFrame[_ProcessedFileListing],
    storage_options: ObjectStoreOptions | None = None,
) -> None:
    """Replace the downloaded-files list with `listing`'s `path` and `last_modified` columns.

    A local write goes through a temporary file and `Path.replace`, so a killed process cannot leave
    a torn list. A torn list would stop the ingest, whereas a torn metadata parquet does not. An
    object-store write replaces the object in one request.

    Args:
        list_of_downloaded_files_path: Local path or remote URI of the list's parquet file.
        listing: The whole bucket listing that the run processed.
        storage_options: Object-store credentials/endpoint for a remote path; ``None``/empty for a
            local path.
    """
    downloaded_files = _DownloadedFiles.validate(
        listing.select("path", "last_modified").sort("path")
    )
    options = typeddict_to_dict(storage_options)
    if is_remote_uri(list_of_downloaded_files_path):
        downloaded_files.write_parquet(
            list_of_downloaded_files_path, compression="zstd", storage_options=options
        )
        return
    if_local_path_then_make_parent_dir(list_of_downloaded_files_path)
    temporary_path = f"{list_of_downloaded_files_path}.tmp"
    downloaded_files.write_parquet(temporary_path, compression="zstd")
    Path(temporary_path).replace(list_of_downloaded_files_path)

select_files_not_yet_downloaded(file_listing, downloaded_files)

Keep the listed files whose (path, last_modified) the downloaded-files list lacks.

A file whose path is known but whose last_modified changed was rewritten by NGED, so it is selected again. A late file is selected too: a file absent from an earlier listing is absent from the list, so a later listing selects it.

Parameters:

Name	Type	Description	Default
file_listing	DataFrame[_ProcessedFileListing]	The whole bucket listing.	required
downloaded_files	DataFrame[_DownloadedFiles]	The listing that the ingest last processed in full.	required

Returns:

Type	Description
DataFrame[_ProcessedFileListing]	The selected files, in no particular order.

Source code in packages/nged_data/src/nged_data/storage.py

def select_files_not_yet_downloaded(
    file_listing: pt.DataFrame[_ProcessedFileListing],
    downloaded_files: pt.DataFrame[_DownloadedFiles],
) -> pt.DataFrame[_ProcessedFileListing]:
    """Keep the listed files whose `(path, last_modified)` the downloaded-files list lacks.

    A file whose path is known but whose `last_modified` changed was rewritten by NGED, so it is
    selected again. A late file is selected too: a file absent from an earlier listing is absent
    from the list, so a later listing selects it.

    Args:
        file_listing: The whole bucket listing.
        downloaded_files: The listing that the ingest last processed in full.

    Returns:
        The selected files, in no particular order.
    """
    # Strip the Patito model so Polars' cross-subclass join check accepts the right-hand frame.
    plain_downloaded_files = pl.DataFrame._from_pydf(downloaded_files._df)
    selected = file_listing.join(plain_downloaded_files, on=["path", "last_modified"], how="anti")
    return pt.DataFrame(selected).set_model(_ProcessedFileListing).validate()

download_and_parse_files(store, paths_df)

Download and parse each listed file, in ascending end_time order.

Two files can cover overlapping periods for the same time_series_id. The function sorts its input by end_time itself. Processing in that order means the more recent file's readings overwrite the older file's duplicate rows, in the unique(..., keep="last") dedupes below.

The function works through the sorted input in chunks of _DOWNLOAD_CHUNK_FILES. Per chunk it fetches all the files concurrently, at most _MAX_REQUESTS_IN_FLIGHT at a time, through asyncio.run, then parses them sequentially in input order. The function is synchronous and needs a thread with no running event loop, which a synchronous Dagster asset has.

Parameters:

Name	Type	Description	Default
store	S3Store	The NGED S3 bucket to download each file from.	required
paths_df	DataFrame[_ProcessedFileListing]	The files to process, typically the output of select_files_not_yet_downloaded.	required

Returns:

Type	Description
DownloadAndParseResult	A DownloadAndParseResult bundling every file's parsed TimeSeriesMetadata and
DownloadAndParseResult	PowerTimeSeries rows, deduplicated across files — see DownloadAndParseResult's
DownloadAndParseResult	docstring for what each field holds. When every file was data-less (NoReadingsInFile),
DownloadAndParseResult	the metadata is still returned and the power frame is empty.

Raises:

Type	Description
NgedFileParseError	if a file is malformed or breaks the TimeSeriesMetadata contract. The message names the file's path. The first failing request also raises, out of asyncio.run.

Source code in packages/nged_data/src/nged_data/storage.py

def download_and_parse_files(
    store: obstore.store.S3Store, paths_df: pt.DataFrame[_ProcessedFileListing]
) -> DownloadAndParseResult:
    """Download and parse each listed file, in ascending `end_time` order.

    Two files can cover overlapping periods for the same `time_series_id`. The function sorts its
    input by `end_time` itself. Processing in that order means the more recent file's readings
    overwrite the older file's duplicate rows, in the `unique(..., keep="last")` dedupes below.

    The function works through the sorted input in chunks of `_DOWNLOAD_CHUNK_FILES`. Per chunk it
    fetches all the files concurrently, at most `_MAX_REQUESTS_IN_FLIGHT` at a time, through
    `asyncio.run`, then parses them sequentially in input order. The function is synchronous and
    needs a thread with no running event loop, which a synchronous Dagster asset has.

    Args:
        store: The NGED S3 bucket to download each file from.
        paths_df: The files to process, typically the output of `select_files_not_yet_downloaded`.

    Returns:
        A `DownloadAndParseResult` bundling every file's parsed `TimeSeriesMetadata` and
        `PowerTimeSeries` rows, deduplicated across files — see `DownloadAndParseResult`'s
        docstring for what each field holds. When every file was data-less (`NoReadingsInFile`),
        the metadata is still returned and the power frame is empty.

    Raises:
        NgedFileParseError: if a file is malformed or breaks the `TimeSeriesMetadata` contract. The
            message names the file's path. The first failing request also raises, out of
            `asyncio.run`.
    """
    paths = paths_df.sort("end_time", "path")["path"].to_list()
    metadata_dfs: list[pt.DataFrame[TimeSeriesMetadata]] = []
    power_time_series_dfs: list[pt.DataFrame[PowerTimeSeries]] = []
    n_implausible_power_rows_dropped = 0
    for chunk_start in range(0, len(paths), _DOWNLOAD_CHUNK_FILES):
        chunk_paths = paths[chunk_start : chunk_start + _DOWNLOAD_CHUNK_FILES]
        chunk_bytes = asyncio.run(_fetch_all(store, chunk_paths))
        for path, json_bytes in zip(chunk_paths, chunk_bytes, strict=True):
            metadata, extracted = _parse_file(path, json_bytes)
            metadata_dfs.append(metadata)
            if extracted is not None:
                power_time_series_dfs.append(extracted.dataframe)
                n_implausible_power_rows_dropped += extracted.n_dropped

    log.info(
        f"{len(metadata_dfs)} new TimeSeriesMetadata DataFrames and {len(power_time_series_dfs)}"
        " new PowerTimeSeries dataframes extracted from NGED JSON data."
    )

    metadata_df = (
        pl.concat(metadata_dfs, how="diagonal")
        .unique(subset="time_series_id", keep="last")
        .sort("time_series_id")
    )
    if power_time_series_dfs:
        time_series_df = (
            pl.concat(power_time_series_dfs)
            .unique(subset=["time_series_id", "time"], keep="last")
            .sort(by=PowerTimeSeries.columns_to_sort_by)
        )
    else:
        time_series_df = pl.DataFrame(
            schema={name: PowerTimeSeries.dtypes[name] for name in PowerTimeSeries.columns}
        )

    return DownloadAndParseResult(
        metadata=TimeSeriesMetadata.validate(metadata_df),
        power_time_series=PowerTimeSeries.validate(time_series_df),
        n_implausible_power_rows_dropped=n_implausible_power_rows_dropped,
    )

time_series_coverage(delta_path, storage_options=None)

Return the earliest/latest observation time on disk per time_series_id.

Returns an empty (but correctly typed) frame if the Delta table does not exist yet. min/max grouped by time_series_id are value aggregations, so they are safe from the Polars 32-bit row-count wraparound even on a very large table (see https://openclimatefix.github.io/nged-substation-forecast/architecture/code-style/#data-handling).

Cost: a full two-column scan-and-aggregate, O(rows in the table). Projection pushdown drops the power column. A group-wise min/max cannot be answered from Parquet row-group statistics, because no engine on our stack does aggregate-from-statistics, so every time/time_series_id value is read. Computing both bounds instead of one takes ~20% more wall-clock time and no extra memory, because the shared scan dominates.

The collect uses the streaming engine to keep peak memory bounded, because this scan runs hourly on a small control-plane VM, in the power_data_is_fresh asset check. The ingest does not run time_series_coverage: select_new_rows uses _existing_power_time_series_keys instead, a scan restricted to the reporting series' own history. The measurement used a synthetic V2 table: 2,500 series, half-hourly, partitioned by time_series_id, holding a year of history (43.8M rows). The streaming engine took ~0.21 s at ~190 MB peak. The in-memory engine peaked at ~1.3 GB for the same result, so streaming uses ~7x less memory.

Cost scales linearly with accumulated history. If the scan ever becomes a problem, both bounds can instead be read from the Delta add-action min.time/max.time file statistics. Delta's transaction log records one add-action per data file, carrying that file's per-column minimum and maximum. That read is metadata-only, O(files): ~0.02 s and <100 MB at the same scale. The statistics read is the same Delta-log-metadata trick used to count whole-table rows without scanning.

delta_path is a local path or remote URI for the power_time_series Delta table; storage_options carries the object-store credentials/endpoint for a remote delta_path.

Source code in packages/nged_data/src/nged_data/storage.py

def time_series_coverage(
    delta_path: str,
    storage_options: ObjectStoreOptions | None = None,
) -> pt.DataFrame[TimeSeriesCoverage]:
    """Return the earliest/latest observation ``time`` on disk per ``time_series_id``.

    Returns an empty (but correctly typed) frame if the Delta table does not exist yet.
    ``min``/``max`` grouped by ``time_series_id`` are value aggregations, so they are safe from
    the Polars 32-bit row-count wraparound even on a very large table (see
    <https://openclimatefix.github.io/nged-substation-forecast/architecture/code-style/#data-handling>).

    Cost: a full two-column scan-and-aggregate, O(rows in the table). Projection pushdown drops
    the ``power`` column. A group-wise ``min``/``max`` cannot be answered from Parquet row-group
    statistics, because no engine on our stack does aggregate-from-statistics, so every
    ``time``/``time_series_id`` value is read. Computing both bounds instead of one takes ~20%
    more wall-clock time and no extra memory, because the shared scan dominates.

    The ``collect`` uses the streaming engine to keep peak memory bounded, because this scan runs
    hourly on a small control-plane VM, in the ``power_data_is_fresh`` asset check. The ingest does
    not run ``time_series_coverage``: ``select_new_rows`` uses ``_existing_power_time_series_keys``
    instead, a scan restricted to the reporting series' own history. The measurement used a
    synthetic V2 table: 2,500 series, half-hourly, partitioned by ``time_series_id``, holding a
    year of history (43.8M rows). The streaming engine took ~0.21 s at ~190 MB peak. The in-memory
    engine peaked at ~1.3 GB for the same result, so streaming uses ~7x less memory.

    Cost scales linearly with accumulated history. If the scan ever becomes a problem, both
    bounds can instead be read from the Delta add-action ``min.time``/``max.time`` file
    statistics. Delta's transaction log records one add-action per data file, carrying that
    file's per-column minimum and maximum. That read is metadata-only, O(files): ~0.02 s and <100
    MB at the same scale. The statistics read is the same Delta-log-metadata trick used to count
    whole-table rows without scanning.

    `delta_path` is a local path or remote URI for the ``power_time_series`` Delta table;
    `storage_options` carries the object-store credentials/endpoint for a remote `delta_path`.
    """
    if not delta_table_exists(delta_path, storage_options):
        log.info(f"{delta_path=} does not exist yet; returning an empty coverage frame.")
        empty = pl.DataFrame(
            schema={name: TimeSeriesCoverage.dtypes[name] for name in TimeSeriesCoverage.columns}
        )
        return pt.DataFrame(empty).set_model(TimeSeriesCoverage).validate()

    return coverage_from_power(
        pt.LazyFrame.from_existing(
            pl.scan_delta(delta_path, storage_options=typeddict_to_dict(storage_options))
        ).set_model(PowerTimeSeries)
    )

coverage_from_power(power)

Return the earliest/latest observation time per time_series_id of a power frame.

The aggregation behind time_series_coverage, split out so a caller can compute coverage of any scan of power, for example the unflagged rows of the cleaned table (scan_cleaned_power).

Parameters:

Name	Type	Description	Default
power	LazyFrame[PowerTimeSeries]	Lazy power observations; only time_series_id and time are read.	required

Returns:

Type	Description
DataFrame[TimeSeriesCoverage]	One row per time_series_id, validated against TimeSeriesCoverage.

Source code in packages/nged_data/src/nged_data/storage.py

def coverage_from_power(power: pt.LazyFrame[PowerTimeSeries]) -> pt.DataFrame[TimeSeriesCoverage]:
    """Return the earliest/latest observation ``time`` per ``time_series_id`` of a power frame.

    The aggregation behind `time_series_coverage`, split out so a caller can compute coverage of
    any scan of power, for example the unflagged rows of the cleaned table (`scan_cleaned_power`).

    Args:
        power: Lazy power observations; only `time_series_id` and `time` are read.

    Returns:
        One row per `time_series_id`, validated against `TimeSeriesCoverage`.
    """
    coverage = (
        pl.LazyFrame._from_pyldf(power._ldf)
        .group_by("time_series_id")
        .agg(first_time=pl.min("time"), last_time=pl.max("time"))
        # Streaming engine: bounds peak memory (~7x lower than in-memory at V2 scale) so the
        # hourly full-table aggregate stays comfortable on a small control-plane VM. See
        # `time_series_coverage`'s docstring.
        .collect(engine="streaming")
    )
    log.info(
        f"Found on-disk coverage for {coverage.height} time_series_ids."
        f" {coverage['last_time'].min()=}. {coverage['last_time'].max()=}"
    )
    return pt.DataFrame(coverage).set_model(TimeSeriesCoverage).validate()

scan_cleaned_power(delta_path, storage_options=None)

Scan the cleaned power table, keeping only the rows that no cleaning rule flagged.

scan_cleaned_power is the one read path for every consumer of observed power except the ingest and its freshness check, so no consumer can forget the drop_reason filter. The raw table stays the record of what NGED delivered.

Parameters:

Name	Type	Description	Default
delta_path	str	Path or URI of the cleaned_power_time_series Delta table.	required
storage_options	ObjectStoreOptions | None	delta-rs object-store options for a remote delta_path.	None

Returns:

Type	Description
LazyFrame[PowerTimeSeries]	A lazy frame with the three PowerTimeSeries columns, whose drop_reason was null.

Source code in packages/nged_data/src/nged_data/storage.py

def scan_cleaned_power(
    delta_path: str,
    storage_options: ObjectStoreOptions | None = None,
) -> pt.LazyFrame[PowerTimeSeries]:
    """Scan the cleaned power table, keeping only the rows that no cleaning rule flagged.

    `scan_cleaned_power` is the one read path for every consumer of observed power except the ingest
    and its freshness check, so no consumer can forget the `drop_reason` filter. The raw table stays
    the record of what NGED delivered.

    Args:
        delta_path: Path or URI of the `cleaned_power_time_series` Delta table.
        storage_options: delta-rs object-store options for a remote `delta_path`.

    Returns:
        A lazy frame with the three `PowerTimeSeries` columns, whose `drop_reason` was null.
    """
    scan = pl.scan_delta(delta_path, storage_options=typeddict_to_dict(storage_options))
    unflagged = scan.filter(pl.col("drop_reason").is_null()).select(*PowerTimeSeries.columns)
    return pt.LazyFrame.from_existing(unflagged).set_model(PowerTimeSeries)

select_new_rows(time_series, delta_path, storage_options=None)

Return rows in time_series genuinely missing from the Delta table.

The filter is an existence check: an anti-join on (time_series_id, time) against _existing_power_time_series_keys. A late file is therefore ingested even when a later reading for the same series is already on disk. So is a file that fills a gap earlier in a series' history. The existence check also makes a re-download safe, because a rewritten file or a crash between the append and the downloaded-files list write can offer readings the table already holds. See _existing_power_time_series_keys's docstring for the cost the existence check trades in return.

When the Delta table does not exist yet, a call returns its input unchanged and scans nothing.

Parameters:

Name	Type	Description	Default
time_series	DataFrame[PowerTimeSeries]	The parsed rows to check.	required
delta_path	str	Local path or remote URI for the power_time_series Delta table.	required
storage_options	ObjectStoreOptions | None	Object-store credentials/endpoint for a remote delta_path.	None

Returns:

Type	Description
DataFrame[PowerTimeSeries]	The rows of time_series that the Delta table lacks, in PowerTimeSeries sort order.

Source code in packages/nged_data/src/nged_data/storage.py

def select_new_rows(
    time_series: pt.DataFrame[PowerTimeSeries],
    delta_path: str,
    storage_options: ObjectStoreOptions | None = None,
) -> pt.DataFrame[PowerTimeSeries]:
    """Return rows in `time_series` genuinely missing from the Delta table.

    The filter is an existence check: an anti-join on `(time_series_id, time)` against
    `_existing_power_time_series_keys`. A late file is therefore ingested even when a later reading
    for the same series is already on disk. So is a file that fills a gap earlier in a series'
    history. The existence check also makes a re-download safe, because a rewritten file or a
    crash between the append and the downloaded-files list write can offer readings the table
    already holds. See
    `_existing_power_time_series_keys`'s docstring for the cost the existence check trades in
    return.

    When the Delta table does not exist yet, a call returns its input unchanged and scans nothing.

    Args:
        time_series: The parsed rows to check.
        delta_path: Local path or remote URI for the ``power_time_series`` Delta table.
        storage_options: Object-store credentials/endpoint for a remote `delta_path`.

    Returns:
        The rows of `time_series` that the Delta table lacks, in `PowerTimeSeries` sort order.
    """
    if not delta_table_exists(delta_path, storage_options):
        log.info(f"{delta_path=} does not exist yet.")
        return time_series

    reporting_ids = time_series["time_series_id"].unique().to_list()
    existing_keys = _existing_power_time_series_keys(delta_path, storage_options, reporting_ids)
    filtered_df = (
        time_series.lazy()
        .join(existing_keys, on=["time_series_id", "time"], how="anti")
        .sort(by=PowerTimeSeries.columns_to_sort_by)
        .collect()
    )
    return pt.DataFrame(filtered_df).set_model(PowerTimeSeries).validate()

upsert_metadata(new_metadata, metadata_path, storage_options=None)

Upserts metadata to a Parquet file, keeping the newest version of each time series.

If the Parquet file does not exist, it saves the new_metadata. If it exists, it merges the new_metadata into it and rewrites the file only if the incoming metadata differs from what is stored. new_metadata is a snapshot of the metadata for the series that reported this run. The snapshot need not carry the same columns, or the same column order, as the stored metadata table. Rows are matched on time_series_id. A series that new_metadata covers is replaced wholesale, so a field the snapshot has stopped carrying is cleared for that series. A series that new_metadata omits keeps its last stored values indefinitely. The metadata table therefore holds every time series we have ever seen, not only the series in the latest snapshot.

This function is not safe under concurrent callers: it assumes it is called by one thread at a time, and takes no lock.

The rewrite is not atomic either. write_parquet overwrites the metadata table in place, with no write-to-temporary-file-and-rename. The metadata table therefore does not get the all-or-nothing commit that Delta gives the tables around it. See principle 10, every write is atomic and idempotent. A crash or an out-of-memory kill part-way through a local write leaves a partial file. pl.read_parquet below is what rejects that partial file on the next run, before TimeSeriesMetadata.validate ever sees it. The error reads ComputeError: parquet: File out of specification: The file must end with PAR1. validate is the guard for the other case: a metadata table that reads back cleanly but is off-contract, from an older writer or a hand-edit. Either way the asset records metadata_upsert_failed, and the metadata table stays broken until an operator acts. A corrupt file is not a missing file, so the create branch below never runs again by itself.

Deleting the metadata parquet makes the next run rebuild the metadata table. read_downloaded_files counts the downloaded-files list as empty when the metadata table does not exist, so the next run downloads every file in NGED's bucket and the metadata table gets every series NGED publishes.

Parameters:

Name	Type	Description	Default
new_metadata	DataFrame[TimeSeriesMetadata]	The new metadata DataFrame.	required
metadata_path	str	Local path or remote URI of the Parquet file where we store our version of the metadata.	required
storage_options	ObjectStoreOptions | None	Object-store credentials/endpoint for a remote metadata_path; None/empty for a local path.	None

Returns:

Type	Description
UpsertMetadataStats	An UpsertMetadataStats. metadata_n_new_TimeSeriesIDs counts the time_series_ids in
UpsertMetadataStats	new_metadata that are new to the metadata table. metadata_n_updated_TimeSeriesIDs counts
UpsertMetadataStats	the time_series_ids already in the metadata table that changed.
UpsertMetadataStats	metadata_updated_TimeSeriesIDs holds the sorted list of changed time_series_ids. Both
UpsertMetadataStats	counts are zero and the id list is omitted when the parquet file was up to date already. The
UpsertMetadataStats	id list is also omitted on a first-ever write, when every id in new_metadata counts as new
UpsertMetadataStats	rather than updated.

Source code in packages/nged_data/src/nged_data/storage.py

def upsert_metadata(
    new_metadata: pt.DataFrame[TimeSeriesMetadata],
    metadata_path: str,
    storage_options: ObjectStoreOptions | None = None,
) -> UpsertMetadataStats:
    """Upserts metadata to a Parquet file, keeping the newest version of each time series.

    If the Parquet file does not exist, it saves the new_metadata. If it exists, it merges the
    new_metadata into it and rewrites the file only if the incoming metadata differs from what is
    stored. ``new_metadata`` is a snapshot of the metadata for the series that reported this run.
    The snapshot need not carry the same columns, or the same column order, as the stored metadata
    table. Rows are matched on ``time_series_id``. A series that ``new_metadata`` covers is replaced
    wholesale, so a field the snapshot has stopped carrying is **cleared** for that series. A series
    that ``new_metadata`` omits keeps its last stored values indefinitely. The metadata table
    therefore holds every time series we have ever seen, not only the series in the latest snapshot.

    This function is not safe under concurrent callers: it assumes it is called by one thread at a
    time, and takes no lock.

    The rewrite is not atomic either. `write_parquet` overwrites the metadata table in place, with
    no write-to-temporary-file-and-rename. The metadata table therefore does not get the
    all-or-nothing commit that Delta gives the tables around it. See [principle 10, every write is
    atomic and idempotent](https://openclimatefix.github.io/nged-substation-forecast/design-philosophy/design-principles/#10-every-write-is-atomic-and-idempotent-and-every-failure-is-confined-to-one-partition).
    A crash or an out-of-memory kill part-way through a local write leaves a partial file.
    `pl.read_parquet` below is what rejects that partial file on the next run, before
    `TimeSeriesMetadata.validate` ever sees it. The error reads `ComputeError: parquet: File out of
    specification: The file must end with PAR1`. `validate` is the guard for the other case: a
    metadata table that reads back cleanly but is off-contract, from an older writer or a hand-edit.
    Either way the asset records `metadata_upsert_failed`, and the metadata table stays broken until
    an operator acts. A corrupt file is not a missing file, so the create branch below never runs
    again by itself.

    Deleting the metadata parquet makes the next run rebuild the metadata table.
    `read_downloaded_files` counts the downloaded-files list as empty
    when the metadata table does not exist, so the next run downloads every file in NGED's bucket
    and the metadata table gets every series NGED publishes.

    Args:
        new_metadata: The new metadata DataFrame.
        metadata_path: Local path or remote URI of the Parquet file where we store our version
            of the metadata.
        storage_options: Object-store credentials/endpoint for a remote `metadata_path`;
            ``None``/empty for a local path.

    Returns:
        An `UpsertMetadataStats`. `metadata_n_new_TimeSeriesIDs` counts the `time_series_id`s in
        `new_metadata` that are new to the metadata table. `metadata_n_updated_TimeSeriesIDs` counts
        the `time_series_id`s already in the metadata table that changed.
        `metadata_updated_TimeSeriesIDs` holds the sorted list of changed `time_series_id`s. Both
        counts are zero and the id list is omitted when the parquet file was up to date already. The
        id list is also omitted on a first-ever write, when every id in `new_metadata` counts as new
        rather than updated.
    """
    COMPRESSION: Final[str] = "zstd"

    # The annotation is not enforced at runtime and this is the package's only public entry point,
    # so check the caller's snapshot rather than trust it.
    new_metadata = TimeSeriesMetadata.validate(new_metadata.sort("time_series_id"))

    if not object_exists(metadata_path, storage_options):
        log.info(f"Metadata file not found at {metadata_path}. Creating new file.")
        # write_parquet doesn't create missing parent directories, so a first-ever run against a
        # fresh local data root would fail here. This create branch runs before any Delta write that
        # would otherwise create the dir. Create the parent for a local metadata_path.
        if_local_path_then_make_parent_dir(metadata_path)
        new_metadata.write_parquet(
            metadata_path,
            compression=COMPRESSION,
            storage_options=typeddict_to_dict(storage_options),
        )
        return UpsertMetadataStats(
            metadata_n_new_TimeSeriesIDs=new_metadata.height,
            metadata_n_updated_TimeSeriesIDs=0,
        )

    existing_metadata = pl.read_parquet(
        metadata_path, storage_options=typeddict_to_dict(storage_options)
    )
    # The stored metadata table is outside this code's control: it can come from an older writer, a
    # hand-edit, or a truncated upload. An off-contract file must therefore not be merged blind into
    # the metadata table we write back. As with any raise from this function, the asset contains it
    # rather than failing: it records `metadata_upsert_failed` and lets the power write proceed (see
    # `defs/assets.py`).
    TimeSeriesMetadata.validate(existing_metadata)

    # `how="diagonal"` because the snapshot and the stored metadata table can differ in both width
    # and column order: four TimeSeriesMetadata fields are `allow_missing`. Aligning the two frames
    # into one also makes the `hash_rows` diff below insensitive to the stored column order. Hashing
    # the two frames separately would not be.
    combined = pl.concat([new_metadata, existing_metadata], how="diagonal")
    new_rows = combined.head(new_metadata.height)
    stored_rows = combined.slice(new_metadata.height)

    # Compare metadata. `metadata_diff` contains all rows in `new_metadata` that do not have an
    # exact match in `existing_metadata`. Adapted from https://stackoverflow.com/a/79888719
    metadata_diff = new_rows.filter(~new_rows.hash_rows().is_in(stored_rows.hash_rows().implode()))
    # The first frame carrying the union of both inputs' columns: the concat adds to the snapshot's
    # rows any `allow_missing` field only the stored metadata table had. All four fields are
    # nullable, so this validate is a shape check on a frame neither validation above saw, not a
    # guard against a known fault. Of the four validate calls in this function this is the weakest,
    # and the first to reconsider if the validate calls get trimmed.
    TimeSeriesMetadata.validate(metadata_diff)

    if metadata_diff.is_empty():
        log.info("TimeSeriesMetadata is up to date.")
        return UpsertMetadataStats(
            metadata_n_new_TimeSeriesIDs=0,
            metadata_n_updated_TimeSeriesIDs=0,
        )

    log.info(
        f"New TimeSeriesMetadata available for {metadata_diff.height} timeseries_ids."
        f" Updating {metadata_path}."
    )

    # Merge metadata. Put new_metadata first so that unique(keep="first") keeps the new version
    merged_metadata = combined.unique(subset="time_series_id", keep="first").sort("time_series_id")

    # The last gate before the stored metadata table is overwritten. `unique` draws rows from both
    # sides of the concat, so no validation above has seen this row set.
    TimeSeriesMetadata.validate(merged_metadata)

    merged_metadata.write_parquet(
        metadata_path, compression=COMPRESSION, storage_options=typeddict_to_dict(storage_options)
    )

    # Compute stats
    new_ids = set(new_metadata["time_series_id"]) - set(existing_metadata["time_series_id"])
    updated_ids = list(
        set(metadata_diff["time_series_id"]).intersection(existing_metadata["time_series_id"])
    )
    return UpsertMetadataStats(
        metadata_n_new_TimeSeriesIDs=len(new_ids),
        metadata_n_updated_TimeSeriesIDs=len(updated_ids),
        metadata_updated_TimeSeriesIDs=sorted(updated_ids),
    )

nged_data.cleaning

Flagging NGED's power telemetry that should not be trained or forecast on.

flag_nged_power is where cleaning rules live. The clean_nged_power_data Dagster asset calls it with the whole raw power_time_series table and writes the result to the cleaned_power_time_series Delta table.

Attributes

CLEANING_CODE_HASH = hashlib.sha256(Path(__file__).read_bytes() + Path(contracts.power_schemas.__file__).read_bytes()).hexdigest() module-attribute

The SHA-256 of this module's source file and of contracts/power_schemas.py, computed at import.

power_schemas.py is included because the substation_zero rule reads TimeSeriesMetadata.SUBSTATION_TYPES from it.

clean_nged_power_data skips a rebuild only when this hash matches the one recorded in the cleaned table's newest write commit, so any edit to a cleaning rule forces a rebuild on the next run, whether or not the edit is committed. Every file a cleaning rule reads its code or constants from must be in this hash.

Classes

Functions:

flag_nged_power(power, metadata)

Flag the rows of NGED's power telemetry that cleaning rejects, by adding drop_reason.

Where this sits. clean_nged_power_data calls this function with the whole raw power table and the TimeSeriesMetadata table, about four times a day, and writes the result to the cleaned table. Every reader of power then sees only the rows whose drop_reason is null.

How to add a rule. Write a boolean Polars expression that is true for a row to flag. Add the expression as one more .when(...).then(pl.lit("<reason>")) branch of the chain below, and add the reason to contracts.power_schemas.DROP_REASONS. The first matching branch wins. A rule needing another TimeSeriesMetadata column adds that column to the select below.

The contract a rule keeps. Return every input row exactly once, flagged rather than deleted. Never change time_series_id or time. Do not call collect inside this function.

Logging. clean_nged_power_data reports to Dagster, per reason, the rows and series flagged and the range of flagged power and time, so a rule needs no logging code. Python's logging output reaches only the step's captured stderr.

The one rule today, substation_zero, flags a reading of exactly 0 from a Primary, BSP, or GSP series. A substation almost never truly reads zero, so a zero is almost always a telemetry fault. A series missing from the TimeSeriesMetadata table has no substation_type and is not flagged.

Example. The output holds every input power row, with a drop_reason column added. These seven rows are real readings from a Primary substation, MARSH LANE 33 11kV S STN, after cleaning:

time_series_id  time (UTC)        power (MVA)  drop_reason
Int32           Datetime          Float32      String
26              2026-03-05 12:30  0.043        null
26              2026-03-05 13:00  0.094        null
26              2026-03-05 13:30  0.045        null
26              2026-03-05 14:00  0.0          substation_zero
26              2026-03-05 14:30  0.392        null
26              2026-03-05 15:00  1.492        null
26              2026-03-05 15:30  2.585        null

The 14:00 reading of exactly 0 MVA is flagged. The flagged row stays in the output. Every reader of power skips the flagged row, so the zero never reaches training, cross-validation, scoring, or live forecasts. A zero from an EHV Customer or HV Customer series would get a null drop_reason and stay in, because a customer site, such as a solar farm at night, can truly read zero. CleanedPowerTimeSeries in contracts.power_schemas defines these four columns and their dtypes.

Parameters:

Name	Type	Description	Default
power	LazyFrame[PowerTimeSeries]	The whole raw power table.	required
metadata	DataFrame[TimeSeriesMetadata]	The TimeSeriesMetadata table.	required

Returns:

Type	Description
LazyFrame[CleanedPowerTimeSeries]	Every row of power, in no guaranteed order, with a drop_reason column: null for a row
LazyFrame[CleanedPowerTimeSeries]	that passed, otherwise the name of the rule that flagged it.

Source code in packages/nged_data/src/nged_data/cleaning.py

def flag_nged_power(
    power: pt.LazyFrame[PowerTimeSeries],
    metadata: pt.DataFrame[TimeSeriesMetadata],
) -> pt.LazyFrame[CleanedPowerTimeSeries]:
    """Flag the rows of NGED's power telemetry that cleaning rejects, by adding `drop_reason`.

    **Where this sits.** `clean_nged_power_data` calls this function with the whole raw power
    table and the `TimeSeriesMetadata` table, about four times a day, and writes the result to the
    cleaned table. Every reader of power then sees only the rows whose `drop_reason` is null.

    **How to add a rule.** Write a boolean Polars expression that is true for a row to flag. Add
    the expression as one more `.when(...).then(pl.lit("<reason>"))` branch of the chain below, and
    add the reason to `contracts.power_schemas.DROP_REASONS`. The first matching branch wins. A
    rule needing another `TimeSeriesMetadata` column adds that column to the `select` below.

    **The contract a rule keeps.** Return every input row exactly once, flagged rather than
    deleted. Never change `time_series_id` or `time`. Do not call `collect` inside this function.

    **Logging.** `clean_nged_power_data` reports to Dagster, per reason, the rows and series
    flagged and the range of flagged power and time, so a rule needs no logging code. Python's
    `logging` output reaches only the step's captured stderr.

    The one rule today, `substation_zero`, flags a reading of exactly 0 from a `Primary`, `BSP`,
    or `GSP` series. A substation almost never truly reads zero, so a zero is almost always a
    telemetry fault. A series missing from the `TimeSeriesMetadata` table has no `substation_type`
    and is not flagged.

    **Example.** The output holds every input power row, with a `drop_reason` column added. These
    seven rows are real readings from a `Primary` substation, MARSH LANE 33 11kV S STN, after
    cleaning:

    ```text
    time_series_id  time (UTC)        power (MVA)  drop_reason
    Int32           Datetime          Float32      String
    26              2026-03-05 12:30  0.043        null
    26              2026-03-05 13:00  0.094        null
    26              2026-03-05 13:30  0.045        null
    26              2026-03-05 14:00  0.0          substation_zero
    26              2026-03-05 14:30  0.392        null
    26              2026-03-05 15:00  1.492        null
    26              2026-03-05 15:30  2.585        null
    ```

    The 14:00 reading of exactly 0 MVA is flagged. The flagged row stays in the output. Every reader
    of power skips the flagged row, so the zero never reaches training, cross-validation, scoring,
    or live forecasts. A zero from an `EHV Customer` or `HV Customer` series would get a null
    `drop_reason` and stay in, because a customer site, such as a solar farm at night, can truly
    read zero. `CleanedPowerTimeSeries` in `contracts.power_schemas` defines these four columns and
    their dtypes.

    Args:
        power: The whole raw power table.
        metadata: The `TimeSeriesMetadata` table.

    Returns:
        Every row of `power`, in no guaranteed order, with a `drop_reason` column: null for a row
        that passed, otherwise the name of the rule that flagged it.
    """
    # Join the `TimeSeriesMetadata` columns the cleaning rules need onto the power rows. An eager
    # `select` returns a plain Polars frame. The power frame is stripped of its `PowerTimeSeries`
    # model, which would otherwise carry through the join to a frame with different columns.
    metadata_columns = metadata.select("time_series_id", "substation_type").lazy()
    joined = pl.LazyFrame._from_pyldf(power._ldf).join(
        metadata_columns, on="time_series_id", how="left"
    )
    # `joined` has one row per power reading, with the columns `time_series_id`, `time`, `power`,
    # and `substation_type` (null for a series missing from the `TimeSeriesMetadata` table).

    flagged = joined.with_columns(
        # `pl.when(condition).then(value)` sets `drop_reason` to `value` on every row where
        # `condition` is true. Add a rule as another `.when(...).then(...)` before `.otherwise`.
        drop_reason=pl.when(
            # The `substation_zero` rule: the reading is exactly zero, and the series is a
            # substation. `&` is a row-by-row "and"; `is_in` is true where `substation_type` is
            # one of the listed values, and null (so not flagged) where `substation_type` is null.
            (pl.col("power") == 0)
            & pl.col("substation_type").is_in(TimeSeriesMetadata.SUBSTATION_TYPES)
        )
        .then(pl.lit("substation_zero"))
        # Every row no rule flagged gets a null `drop_reason`. A bare `pl.lit(None)` has the `Null`
        # dtype, which would not match the contract's String.
        .otherwise(pl.lit(None, dtype=pl.String))
    ).select("time_series_id", "time", "power", "drop_reason")
    return pt.LazyFrame.from_existing(flagged).set_model(CleanedPowerTimeSeries)