Skip to content

heiplanet_data.pipeline module⚓︎

heiplanet_data.pipeline ⚓︎

Preprocessing step registry and pipeline orchestration.

This module wires the individual transformations from heiplanet_data.converters, heiplanet_data.regrid, and heiplanet_data.temporal into a settings-driven pipeline:

  • each preprocessing step is a small wrapper function registered with the register_step decorator and an explicit order value that fixes the execution sequence (and the order of filename suffixes),
  • preprocess_data_file is the public entry point: it loads the settings for a data source, runs all enabled steps over a NetCDF file, and writes the result (plus the settings used) to the output directory.

To add a new preprocessing method, write one @register_step function here and add its keys to setting_schema.json and the settings JSON files; the orchestrator does not need to change.

Functions:

Attributes:

StepFn module-attribute ⚓︎

StepFn = Callable[[Dataset, str, Dict[str, Any], Logger], Tuple[Dataset, str]]

logger module-attribute ⚓︎

logger = getLogger(__name__)

preprocess_data_file ⚓︎

preprocess_data_file(netcdf_file, source='era5', settings='default', new_settings=None, unique_tag=None)

Preprocess the dataset based on provided settings. If the settings path is "default", use the default settings of the source. The settings and preprocessed files are saved in the directory, which is specified by the settings file and unique number.

Parameters:

  • netcdf_file (Path) –

    Path to the NetCDF file to preprocess.

  • source (Literal['era5', 'isimip'], default: 'era5' ) –

    Source of the data. Defaults to "era5".

  • settings (Path | str, default: 'default' ) –

    Path to the settings file or "default" for default settings.

  • new_settings (Dict[str, Any] | None, default: None ) –

    Additional settings to overwrite defaults. Defaults to None.

  • unique_tag (str | None, default: None ) –

    Unique tag to append to the output file name and settings file. Defaults to None.

Returns: Tuple[xr.Dataset, str]: Preprocessed dataset and the name of the preprocessed file.

register_step ⚓︎

register_step(name, order)

Register a preprocessing step so _apply_preprocessing runs it.

Parameters:

  • name (str) –

    Unique step name (matches its settings key).

  • order (int) –

    Execution order; steps with a lower value run first.

Returns:

  • Callable[[StepFn], StepFn] –

    Callable[[StepFn], StepFn]: Decorator that records the step.