Tutorial: verification-games workflow recipes

These recipes condense common ogcat patterns from the verification-games notebooks into small copy/paste examples. They use temporary directories and tiny synthetic local data, so they do not depend on the verification-games repositories or shared ACRG data paths.

The examples are written as notebook-style cells that can be run in order. The xarray cells are optional; they require xarray and a local NetCDF or Zarr backend in the active environment.

Create and reopen a local catalog

Start by creating a tiny catalog with named schemas for raw fluxes, derived fluxes, and model output. The schema fields are intentionally broad: they make common search and display workflows easy without forcing every project to use the same metadata vocabulary.

from pathlib import Path
from tempfile import TemporaryDirectory

from ogcat import (
    ArtifactLocator,
    Catalog,
    CatalogSpec,
    MetadataFieldDescription,
    RecordSchema,
)
from ogcat.example_data import (
    EXAMPLE_FLUX_FILENAME,
    example_flux_cdl,
    example_flux_collection_path,
    write_example_flux_netcdf_or_placeholder,
)
from ogcat.writers import memory_source, source_writer


def metadata_field(name: str, description: str, *, required: bool = False) -> MetadataFieldDescription:
    """Return a concise metadata field description."""
    return MetadataFieldDescription(name=name, description=description, required=required)


common_fields = [
    metadata_field("title", "Short human-readable description."),
    metadata_field("product", "Product, model, or inventory family.", required=True),
    metadata_field("species", "Species or atmospheric quantity."),
    metadata_field("domain", "Spatial domain."),
    metadata_field("year", "Single year or compact year range."),
    metadata_field("keywords", "Search keywords."),
    metadata_field("provenance", "raw or derived.", required=True),
    metadata_field("format", "Normalised format such as netcdf, zarr, or text.", required=True),
    metadata_field("processing_stage", "Workflow stage that produced this artifact."),
]

spec = CatalogSpec(
    catalog_name="verification-games-demo",
    files_root="data",
    record_schemas={
        "raw_flux": RecordSchema(
            description="Raw or externally supplied flux data.",
            display_fields=["id", "product", "species", "year", "path"],
            directory_template="flux/raw",
            filename_template="{title_slug|original_stem}{original_suffix}",
            metadata_fields=common_fields,
        ),
        "derived_flux": RecordSchema(
            description="Processed flux data and intermediate products.",
            display_fields=["id", "product", "processing_stage", "format", "path"],
            directory_template="flux/derived",
            filename_template="{title_slug|original_stem}{original_suffix}",
            metadata_fields=[
                *common_fields,
                metadata_field("inputs", "Input record ids or external input labels."),
                metadata_field("modifications", "Short description of processing changes."),
            ],
        ),
        "verification_games_obs": RecordSchema(
            description="Synthetic observation or forward-model output.",
            display_fields=["id", "species", "domain", "processing_stage", "path"],
            directory_template="verification_games_obs",
            filename_template="{title_slug|original_stem}{original_suffix}",
            metadata_fields=[
                *common_fields,
                metadata_field("inputs", "Input record ids or external input labels."),
                metadata_field("modifications", "Short description of processing changes."),
                metadata_field("site", "Observation or footprint site code."),
            ],
        ),
    },
)

work = TemporaryDirectory(prefix="ogcat-verification-games-")
root = Path(work.name)
external_dir = root / "external"
scratch_dir = root / "scratch"
external_dir.mkdir()
scratch_dir.mkdir()

catalog = Catalog.create(root / "catalog", spec)
catalog = Catalog.open(root / "catalog")
print(catalog.describe()["record_schemas"])

The helper module used below provides a generic HPC-style path and a small CDL-style description based on the kinds of ncdump -h and xr.Dataset printouts used in the notebooks:

print(example_flux_collection_path())
print(example_flux_cdl())

Choose the right add method

Method	Use it when	Storage effect
add_reference(...)	The artifact already exists and should stay where it is.	Records a local path, URI, or URL path; does not copy, move, or write data.
add_collection(...)	One logical dataset is represented by several members under a directory, URI, or URL path root.	Records one collection artifact with member-pattern metadata; does not scan, copy, move, or open member files.
add_file(...)	ogcat should manage a local copy or move a finished file into catalog storage.	Copies or moves one local file into the catalog object storage root and, by default, creates a readable template symlink under files_root.
add_artifact(...) with a StoragePlan and writer	Domain code should create the output while ogcat handles planning, record creation, and rollback hooks.	The writer materialises the planned file or directory, then ogcat records it.

For the detailed storage model, see Locators and storage.

Add an existing path as a reference

Use add_reference(...) for mounted shared files, generated outputs that have already been moved into place, or remote identifiers that domain code will open later. The file below is deliberately tiny; in a real workflow it might be a large GridFED zip, NetCDF file, or Zarr directory.

raw_path = write_example_flux_netcdf_or_placeholder(external_dir / EXAMPLE_FLUX_FILENAME)

raw_record = catalog.add_reference(
    raw_path,
    record_type="raw_flux",
    metadata={
        "title": "GridFED TOTAL 2021 raw file",
        "product": "GridFED",
        "species": ["co2", "o2"],
        "domain": "global",
        "year": 2021,
        "keywords": ["fossil", "ff"],
        "provenance": "raw",
        "format": "netcdf",
        "processing_stage": "downloaded",
    },
)

print(raw_record.id, raw_record.locator.value)

URI references use the same method when ogcat should not interpret or inspect the target:

catalog.add_reference(
    uri="https://example.org/data/gridfed-total-2020.nc",
    record_type="raw_flux",
    metadata={
        "title": "Remote GridFED TOTAL 2020 raw file",
        "product": "GridFED",
        "species": ["co2", "o2"],
        "domain": "global",
        "year": 2020,
        "keywords": ["fossil", "ff", "remote"],
        "provenance": "raw",
        "format": "netcdf",
        "processing_stage": "remote_catalogue_entry",
    },
)

Add a directory-backed collection

Use add_collection(...) when users should search for one dataset even though the data live in several member files. This is common for yearly or monthly NetCDF series that are opened together with xarray.open_mfdataset.

footprint_dir = external_dir / "name-footprints" / "EASTASIA" / "BCOB-10magl" / "inert"
footprint_dir.mkdir(parents=True)
for month in ["202301", "202302"]:
    (footprint_dir / f"BCOB-10magl_NAME_UMG_EASTASIA_inert_{month}.nc").write_text(
        "tiny placeholder\n",
        encoding="utf-8",
    )

footprint_collection = catalog.add_collection(
    footprint_dir,
    record_type="verification_games_obs",
    metadata={
        "title": "BCOB EASTASIA NAME footprint series",
        "product": "NAME",
        "species": "inert",
        "domain": "EASTASIA",
        "year": "2023",
        "keywords": ["footprint", "collection"],
        "provenance": "raw",
        "format": "netcdf",
        "processing_stage": "downloaded",
        "site": "BCOB",
    },
    collection_pattern="BCOB-10magl_NAME_UMG_EASTASIA_inert_*.nc",
    member_format="netcdf",
    member_suffixes=[".nc"],
    reader_hint="xarray.open_mfdataset",
)

print(footprint_collection.locator.kind, footprint_collection.derived_metadata["classification"])

Add a managed copy

Use add_file(...) when the catalog should own the stored file. The default operation is a copy; pass operation="move" when the scratch file should be removed after a successful catalog add.

note_path = scratch_dir / "mhd_co2_screening.txt"
note_path.write_text("small screening output\n", encoding="utf-8")

managed_record = catalog.add_file(
    note_path,
    record_type="verification_games_obs",
    metadata={
        "title": "MHD CO2 screening output",
        "product": "screening",
        "species": "co2",
        "domain": "EUROPE",
        "year": 2021,
        "keywords": ["screening", "copy"],
        "provenance": "derived",
        "format": "text",
        "processing_stage": "screened_site_output",
        "inputs": [raw_record.id],
        "site": "MHD",
    },
)

print(managed_record.path())

Search, get one record, and open with xarray

The notebooks often search with enough metadata to identify exactly one input, then open the record’s locator with xarray.

selected = catalog.get_one(
    where={
        "product": "GridFED",
        "year": 2021,
        "provenance": "raw",
        "format": "netcdf",
    },
    contains={"keywords": "fossil"},
)
print(selected.id, selected.locator.value)

If xarray and a NetCDF writer backend are installed, this cell turns the fake path into a tiny real NetCDF and opens it through the selected record. If those optional pieces are missing, the helper writes a readable .nc placeholder and this cell reports that xarray cannot open it.

try:
    import xarray as xr
except ImportError:
    xr = None

if xr is not None:
    try:
        write_example_flux_netcdf_or_placeholder(raw_path)
        with xr.open_dataset(selected.locator.value) as ds:
            print(dict(ds.sizes))
    except Exception as exc:
        print(f"xarray NetCDF example skipped: {exc}")

For a Zarr artifact, use the same record lookup and the stored reader_hint:

zarr_path = scratch_dir / "staged_flux.zarr"
zarr_path.mkdir(exist_ok=True)

zarr_record = catalog.add_reference(
    zarr_path,
    record_type="derived_flux",
    metadata={
        "title": "Staged flux Zarr",
        "product": "PARIS_CTE-HR_filled_fluxes_staged",
        "species": ["co2", "o2"],
        "domain": "EUROPE",
        "year": "202012-202112",
        "keywords": ["flux_stage", "forward_model_input"],
        "provenance": "derived",
        "format": "zarr",
        "processing_stage": "forward_model_input",
        "inputs": [raw_record.id],
        "modifications": "Stacked source fluxes and filled missing values before forward modelling.",
    },
    derived_metadata={
        "reader_hint": "xarray.open_zarr",
        "source_record_ids": [raw_record.id],
        "source_paths": [str(raw_record.locator.value)],
        "modifications": "Stacked source fluxes and filled missing values before forward modelling.",
    },
    naming_metadata={"target_kind": "directory"},
)

if zarr_record.derived_metadata.get("reader_hint") == "xarray.open_zarr":
    print(f"Open with xr.open_zarr({zarr_record.locator.value!r})")

Search and display fields

search(...) returns a CatalogRecordSet by default. Use rows(...) for JSON-friendly notebook output, display_rows(...) for terminal-style strings, or to_dataframe(...) when pandas is installed.

derived = catalog.search(
    where={"provenance": "derived"},
    contains={"keywords": "forward_model_input"},
    as_record_set=True,
)

fields = ["id", "product", "processing_stage", "format", "path"]
print(derived.rows(fields))
print(derived.display_rows(fields))

try:
    df = derived.to_dataframe(fields)
    print(df)
except ImportError:
    print("Install pandas to use CatalogRecordSet.to_dataframe().")

For record-set details, see Search and record sets.

Store generated artifacts after computation

A common notebook pattern is to do heavy work outside ogcat, then register the small set of final outputs serially.

Register an already-written artifact

When the computation has already written and validated the output path, record it with add_reference(...) and store provenance in metadata.

staged_output = scratch_dir / "already_written_stage.zarr"
staged_output.mkdir(exist_ok=True)

registered_stage = catalog.add_reference(
    staged_output,
    record_type="derived_flux",
    metadata={
        "title": "Already-written staged flux store",
        "product": "PARIS_CTE-HR_filled_fluxes_staged",
        "species": ["co2", "o2"],
        "domain": "EUROPE",
        "year": "202012-202112",
        "keywords": ["flux_stage", "filled_nan_zero"],
        "provenance": "derived",
        "format": "zarr",
        "processing_stage": "forward_model_input",
        "inputs": [raw_record.id],
        "modifications": "Stacked source fluxes into a forward-model input store.",
    },
    derived_metadata={
        "reader_hint": "xarray.open_zarr",
        "source_record_ids": [raw_record.id],
        "source_paths": [str(raw_record.locator.value)],
        "sizes": {"time": 1, "source": 1},
        "chunks": {"time": 1, "source": 1},
        "modifications": "Stacked source fluxes into a forward-model input store.",
    },
    naming_metadata={"target_kind": "directory"},
)

Move a finished file into managed storage

When the output is a single finished file in scratch space, use add_file(..., operation="move").

finished_path = scratch_dir / "mhd_pollution_event.txt"
finished_path.write_text("finished fake model output\n", encoding="utf-8")

moved_output = catalog.add_file(
    finished_path,
    record_type="verification_games_obs",
    operation="move",
    metadata={
        "title": "MHD pollution event intermediate",
        "product": "modelled_pollution_event",
        "species": "co2_o2",
        "domain": "EUROPE",
        "year": "202101",
        "keywords": ["pollution_events", "intermediate"],
        "provenance": "derived",
        "format": "text",
        "processing_stage": "raw_forward_model_output",
        "inputs": [registered_stage.id],
        "modifications": "Computed footprint dot flux for one site-month before baseline correction.",
        "site": "MHD",
    },
)
print(finished_path.exists(), moved_output.path())

Let a writer create the artifact

Use writer-backed add_artifact(...) when ogcat should plan the target path and domain code should create the artifact during the catalog operation.

normalised_flux_modifications = (
    "Normalized the raw CO2 agriculture flux to a single CF-style flux variable "
    "and preserved coordinate metadata for a small documentation example."
)

normalised_flux_metadata = {
    "title": "Tiny normalized CO2 agriculture flux",
    "product": "EDGAR",
    "species": "co2",
    "domain": "EUROPE",
    "year": 2021,
    "keywords": ["agriculture", "normalized_total_flux"],
    "provenance": "derived",
    "format": "netcdf",
    "processing_stage": "normalized_total_flux",
    "inputs": [raw_record.id],
    "modifications": normalised_flux_modifications,
}

normalised_flux_target = catalog.root / catalog.spec.files_root / "flux" / "derived" / "tiny-normalized-flux.nc"
normalised_flux_plan = catalog.plan_artifact_storage(
    record_type="derived_flux",
    locator=ArtifactLocator.from_path(
        normalised_flux_target,
        relative_path="data/flux/derived/tiny-normalized-flux.nc",
    ),
    target_kind="file",
    write_mode="write",
    metadata=normalised_flux_metadata,
)


def write_normalised_flux(source, target: Path) -> dict[str, object]:
    """Write a tiny NetCDF-shaped flux artifact and return derived metadata."""
    write_example_flux_netcdf_or_placeholder(target)
    return {
        "reader_hint": "xarray.open_dataset",
        "source_record_ids": source.metadata["source_record_ids"],
        "source_paths": source.metadata["source_paths"],
        "output_variables": ["flux"],
        "sizes": {"time": 1, "lat": 2, "lon": 3},
        "units": "kg m-2 s-1",
        "processing_stage": "normalized_total_flux",
        "history": source.metadata["modifications"],
        "modifications": source.metadata["modifications"],
    }


normalised_flux_record = catalog.add_artifact(
    record_type="derived_flux",
    storage_plan=normalised_flux_plan,
    metadata=normalised_flux_metadata,
    source=memory_source(
        {"stage": "normalized_total_flux"},
        kind="normalised_flux_payload",
        metadata={
            "source_record_ids": [raw_record.id],
            "source_paths": [str(raw_record.locator.value)],
            "modifications": normalised_flux_modifications,
        },
    ),
    artifact_writer=source_writer(
        write_normalised_flux,
        target_kind="file",
        source_kind="normalised_flux_payload",
    ),
)
print(normalised_flux_record.path())

Raw and derived metadata conventions

Use user metadata for fields that users search and select on:

Field	Suggested use
provenance	Use "raw" for original or externally supplied data and "derived" for processed outputs.
inputs	Store input record ids, external labels, or both. Keep values JSON-compatible.
processing_stage	Name the workflow stage, such as downloaded, normalized_total_flux, or forward_model_input.
format	Store the normalised format users search for, such as netcdf, zarr, zip, or text.
modifications	Store a short, human-readable processing note that explains what changed and why. For NetCDF outputs, also append the same idea to the dataset history attribute.
reader_hint	Prefer this in derived metadata when it describes how to reopen the artifact, such as xarray.open_dataset or xarray.open_zarr.

Use derived metadata for facts extracted from or produced alongside the artifact:

derived_metadata = {
    "reader_hint": "xarray.open_dataset",
    "source_record_ids": [raw_record.id],
    "source_paths": [str(raw_record.locator.value)],
    "output_variables": ["flux"],
    "sizes": {"time": 1, "lat": 1, "lon": 1},
    "chunks": {"time": 1, "lat": 1, "lon": 1},
    "modifications": "Regridded source fluxes and filled missing values with 0.0.",
}

For multi-input products, named source paths are often easier to audit months later than a bare list:

scenario_source_paths = {
    "primary_o2": "/hpc/shared/atmos/verification-games/data/raw/PARIS_ATEN_O2.nc",
    "FF": str(raw_record.locator.value),
    "GPP": "/hpc/shared/atmos/some_flux_collection/sib4_gpp_2021.nc",
    "TER": "/hpc/shared/atmos/some_flux_collection/sib4_ter_2021.nc",
    "ocean": "/hpc/shared/atmos/some_flux_collection/cesm_scaled_ocean_2021.nc",
}

For general field-discovery and search-containment rules, see Metadata and validation, Search and record sets, and CLI.

When you are done with the temporary example catalog:

work.cleanup()