# SPDX-FileCopyrightText: Copyright © 2025 Idiap Research Institute <contact@idiap.ch>
#
# SPDX-License-Identifier: GPL-3.0-or-later
"""Tools for command-line applications."""
import csv
import logging
import pathlib
import shutil
import typing
import click
import compact_json
import numpy
from ..scores import Scores
from ..solutions import Solutions
logger = logging.getLogger(__name__)
[docs]
def prepare_and_backup(path: pathlib.Path) -> None:
"""Ensure parent directory exists and back-up copies.
This function will check that the directory leading to a file path exists and will
created it otherwise. It will also check if the file does not already exists, and
back it up otherwise.
Parameters
----------
path
The full path of the file to ensure the parent directory exists and that it is
properly backed-up if necessary.
"""
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists():
backup = path.parent / (path.name + "~")
shutil.copy(path, backup)
[docs]
def save_json_with_backup(path: pathlib.Path, data: dict) -> None:
"""Save a dictionary into a JSON file with path checking and backup.
This function will save a dictionary into a JSON file. It will check
the existence of the directory leading to the file and create it if
necessary. If the file already exists on the destination folder, it is
backed-up before a new file is created with the new contents.
Parameters
----------
path
The full path where to save the JSON data.
data
The data to save on the JSON file.
"""
formatter = compact_json.Formatter()
# only only 2 indent spaces for further levels
formatter.indent_spaces = 2
# controls how much nesting can happen
formatter.max_inline_complexity = 2
# controls the maximum line width (has priority over nesting)
formatter.max_inline_length = 88
# remove any trailing whitespaces
formatter.omit_trailing_whitespace = True
prepare_and_backup(path)
formatter.dump(data, str(path))
[docs]
def save_csv_with_backups(path: pathlib.Path, data: dict[str, typing.Any]):
"""Save a dictionary into a CSV file with path checking and backup.
This function will save a dictionary into a CSV file. It will check
the existence of the directory leading to the file and create it if
necessary. If the file already exists on the destination folder, it is
backed-up before a new file is created with the new contents.
Parameters
----------
path
The full path where to save the CSV data.
data
The data to save on the CSV file.
"""
prepare_and_backup(path)
with path.open("w") as csv_file:
csv_writer = csv.writer(csv_file)
csv_writer.writerow(data.keys())
for d in data.values():
csv_writer.writerow(d)
[docs]
def validate_scores_json(
ctx: click.Context, param: click.Parameter, value: typing.Sequence[pathlib.Path]
) -> list[
tuple[pathlib.Path, Scores]
| tuple[tuple[pathlib.Path, Scores], tuple[pathlib.Path, Solutions]]
]:
"""
Validate one or more JSON files against a predefined data model.
This function is intended to be used as a Click argument callback. It opens the
given file path, parses its JSON content, and validates it against the library data
model. If any validation error occurs, a `click.BadParameter` is raised to inform
the user.
Parameters
----------
ctx
The Click execution context.
param
The parameter that triggered the callback.
value
Path to the JSON files to load and validate.
Returns
-------
The parsed and validated JSON content as a list, where each element is either a
a tuple mapping filename to scores, or a tuple containing two entries, the first
mapping a filename to scores and a second (to be used for selecting *a priori*
thresholds and systems) mapping a filename to pre-calculated solutions.
Raises
------
click.BadParameter
If the file cannot be read, contains invalid JSON, or fails data model
validation.
"""
retval: list[
tuple[pathlib.Path, Scores]
| tuple[tuple[pathlib.Path, Scores], tuple[pathlib.Path, Solutions]]
] = []
def _load_scores(v):
try:
assert v.stem not in [k[0] for k in retval]
logger.info(f"Loading scores for system `{v}`...")
return Scores.load(v)
except AssertionError as e:
raise click.BadParameter(f"The same path was passed more than once: {e}")
except __import__("pydantic").ValidationError as e:
raise click.BadParameter(f"Score data model validation failed: {e}")
def _load_solutions(v):
try:
logger.info(f"Loading solutions for system `{v}`...")
return Solutions.load(v)
except __import__("pydantic").ValidationError as e:
raise click.BadParameter(f"Solution data model validation failed: {e}")
if isinstance(value, pathlib.PosixPath):
value = (value,)
for v in value:
system_groups = [pathlib.Path(p) for p in str(v).split("@")]
scores = _load_scores(system_groups[0])
if len(system_groups) == 1:
retval.append((system_groups[0], scores))
elif len(system_groups) == 2:
retval.append(
(
(system_groups[0], scores),
(system_groups[1], _load_solutions(system_groups[1])),
)
)
else:
raise click.BadParameter(f"Too many systems specified in group: {v}")
return retval
[docs]
def validate_solutions_json(
ctx: click.Context, param: click.Parameter, value: typing.Sequence[pathlib.Path]
) -> dict[str, Solutions]:
"""
Validate one or more JSON files against a predefined data model.
This function is intended to be used as a Click argument callback. It opens the
given file path, parses its JSON content, and validates it against the library data
model. If any validation error occurs, a `click.BadParameter` is raised to inform
the user.
Parameters
----------
ctx
The Click execution context.
param
The parameter that triggered the callback.
value
Path to the JSON files to load and validate.
Returns
-------
The parsed and validated JSON content as a dictionary, where keys represent the
basename of files without extension.
Raises
------
click.BadParameter
If the file cannot be read, contains invalid JSON, or fails data model
validation.
"""
retval: dict[str, Solutions] = {}
if isinstance(value, pathlib.PosixPath):
value = (value,)
for v in value:
try:
assert v.stem not in retval
logger.info(f"Loading solutions for system `{v}`...")
retval[v.stem] = Solutions.load(v)
except AssertionError as e:
raise click.BadParameter(f"The same path was passed more than once: {e}")
except __import__("pydantic").ValidationError as e:
raise click.BadParameter(f"Solution data model validation failed: {e}")
return retval
[docs]
def validate_metrics(
ctx: click.Context, param: click.Parameter, value: typing.Sequence[str]
) -> tuple[str, ...]:
"""Validate a user defined metric for support.
Parameters
----------
ctx
The Click execution context.
param
The parameter that triggered the callback.
value
The value to validate.
Returns
-------
The validated metrics.
Raises
------
click.BadParameter
If one or more of the provided metrics do not validate correctly.
"""
from .. import metrics
invalid = []
for name in value:
try:
metrics.parse_metric(name)
except ValueError:
invalid.append(f"`{name}`")
if invalid:
raise click.BadParameter(f"invalid metric names: {', '.join(invalid)}")
return tuple(value)
[docs]
def validate_thresholds(
ctx: click.Context, param: click.Parameter, value: int | None
) -> None | list[float]:
"""Validate a threshold setup.
Parameters
----------
ctx
The Click execution context.
param
The parameter that triggered the callback.
value
The value to validate.
Returns
-------
The validated threshold, as our API likes it.
Raises
------
click.BadParameter
If the threshold cannot be validated.
"""
if value is None:
return value
return typing.cast(list[float], list(numpy.linspace(0, 1, value, dtype=float)))
[docs]
def verbosity_option(
logger: logging.Logger,
short_name: str = "v",
name: str = "verbose",
dflt: int = 0,
**kwargs: typing.Any,
) -> typing.Callable[..., typing.Any]:
"""Click-option decorator that adds a ``-v``/``--verbose`` option to a cli.
This decorator adds a click option to your CLI to set the log-level on a
provided :py:class:`logging.Logger`. You must specifically determine the
logger that will be affected by this CLI option, via the ``logger`` option.
.. code-block:: python
@verbosity_option(logger=logger)
The verbosity option has the "count" type, and has a default value of 0.
At each time you provide ``-v`` options on the command-line, this value is
increased by one. For example, a CLI setting of ``-vvv`` will set the
value of this option to 3. This is the mapping between the value of this
option (count of ``-v`` CLI options passed) and the log-level set at the
provided logger:
* 0 (no ``-v`` option provided): ``logger.setLevel(logging.ERROR)``
* 1 (``-v``): ``logger.setLevel(logging.WARNING)``
* 2 (``-vv``): ``logger.setLevel(logging.INFO)``
* 3 (``-vvv`` or more): ``logger.setLevel(logging.DEBUG)``
Arguments:
logger: The :py:class:`logging.Logger` to be set.
short_name: Short name of the option. If not set, then use ``v``
name: Long name of the option. If not set, then use ``verbose`` --
this will also become the name of the contextual parameter for click.
dlft: The default verbosity level to use (defaults to 0).
**kwargs: Further keyword-arguments to be forwarded to the underlying
:py:func:`click.option`
Returns
-------
A callable, that follows the :py:mod:`click`-framework policy for
option decorators. Use it accordingly.
"""
def custom_verbosity_option(f):
def callback(ctx, _, value):
ctx.meta[name] = value
log_level: int = { # type: ignore
0: logging.ERROR,
1: logging.WARNING,
2: logging.INFO,
3: logging.DEBUG,
}[value]
# one‐time handler setup
if not getattr(logger, "_verbosity_configured", False):
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter(
"%(levelname)s %(name)s: %(message)s",
)
)
logger.addHandler(handler)
logger._verbosity_configured = True # type: ignore[attr-defined]
logger.setLevel(log_level)
logger.debug(f'Level of Logger("{logger.name}") was set to {log_level}')
return value
return click.option(
f"-{short_name}",
f"--{name}",
count=True,
type=click.IntRange(min=0, max=3, clamp=True),
default=dflt,
show_default=True,
help=(
f"Increase the verbosity level from 0 (only error and "
f"critical) messages will be displayed, to 1 (like 0, but adds "
f"warnings), 2 (like 1, but adds info messags), and 3 (like 2, "
f"but also adds debugging messages) by adding the --{name} "
f"option as often as desired (e.g. '-vvv' for debug)."
),
callback=callback,
**kwargs,
)(f)
return custom_verbosity_option