Source code for esmvalcore.preprocessor._derive

"""Automatically derive variables."""

import importlib
import logging
from copy import deepcopy
from pathlib import Path

import iris

logger = logging.getLogger(__name__)

def _get_all_derived_variables():
    """Get all possible derived variables.

        All derived variables with `short_name` (keys) and the associated
        python classes (values).

    derivers = {}
    for path in Path(__file__).parent.glob('[a-z]*.py'):
        short_name = path.stem
        module = importlib.import_module(
        derivers[short_name] = getattr(module, 'DerivedVariable')
    return derivers

ALL_DERIVED_VARIABLES = _get_all_derived_variables()


def get_required(short_name, project):
    """Return all required variables for derivation.

    Get all information (at least `short_name`) required for derivation.

    short_name : str
        `short_name` of the variable to derive.
    project : str
        `project` of the variable to derive.

        List of dictionaries (including at least the key `short_name`).

    if short_name.lower() not in ALL_DERIVED_VARIABLES:
        raise NotImplementedError(
            f"Cannot derive variable '{short_name}', no derivation script "
    DerivedVariable = ALL_DERIVED_VARIABLES[short_name.lower()]  # noqa: N806
    variables = deepcopy(DerivedVariable().required(project))
    return variables

[docs] def derive(cubes, short_name, long_name, units, standard_name=None): """Derive variable. Parameters ---------- cubes: iris.cube.CubeList Includes all the needed variables for derivation defined in :func:`get_required`. short_name: str short_name long_name: str long_name units: str units standard_name: str, optional standard_name Returns ------- iris.cube.Cube The new derived variable. """ if short_name == cubes[0].var_name: return cubes[0] cubes = iris.cube.CubeList(cubes) # Derive variable DerivedVariable = ALL_DERIVED_VARIABLES[short_name.lower()] # noqa: N806 try: cube = DerivedVariable().calculate(cubes) except Exception as exc: msg = (f"Derivation of variable '{short_name}' failed. If you used " f"the option '--skip_nonexistent' for running your recipe, " f"this might be caused by missing input data for derivation") raise ValueError(msg) from exc # Set standard attributes cube.var_name = short_name cube.standard_name = standard_name if standard_name else None cube.long_name = long_name for temp in cubes: if 'source_file' in temp.attributes: cube.attributes['source_file'] = temp.attributes['source_file'] # Check/convert units if cube.units is None or cube.units == units: cube.units = units elif cube.units.is_no_unit() or cube.units.is_unknown(): logger.warning( "Units of cube after executing derivation script of '%s' are " "'%s', automatically setting them to '%s'. This might lead to " "incorrect data", short_name, cube.units, units) cube.units = units elif cube.units.is_convertible(units): cube.convert_units(units) else: raise ValueError( f"Units '{cube.units}' after executing derivation script of " f"'{short_name}' cannot be converted to target units '{units}'") return cube