Source code for esmvalcore.preprocessor._bias

"""Preprocessor functions to calculate biases from data."""
from __future__ import annotations

import logging
from import Iterable
from typing import TYPE_CHECKING, Literal, Optional

import dask.array as da
from iris.cube import Cube, CubeList

from ._io import concatenate

    from esmvalcore.preprocessor import PreprocessorFile

logger = logging.getLogger(__name__)

BiasType = Literal['absolute', 'relative']

[docs] def bias( products: set[PreprocessorFile] | Iterable[Cube], ref_cube: Optional[Cube] = None, bias_type: BiasType = 'absolute', denominator_mask_threshold: float = 1e-3, keep_reference_dataset: bool = False, ) -> set[PreprocessorFile] | CubeList: """Calculate biases relative to a reference dataset. The reference dataset needs to be broadcastable to all input `products`. This supports `iris' rich broadcasting abilities < html#calculating-a-cube-anomaly>`__. To ensure this, the preprocessors :func:`esmvalcore.preprocessor.regrid` and/or :func:`esmvalcore.preprocessor.regrid_time` might be helpful. Notes ----- The reference dataset can be specified with the `ref_cube` argument. If `ref_cube` is ``None``, exactly one input dataset in the `products` set needs to have the facet ``reference_for_bias: true`` defined in the recipe. Please do **not** specify the option `ref_cube` when using this preprocessor function in a recipe. Parameters ---------- products: Input datasets/cubes for which the bias is calculated relative to a reference dataset/cube. ref_cube: Cube which is used as reference for the bias calculation. If ``None``, `products` needs to be a :obj:`set` of `~esmvalcore.preprocessor.PreprocessorFile` objects and exactly one dataset in `products` needs the facet ``reference_for_bias: true``. bias_type: Bias type that is calculated. Must be one of ``'absolute'`` (dataset - ref) or ``'relative'`` ((dataset - ref) / ref). denominator_mask_threshold: Threshold to mask values close to zero in the denominator (i.e., the reference dataset) during the calculation of relative biases. All values in the reference dataset with absolute value less than the given threshold are masked out. This setting is ignored when ``bias_type`` is set to ``'absolute'``. Please note that for some variables with very small absolute values (e.g., carbon cycle fluxes, which are usually :math:`< 10^{-6}` kg m :math:`^{-2}` s :math:`^{-1}`) it is absolutely essential to change the default value in order to get reasonable results. keep_reference_dataset: If ``True``, keep the reference dataset in the output. If ``False``, drop the reference dataset. Ignored if `ref_cube` is given. Returns ------- set[PreprocessorFile] | CubeList Output datasets/cubes. Will be a :obj:`set` of :class:`~esmvalcore.preprocessor.PreprocessorFile` objects if `products` is also one, a :class:`~iris.cube.CubeList` otherwise. Raises ------ ValueError Not exactly one input datasets contains the facet ``reference_for_bias: true`` if ``ref_cube=None``; ``ref_cube=None`` and the input products are given as iterable of :class:`~iris.cube.Cube` objects; ``bias_type`` is not one of ``'absolute'`` or ``'relative'``. """ ref_product = None all_cubes_given = all(isinstance(p, Cube) for p in products) # Get reference cube if not explicitly given if ref_cube is None: if all_cubes_given: raise ValueError( "A list of Cubes is given to this preprocessor; please " "specify a `ref_cube`" ) (ref_cube, ref_product) = _get_ref(products, 'reference_for_bias') else: ref_product = None # Mask reference cube appropriately for relative biases if bias_type == 'relative': ref_cube = ref_cube.copy() = ref_cube.core_data(), -denominator_mask_threshold, denominator_mask_threshold, ) # If input is an Iterable of Cube objects, calculate bias for each element if all_cubes_given: cubes = [_calculate_bias(c, ref_cube, bias_type) for c in products] return CubeList(cubes) # Otherwise, iterate over all input products, calculate bias and adapt # metadata and provenance information accordingly output_products = set() for product in products: if product == ref_product: continue cube = concatenate(product.cubes) # Calculate bias cube = _calculate_bias(cube, ref_cube, bias_type) # Adapt metadata and provenance information product.attributes['units'] = str(cube.units) if ref_product is not None: product.wasderivedfrom(ref_product) product.cubes = CubeList([cube]) output_products.add(product) # Add reference dataset to output if desired if keep_reference_dataset and ref_product is not None: output_products.add(ref_product) return output_products
def _get_ref(products, ref_tag: str) -> tuple[Cube, PreprocessorFile]: """Get reference cube and product.""" ref_products = [] for product in products: if product.attributes.get(ref_tag, False): ref_products.append(product) if len(ref_products) != 1: raise ValueError( f"Expected exactly 1 dataset with '{ref_tag}: true', found " f"{len(ref_products):d}" ) ref_product = ref_products[0] # Extract reference cube # Note: For technical reasons, product objects contain the member # ``cubes``, which is a list of cubes. However, this is expected to be a # list with exactly one element due to the call of concatenate earlier in # the preprocessing chain of ESMValTool. To make sure that this # preprocessor can also be used outside the ESMValTool preprocessing chain, # an additional concatenate call is added here. ref_cube = concatenate(ref_product.cubes) return (ref_cube, ref_product) def _calculate_bias(cube: Cube, ref_cube: Cube, bias_type: BiasType) -> Cube: """Calculate bias for a single cube relative to a reference cube.""" cube_metadata = cube.metadata if bias_type == 'absolute': cube = cube - ref_cube new_units = cube.units elif bias_type == 'relative': cube = (cube - ref_cube) / ref_cube new_units = '1' else: raise ValueError( f"Expected one of ['absolute', 'relative'] for bias_type, got " f"'{bias_type}'" ) cube.metadata = cube_metadata cube.units = new_units return cube