"""
Indicator collections
=====================
An indicator collection is a structure holding multiple indicators. It can be created through a yaml configuration file.
YAML file structure
~~~~~~~~~~~~~~~~~~~
Indicator-defining yaml files are structured in the following way. Most entries of the `indicators` section are
mirroring attributes of the :py:class:`xclim.core.indicator.Indicator`, please refer to its documentation for more
details on each.
.. code-block:: yaml
module: <module name> # Defaults to the file name
realm: <realm> # If given here, applies to all indicators that do not already provide it.
keywords:
- <keyword> # Merged with indicator-specific keywords (appended to the list)
references: <references> # Merged with indicator-specific references (joined with a new line)
base: <base indicator class> # Defaults to "Daily" and applies to all indicators that do not give it.
doc: <module docstring> # Defaults to a minimal header, only valid if the module doesn't already exist.
variables: # Optional section if indicators declared below rely on variables unknown to xclim
# (not in `xclim.core.VARIABLES`)
# The variables are not module-dependent and will overwrite any already existing with the same name.
<varname>:
canonical_units: <units> # required
description: <description> # required
standard_name: <expected standard_name> # optional
cell_methods: <expected cell_methods> # optional
# The `bases` and `indicators` sections have the same syntax. Indicators defined in the `bases` section
# will only be created as classes and not instances. They will not be included in the IndicatorCollection's items,
# but rather in its `bases` property. This is useful for creating a base from which multiple indicators are declared
# in the `indicators`section.
bases:
indicators:
<identifier>: # The actual indicator identifier will be prepended by the module name.
# From which Indicator to inherit
base: <base indicator class> # Defaults to module-wide base class
# See :ref:`Base class specification` below.
# General metadata, usually parsed from the `compute`s docstring when possible.
realm: <realm> # defaults to module-wide realm. One of "atmos", "land", "seaIce", "ocean".
title: <title>
abstract: <abstract>
keywords:
- <keyword> # merged to module-wide keywords.
references: <references> # newline-seperated, merged to module-wide references.
notes: <notes>
# Other options (not all indicator classes support them)
missing: <missing method name>
missing_options: <missing options mapping>
allowed_periods: [<list>, <of>, <allowed>, <periods>]
context: <context> # A unit context enabled during the conversion of the compute's output to the requested units
# Compute function
compute: <function name> # See :ref:`Compute function specification`, below.
input: # When "compute" is a generic function, this is a mapping from argument name to the expected variable.
# It will change the expected name of the variable as well as its units/dimensionality.
# Can refer to a variable declared in the `variables` section above or in `xclim.core.VARIABLES`.
# See also :ref:`Inputs` below.
<var name in compute> : <variable official name>
...
# Parameters
parameters:
<param name>: <param data> # Simplest case, to inject parameters in the compute function.
# Kwargs-like parameters like ``indexer`` must be injected as a dictionary here.
<param name>: # To change parameters metadata or to declare units when "compute" is a generic function.
default: <param default>
description: <param description>
name: <param name> # Change the name of the parameter (similar to what `input` does for variables)
kind: <param kind> # Override the parameter kind. This is mostly useful for transforming an
# optional variable into a required one by passing ``kind: 0``.
...
# Output metadata
outputs: # List of mappings, one for each output.
- var_name: <var name> # Name to give to the output
units: <units> # Units to convert the output to and assign as attribute
attrs: # Mapping of attributes to assign to the output, can be templated strings.
long_name: <...>
description: <...>
- ...
... # and so on.
All fields are optional. Other fields found in the yaml file will trigger errors when validation is activated.
When a module is built from a yaml file, the yaml is first validated against the schema (see xclim/data/schema.yml)
using the YAMALE library (:cite:p:`lopker_yamale_2022`). See the "Extending xclim" notebook for more info.
Base class specification
^^^^^^^^^^^^^^^^^^^^^^^^
There are multiple ways to specify a base class when defining an indicator. In priority order:
- If ``base`` starts with a '.' (ex: `.RXXp`), the base class is taken from the current module.
+ It is first searched in `bases` section.
+ If not found, it is searched as another indicator declared _above_ the current definition.
- The name is searched in the base class registry, :py:data:`xclim.core.indicator.base_registry` (example: ``Daily``).
- The name is searched in the indicator registry, :py:data:`xclim.core.indicator.registry` (example: ``prcptot``).
- If ``base`` contains a '.' :
+ If the first element is one of xclim's indicators submodules (ex: ``atmos.precip_accumulation``), that indicator
is used as a base. Any submodule of ``xclim.indicators`` are possible.
+ Otherwise, that path is loaded with python's normal import mechanism.
(ex: ``mymodule.submod.MyIndicator`` isloaded as ``from mymodule.submod import MyIndicator``).
If the field ``base`` is not given, it defaults to the module-wide ``base``, which itself defaults to
:py:class:`xclim.core.indicator.Daily``.
Compute function specification
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Similar to the ``base`` field, there are multiple ways to refer to a compute function in the ``compute`` field.
In priority order:
- If a module or mapping of compute functions was passed to :py:meth:`IndicatorCollection.from_yaml`,
the name is searched there (ex: `extreme_precip_accumulation_and_days`).
- The name is searched in :py:mod:`xclim.compute.generic` (ex: ``statistics``).
- The name is searched in `:py:mod:`xclim.compute` (ex: ``corn_heat_units``).
It may contain a '.' to denote a submodule (ex: ``generic.statistics``).
- Otherwise, it is loaded with python's normal import mechanism
(ex: ``mymodule.submod.my_function`` is loaded as ``from mymodule.submod import my_function``).
Inputs
^^^^^^
As xclim has strict definitions of possible input variables (see :py:data:`xclim.core.VARIABLES`),
the mapping of `indicators.<identifier>.input` simply links an argument name from the function given in "compute"
to one of those official variables.
"""
from __future__ import annotations
import warnings
from collections.abc import Callable
from importlib import import_module
from os import PathLike
from pathlib import Path
from types import ModuleType
from typing import Literal
import yamale
from yaml import safe_load
import xclim.compute
import xclim.compute.generic
import xclim.indicators
from xclim.core import VARIABLES, raise_warn_or_log
from xclim.core.indicator import Daily, Indicator, base_registry, registry
from xclim.core.locales import load_locale, read_locale_file
from xclim.core.utils import load_module
[docs]
class IndicatorCollection(dict): # numpydoc ignore=PR01
"""A collection of indicators."""
def __init__(
self,
indicators: dict[str, Indicator],
name: str | None = None,
bases: dict[str, type] = None,
doc: str | None = None,
):
"""
Create an IndicatorCollection.
Parameters
----------
indicators : dict of Indicator
Indicators to put in the new collection.
name : str, optional
The name of the module.
bases : dict, optional
Base indicator classes used in definitions of this collection.
doc : str, optional
Documentation of the collection. Defaults to a simple header.
"""
self.name = name
self.bases = bases or {}
self.__doc__ = doc or f"{name.capitalize()} indicators\n" + "=" * (len(name) + 11)
super().__init__(**indicators)
[docs]
def iter_indicators(self):
"""Iterate over the (name, indicator) pairs in this collection."""
yield from self.items()
[docs]
@classmethod
def from_yaml( # noqa: C90
cls,
filename: PathLike,
name: str | None = None,
computes: dict[str, Callable] | ModuleType | PathLike | None = None,
translations: dict[str, dict | PathLike] | None = None,
mode: Literal["raise", "warn", "ignore"] = "raise",
encoding: str = "UTF8",
validate: bool | PathLike = True,
register: bool = False,
):
"""
Build an indicator collection from a YAML file.
When given only a base filename (no 'yml' extension), this tries to find custom indicators in a module
of the same name (*.py) and translations in json files (*.<lang>.json), see Notes.
Indicator created here will have the name of the module prepended to their identifier (ex: `{mod}.{baseId}`).
The base identifier being the key name within the `indicators` mapping in the yaml.
Parameters
----------
filename : PathLike
Path to a YAML file or to the stem of all module files.
See Notes for behaviour when passing a basename only.
name : str, optional
The name of the new or existing module, defaults to the basename of the file (e.g: `atmos.yml` -> `atmos`).
computes : Mapping of callables or module or path, optional
A mapping or module of compute functions or a python file declaring such a module. When creating the
indicator, the name in the `compute` field is first sought here, then the indicator class will search
in :py:mod:`xclim.compute.generic` and finally in :py:mod:`xclim.compute`.
translations : Mapping of dicts or path, optional
Translated metadata for the new indicators. Keys of the mapping must be two-character language tags.
Values can be translations dictionaries as defined in :py:mod:`xclim.core.locales`.
They can also be a path to a JSON file defining the translations.
mode : {'raise', 'warn', 'ignore'}
How to deal with broken indicator definitions.
encoding : str
The encoding used to open the `.yaml` and `.json` files.
It defaults to UTF-8, overriding python's mechanism which is machine dependent.
validate : bool or PathLike
If True (default), the yaml module is validated against the `xclim` schema.
Can also be the path to a YAML schema against which to validate;
Or False, in which case validation is simply skipped.
register : bool
If True, the indicators created here are registered in xclim's indicators registry
:py:data:`~xclim.core.indicator.registry` upon creation, using the collection's name
prepended to their identifier as key, as explained above.
Defaults to False, making collections independent from xclim's registry.
This does not change the behaviour of registering new variables, which are always added
to xclim's central :py:data:`xclim.core.VARIABLES`.
Returns
-------
IndicatorCollection
A collection of indicators.
See Also
--------
xclim.core.indicator.Indicator : Indicator build logic.
Notes
-----
When the given `filename` has no suffix (usually '.yaml' or '.yml'), the function will try to load
custom compute functions definitions from a file with the same name but with a `.py` extension. Similarly,
it will try to load translations in `*.<lang>.json` files, where `<lang>` is the IETF language tag.
Note that the file name *can not* contain a dot (``.``) for this logic to work.
For example. a set of custom indicators could be fully described by the following files:
- `example.yml` : defining the indicator's metadata.
- `example.py` : defining a few compute functions.
- `example.fr.json` : French translations
"""
filepath = Path(filename)
# A stem was passed, try to load module, functions and translations with same name but different suffixes
is_stem = filepath.suffix not in [".yml", ".yaml"]
if is_stem:
yml_path = filepath.with_suffix(".yml")
else:
yml_path = filepath
# Read YAML file
with yml_path.open(encoding=encoding) as f:
yml = safe_load(f)
if validate is not False:
cls._validate_yaml(
validate if validate is not True else (Path(__file__).parent.parent / "data" / "schema.yml"),
yml_path,
encoding,
)
# Load values from top-level in yml.
# Priority of arguments differ.
coll_name = name or yml.get("name", filepath.stem)
default_base = registry.get(yml.get("base"), base_registry.get(yml.get("base"), Daily))
doc = yml.get("doc")
if is_stem and computes is None and (ind_file := filepath.with_suffix(".py")).is_file():
# No suffix means we try to automatically detect the python file
computes = ind_file
if isinstance(computes, str | Path):
computes = load_module(computes, name=coll_name)
_translations: dict[str, dict] = {}
if is_stem and translations is None:
# No suffix mean we try to automatically detect the json files.
for loc_file in filepath.parent.glob(f"{filepath.stem}.*.json"):
locale = loc_file.suffixes[0][1:]
_translations[locale] = read_locale_file(loc_file, module=coll_name, encoding=encoding)
elif translations is not None:
# A mapping was passed, we read paths if any.
_translations = {
lng: (
read_locale_file(trans, module=coll_name, encoding=encoding)
if isinstance(trans, str | Path)
else trans
)
for lng, trans in translations.items()
}
# Module-wide default values for some attributes
defkwargs = {
# Only used in case the indicator definition does not give them.
"realm": yml.get("realm", "atmos"),
# Merged with a space
"keywords": yml.get("keywords", []),
# Merged with a new line
"references": yml.get("references"),
}
# Parse the variables:
for varname, vardata in yml.get("variables", {}).items():
if varname in VARIABLES and VARIABLES[varname] != vardata:
warnings.warn(
f"Variable {varname} from collection {coll_name} "
"will overwrite the one already defined in `xclim.core.VARIABLES`"
)
VARIABLES[varname] = vardata.copy()
# Parse the indicators:
mapping = {}
bases = {}
# This because we enforce indicators being required and bases being optional
for section, sectiondata in [("bases", yml.get("bases", {})), ("indicators", yml["indicators"])]:
for identifier, data in sectiondata.items():
try:
# Get base class
base = default_base
if (basename := data.pop("base", None)) is not None:
base = cls._find_base_class(basename, mapping, bases)
if (funcname := data.pop("compute", None)) is not None:
data["compute"] = cls._find_compute_function(funcname, computes)
if data.get("references") and defkwargs.get("references"):
data["references"] = f"{data['references']}\n{defkwargs['references']}"
elif defkwargs.get("references"):
data["references"] = defkwargs["references"]
data["keywords"] = [*defkwargs.get("keywords", []), *data.get("keywords", [])]
data.setdefault("realm", defkwargs.get("realm"))
ind = base(
identifier=f"{coll_name}.{identifier}",
module=coll_name,
register=register and (section == "indicators"),
**data,
)
if section == "bases":
bases[identifier] = ind.__class__
else:
mapping[identifier] = ind
except Exception as err: # pylint: disable=broad-except
raise_warn_or_log(err, mode, msg=f"Constructing {identifier} failed with {err!r}")
coll = cls(mapping, name=coll_name, bases=bases, doc=doc)
# If there are translations, load them
if _translations:
for locale, loc_dict in _translations.items():
load_locale(loc_dict, locale)
return coll
[docs]
@staticmethod
def _validate_yaml(schema_path, yml_path, encoding):
# Read schema
schema = yamale.make_schema(schema_path)
# Validate - a YamaleError will be raised if the module does not comply with the schema.
yamale.validate(schema, yamale.make_data(content=yml_path.read_text(encoding=encoding)))
[docs]
@staticmethod
def _find_base_class(name, mapping, bases):
try:
if name.startswith("."):
# A starting dot means the base has been declared above.
if name[1:] in bases:
base = bases[name[1:]]
else:
base = mapping[name[1:]].__class__
elif name in base_registry:
base = base_registry[name]
elif name in registry:
base = registry[name].__class__
elif "." in name:
# A dot not at the start means a qualified name either relative to xclim.indicators or full
xclim_mods = [
mod
for mod in dir(xclim.indicators)
if isinstance(getattr(xclim.indicators, mod), (IndicatorCollection, ModuleType))
]
if name.count(".") == 1 and name.split(".")[0] in xclim_mods:
name = f"xclim.indicators.{name}"
modname, indname = name.rsplit(".", 1)
base = getattr(import_module(modname), indname)
if isinstance(base, Indicator):
base = base.__class__
else:
raise KeyError(name)
except Exception as err:
raise ValueError(f"Can't find base class {name}.") from err
return base
[docs]
@staticmethod
def _find_compute_function(name, computes):
func = None
if computes is not None:
func = getattr(computes, name, None)
if func is None and hasattr(computes, "__getitem__") and name in computes:
func = computes[name]
if func is None:
if "." in name:
modname, name = name.split(".")
submod = getattr(xclim.compute, modname, None)
func = getattr(submod, name, None)
else:
func = getattr(xclim.compute.generic, name, getattr(xclim.compute, name, None))
if func is None and "." in name:
modname, funcname = name.rsplit(".", 1)
try:
func = getattr(import_module(modname), funcname)
except ModuleNotFoundError as err:
raise_warn_or_log(err, "log", msg=f"Failed importing {funcname} from {modname}.")
if func is None:
raise ValueError(f"Can't find compute function '{name}'.")
return func
def __dir__(self):
"""Autocompletion support for indicators."""
return self.keys()
def __getattr__(self, k):
"""Access indicators as properties of a module (Obj.name)."""
if k in self.keys():
return self[k]
super().__getattribute__(k)