11784708548

Committed 11 Nov 2024 07:09PM UTC coverage: 89.75% (+0.4%) from 89.398%

Build # 11784708548

Build Type

Pull #1971

github

Committed by

web-flow

Commit Message

Merge ebf45886b into d5aefa4a9

Pull Request Pull Request #1971: Employ a `src`-based layout for code base

Run Details

4 of 6 new or added lines in 2 files covered. (66.67%)

9378 of 10449 relevant lines covered (89.75%)

7.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.4

/src/xclim/cli.py

"""
=============================
Command Line Interface module
=============================
"""

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 MissingVariableError
from xclim.core.dataflags import DataQualityException, data_flags, ecad_compliant
from xclim.core.utils import InputKind
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,
)

distributed = False
try:
    from dask.distributed import Client, progress  # pylint: disable=ungrouped-imports

    distributed = True
except ImportError:  # noqa: S110
    # Distributed is not a dependency of xclim
    pass


def _get_indicator(indicator_name):
    try:
        return xc.core.indicator.registry[indicator_name.upper()].get_instance()  # noqa
    except KeyError as e:
        raise click.BadArgumentUsage(
            f"Indicator '{indicator_name}' not found in xclim."
        ) from e


def _get_input(ctx):
    """Return the input dataset stored in the given context.

    If the dataset is not open, opens it with open_dataset if a single path was given,
    or with `open_mfdataset` if a tuple or glob path was given.
    """
    arg = ctx.obj["input"]
    if arg is None:
        raise click.BadOptionUsage("input", "No input file name given.", ctx.parent)
    if isinstance(arg, xr.Dataset):
        return arg
    if isinstance(arg, tuple) or "*" in arg:
        ctx.obj["xr_kwargs"].setdefault("combine", "by_coords")
        ds = xr.open_mfdataset(arg, **ctx.obj["xr_kwargs"])
    else:
        ctx.obj["xr_kwargs"].pop("combine", None)
        ds = xr.open_dataset(arg, **ctx.obj["xr_kwargs"])
    ctx.obj["input"] = ds
    return ds


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"]


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

    if isinstance(out, tuple):
        dsout = dsout.assign(**{var.name: var for var in out})
    else:
        dsout = dsout.assign({out.name: out})
    ctx.obj["ds_out"] = dsout


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):
    """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):
    """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} ...")
    click.echo(
        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):
    """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):
    """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 indices(info):  # noqa
    """List all indicators."""
    formatter = click.HelpFormatter()
    formatter.write_heading("Listing all available indicators for computation.")
    rows = []
    for name, indcls in xc.core.indicator.registry.items():  # noqa
        left = click.style(name.lower(), fg="yellow")
        right = ", ".join(
            [var.get("long_name", var["var_name"]) for var in indcls.cf_attrs]
        )
        if indcls.cf_attrs[0]["var_name"] != name.lower():
            right += (
                " (" + ", ".join([var["var_name"] for var in indcls.cf_attrs]) + ")"
            )
        if info:
            right += "\n" + indcls.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):
    """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())


def _format_dict(data, formatter, key_fg="blue", spaces=2):
    for attr, val in data.items():
        if isinstance(val, list):
            for isub, sub in enumerate(val):
                formatter.write_text(
                    click.style(" " * spaces + f"{attr} (#{isub + 1})", fg=key_fg)
                )
                _format_dict(sub, formatter, key_fg=key_fg, spaces=spaces + 2)
        elif isinstance(val, dict):
            formatter.write_text(click.style(" " * spaces + f"{attr}:", fg=key_fg))
            _format_dict(val, formatter, key_fg=key_fg, spaces=spaces + 2)
        else:
            formatter.write_text(
                click.style(" " * spaces + attr + " :", fg=key_fg) + " " + str(val)
            )


class XclimCli(click.MultiCommand):
    """Main cli class."""

    def list_commands(self, ctx):
        """Return the available commands (other than the indicators)."""
        return (
            "indices",
            "info",
            "dataflags",
            "prefetch_testing_data",
            "release_notes",
            "show_version_info",
        )

    def get_command(self, ctx, cmd_name):
        """Return the requested command."""
        command = {
            "dataflags": dataflags,
            "indices": indices,
            "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,
    help="Command line tool to compute indices 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):
    """Entry point for the command line interface.

    Manages the global options.
    """
    if not kwargs["verbose"]:
        warnings.simplefilter("ignore", FutureWarning)
        warnings.simplefilter("ignore", DeprecationWarning)

    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


@cli.result_callback()
@click.pass_context
def write_file(ctx, *args, **kwargs):
    """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.

