Skip to article frontmatterSkip to article content
Site not loading correctly?

This may be due to an incorrect BASE_URL configuration. See the MyST Documentation for reference.

Source Data Module Developer’s Guide

Module Design Philosophy

The cstar_forge.forge.source_data module provides a registry-based framework for managing heterogeneous source datasets used in ROMS preprocessing. The design emphasizes:

The module is split across two files:

Core Architecture

The module consists of three main components:

  1. Registry System: Decorator-based registration of dataset handlers (source_data.py)

  2. SourceData Class: Main interface for preparing and accessing datasets (source_data.py)

  3. Source Name Mapping: Translation layer between logical names and dataset keys (source_registry.py, re-exported from source_data.py)

User Request ("GLORYS") 
    ↓
SOURCE_ALIAS mapping
    ↓
Dataset Key ("GLORYS_REGIONAL")
    ↓
DATASET_REGISTRY lookup
    ↓
DatasetHandler (function + requirements)
    ↓
Handler execution → Path(s) to prepared data

Core Objects

SourceData (Dataclass)

The main interface for preparing and accessing source datasets.

Constructor:

SourceData(
    datasets: list[str],                        # Dataset names to prepare
    clobber: bool = False,                       # Force re-download if True
    grid: object | None = None,
    grid_name: str | None = None,
    start_time: object | None = None,
    end_time: object | None = None,
    source_data_dir: Path | None = None,         # Cache root — injected by the caller
    resolved_datasets: dict[str, dict] | None = None,  # Optional frozen key/metadata snapshot
)

SourceData no longer resolves a cache directory from cstar_forge.config internally — the caller (typically ForgeExecutor) must inject source_data_dir, e.g. from a HostPaths.source_data_cache built by cstar_forge.config.resolve_host(). If source_data_dir is None, prepare_all() raises ValueError for the first dataset it actually stages (unstaged/streamable skips happen first).

Key Attributes:

Key Methods:

Lifecycle:

  1. Initialization: Normalizes dataset names through SOURCE_ALIAS, validates against DATASET_REGISTRY | UNSTAGED_DATASETS

  2. Preparation: prepare_all() iterates through datasets, skips UNSTAGED_DATASETS (e.g. ETOPO5, DAI — provided by roms-tools itself, never staged by Forge) and, unless include_streamable=True, skips STREAMABLE_SOURCES; for the rest it checks requirements and calls handler functions

  3. Storage: Handler return values (Path, List[Path], or Dict) are stored in self.paths. prepare_all() overwrites self.paths[key] with the handler’s return value, so a handler’s own self.paths[...] assignment only matters when the handler is called directly

DatasetHandler (Class)

Container for a dataset preparation function and its dependency requirements.

class DatasetHandler:
    def __init__(self, func: Callable[["SourceData"], Path], requires: List[str]):
        self.func = func      # Handler function
        self.requires = requires  # Required SourceData attributes

Purpose:

Registry Framework

Registration Decorator

The @register_dataset decorator registers dataset preparation functions:

@register_dataset(
    name: str,                    # Dataset key (e.g., "GLORYS_REGIONAL")
    requires: Optional[List[str]] = None  # Required SourceData attributes
)
def _prepare_dataset(self: SourceData) -> Union[Path, List[Path], Dict]:
    """Handler function that prepares the dataset."""
    # Implementation...
    return path_or_paths

Registration Process:

  1. Decorator captures function and requirements

  2. Creates DatasetHandler instance

  3. Stores in DATASET_REGISTRY with uppercase key

Example:

@register_dataset(
    "GLORYS_REGIONAL",
    requires=["grid", "grid_name", "start_time", "end_time"]
)
def _prepare_glorys_regional(self: SourceData) -> List[Path]:
    """Download daily regional GLORYS subsets."""
    bounds = rt.get_glorys_bounds(self.grid)
    paths = self._prepare_glorys_daily(is_regional=True, bounds=bounds)
    self.paths["GLORYS_REGIONAL"] = paths[0] if len(paths) == 1 else paths
    return paths

Registry Dictionary

DATASET_REGISTRY: Dict[str, DatasetHandler] (defined in source_data.py, alongside the handlers it registers) maps dataset keys to their handlers.

Handler Function Signature

Handler functions must:

Source Name Mapping

Logical Names vs. Dataset Keys

Users specify logical source names in a ForcingSpec (catalog/ForcingSpec/<name>/Forcing.yaml); the topography source is Domain-level (model.yaml no longer carries source selection):

SOURCE_ALIAS Dictionary

Defined in cstar_forge/forge/source_registry.py (re-exported from source_data.py). Maps logical names to dataset registry keys:

SOURCE_ALIAS: dict[str, str] = {
    "ERA5": "ERA5",
    "GLORYS": "GLORYS_REGIONAL",       # defaults to regional; see resolve_dataset_key
    "GLORYS_GLOBAL": "GLORYS_GLOBAL",
    "GLORYS_REGIONAL": "GLORYS_REGIONAL",
    "UNIFIED": "UNIFIED_BGC",
    "UNIFIED_BGC": "UNIFIED_BGC",
    "SRTM15": "SRTM15",
    "MBL_CO2": "MBL_CO2",
    "TPXO": "TPXO",
    "WOA": "WOA",
    "DAI": "DAI",
    "GLOFAS": "GLOFAS",
    "EMOD": "EMOD",
    "RIVR2O": "RIVR2O",
    "CONSTANTS": "CONSTANTS",
}

Mapping Function:

def map_source_to_dataset_key(name: str) -> str:
    """Map logical name to dataset key, or return uppercased name if no alias."""
    return SOURCE_ALIAS.get(name.upper(), name.upper())

resolve_dataset_key(name, glorys_layout=None) wraps this and additionally handles the one case the table can’t: logical "GLORYS" disambiguated by an explicit glorys_layout ("global" vs. "regional", defaulting to regional).

Normalization:

Streamable and Unstaged Sources

Some datasets don’t require local caching (e.g., ERA5, DAI, CONSTANTS). Listed in STREAMABLE_SOURCES (source_registry.py):

STREAMABLE_SOURCES = ["ERA5", "DAI", "CONSTANTS"]

A distinct set, UNSTAGED_DATASETS = {"ETOPO5", "DAI"}, covers recognized keys that Forge never stages at all — no @register_dataset handler exists because something else supplies the file (roms-tools auto-fetches ETOPO5 at grid-build time; DAI is streamed). SourceData.__post_init__() treats these as valid-but-skipped, distinguishing them from a genuinely unknown/typo’d name, which still raises. prepare_all() always skips them, even with include_streamable=True.

Adding a New Dataset

Step 1: Implement Handler Function

@register_dataset("MY_DATASET", requires=["grid", "grid_name"])
def _prepare_my_dataset(self: SourceData) -> Path:
    """
    Prepare MY_DATASET for the given grid.
    
    Returns
    -------
    Path
        Path to the prepared dataset file.
    """
    dataset_dir = self.source_data_dir / "MY_DATASET"
    dataset_dir.mkdir(parents=True, exist_ok=True)
    path = dataset_dir / f"my_dataset_{self.grid_name}.nc"
    
    needs_download = self.clobber or (not path.exists())
    
    if needs_download:
        if path.exists():
            print(f"⚠️  Clobber=True: removing existing file {path.name}")
            path.unlink()
        
        print(f"⬇️  Downloading MY_DATASET → {path}")
        # Download/prepare logic here...
    else:
        print(f"✔️  Using existing MY_DATASET: {path}")
    
    self.paths["MY_DATASET"] = path
    return path

Step 2: Add Source Alias (if needed)

If users should reference it by a logical name, add the entry to SOURCE_ALIAS in cstar_forge/forge/source_registry.py:

"MY_SOURCE": "MY_DATASET",

Step 3: Add to Streamable Sources (if applicable)

If the dataset doesn’t need local caching, add it to STREAMABLE_SOURCES in cstar_forge/forge/source_registry.py:

"MY_DATASET",

Step 4: Add a DATASET_METADATA Entry

Add a provenance entry to DATASET_METADATA in cstar_forge/forge/source_registry.py ({"dataset_id": ...} or {"url": ...}; {} for user-staged datasets) so provenance is snapshotted into ForgeBlueprint.forcing.resolved_datasets.

Example Dataset Handlers

Simple Download Handler (SRTM15)

@register_dataset("SRTM15")
def _prepare_srtm15(self: SourceData) -> Path:
    """Download SRTM15 bathymetry."""
    path = self.source_data_dir / "SRTM15" / f"SRTM15_{SRTM15_VERSION}.nc"
    
    if self.clobber or not path.exists():
        # Download logic...
        pass
    
    self.srtm15_path = path
    return path

Characteristics:

Time-Dependent Handler (GLORYS)

@register_dataset(
    "GLORYS_REGIONAL",
    requires=["grid", "grid_name", "start_time", "end_time"]
)
def _prepare_glorys_regional(self: SourceData) -> List[Path]:
    """Download daily regional GLORYS subsets."""
    bounds = rt.get_glorys_bounds(self.grid)
    paths = self._prepare_glorys_daily(is_regional=True, bounds=bounds)
    # Store single path or list depending on count
    self.paths["GLORYS_REGIONAL"] = paths[0] if len(paths) == 1 else paths
    return paths

Characteristics:

User-Provided Dataset Handler (TPXO)

@register_dataset("TPXO")
def _prepare_tpxo(self: SourceData) -> Dict[str, Path]:
    """Verify user-provided TPXO tidal data exists."""
    tpxo_path = self.source_data_dir / "TPXO" / "TPXO10.v2a"
    
    # Verify files exist
    tpxo_dict = {
        "grid": tpxo_path / "grid_tpxo10v2a.nc",
        "h": tpxo_path / "h_tpxo10.v2a.nc",
        "u": tpxo_path / "u_tpxo10.v2a.nc",
    }
    
    # Validation logic...
    
    self.paths["TPXO"] = tpxo_path
    return tpxo_dict

Characteristics:

Usage Pattern

from cstar_forge.forge.source_data import SourceData
from cstar_forge.config import resolve_host
from datetime import datetime

host = resolve_host(working_dir="~/cstar/_forge_bp_runs/my-grid")

# Create SourceData instance
src = SourceData(
    datasets=["GLORYS", "UNIFIED", "SRTM15"],  # Logical names
    clobber=False,
    grid=my_grid,
    grid_name="my-grid",
    start_time=datetime(2024, 1, 1),
    end_time=datetime(2024, 1, 2),
    source_data_dir=host.source_data_cache,
)

# Prepare all datasets
src.prepare_all()

# Access prepared paths
glorys_path = src.path_for_source("GLORYS")  # Returns Path or List[Path]
unified_path = src.path_for_source("UNIFIED")  # Returns Path
srtm15_path = src.path_for_source("SRTM15")  # Returns Path

# Or access directly
glorys_path = src.paths["GLORYS_REGIONAL"]

Design Patterns

Dependency Injection

Required attributes are injected into SourceData and accessed by handlers via self. This enables:

Caching Strategy

Return Value Flexibility

Handlers can return:

All are stored in self.paths[dataset_key] for uniform access.

Developer Guide
Source Data Example
Developer Guide
Input Data Generation Overview