cli

cli

CLI entry point for hydro-param.

Provide the top-level hydro-param console script and all subcommands using the cyclopts framework. The CLI is the primary user interface for configuration-driven hydrologic parameterization workflows.

Commands

hydro-param init [project-dir] Scaffold a project directory with standard layout. hydro-param datasets list List available datasets grouped by category. hydro-param datasets info Show full details for a single dataset. hydro-param datasets download Download dataset files via AWS CLI. hydro-param run Execute the generic parameterization pipeline (stages 1--5). hydro-param pywatershed run Generate a complete pywatershed model setup (two-phase workflow). hydro-param pywatershed validate Validate a pywatershed parameter NetCDF file. hydro-param gfv11 download --output-dir

[--items {data-layers,tgf-topo,all}] Download GFv1.1 NHM data layer rasters from ScienceBase.

Notes

All model-specific logic (unit conversions, variable renaming, derivation) lives in plugin modules (derivations/pywatershed.py, formatters/pywatershed.py). The generic run command produces a raw Standardized Internal Representation (SIR); the pywatershed run command adds a second phase of model-specific post-processing.

See Also

docs/design.md : Full architecture and CLI specification (§11.9). hydro_param.pipeline : Generic 5-stage pipeline orchestrator. hydro_param.derivations.pywatershed : pywatershed derivation plugin.

datasets_list

datasets_list(*, registry: Path | None = None) -> None

Display all registered datasets grouped by category.

Print a formatted table of datasets to stdout, showing name, description, access strategy, and availability status. Datasets are grouped under their category heading (e.g., Topography, Land Cover, Soils).

PARAMETER	DESCRIPTION
registry	Path to a custom registry YAML file or directory. When omitted, the bundled default registry is used. TYPE: Path | None DEFAULT: None

Source code in src/hydro_param/cli.py

@datasets_app.command(name="list")
def datasets_list(*, registry: Path | None = None) -> None:
    """Display all registered datasets grouped by category.

    Print a formatted table of datasets to stdout, showing name,
    description, access strategy, and availability status.  Datasets
    are grouped under their category heading (e.g., Topography,
    Land Cover, Soils).

    Parameters
    ----------
    registry
        Path to a custom registry YAML file or directory.  When
        omitted, the bundled default registry is used.
    """
    reg = _load_registry(registry)

    # Group by category
    by_category: dict[str, list[tuple[str, DatasetEntry]]] = {}
    for name, entry in reg.datasets.items():
        cat = entry.category or "uncategorized"
        by_category.setdefault(cat, []).append((name, entry))

    print("  Format: [strategy | years | access | variables]")
    for category, entries in sorted(by_category.items()):
        print(f"\n{category.replace('_', ' ').title()}:")
        for name, entry in sorted(entries):
            desc = entry.description or ""
            status = _access_status(entry)
            yr = f"{entry.year_range[0]}-{entry.year_range[1]}" if entry.year_range else "\u2014"
            temporal = f", {entry.time_step}" if entry.temporal and entry.time_step else ""
            var_names = [v.name for v in (entry.variables or [])]
            var_names += [v.name for v in (entry.derived_variables or [])]
            var_list = ", ".join(var_names) if var_names else "\u2014"
            print(
                f"  {name:<20s} {desc:<50s} "
                f"[{entry.strategy} | {yr} | {status}{temporal} | {var_list}]"
            )

datasets_info

datasets_info(
    name: str, *, registry: Path | None = None
) -> None

Show full details for a single dataset.

Print comprehensive information including description, strategy, CRS, variables (continuous and derived), download instructions, and a pipeline config snippet for easy copy-paste.

PARAMETER	DESCRIPTION
name	Dataset name as it appears in the registry (e.g., "dem_3dep_10m", "nlcd_osn_lndcov"). TYPE: str
registry	Path to a custom registry YAML file or directory. When omitted, the bundled default registry is used. TYPE: Path | None DEFAULT: None

RAISES	DESCRIPTION
SystemExit	If the dataset name is not found in the registry (exit code 1).

Source code in src/hydro_param/cli.py

