"""
=============================
Command Line Interface module
=============================
""" # numpydoc ignore=SS03,SS06
from __future__ import annotations
import sys
import warnings
import click
import xarray as xr
from dask.diagnostics.progress import ProgressBar
import xclim as xc
from xclim.core import InputKind, MissingVariableError, indicator
from xclim.core.dataflags import DataQualityException, data_flags, ecad_compliant
from xclim.testing.utils import (
TESTDATA_BRANCH,
TESTDATA_CACHE_DIR,
TESTDATA_REPO_URL,
default_testdata_cache,
default_testdata_repo_url,
default_testdata_version,
populate_testing_data,
publish_release_notes,
show_versions,
)
CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]}
# Distributed is not a dependency of xclim
distributed = False
try:
from dask.distributed import Client, progress # pylint: disable=ungrouped-imports
distributed = True
except ImportError:
Client, progress = None, None
[docs]
def _get_indicator(indicator_name):
try:
return indicator.registry[indicator_name]
except KeyError as e:
raise click.BadArgumentUsage(f"Indicator '{indicator_name}' not found in xclim.") from e
[docs]
def _get_output(ctx):
"""
Return the output dataset stored in the given context.
If the output dataset doesn't exist, create it.
"""
if "ds_out" not in ctx.obj:
ds_in = _get_input(ctx)
ctx.obj["ds_out"] = xr.Dataset(attrs=ds_in.attrs)
if ctx.obj["output"] is None:
raise click.BadOptionUsage("output", "No output file name given.", ctx.parent)
return ctx.obj["ds_out"]
[docs]
def _process_indicator(indicator, ctx, **params):
"""
Add given climate indicator to the output dataset from variables in the input dataset.
Computation is not triggered here if dask is enabled.
"""
if ctx.obj["verbose"]:
click.echo(f"Processing : {indicator.identifier}")
dsin = _get_input(ctx)
dsout = _get_output(ctx)
for key, val in params.items():
if val == "None" or val is None:
params[key] = None
elif ctx.obj["verbose"]:
click.echo(f"Parsed {key} = {val}")
params["ds"] = dsin
try:
out = indicator(**params)
except MissingVariableError as err:
raise click.BadArgumentUsage(err.args[0]) from err
dsout = xr.merge([out, dsout], combine_attrs="override")
ctx.obj["ds_out"] = dsout
[docs]
def _create_command(indicator_name):
"""Generate a Click.Command from an xclim Indicator."""
indicator = _get_indicator(indicator_name)
params = []
for name, param in indicator.parameters.items():
if name in ["ds"] or param.kind == InputKind.KWARGS:
continue
choices = "" if "choices" not in param else f" Choices: {param.choices}"
params.append(
click.Option(
param_decls=[f"--{name}"],
default=param.default,
show_default=True,
help=param.description + choices,
metavar=(
"VAR_NAME"
if param.kind
in [
InputKind.VARIABLE,
InputKind.OPTIONAL_VARIABLE,
]
else "TEXT"
),
)
)
@click.pass_context
def _process(ctx, **kwargs):
return _process_indicator(indicator, ctx, **kwargs)
return click.Command(
indicator_name,
callback=_process,
params=params,
help=indicator.abstract,
short_help=indicator.title,
)
@click.command(short_help="Print versions of dependencies for debugging purposes.")
@click.pass_context
def show_version_info(ctx): # numpydoc ignore=PR01
"""Print versions of dependencies for debugging purposes."""
click.echo(show_versions())
ctx.exit()
@click.command(short_help="Prefetch xclim testing data for development purposes.")
@click.option(
"-r",
"--repo",
help="The xclim-testdata repo to be fetched and cached. If not specified, defaults to "
f"`XCLIM_TESTDATA_REPO_URL` (if set) or `{default_testdata_repo_url}`.",
)
@click.option(
"-b",
"--branch",
help="The xclim-testdata branch to be fetched and cached. If not specified, defaults to "
f"`XCLIM_TESTDATA_BRANCH` (if set) or `{default_testdata_version}`.",
)
@click.option(
"-c",
"--cache-dir",
help="The xclim-testdata branch to be fetched and cached. If not specified, defaults to "
f"`XCLIM_TESTDATA_CACHE` (if set) or `{default_testdata_cache}`.",
)
@click.pass_context
def prefetch_testing_data(ctx, repo, branch, cache_dir): # numpydoc ignore=PR01
"""Prefetch xclim testing data for development purposes."""
if repo:
testdata_repo = repo
else:
testdata_repo = TESTDATA_REPO_URL
if branch:
testdata_branch = branch
else:
testdata_branch = TESTDATA_BRANCH
if cache_dir:
testdata_cache_dir = cache_dir
else:
testdata_cache_dir = TESTDATA_CACHE_DIR
click.echo(f"Gathering testing data from {testdata_repo}/{testdata_branch} ...")
populate_testing_data(repo=testdata_repo, branch=testdata_branch, local_cache=testdata_cache_dir)
click.echo(f"Testing data saved to `{testdata_cache_dir}`.")
ctx.exit()
@click.command(short_help="Print history for publishing purposes.")
@click.option("-m", "--md", is_flag=True, help="Prints the history in Markdown format.")
@click.option("-r", "--rst", is_flag=True, help="Prints the history in ReStructuredText format.")
@click.option("-c", "--changes", help="Pass a custom changelog file to be used instead.")
@click.pass_context
def release_notes(ctx, md, rst, changes): # numpydoc ignore=PR01
"""Generate the release notes history for publishing purposes."""
if md and rst:
raise click.BadArgumentUsage("Cannot return both Markdown and ReStructuredText in same release_notes call.")
if md:
style = "md"
elif rst:
style = "rst"
else:
raise click.BadArgumentUsage("Must specify Markdown (-m) or ReStructuredText (-r).")
if changes:
click.echo(f"{publish_release_notes(style, changes=changes)}")
else:
click.echo(f"{publish_release_notes(style)}")
ctx.exit()
@click.command(short_help="Run data flag checks for input variables.")
@click.argument("variables", required=False, nargs=-1)
@click.option(
"-r",
"--raise-flags",
is_flag=True,
help="Print an exception in the event that a variable is found to have quality control issues.",
)
@click.option(
"-a",
"--append",
is_flag=True,
help="Return the netCDF dataset with the `ecad_qc_flag` array appended as a data_var.",
)
@click.option(
"-d",
"--dims",
default="all",
help='Dimensions upon which aggregation should be performed. Default: "all". Ignored if no variable provided.',
)
@click.option(
"-f",
"--freq",
default=None,
help="Resampling periods frequency used for aggregation. Default: None. Ignored if no variable provided.",
)
@click.pass_context
def dataflags(ctx, variables, raise_flags, append, dims, freq): # numpydoc ignore=PR01
"""Run quality control checks on input data variables and flag for quality control issues or suspicious values."""
ds = _get_input(ctx)
flagged = xr.Dataset()
output = ctx.obj["output"]
if dims == "none":
dims = None
if output and raise_flags:
ctx.fail(
click.BadOptionUsage(
"raise_flags",
"Cannot use 'raise_flags' with output netCDF.",
ctx.parent,
)
)
if not output and not raise_flags:
ctx.fail(
click.BadOptionUsage(
"raise_flags",
"Must specify output or call with 'raise_flags'.",
ctx.parent,
)
)
if variables:
exit_code = 0
for v in variables:
try:
flagged_var = data_flags(ds[v], ds, dims=dims, freq=freq, raise_flags=raise_flags)
if output:
flagged = xr.merge([flagged, flagged_var])
except DataQualityException as e:
exit_code = 1
tb = sys.exc_info()
click.echo(e.with_traceback(tb[2]))
if raise_flags:
ctx.exit(exit_code)
else:
try:
flagged = ecad_compliant(ds, dims=dims, raise_flags=raise_flags, append=append)
if raise_flags:
click.echo("Dataset passes quality control checks!")
ctx.exit()
except DataQualityException as e:
tb = sys.exc_info()
click.echo(e.with_traceback(tb[2]))
ctx.exit(1)
if output:
ctx.obj["ds_out"] = flagged
@click.command(short_help="List indicators.")
@click.option("-i", "--info", is_flag=True, help="Prints more details for each indicator.")
def indicators(info): # numpydoc ignore=PR01
"""List all indicators."""
formatter = click.HelpFormatter()
formatter.write_heading("Listing all available indicators for computation.")
rows = []
for name, ind in indicator.registry.items():
left = click.style(name.lower(), fg="yellow")
right = ", ".join([outmeta.attrs.get("long_name", outmeta.var_name) for outmeta in ind.outputs])
if ind.outputs[0].var_name.lower() != name:
right += " (" + ", ".join([outmeta.var_name for outmeta in ind.outputs]) + ")"
if info:
right += "\n" + ind.abstract
rows.append((left, right))
rows.sort(key=lambda row: row[0])
formatter.write_dl(rows)
click.echo(formatter.getvalue())
@click.command()
@click.argument("indicator", nargs=-1)
@click.pass_context
def info(ctx, indicator): # numpydoc ignore=PR01
"""Give information about INDICATOR."""
for indname in indicator:
ind = _get_indicator(indname)
command = _create_command(indname)
formatter = click.HelpFormatter()
with formatter.section(click.style("Indicator", fg="blue") + click.style(f" {indname}", fg="yellow")):
data = ind.json()
data.pop("parameters")
_format_dict(data, formatter, key_fg="blue", spaces=2)
command.format_options(ctx, formatter)
click.echo(formatter.getvalue())
[docs]
class XclimCli(click.Group):
"""Main cli class."""
[docs]
def list_commands(self, ctx) -> tuple[str, str, str, str, str, str]: # numpydoc ignore=PR01,RT01
"""Return the available commands (other than the indicators)."""
return (
"list",
"info",
"dataflags",
"prefetch_testing_data",
"release_notes",
"show_version_info",
)
[docs]
def get_command(self, ctx, cmd_name) -> click.Command: # numpydoc ignore=PR01,RT01
"""Return the requested command."""
command = {
"dataflags": dataflags,
"list": indicators,
"info": info,
"prefetch_testing_data": prefetch_testing_data,
"release_notes": release_notes,
"show_version_info": show_version_info,
}.get(cmd_name)
if command is None:
command = _create_command(cmd_name)
return command
@click.command(
cls=XclimCli,
chain=True,
context_settings=CONTEXT_SETTINGS,
help="Command line tool to compute indicators on netCDF datasets. Indicators are referred to by their "
"(case-insensitive) identifier, as in xclim.core.indicator.registry.",
invoke_without_command=True,
subcommand_metavar="INDICATOR1 [OPTIONS] ... [INDICATOR2 [OPTIONS] ... ] ...",
)
@click.option(
"-i",
"--input",
help="Input files. Can be a netCDF path or a glob pattern.",
multiple=True,
)
@click.option("-o", "--output", help="Output filepath. A new file will be created")
@click.option("-v", "--verbose", help="Print details about context and progress.", count=True)
@click.option("-V", "--version", is_flag=True, help="Prints xclim's version number and exits")
@click.option(
"--dask-nthreads",
type=int,
help="Start a dask.distributed Client with this many threads and 1 worker. "
"If not specified, the local scheduler is used. If specified, '--dask-maxmem' must also be given",
)
@click.option(
"--dask-maxmem",
help="Memory limit for the dask.distributed Client as a human readable string (ex: 4GB). "
"If specified, '--dask-nthreads' must also be specified.",
)
@click.option(
"--chunks",
help="Chunks to use when opening the input dataset(s). "
"Given as <dim1>:num,<dim2:num>. Ex: time:365,lat:168,lon:150.",
)
@click.option(
"--engine",
help="Engine to use when opening the input dataset(s). If not specified, xarray decides.",
)
@click.pass_context
def cli(ctx, **kwargs): # numpydoc ignore=PR01
"""
Entry point for the command line interface.
Manages the global options.
"""
if not kwargs["verbose"]:
warnings.simplefilter("ignore", FutureWarning)
warnings.simplefilter("ignore", DeprecationWarning)
warnings.simplefilter("ignore", RuntimeWarning)
if kwargs["version"]:
click.echo(f"xclim {xc.__version__}")
elif ctx.invoked_subcommand is None:
raise click.UsageError("Missing command.", ctx)
if len(kwargs["input"]) == 0:
kwargs["input"] = None
elif len(kwargs["input"]) == 1:
kwargs["input"] = kwargs["input"][0]
if kwargs["dask_nthreads"] is not None:
if not distributed:
raise click.BadOptionUsage(
"dask_nthreads",
"Dask's distributed scheduler is not installed, only the "
"local scheduler (non-customizable) can be used.",
ctx,
)
if kwargs["dask_maxmem"] is None:
raise click.BadOptionUsage(
"dask_nthreads",
"'--dask-maxmem' must be given if '--dask-nthreads' is given.",
ctx,
)
client = Client(
n_workers=1,
threads_per_worker=kwargs["dask_nthreads"],
memory_limit=kwargs["dask_maxmem"],
)
click.echo(
"Dask client started. The dashboard is available at http://127.0.0.1:"
f"{client.scheduler_info()['services']['dashboard']}/status"
)
if kwargs["chunks"] is not None:
kwargs["chunks"] = {dim: int(num) for dim, num in map(lambda x: x.split(":"), kwargs["chunks"].split(","))}
kwargs["xr_kwargs"] = {
"chunks": kwargs["chunks"] or {},
}
ctx.obj = kwargs
[docs]
@cli.result_callback()
@click.pass_context
def write_file(ctx, *_, **kwargs): # numpydoc ignore=PR01
"""Write the output dataset to file."""
if ctx.obj["output"] is not None:
if ctx.obj["verbose"]:
click.echo(f"Writing to file {ctx.obj['output']}")
with ProgressBar():
r = ctx.obj["ds_out"].to_netcdf(ctx.obj["output"], engine=kwargs["engine"], compute=False)
if ctx.obj["dask_nthreads"] is not None:
progress(r.data)
r.compute()
if ctx.obj["dask_nthreads"] is not None:
click.echo("") # Distributed's progress doesn't print a final \n.