cli_options

Climate Data CLI Options

This module provides a set of CLI options for extracting climate data from the ERA5 and CMIP6 datasets. These options are used to specify the data to extract, such as the year, month, variable, and dataset. It also provides global variables representing the full space of valid values for these options.

resolve_run_mode_root(param_name: str, value: str, run_mode: str, *, aggregate: bool = False) -> str

Resolve a directory option to the run-mode root unless the user overrode it.

--run-mode selects the storage-root profile; an explicitly-passed --<param> still wins (detected via click's parameter source).

Source code in src/climate_data/cli_options.py

def resolve_run_mode_root(
    param_name: str, value: str, run_mode: str, *, aggregate: bool = False
) -> str:
    """Resolve a directory option to the run-mode root unless the user overrode it.

    ``--run-mode`` selects the storage-root profile; an explicitly-passed
    ``--<param>`` still wins (detected via click's parameter source).
    """
    ctx = click.get_current_context(silent=True)
    overridden = (
        ctx is not None
        and ctx.get_parameter_source(param_name)
        is not click.core.ParameterSource.DEFAULT
    )
    if overridden:
        return value
    root = cdc.aggregate_root(run_mode) if aggregate else cdc.model_root(run_mode)
    return str(root)

with_agg_measure(*, allow_all: bool = False) -> Callable[[Callable[P, T]], Callable[P, T]]

Add aggregation measure option to a command.

Source code in src/climate_data/cli_options.py

def with_agg_measure[**P, T](
    *,
    allow_all: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add aggregation measure option to a command."""
    return with_choice(
        "agg-measure",
        allow_all=allow_all,
        choices=cdc.AGGREGATION_MEASURES,
        help="Climate measure to process.",
    )

with_agg_scenario(*, allow_all: bool = False) -> Callable[[Callable[P, T]], Callable[P, T]]

Add aggregation scenario option to a command.

Source code in src/climate_data/cli_options.py

def with_agg_scenario[**P, T](
    *,
    allow_all: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add aggregation scenario option to a command."""
    return with_choice(
        "agg-scenario",
        allow_all=allow_all,
        choices=cdc.AGGREGATION_SCENARIOS,
        help="Climate scenario to process.",
    )

with_agg_version() -> Callable[[Callable[P, T]], Callable[P, T]]

Add aggregation version option to a command.

Source code in src/climate_data/cli_options.py

def with_agg_version[**P, T]() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add aggregation version option to a command."""
    return click.option(
        "--agg-version",
        help="Aggregation version to process.",
        required=True,
    )

with_block_key(*, allow_all: bool = False) -> Callable[[Callable[P, T]], Callable[P, T]]

Add block key option to a command.

Source code in src/climate_data/cli_options.py

def with_block_key[**P, T](
    *,
    allow_all: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add block key option to a command."""
    return with_choice(
        "block-key",
        allow_all=allow_all,
        choices=None,  # Will be populated at runtime
        help="Block key to process.",
    )

with_concurrency_limit(*, default: int | None = None) -> Callable[[Callable[P, T]], Callable[P, T]]

Add the jobmon concurrency-limit option to a command.

Pass default to throttle a runner out of the box; leaving it unset means the option defaults to None, which run_parallel_maybe_dry_run drops so jobmon applies its own default (10000, effectively unthrottled).

Source code in src/climate_data/cli_options.py

def with_concurrency_limit[**P, T](
    *,
    default: int | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add the jobmon concurrency-limit option to a command.

    Pass ``default`` to throttle a runner out of the box; leaving it unset means
    the option defaults to ``None``, which ``run_parallel_maybe_dry_run`` drops
    so jobmon applies its own default (10000, effectively unthrottled).
    """
    return click.option(
        "--concurrency-limit",
        type=click.INT,
        default=default,
        show_default=True,
        help=(
            "Cap on tasks running at once, to keep write latency on shared "
            "storage manageable. jobmon's own default is 10000, which is "
            "effectively unthrottled."
        ),
    )

with_debias_method() -> Callable[[Callable[P, T]], Callable[P, T]]

Add the option selecting the Jensen de-bias applied to a multiplicative anomaly.

Source code in src/climate_data/cli_options.py

def with_debias_method[**P, T]() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add the option selecting the Jensen de-bias applied to a multiplicative anomaly."""
    return click.option(
        "--debias-method",
        type=click.Choice(list(cdc.DEBIAS_METHODS)),
        default="none",
        show_default=True,
        help=(
            "Correct the ratio-estimator (Jensen) bias in the multiplicative anomaly: "
            "'none' (ship as-is), 'loo' (leave-one-out, a direct out-of-sample estimate), "
            "or 'analytic' (second-order expansion). Only valid for "
            f"{', '.join(cdc.DEBIAS_VARIABLES)}."
        ),
    )

with_dry_day_rule() -> Callable[[Callable[P, T]], Callable[P, T]]

Add the option selecting how days the driving model reports as dry are treated.

Source code in src/climate_data/cli_options.py

def with_dry_day_rule[**P, T]() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add the option selecting how days the driving model reports as dry are treated."""
    return click.option(
        "--dry-day-rule",
        type=click.Choice(list(cdc.DRY_DAY_RULES)),
        default="none",
        show_default=True,
        help=(
            "Treatment of dry model days in the multiplicative anomaly: 'none' (ship as-is) "
            "or 'preserve' (zero the anomaly on dry model days and renormalise the "
            "cell-month, leaving the monthly total unchanged). Only valid for "
            f"{', '.join(cdc.DRY_DAY_VARIABLES)}."
        ),
    )

with_dry_run() -> Callable[[Callable[P, T]], Callable[P, T]]

Add dry-run flag to a command.

Source code in src/climate_data/cli_options.py

def with_dry_run[**P, T]() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add dry-run flag to a command."""
    return click.option(
        "--dry-run/--no-dry-run",
        "-n",
        default=False,
        show_default=True,
        help="Print sbatch-like previews instead of submitting jobs.",
    )

with_hierarchy(choices: Collection[str] = cdc.HIERARCHY_MAP, *, allow_all: bool = False, default: str | None = None) -> Callable[[Callable[P, T]], Callable[P, T]]

Add hierarchy option to a command.

Pass default to give the (single-value) option a default; this builds the option directly since with_choice always supplies its own default.

Source code in src/climate_data/cli_options.py

def with_hierarchy[**P, T](
    choices: Collection[str] = cdc.HIERARCHY_MAP,
    *,
    allow_all: bool = False,
    default: str | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add hierarchy option to a command.

    Pass ``default`` to give the (single-value) option a default; this builds
    the option directly since ``with_choice`` always supplies its own default.
    """
    if default is not None:
        return click.option(
            "--hierarchy",
            type=click.Choice(list(choices)),
            default=default,
            show_default=True,
            help="Hierarchy to process.",
        )
    return with_choice(
        "hierarchy",
        allow_all=allow_all,
        choices=choices,
        help="Hierarchy to process.",
        convert=allow_all,
    )

with_location_id() -> Callable[[Callable[P, T]], Callable[P, T]]

Add location ID option to a command.

Source code in src/climate_data/cli_options.py

def with_location_id[**P, T]() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add location ID option to a command."""
    return click.option(
        "--location-id",
        "-l",
        type=click.INT,
        help="Location ID to process.",
    )

with_run_mode() -> Callable[[Callable[P, T]], Callable[P, T]]

Add the run-mode option selecting the storage-root profile.

Source code in src/climate_data/cli_options.py

def with_run_mode[**P, T]() -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Add the run-mode option selecting the storage-root profile."""
    return click.option(
        "--run-mode",
        type=click.Choice(list(cdc.RUN_MODES)),
        default="forecast",
        show_default=True,
        help=(
            "Storage-root profile: 'forecast' (production roots) or 'historical' "
            "(geospatial roots for the GBD historical exposure product)."
        ),
    )

with_year(years: Collection[str], *, allow_all: bool = False) -> Callable[[Callable[P, T]], Callable[P, T]]

Create a CLI option for selecting a year.

Source code in src/climate_data/cli_options.py

def with_year[**P, T](
    years: Collection[str],
    *,
    allow_all: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
    """Create a CLI option for selecting a year."""
    return with_choice(
        "year",
        "y",
        allow_all=allow_all,
        choices=years,
        help="Year to extract data for.",
        convert=allow_all,
    )