@datasets_app.command(name="info")
def datasets_info(name: str, *, registry: Path | None = None) -> None:
    """Show full details for a single dataset.

    Print comprehensive information including description, strategy,
    CRS, variables (continuous and derived), download instructions,
    and a pipeline config snippet for easy copy-paste.

    Parameters
    ----------
    name
        Dataset name as it appears in the registry (e.g.,
        ``"dem_3dep_10m"``, ``"nlcd_osn_lndcov"``).
    registry
        Path to a custom registry YAML file or directory.  When
        omitted, the bundled default registry is used.

    Raises
    ------
    SystemExit
        If the dataset name is not found in the registry (exit code 1).
    """
    reg = _load_registry(registry)
    try:
        entry = reg.get(name)
    except KeyError as exc:
        print(str(exc), file=sys.stderr)
        raise SystemExit(1) from None

    print(f"Dataset: {name}")
    if entry.description:
        print(f"Description: {entry.description}")
    print(f"Strategy: {entry.strategy} ({_access_status(entry)})")
    print(f"CRS: {entry.crs}")
    if entry.category:
        print(f"Category: {entry.category}")
    if entry.temporal:
        print(f"Time step: {entry.time_step}")
        if entry.year_range:
            print(f"Available years: {entry.year_range[0]}-{entry.year_range[1]}")

    if entry.variables:
        print("\nVariables:")
        for v in entry.variables:
            cat_tag = " (categorical)" if v.categorical else ""
            print(f"  {v.name}{cat_tag}")
            if v.long_name:
                print(f"    {v.long_name}")

    if entry.derived_variables:
        print("\nDerived Variables:")
        for dv in entry.derived_variables:
            print(f"  {dv.name} (from {dv.source}, method={dv.method})")
            if dv.long_name:
                print(f"    {dv.long_name}")

    if entry.download:
        dl = entry.download
        if dl.files:
            # Multi-file dataset
            years = sorted({f.year for f in dl.files})
            variables = sorted({f.variable for f in dl.files})
            print(f"\nDownload: {len(dl.files)} files available")
            print(f"  Years: {', '.join(str(y) for y in years)}")
            print(f"  Products: {', '.join(variables)}")
            print("\n  Files:")
            for f in sorted(dl.files, key=lambda x: (x.year, x.variable)):
                size = f"  ~{f.size_gb} GB" if f.size_gb else ""
                print(f"    {f.year} {f.variable:<20s}{size}")
                print(f"      {f.url}")
            print("\n  Download with:")
            print(f"    hydro-param datasets download {name} --years {years[-1]}")
        elif dl.url_template:
            # Template-based dataset
            start, end = dl.year_range
            total = (end - start + 1) * len(dl.variables_available)
            example_url = dl.url_template.format(variable=dl.variables_available[0], year=end)
            print(f"\nDownload: {total} files via URL template")
            print(f"  Years: {start}-{end}")
            print(f"  Products: {', '.join(dl.variables_available)}")
            print(f"  Requester-pays: {'yes' if dl.requester_pays else 'no'}")
            print(f"\n  Example URL:\n    {example_url}")
            if dl.notes:
                print(f"  {dl.notes.strip()}")
            first_var = dl.variables_available[0]
            print("\n  Download with:")
            print(f"    hydro-param datasets download {name} --years {end} --variables {first_var}")
        elif dl.url:
            # Single-file dataset
            print("\nDownload:")
            print(f"  URL: {dl.url}")
            if dl.size_gb:
                print(f"  Size: ~{dl.size_gb} GB")
            if dl.format:
                print(f"  Format: {dl.format}")
            if dl.notes:
                print(f"  {dl.notes.strip()}")
            print("\n  Download with:")
            print(f"    hydro-param datasets download {name}")

    print("\nTo use in your pipeline config:")
    print("  datasets:")
    print(f"    - name: {name}")
    if entry.strategy == "local_tiff":
        print("      source: /path/to/your/downloaded/file.tif")
    if entry.variables:
        var_names = [v.name for v in entry.variables]
        print(f"      variables: [{', '.join(var_names)}]")

datasets_download

datasets_download(
    name: str,
    *,
    dest: Path | None = None,
    years: str | None = None,
    variables: str | None = None,
    registry: Path | None = None,
) -> None

Download dataset files via the AWS CLI.

Fetch remote dataset files (typically from S3) using aws s3 cp. Supports single-file, multi-file (explicit file list), and template-based (URL pattern with year/variable placeholders) download configurations.

When run inside an initialised hydro-param project (detected via .hydro-param marker), files are automatically routed to the data/<category>/ subdirectory.

PARAMETER	DESCRIPTION
name	Dataset name as it appears in the registry (e.g., "polaris_30m"). TYPE: str
dest	Destination directory for downloaded files. When omitted inside an initialised project, files are routed to data/<category>/ automatically. Otherwise defaults to the current directory. TYPE: Path | None DEFAULT: None
years	Comma-separated list of years to download (multi-file datasets). Example: "2019,2020,2021". TYPE: str | None DEFAULT: None
variables	Comma-separated list of variables/products to download (multi-file datasets). Example: "silt,sand,clay". TYPE: str | None DEFAULT: None
registry	Path to a custom registry YAML file or directory. When omitted, the bundled default registry is used. TYPE: Path | None DEFAULT: None

RAISES	DESCRIPTION
SystemExit	If the dataset is not found, has no download info, AWS CLI is not installed, or a download fails (exit code 1).

Notes

Requires the AWS CLI (aws) to be installed and available on PATH. For requester-pays buckets (e.g., s3://usgs-landcover), valid AWS credentials are needed. Anonymous access is used otherwise.

Source code in src/hydro_param/cli.py

@datasets_app.command(name="download")
def datasets_download(
    name: str,
    *,
    dest: Path | None = None,
    years: str | None = None,
    variables: str | None = None,
    registry: Path | None = None,
) -> None:
    """Download dataset files via the AWS CLI.

    Fetch remote dataset files (typically from S3) using ``aws s3 cp``.
    Supports single-file, multi-file (explicit file list), and
    template-based (URL pattern with year/variable placeholders)
    download configurations.

    When run inside an initialised hydro-param project (detected via
    ``.hydro-param`` marker), files are automatically routed to the
    ``data/<category>/`` subdirectory.

    Parameters
    ----------
    name
        Dataset name as it appears in the registry (e.g.,
        ``"polaris_30m"``).
    dest
        Destination directory for downloaded files.  When omitted inside
        an initialised project, files are routed to ``data/<category>/``
        automatically.  Otherwise defaults to the current directory.
    years
        Comma-separated list of years to download (multi-file datasets).
        Example: ``"2019,2020,2021"``.
    variables
        Comma-separated list of variables/products to download
        (multi-file datasets).  Example: ``"silt,sand,clay"``.
    registry
        Path to a custom registry YAML file or directory.  When
        omitted, the bundled default registry is used.

    Raises
    ------
    SystemExit
        If the dataset is not found, has no download info, AWS CLI is
        not installed, or a download fails (exit code 1).

    Notes
    -----
    Requires the AWS CLI (``aws``) to be installed and available on
    ``PATH``.  For requester-pays buckets (e.g., ``s3://usgs-landcover``),
    valid AWS credentials are needed.  Anonymous access is used otherwise.
    """
    reg = _load_registry(registry)
    try:
        entry = reg.get(name)
    except KeyError as exc:
        print(str(exc), file=sys.stderr)
        raise SystemExit(1) from None

    if entry.download is None:
        print(f"Error: Dataset '{name}' has no download information.", file=sys.stderr)
        print("It may be a remote dataset that does not require downloading.", file=sys.stderr)
        raise SystemExit(1)

    # Resolve destination: auto-route when inside an initialised project
    if dest is None:
        project_root = find_project_root()
        if project_root is not None and entry.category:
            dest = project_root / "data" / entry.category
            print(f"Project detected: downloading to {dest}")
        else:
            dest = Path(".")

    # Check for AWS CLI
    if shutil.which("aws") is None:
        print(
            "Error: AWS CLI not found. Install it to download datasets.\n"
            "  https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html",
            file=sys.stderr,
        )
        raise SystemExit(1)

    dest.mkdir(parents=True, exist_ok=True)

    if entry.download.files or entry.download.url_template:
        _download_multi_file(name, entry, dest, years, variables)
    elif entry.download.url:
        _download_single_file(
            entry.download.url, dest, requester_pays=entry.download.requester_pays
        )
    else:
        print(f"Error: Dataset '{name}' has no download URLs.", file=sys.stderr)
        raise SystemExit(1)

run_cmd

run_cmd(
    config: Path,
    *,
    registry: Path | None = None,
    resume: bool = False,
) -> None

Execute the generic parameterization pipeline.

Run stages 1--5 (resolve fabric, resolve datasets, compute weights, process datasets, normalize output) to produce a normalized Standardized Internal Representation (SIR) with canonical variable names and converted units. This command is model-agnostic; use pywatershed run for model-specific post-processing.

PARAMETER	DESCRIPTION
config	Path to the pipeline YAML config file (e.g., configs/examples/delaware_2yr.yml). TYPE: Path
registry	Path to a custom dataset registry YAML file or directory. When omitted, the bundled default registry is used. TYPE: Path | None DEFAULT: None
resume	Enable manifest-based resume: skip datasets whose outputs are already complete and whose config fingerprint has not changed. Compares SHA-256 fingerprints of dataset request + registry entry + processing options. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
SystemExit	If the pipeline raises any exception (exit code 1).

See Also

hydro_param.pipeline.run_pipeline_from_config : Pipeline entry point. hydro_param.manifest : Manifest-based resume logic.

Source code in src/hydro_param/cli.py

@app.command(name="run")
def run_cmd(config: Path, *, registry: Path | None = None, resume: bool = False) -> None:
    """Execute the generic parameterization pipeline.

    Run stages 1--5 (resolve fabric, resolve datasets, compute weights,
    process datasets, normalize output) to produce a normalized
    Standardized Internal Representation (SIR) with canonical variable
    names and converted units.  This command is model-agnostic; use ``pywatershed run`` for
    model-specific post-processing.

    Parameters
    ----------
    config
        Path to the pipeline YAML config file (e.g.,
        ``configs/examples/delaware_2yr.yml``).
    registry
        Path to a custom dataset registry YAML file or directory.
        When omitted, the bundled default registry is used.
    resume
        Enable manifest-based resume: skip datasets whose outputs are
        already complete and whose config fingerprint has not changed.
        Compares SHA-256 fingerprints of dataset request + registry
        entry + processing options.

    Raises
    ------
    SystemExit
        If the pipeline raises any exception (exit code 1).

    See Also
    --------
    hydro_param.pipeline.run_pipeline_from_config : Pipeline entry point.
    hydro_param.manifest : Manifest-based resume logic.
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )

    try:
        cfg = load_config(config)
        if resume:
            cfg = cfg.model_copy(
                update={"processing": cfg.processing.model_copy(update={"resume": True})}
            )
        reg = _load_registry(registry)
        run_pipeline_from_config(cfg, reg)
    except Exception as exc:
        logger.exception("Pipeline failed.")
        raise SystemExit(1) from exc

init_cmd

init_cmd(
    project_dir: Path = Path("."),
    *,
    force: bool = False,
    registry: Path | None = None,
) -> None

Scaffold a new hydro-param project directory.

Create a standard directory structure with a template pipeline configuration, data directories organised by dataset category (e.g., data/topography/, data/soils/), and a .gitignore. A .hydro-param marker file is written to identify the project root for automatic path resolution in other commands (e.g., datasets download auto-routes files to data/<category>/).

PARAMETER	DESCRIPTION
project_dir	Directory to initialise. Defaults to the current directory. TYPE: Path DEFAULT: Path('.')
force	Re-initialise an existing project: create missing directories and refresh the marker file, but preserve any existing pipeline.yml. TYPE: bool DEFAULT: False
registry	Path to a custom registry YAML for category discovery. When omitted, the bundled default registry is used to determine which data/<category>/ subdirectories to create. TYPE: Path | None DEFAULT: None

See Also

hydro_param.project.init_project : Implementation of project scaffolding.

Source code in src/hydro_param/cli.py

@app.command(name="init")
def init_cmd(
    project_dir: Path = Path("."),
    *,
    force: bool = False,
    registry: Path | None = None,
) -> None:
    """Scaffold a new hydro-param project directory.

    Create a standard directory structure with a template pipeline
    configuration, data directories organised by dataset category
    (e.g., ``data/topography/``, ``data/soils/``), and a ``.gitignore``.
    A ``.hydro-param`` marker file is written to identify the project
    root for automatic path resolution in other commands (e.g.,
    ``datasets download`` auto-routes files to ``data/<category>/``).

    Parameters
    ----------
    project_dir
        Directory to initialise.  Defaults to the current directory.
    force
        Re-initialise an existing project: create missing directories
        and refresh the marker file, but preserve any existing
        ``pipeline.yml``.
    registry
        Path to a custom registry YAML for category discovery.  When
        omitted, the bundled default registry is used to determine
        which ``data/<category>/`` subdirectories to create.

    See Also
    --------
    hydro_param.project.init_project : Implementation of project scaffolding.
    """
    init_project(project_dir, force=force, registry_path=registry)

pws_run_cmd

pws_run_cmd(config: Path) -> None

Generate a complete pywatershed model setup from existing SIR output.

Consume pre-built SIR (Standardized Internal Representation) output produced by hydro-param run and derive all PRMS parameters needed for a pywatershed (NHM-PRMS) simulation.

This command executes Phase 2 only -- it does not re-run the generic pipeline. Run hydro-param run pipeline.yml first to produce SIR output, then run this command to derive model-specific parameters.

Output files produced:

• parameters.nc -- static PRMS parameters loadable by pws.Parameters.from_netcdf()
• forcing/<var>.nc -- one file per climate variable (prcp, tmax, tmin) in PRMS units (inches, degrees F)
• soltab.nc -- potential solar radiation tables (nhru x 366)
• control.yml -- simulation time period configuration

PARAMETER	DESCRIPTION
config	Path to a pywatershed run config YAML file (v4.0). See PywatershedRunConfig for the expected schema. TYPE: Path

RAISES	DESCRIPTION
SystemExit	If config loading, SIR loading, or derivation fails (exit code 1).

See Also

hydro_param.pywatershed_config.PywatershedRunConfig : Config schema. hydro_param.derivations.pywatershed.PywatershedDerivation : Derivation plugin. hydro_param.formatters.pywatershed.PywatershedFormatter : Output formatter.

Source code in src/hydro_param/cli.py

@pws_app.command(name="run")
def pws_run_cmd(config: Path) -> None:
    """Generate a complete pywatershed model setup from existing SIR output.

    Consume pre-built SIR (Standardized Internal Representation) output
    produced by ``hydro-param run`` and derive all PRMS parameters needed
    for a pywatershed (NHM-PRMS) simulation.

    This command executes Phase 2 only -- it does **not** re-run the
    generic pipeline.  Run ``hydro-param run pipeline.yml`` first to
    produce SIR output, then run this command to derive model-specific
    parameters.

    Output files produced:

    - ``parameters.nc`` -- static PRMS parameters loadable by
      ``pws.Parameters.from_netcdf()``
    - ``forcing/<var>.nc`` -- one file per climate variable (prcp,
      tmax, tmin) in PRMS units (inches, degrees F)
    - ``soltab.nc`` -- potential solar radiation tables (nhru x 366)
    - ``control.yml`` -- simulation time period configuration

    Parameters
    ----------
    config
        Path to a pywatershed run config YAML file (v4.0).  See
        ``PywatershedRunConfig`` for the expected schema.

    Raises
    ------
    SystemExit
        If config loading, SIR loading, or derivation fails (exit code 1).

    See Also
    --------
    hydro_param.pywatershed_config.PywatershedRunConfig : Config schema.
    hydro_param.derivations.pywatershed.PywatershedDerivation : Derivation plugin.
    hydro_param.formatters.pywatershed.PywatershedFormatter : Output formatter.
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )

    import geopandas as gpd
    import xarray as xr

    from hydro_param.derivations.pywatershed import PywatershedDerivation
    from hydro_param.plugins import DerivationContext, get_formatter
    from hydro_param.pywatershed_config import load_pywatershed_config
    from hydro_param.sir_accessor import SIRAccessor

    try:
        pws_config = load_pywatershed_config(config)
    except Exception as exc:
        logger.exception("Failed to load pywatershed config '%s'.", config)
        raise SystemExit(1) from exc

    logger.info("pywatershed config validated: %s", config)
    logger.info("  Time: %s to %s", pws_config.time.start, pws_config.time.end)
    logger.info("  SIR path: %s", pws_config.sir_path)
    logger.info("  Output: %s", pws_config.output.path)

    # ── Resolve SIR path ──
    sir_path = pws_config.sir_path
    if not sir_path.is_absolute():
        sir_path = config.parent / sir_path
    sir_path = sir_path.resolve()

    try:
        sir = SIRAccessor(sir_path)
    except FileNotFoundError as exc:
        logger.error("SIR output not found: %s", exc)
        logger.error("Run 'hydro-param run pipeline.yml' first to produce SIR output.")
        raise SystemExit(1) from exc

    # ── Validate available datasets against registry ──
    pws_config.validate_available_fields()

    # ── Log declared parameter entries for diagnostics ──
    declared = pws_config.declared_entries()
    if declared:
        logger.info("Config declares %d parameter entries:", len(declared))
        for name, entry in declared.items():
            var = entry.variable or entry.variables
            logger.info("  %s <- %s.%s", name, entry.source, var)
    else:
        logger.warning("No parameter entries declared in config — derivation will use SIR as-is.")

    # ── Load fabric ──
    fabric_path = pws_config.domain.fabric_path
    if not fabric_path.exists():
        logger.error(
            "Fabric file not found: '%s'. Check domain.fabric_path in '%s'.",
            fabric_path,
            config,
        )
        raise SystemExit(1)
    try:
        fabric = gpd.read_file(fabric_path)
    except Exception as exc:
        logger.exception("Failed to read fabric file '%s'.", fabric_path)
        raise SystemExit(1) from exc

    # ── Load optional segments / waterbodies ──
    segments = None
    if pws_config.domain.segment_path is not None:
        seg_path = pws_config.domain.segment_path
        if not seg_path.exists():
            logger.error(
                "Segment file not found: '%s'. Check domain.segment_path in '%s'.",
                seg_path,
                config,
            )
            raise SystemExit(1)
        try:
            segments = gpd.read_file(seg_path)
        except Exception as exc:
            logger.exception(
                "Failed to read segment file '%s'. Ensure it is a valid GeoPackage or GeoParquet.",
                seg_path,
            )
            raise SystemExit(1) from exc

    waterbodies = None
    if pws_config.domain.waterbody_path is not None:
        wb_path = pws_config.domain.waterbody_path
        if not wb_path.exists():
            logger.error(
                "Waterbody file not found: '%s'. Check domain.waterbody_path in '%s'.",
                wb_path,
                config,
            )
            raise SystemExit(1)
        try:
            waterbodies = gpd.read_file(wb_path)
        except Exception as exc:
            logger.exception(
                "Failed to read waterbody file '%s'. "
                "Ensure it is a valid GeoPackage or GeoParquet.",
                wb_path,
            )
            raise SystemExit(1) from exc
        if "ftype" not in waterbodies.columns:
            logger.error(
                "Waterbody file '%s' is missing required 'ftype' column. "
                "Expected NHDPlus waterbody polygons with 'ftype' values "
                "like 'LakePond' and 'Reservoir'. Found columns: %s",
                wb_path,
                sorted(waterbodies.columns.tolist()),
            )
            raise SystemExit(1)

    # ── Load temporal data from SIR ──
    temporal: dict[str, xr.Dataset] = {}
    try:
        for name in sir.available_temporal():
            try:
                temporal[name] = sir.load_temporal(name)
            except (OSError, KeyError) as exc:
                logger.error("Failed to load temporal SIR data '%s': %s", name, exc)
                logger.error("Re-run 'hydro-param run pipeline.yml' to regenerate SIR output.")
                raise SystemExit(1) from exc

        if temporal:
            logger.info(
                "Loaded %d temporal datasets: %s",
                len(temporal),
                list(temporal.keys()),
            )
        else:
            logger.info(
                "No temporal data in SIR; forcing generation will be skipped "
                "and PET/transpiration will use scalar defaults."
            )

        # ── Derive parameters ──
        logger.info("Deriving pywatershed parameters from SIR")

        derivation_config: dict = {}
        if pws_config.parameter_overrides.values:
            derivation_config["parameter_overrides"] = {
                "values": pws_config.parameter_overrides.values,
            }

        # Build precomputed map from static_datasets declared entries.
        # This tells the derivation plugin which parameters have pre-computed
        # SIR values (e.g., GFv1.1 rasters) that should be used directly
        # instead of being derived from source datasets.
        precomputed = _build_precomputed_map(pws_config)

        ctx = DerivationContext(
            sir=sir,
            fabric=fabric,
            segments=segments,
            waterbodies=waterbodies,
            fabric_id_field=pws_config.domain.id_field,
            segment_id_field=pws_config.domain.segment_id_field,
            config=derivation_config,
            precomputed=precomputed or None,
            temporal=temporal,
        )

        plugin = PywatershedDerivation()
        derived = plugin.derive(ctx)
    except SystemExit:
        raise
    except Exception as exc:
        logger.exception("Parameter derivation failed.")
        raise SystemExit(1) from exc
    finally:
        for ds in temporal.values():
            try:
                ds.close()
            except Exception:
                logger.debug("Failed to close temporal dataset", exc_info=True)

    # ── Format and write ──
    formatter = get_formatter("pywatershed")
    formatter_config = {
        "parameter_file": pws_config.output.parameter_file,
        "forcing_dir": pws_config.output.forcing_dir,
        "soltab_file": pws_config.output.soltab_file,
        "control_file": pws_config.output.control_file,
        "start": pws_config.time.start,
        "end": pws_config.time.end,
    }
    try:
        formatter.write(derived, pws_config.output.path, formatter_config)
    except Exception as exc:
        logger.exception("Failed to write pywatershed output to '%s'.", pws_config.output.path)
        raise SystemExit(1) from exc

    soltab_path = Path(pws_config.output.path) / pws_config.output.soltab_file
    if not soltab_path.exists():
        logger.info(
            "soltab.nc was not produced. Ensure the topography dataset includes "
            "elevation, slope, and aspect variables. Solar radiation tables will "
            "not be available for this run."
        )
    elif soltab_path.stat().st_size == 0:
        logger.warning(
            "soltab.nc exists but is empty (0 bytes). This may indicate a write "
            "failure. Solar radiation tables may not be usable."
        )

    logger.info("pywatershed model setup complete: %s", pws_config.output.path)

pws_validate_cmd

pws_validate_cmd(param_file: Path) -> None

Validate a pywatershed parameter file against metadata constraints.

Check that required PRMS parameters are present and that values fall within the valid ranges defined in the bundled parameter_metadata.yml. Print a summary of issues found or a success message.

PARAMETER	DESCRIPTION
param_file	Path to a pywatershed parameter NetCDF file (e.g., output/parameters.nc). TYPE: Path

RAISES	DESCRIPTION
SystemExit	If the file cannot be opened (exit code 1) or validation finds any issues (exit code 1).

See Also

PywatershedFormatter.validate : Underlying validation logic.

Source code in src/hydro_param/cli.py

@pws_app.command(name="validate")
def pws_validate_cmd(
    param_file: Path,
) -> None:
    """Validate a pywatershed parameter file against metadata constraints.

    Check that required PRMS parameters are present and that values
    fall within the valid ranges defined in the bundled
    ``parameter_metadata.yml``.  Print a summary of issues found or
    a success message.

    Parameters
    ----------
    param_file
        Path to a pywatershed parameter NetCDF file (e.g.,
        ``output/parameters.nc``).

    Raises
    ------
    SystemExit
        If the file cannot be opened (exit code 1) or validation
        finds any issues (exit code 1).

    See Also
    --------
    PywatershedFormatter.validate : Underlying validation logic.
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )

    import xarray as xr

    from hydro_param.formatters.pywatershed import PywatershedFormatter

    try:
        ds = xr.open_dataset(param_file)
    except Exception as exc:
        print(f"Error: Could not open '{param_file}': {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

    formatter = PywatershedFormatter()
    warnings = formatter.validate(ds)
    ds.close()

    if warnings:
        print(f"Validation found {len(warnings)} issue(s):")
        for w in warnings:
            print(f"  - {w}")
        raise SystemExit(1)
    else:
        print("Validation passed: all checks OK.")

gfv11_download_cmd

gfv11_download_cmd(
    *, output_dir: Path, items: str = "all"
) -> None

Download GFv1.1 NHM data layer rasters from ScienceBase.

Stream-download and unzip the original CONUS-wide rasters used to parameterize NHM v1.1 from two public ScienceBase items (~15 GB total). Files are organised into categorised subdirectories (soils, land_cover, topo, etc.) under output_dir.

Existing extracted files are skipped automatically. Zip archives are cleaned up after successful extraction.

PARAMETER	DESCRIPTION
output_dir	Shared data directory for downloaded rasters. Subdirectories (soils/, land_cover/, topo/, etc.) are created automatically. TYPE: Path
items	Which ScienceBase item(s) to download. One of "all" (default), "data-layers", or "tgf-topo". TYPE: str DEFAULT: 'all'

RAISES	DESCRIPTION
SystemExit	If items is not a valid choice or a fatal download error occurs (exit code 1).

See Also

hydro_param.gfv11 : Download implementation module.

Source code in src/hydro_param/cli.py

@gfv11_app.command(name="download")
def gfv11_download_cmd(
    *,
    output_dir: Path,
    items: str = "all",
) -> None:
    """Download GFv1.1 NHM data layer rasters from ScienceBase.

    Stream-download and unzip the original CONUS-wide rasters used to
    parameterize NHM v1.1 from two public ScienceBase items (~15 GB
    total).  Files are organised into categorised subdirectories
    (soils, land_cover, topo, etc.) under *output_dir*.

    Existing extracted files are skipped automatically.  Zip archives
    are cleaned up after successful extraction.

    Parameters
    ----------
    output_dir
        Shared data directory for downloaded rasters.  Subdirectories
        (``soils/``, ``land_cover/``, ``topo/``, etc.) are created
        automatically.
    items
        Which ScienceBase item(s) to download.  One of ``"all"``
        (default), ``"data-layers"``, or ``"tgf-topo"``.

    Raises
    ------
    SystemExit
        If *items* is not a valid choice or a fatal download error
        occurs (exit code 1).

    See Also
    --------
    hydro_param.gfv11 : Download implementation module.
    """
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%H:%M:%S",
    )

    valid_items = {"all", "data-layers", "tgf-topo"}
    if items not in valid_items:
        print(
            f"Error: --items must be one of {sorted(valid_items)}, got '{items}'",
            file=sys.stderr,
        )
        raise SystemExit(1)

    import requests as _requests  # type: ignore[import-untyped]

    from hydro_param.gfv11 import DownloadError, GFv11Items, download_gfv11

    try:
        download_gfv11(output_dir, items=cast(GFv11Items, items))
    except DownloadError as exc:
        logger.error("GFv1.1 download incomplete: %s", exc)
        raise SystemExit(1) from exc
    except (_requests.RequestException, ValueError) as exc:
        logger.error("GFv1.1 download failed due to a network/API error: %s", exc)
        logger.error("Check your internet connection and try again.")
        raise SystemExit(1) from exc
    except OSError as exc:
        logger.error("GFv1.1 download failed due to a filesystem error: %s", exc)
        raise SystemExit(1) from exc

main

main() -> None

Invoke the cyclopts CLI application.

This is the entry point registered as the hydro-param console script in pyproject.toml. Delegates to the cyclopts App for argument parsing and command dispatch.

Source code in src/hydro_param/cli.py

def main() -> None:
    """Invoke the cyclopts CLI application.

    This is the entry point registered as the ``hydro-param`` console
    script in ``pyproject.toml``.  Delegates to the cyclopts ``App``
    for argument parsing and command dispatch.
    """
    app()