1	"""
2	=============================
3	Command Line Interface module
4	=============================
5	"""
6
7	from __future__ import annotations	9✔
8
9	import sys	9✔
10	import warnings	9✔
11
12	import click	9✔
13	import xarray as xr	9✔
14	from dask.diagnostics.progress import ProgressBar	9✔
15
16	import xclim as xc	9✔
17	from xclim.core import MissingVariableError	9✔
18	from xclim.core.dataflags import DataQualityException, data_flags, ecad_compliant	9✔
19	from xclim.core.utils import InputKind	9✔
20	from xclim.testing.utils import (	9✔
21	TESTDATA_BRANCH,
22	TESTDATA_CACHE_DIR,
23	TESTDATA_REPO_URL,
24	default_testdata_cache,
25	default_testdata_repo_url,
26	default_testdata_version,
27	populate_testing_data,
28	publish_release_notes,
29	show_versions,
30	)
31
32	distributed = False	9✔
33	try:	9✔
34	from dask.distributed import Client, progress # pylint: disable=ungrouped-imports	9✔
35
36	distributed = True	2✔
37	except ImportError: # noqa: S110	7✔
38	# Distributed is not a dependency of xclim
39	pass	7✔
40
41
42	def _get_indicator(indicator_name):	9✔
43	try:	9✔
44	return xc.core.indicator.registry[indicator_name.upper()].get_instance() # noqa	9✔
45	except KeyError as e:	9✔
46	raise click.BadArgumentUsage(	9✔
47	f"Indicator '{indicator_name}' not found in xclim."
48	) from e
49
50
51	def _get_input(ctx):	9✔
52	"""Return the input dataset stored in the given context.
53
54	If the dataset is not open, opens it with open_dataset if a single path was given,
55	or with `open_mfdataset` if a tuple or glob path was given.
56	"""
57	arg = ctx.obj["input"]	9✔
58	if arg is None:	9✔
59	raise click.BadOptionUsage("input", "No input file name given.", ctx.parent)	9✔
60	if isinstance(arg, xr.Dataset):	9✔
61	return arg	9✔
62	if isinstance(arg, tuple) or "*" in arg:	9✔
63	ctx.obj["xr_kwargs"].setdefault("combine", "by_coords")	9✔
64	ds = xr.open_mfdataset(arg, **ctx.obj["xr_kwargs"])	9✔
65	else:
66	ctx.obj["xr_kwargs"].pop("combine", None)	9✔
67	ds = xr.open_dataset(arg, **ctx.obj["xr_kwargs"])	9✔
68	ctx.obj["input"] = ds	9✔
69	return ds	9✔
70
71
72	def _get_output(ctx):	9✔
73	"""Return the output dataset stored in the given context.
74
75	If the output dataset doesn't exist, create it.
76	"""
77	if "ds_out" not in ctx.obj:	9✔
78	ds_in = _get_input(ctx)	9✔
79	ctx.obj["ds_out"] = xr.Dataset(attrs=ds_in.attrs)	9✔
80	if ctx.obj["output"] is None:	9✔
81	raise click.BadOptionUsage(	9✔
82	"output", "No output file name given.", ctx.parent
83	)
84	return ctx.obj["ds_out"]	9✔
85
86
87	def _process_indicator(indicator, ctx, **params):	9✔
88	"""Add given climate indicator to the output dataset from variables in the input dataset.
89
90	Computation is not triggered here if dask is enabled.
91	"""
92	if ctx.obj["verbose"]:	9✔
93	click.echo(f"Processing : {indicator.identifier}")	9✔
94	dsin = _get_input(ctx)	9✔
95	dsout = _get_output(ctx)	9✔
96
97	for key, val in params.items():	9✔
98	if val == "None" or val is None:	9✔
99	params[key] = None	×
100	elif ctx.obj["verbose"]:	9✔
101	click.echo(f"Parsed {key} = {val}")	9✔
102	params["ds"] = dsin	9✔
103
104	try:	9✔
105	out = indicator(**params)	9✔
106	except MissingVariableError as err:	9✔
107	raise click.BadArgumentUsage(err.args[0]) from err	9✔
108
109	if isinstance(out, tuple):	9✔
110	dsout = dsout.assign(**{var.name: var for var in out})	9✔
111	else:
112	dsout = dsout.assign({out.name: out})	9✔
113	ctx.obj["ds_out"] = dsout	9✔
114
115
116	def _create_command(indicator_name):	9✔
117	"""Generate a Click.Command from an xclim Indicator."""
118	indicator = _get_indicator(indicator_name)	9✔
119	params = []	9✔
120	for name, param in indicator.parameters.items():	9✔
121	if name in ["ds"] or param.kind == InputKind.KWARGS:	9✔
122	continue	9✔
123	choices = "" if "choices" not in param else f" Choices: {param.choices}"	9✔
124	params.append(	9✔
125	click.Option(
126	param_decls=[f"--{name}"],
127	default=param.default,
128	show_default=True,
129	help=param.description + choices,
130	metavar=(
131	"VAR_NAME"
132	if param.kind
133	in [
134	InputKind.VARIABLE,
135	InputKind.OPTIONAL_VARIABLE,
136	]
137	else "TEXT"
138	),
139	)
140	)
141
142	@click.pass_context	9✔
143	def _process(ctx, **kwargs):	9✔
144	return _process_indicator(indicator, ctx, **kwargs)	9✔
145
146	return click.Command(	9✔
147	indicator_name,
148	callback=_process,
149	params=params,
150	help=indicator.abstract,
151	short_help=indicator.title,
152	)
153
154
155	@click.command(short_help="Print versions of dependencies for debugging purposes.")	9✔
156	@click.pass_context	9✔
157	def show_version_info(ctx):	9✔
158	"""Print versions of dependencies for debugging purposes."""
159	click.echo(show_versions())	9✔
160	ctx.exit()	9✔
161
162
163	@click.command(short_help="Prefetch xclim testing data for development purposes.")	9✔
164	@click.option(	9✔
165	"-r",
166	"--repo",
167	help="The xclim-testdata repo to be fetched and cached. If not specified, defaults to "
168	f"`XCLIM_TESTDATA_REPO_URL` (if set) or `{default_testdata_repo_url}`.",
169	)
170	@click.option(	9✔
171	"-b",
172	"--branch",
173	help="The xclim-testdata branch to be fetched and cached. If not specified, defaults to "
174	f"`XCLIM_TESTDATA_BRANCH` (if set) or `{default_testdata_version}`.",
175	)
176	@click.option(	9✔
177	"-c",
178	"--cache-dir",
179	help="The xclim-testdata branch to be fetched and cached. If not specified, defaults to "
180	f"`XCLIM_TESTDATA_CACHE` (if set) or `{default_testdata_cache}`.",
181	)
182	@click.pass_context	9✔
183	def prefetch_testing_data(ctx, repo, branch, cache_dir):	9✔
184	"""Prefetch xclim testing data for development purposes."""
185	if repo:	×
186	testdata_repo = repo	×
187	else:
188	testdata_repo = TESTDATA_REPO_URL	×
189	if branch:	×
190	testdata_branch = branch	×
191	else:
192	testdata_branch = TESTDATA_BRANCH	×
193	if cache_dir:	×
194	testdata_cache_dir = cache_dir	×
195	else:
196	testdata_cache_dir = TESTDATA_CACHE_DIR	×
197
198	click.echo(f"Gathering testing data from {testdata_repo}/{testdata_branch} ...")	×
199	click.echo(	×
200	populate_testing_data(
201	repo=testdata_repo, branch=testdata_branch, local_cache=testdata_cache_dir
202	)
203	)
204	click.echo(f"Testing data saved to `{testdata_cache_dir}`.")	×
205	ctx.exit()	×
206
207
208	@click.command(short_help="Print history for publishing purposes.")	9✔
209	@click.option("-m", "--md", is_flag=True, help="Prints the history in Markdown format.")	9✔
210	@click.option(	9✔
211	"-r", "--rst", is_flag=True, help="Prints the history in ReStructuredText format."
212	)
213	@click.option(	9✔
214	"-c", "--changes", help="Pass a custom changelog file to be used instead."
215	)
216	@click.pass_context	9✔
217	def release_notes(ctx, md, rst, changes):	9✔
218	"""Generate the release notes history for publishing purposes."""
219	if md and rst:	9✔
220	raise click.BadArgumentUsage(	9✔
221	"Cannot return both Markdown and ReStructuredText in same release_notes call."
222	)
223	if md:	9✔
224	style = "md"	9✔
225	elif rst:	9✔
226	style = "rst"	9✔
227	else:
228	raise click.BadArgumentUsage(	9✔
229	"Must specify Markdown (-m) or ReStructuredText (-r)."
230	)
231
232	if changes:	9✔
233	click.echo(f"{publish_release_notes(style, changes=changes)}")	9✔
234	else:
NEW 235	click.echo(f"{publish_release_notes(style)}")	×
236	ctx.exit()	9✔
237
238
239	@click.command(short_help="Run data flag checks for input variables.")	9✔
240	@click.argument("variables", required=False, nargs=-1)	9✔
241	@click.option(	9✔
242	"-r",
243	"--raise-flags",
244	is_flag=True,
245	help="Print an exception in the event that a variable is found to have quality control issues.",
246	)
247	@click.option(	9✔
248	"-a",
249	"--append",
250	is_flag=True,
251	help="Return the netCDF dataset with the `ecad_qc_flag` array appended as a data_var.",
252	)
253	@click.option(	9✔
254	"-d",
255	"--dims",
256	default="all",
257	help='Dimensions upon which aggregation should be performed. Default: "all". Ignored if no variable provided.',
258	)
259	@click.option(	9✔
260	"-f",
261	"--freq",
262	default=None,
263	help="Resampling periods frequency used for aggregation. Default: None. Ignored if no variable provided.",
264	)
265	@click.pass_context	9✔
266	def dataflags(ctx, variables, raise_flags, append, dims, freq):	9✔
267	"""Run quality control checks on input data variables and flag for quality control issues or suspicious values."""
268	ds = _get_input(ctx)	9✔
269	flagged = xr.Dataset()	9✔
270	output = ctx.obj["output"]	9✔
271	if dims == "none":	9✔
272	dims = None	×
273
274	if output and raise_flags:	9✔
275	ctx.fail(	×
276	click.BadOptionUsage(
277	"raise_flags",
278	"Cannot use 'raise_flags' with output netCDF.",
279	ctx.parent,
280	)
281	)
282	if not output and not raise_flags:	9✔
283	ctx.fail(	×
284	click.BadOptionUsage(
285	"raise_flags",
286	"Must specify output or call with 'raise_flags'.",
287	ctx.parent,
288	)
289	)
290
291	if variables:	9✔
292	exit_code = 0	9✔
293	for v in variables:	9✔
294	try:	9✔
295	flagged_var = data_flags(	9✔
296	ds[v], ds, dims=dims, freq=freq, raise_flags=raise_flags
297	)
298	if output:	9✔
299	flagged = xr.merge([flagged, flagged_var])	9✔
300	except DataQualityException as e:	×
301	exit_code = 1	×
302	tb = sys.exc_info()	×
303	click.echo(e.with_traceback(tb[2]))	×
304	if raise_flags:	9✔
305	ctx.exit(exit_code)	×
306	else:
307	try:	3✔
308	flagged = ecad_compliant(	3✔
309	ds, dims=dims, raise_flags=raise_flags, append=append
310	)
311	if raise_flags:	3✔
312	click.echo("Dataset passes quality control checks!")	3✔
313	ctx.exit()	3✔
314	except DataQualityException as e:	3✔
315	tb = sys.exc_info()	×
316	click.echo(e.with_traceback(tb[2]))	×
317	ctx.exit(1)	×
318
319	if output:	9✔
320	ctx.obj["ds_out"] = flagged	9✔
321
322
323	@click.command(short_help="List indicators.")	9✔
324	@click.option(	9✔
325	"-i", "--info", is_flag=True, help="Prints more details for each indicator."
326	)
327	def indices(info): # noqa	9✔
328	"""List all indicators."""
329	formatter = click.HelpFormatter()	9✔
330	formatter.write_heading("Listing all available indicators for computation.")	9✔
331	rows = []	9✔
332	for name, indcls in xc.core.indicator.registry.items(): # noqa	9✔
333	left = click.style(name.lower(), fg="yellow")	9✔
334	right = ", ".join(	9✔
335	[var.get("long_name", var["var_name"]) for var in indcls.cf_attrs]
336	)
337	if indcls.cf_attrs[0]["var_name"] != name.lower():	9✔
338	right += (	9✔
339	" (" + ", ".join([var["var_name"] for var in indcls.cf_attrs]) + ")"
340	)
341	if info:	9✔
342	right += "\n" + indcls.abstract	×
343	rows.append((left, right))	9✔
344	rows.sort(key=lambda row: row[0])	9✔
345	formatter.write_dl(rows)	9✔
346	click.echo(formatter.getvalue())	9✔
347
348
349	@click.command()	9✔
350	@click.argument("indicator", nargs=-1)	9✔
351	@click.pass_context	9✔
352	def info(ctx, indicator):	9✔
353	"""Give information about INDICATOR."""
354	for indname in indicator:	9✔
355	ind = _get_indicator(indname)	9✔
356	command = _create_command(indname)	9✔
357	formatter = click.HelpFormatter()	9✔
358	with formatter.section(	9✔
359	click.style("Indicator", fg="blue")
360	+ click.style(f" {indname}", fg="yellow")
361	):
362	data = ind.json()	9✔
363	data.pop("parameters")	9✔
364	_format_dict(data, formatter, key_fg="blue", spaces=2)	9✔
365
366	command.format_options(ctx, formatter)	9✔
367
368	click.echo(formatter.getvalue())	9✔
369
370
371	def _format_dict(data, formatter, key_fg="blue", spaces=2):	9✔
372	for attr, val in data.items():	9✔
373	if isinstance(val, list):	9✔
374	for isub, sub in enumerate(val):	9✔
375	formatter.write_text(	9✔
376	click.style(" " * spaces + f"{attr} (#{isub + 1})", fg=key_fg)
377	)
378	_format_dict(sub, formatter, key_fg=key_fg, spaces=spaces + 2)	9✔
379	elif isinstance(val, dict):	9✔
380	formatter.write_text(click.style(" " * spaces + f"{attr}:", fg=key_fg))	×
381	_format_dict(val, formatter, key_fg=key_fg, spaces=spaces + 2)	×
382	else:
383	formatter.write_text(	9✔
384	click.style(" " * spaces + attr + " :", fg=key_fg) + " " + str(val)
385	)
386
387
388	class XclimCli(click.MultiCommand):	9✔
389	"""Main cli class."""
390
391	def list_commands(self, ctx):	9✔
392	"""Return the available commands (other than the indicators)."""
393	return (	×
394	"indices",
395	"info",
396	"dataflags",
397	"prefetch_testing_data",
398	"release_notes",
399	"show_version_info",
400	)
401
402	def get_command(self, ctx, cmd_name):	9✔
403	"""Return the requested command."""
404	command = {	9✔
405	"dataflags": dataflags,
406	"indices": indices,
407	"info": info,
408	"prefetch_testing_data": prefetch_testing_data,
409	"release_notes": release_notes,
410	"show_version_info": show_version_info,
411	}.get(cmd_name)
412	if command is None:	9✔
413	command = _create_command(cmd_name)	9✔
414	return command	9✔
415
416
417	@click.command(	9✔
418	cls=XclimCli,
419	chain=True,
420	help="Command line tool to compute indices on netCDF datasets. Indicators are referred to by their "
421	"(case-insensitive) identifier, as in xclim.core.indicator.registry.",
422	invoke_without_command=True,
423	subcommand_metavar="INDICATOR1 [OPTIONS] ... [INDICATOR2 [OPTIONS] ... ] ...",
424	)
425	@click.option(	9✔
426	"-i",
427	"--input",
428	help="Input files. Can be a netCDF path or a glob pattern.",
429	multiple=True,
430	)
431	@click.option("-o", "--output", help="Output filepath. A new file will be created")	9✔
432	@click.option(	9✔
433	"-v", "--verbose", help="Print details about context and progress.", count=True
434	)
435	@click.option(	9✔
436	"-V", "--version", is_flag=True, help="Prints xclim's version number and exits"
437	)
438	@click.option(	9✔
439	"--dask-nthreads",
440	type=int,
441	help="Start a dask.distributed Client with this many threads and 1 worker. "
442	"If not specified, the local scheduler is used. If specified, '--dask-maxmem' must also be given",
443	)
444	@click.option(	9✔
445	"--dask-maxmem",
446	help="Memory limit for the dask.distributed Client as a human readable string (ex: 4GB). "
447	"If specified, '--dask-nthreads' must also be specified.",
448	)
449	@click.option(	9✔
450	"--chunks",
451	help="Chunks to use when opening the input dataset(s). "
452	"Given as <dim1>:num,<dim2:num>. Ex: time:365,lat:168,lon:150.",
453	)
454	@click.option(	9✔
455	"--engine",
456	help="Engine to use when opening the input dataset(s). "
457	"If not specified, xarray decides.",
458	)
459	@click.pass_context	9✔
460	def cli(ctx, **kwargs):	9✔
461	"""Entry point for the command line interface.
462
463	Manages the global options.
464	"""
465	if not kwargs["verbose"]:	9✔
466	warnings.simplefilter("ignore", FutureWarning)	9✔
467	warnings.simplefilter("ignore", DeprecationWarning)	9✔
468
469	if kwargs["version"]:	9✔
470	click.echo(f"xclim {xc.__version__}")	9✔
471	elif ctx.invoked_subcommand is None:	9✔
472	raise click.UsageError("Missing command.", ctx)	9✔
473
474	if len(kwargs["input"]) == 0:	9✔
475	kwargs["input"] = None	9✔
476	elif len(kwargs["input"]) == 1:	9✔
477	kwargs["input"] = kwargs["input"][0]	9✔
478
479	if kwargs["dask_nthreads"] is not None:	9✔
480	if not distributed:	9✔
481	raise click.BadOptionUsage(	7✔
482	"dask_nthreads",
483	"Dask's distributed scheduler is not installed, only the "
484	"local scheduler (non-customizable) can be used.",
485	ctx,
486	)
487	if kwargs["dask_maxmem"] is None:	2✔
488	raise click.BadOptionUsage(	2✔
489	"dask_nthreads",
490	"'--dask-maxmem' must be given if '--dask-nthreads' is given.",
491	ctx,
492	)
493
494	client = Client(	×
495	n_workers=1,
496	threads_per_worker=kwargs["dask_nthreads"],
497	memory_limit=kwargs["dask_maxmem"],
498	)
499	click.echo(	×
500	"Dask client started. The dashboard is available at http://127.0.0.1:"
501	f"{client.scheduler_info()['services']['dashboard']}/status"
502	)
503	if kwargs["chunks"] is not None:	9✔
504	kwargs["chunks"] = {	9✔
505	dim: int(num)
506	for dim, num in map(lambda x: x.split(":"), kwargs["chunks"].split(","))
507	}
508
509	kwargs["xr_kwargs"] = {	9✔
510	"chunks": kwargs["chunks"] or {},
511	}
512	ctx.obj = kwargs	9✔
513
514
515	@cli.result_callback()	9✔
516	@click.pass_context	9✔
517	def write_file(ctx, args, *kwargs):	9✔
518	"""Write the output dataset to file."""
519	if ctx.obj["output"] is not None:	9✔
520	if ctx.obj["verbose"]:	9✔
521	click.echo(f"Writing to file {ctx.obj['output']}")	9✔
522	with ProgressBar():	9✔
523	r = ctx.obj["ds_out"].to_netcdf(	9✔
524	ctx.obj["output"], engine=kwargs["engine"], compute=False
525	)
526	if ctx.obj["dask_nthreads"] is not None:	9✔
527	progress(r.data)	×
528	r.compute()	9✔
529	if ctx.obj["dask_nthreads"] is not None:	9✔
530	click.echo("") # Distributed's progress doesn't print a final \n.	×

Ouranosinc / xclim / 11784708548

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous