diff --git a/src/icclim/ecad/ecad_indices.py b/src/icclim/ecad/ecad_indices.py index a407c2ac..bdbcf968 100644 --- a/src/icclim/ecad/ecad_indices.py +++ b/src/icclim/ecad/ecad_indices.py @@ -1,6 +1,10 @@ from __future__ import annotations -from icclim.ecad.xclim_binding import XclimBinding +from icclim.ecad.xclim_binding import ( + GrowingSeasonLength, + StandardizedPrecipitationIndex3, + StandardizedPrecipitationIndex6, +) from icclim.generic_indices.registry import GenericIndicatorRegistry from icclim.generic_indices.standard_variable import StandardVariableRegistry from icclim.generic_indices.thresholds.factory import build_threshold @@ -734,7 +738,7 @@ def to_list(cls: type) -> list[str]: ) GSL = StandardIndex( reference=ECAD_REFERENCE, - indicator=XclimBinding.GrowingSeasonLength(), + indicator=GrowingSeasonLength(), definition="Growing season length.", source=ECAD_ATBD, short_name="GSL", @@ -744,7 +748,7 @@ def to_list(cls: type) -> list[str]: ) SPI6 = StandardIndex( reference=ECAD_REFERENCE, - indicator=XclimBinding.StandardizedPrecipitationIndex6(), + indicator=StandardizedPrecipitationIndex6(), definition="6-Month Standardized Precipitation Index.", source=ECAD_ATBD, short_name="SPI6", @@ -755,7 +759,7 @@ def to_list(cls: type) -> list[str]: ) SPI3 = StandardIndex( reference=ECAD_REFERENCE, - indicator=XclimBinding.StandardizedPrecipitationIndex3(), + indicator=StandardizedPrecipitationIndex3(), definition="3-Month Standardized Precipitation Index.", source=ECAD_ATBD, short_name="SPI3", diff --git a/src/icclim/ecad/xclim_binding.py b/src/icclim/ecad/xclim_binding.py index 6b17857a..eabb6ea3 100644 --- a/src/icclim/ecad/xclim_binding.py +++ b/src/icclim/ecad/xclim_binding.py @@ -15,98 +15,108 @@ from icclim.models.index_config import IndexConfig -class XclimBinding: - class GrowingSeasonLength(Indicator): - """xclim.growing_season_length. - - Fake icclim indicator that redirect to xclim `growing_season_length` indicator. - """ - - name = "growing_season_length" - standard_name = xclim.atmos.growing_season_length.standard_name - long_name = "ECAD Growing Season Length (Tmean > 5 degree_Celsius)" - cell_methods = "" - - def __call__(self, config: IndexConfig) -> xarray.DataArray: - study, threshold = get_single_var(config.climate_variables) - return xclim.atmos.growing_season_length( - tas=study, - thresh="5 degree_Celsius", - window=6, - mid_date="07-01", - freq=config.frequency.pandas_freq, - ) - - def preprocess(self, *args, **kwargs) -> list[xarray.DataArray]: - """Not implemented as xclim indicator already handle pre/post processing.""" - raise NotImplementedError - - def postprocess(self, *args, **kwargs) -> xarray.DataArray: - """Not implemented as xclim indicator already handle pre/post processing.""" - raise NotImplementedError - - class StandardizedPrecipitationIndex3(Indicator): - """ - Fake icclim indicator that redirect to xclim `standardized_precipitation_index` - indicator, with 3 MS configured. - """ - - name = "standardized_precipitation_index_3" - standard_name = xclim.atmos.standardized_precipitation_index.standard_name - long_name = "3-Month Standardized Precipitation Index (SPI3)" - cell_methods = "" - - def __call__(self, config: IndexConfig) -> xarray.DataArray: - if config.frequency is not FrequencyRegistry.YEAR: # year is default freq - msg = "`slice_mode` cannot be configured when computing SPI3" - raise InvalidIcclimArgumentError(msg) - study, ref = get_couple_of_var(config.climate_variables, "SPI") - return xclim.atmos.standardized_precipitation_index( - pr=study, - pr_cal=ref, - freq="MS", - window=3, - dist="gamma", - method="APP", - ) - - def preprocess(self, *args, **kwargs) -> list[xarray.DataArray]: - """Not implemented as xclim indicator already handle pre/post processing.""" - raise NotImplementedError - - def postprocess(self, *args, **kwargs) -> xarray.DataArray: - """Not implemented as xclim indicator already handle pre/post processing.""" - raise NotImplementedError - - class StandardizedPrecipitationIndex6(Indicator): - """ - Fake icclim indicator that redirect to xclim `standardized_precipitation_index` - indicator, with 6 MS configured. - """ - - name = "standardized_precipitation_index_6" - standard_name = xclim.atmos.standardized_precipitation_index.standard_name - long_name = "6-Month Standardized Precipitation Index (SPI6)" - cell_methods = "" - - def __call__(self, config: IndexConfig) -> xarray.DataArray: - if config.frequency is not FrequencyRegistry.YEAR: # year is default freq - msg = "`slice_mode` cannot be configured when computing SPI6" - raise InvalidIcclimArgumentError(msg) - study, ref = get_couple_of_var(config.climate_variables, "SPI") - return xclim.atmos.standardized_precipitation_index( - pr=study, - pr_cal=ref, - freq="MS", - window=6, - dist="gamma", - method="APP", - ) - - def preprocess(self, *args, **kwargs) -> list[xarray.DataArray]: - """Not implemented as xclim indicator already handle pre/post processing.""" - raise NotImplementedError - - def postprocess(self, *args, **kwargs) -> xarray.DataArray: - """Not implemented as xclim indicator already handle pre/post processing.""" - raise NotImplementedError +class GrowingSeasonLength(Indicator): + """xclim.growing_season_length. + + Fake icclim indicator that redirect to xclim `growing_season_length` indicator. + """ + + name = "growing_season_length" + standard_name = xclim.atmos.growing_season_length.standard_name + long_name = "ECAD Growing Season Length (Tmean > 5 degree_Celsius)" + cell_methods = "" + + def __call__(self, config: IndexConfig) -> xarray.DataArray: + study, _ = get_single_var(config.climate_variables) + return xclim.atmos.growing_season_length( + tas=study, + thresh="5 degree_Celsius", + window=6, + mid_date="07-01", + freq=config.frequency.pandas_freq, + ) + + def preprocess(self, *args, **kwargs) -> list[xarray.DataArray]: + """Not implemented as xclim indicator already handle pre/post processing.""" + raise NotImplementedError + + def postprocess(self, *args, **kwargs) -> xarray.DataArray: + """Not implemented as xclim indicator already handle pre/post processing.""" + raise NotImplementedError + + def __eq__(self, other: object) -> bool: + return isinstance(other, GrowingSeasonLength) + + +class StandardizedPrecipitationIndex3(Indicator): + """ + Fake icclim indicator that redirect to xclim `standardized_precipitation_index` + indicator, with 3 MS configured. + """ + + name = "standardized_precipitation_index_3" + standard_name = xclim.atmos.standardized_precipitation_index.standard_name + long_name = "3-Month Standardized Precipitation Index (SPI3)" + cell_methods = "" + + def __call__(self, config: IndexConfig) -> xarray.DataArray: + if config.frequency is not FrequencyRegistry.YEAR: # year is default freq + msg = "`slice_mode` cannot be configured when computing SPI3" + raise InvalidIcclimArgumentError(msg) + study, ref = get_couple_of_var(config.climate_variables, "SPI") + return xclim.atmos.standardized_precipitation_index( + pr=study, + pr_cal=ref, + freq="MS", + window=3, + dist="gamma", + method="APP", + ) + + def preprocess(self, *args, **kwargs) -> list[xarray.DataArray]: + """Not implemented as xclim indicator already handle pre/post processing.""" + raise NotImplementedError + + def postprocess(self, *args, **kwargs) -> xarray.DataArray: + """Not implemented as xclim indicator already handle pre/post processing.""" + raise NotImplementedError + + def __eq__(self, other: object) -> bool: + return isinstance(other, StandardizedPrecipitationIndex3) + + +class StandardizedPrecipitationIndex6(Indicator): + """ + Fake icclim indicator that redirect to xclim `standardized_precipitation_index` + indicator, with 6 MS configured. + """ + + name = "standardized_precipitation_index_6" + standard_name = xclim.atmos.standardized_precipitation_index.standard_name + long_name = "6-Month Standardized Precipitation Index (SPI6)" + cell_methods = "" + + def __call__(self, config: IndexConfig) -> xarray.DataArray: + if config.frequency is not FrequencyRegistry.YEAR: # year is default freq + msg = "`slice_mode` cannot be configured when computing SPI6" + raise InvalidIcclimArgumentError(msg) + study, ref = get_couple_of_var(config.climate_variables, "SPI") + return xclim.atmos.standardized_precipitation_index( + pr=study, + pr_cal=ref, + freq="MS", + window=6, + dist="gamma", + method="APP", + ) + + def preprocess(self, *args, **kwargs) -> list[xarray.DataArray]: + """Not implemented as xclim indicator already handle pre/post processing.""" + raise NotImplementedError + + def postprocess(self, *args, **kwargs) -> xarray.DataArray: + """Not implemented as xclim indicator already handle pre/post processing.""" + raise NotImplementedError + + def __eq__(self, other: object) -> bool: + return isinstance(other, StandardizedPrecipitationIndex6) diff --git a/src/icclim/generic_indices/indicator.py b/src/icclim/generic_indices/indicator.py index b58941cf..d4b4c902 100644 --- a/src/icclim/generic_indices/indicator.py +++ b/src/icclim/generic_indices/indicator.py @@ -1,6 +1,7 @@ from __future__ import annotations import contextlib +from copy import deepcopy from functools import reduce from typing import TYPE_CHECKING, Any, Callable @@ -69,12 +70,12 @@ def __init__( self._missing = missing_method.execute if self.missing_options: missing_method.validate(**self.missing_options) - local = INDICATORS_TEMPLATES_EN + en_indicator_templates = deepcopy(INDICATORS_TEMPLATES_EN[name]) self.name = name self.process = process - self.standard_name = local[name]["standard_name"] - self.cell_methods = local[name]["cell_methods"] - self.long_name = local[name]["long_name"] + self.standard_name = en_indicator_templates["standard_name"] + self.cell_methods = en_indicator_templates["cell_methods"] + self.long_name = en_indicator_templates["long_name"] self.check_vars = check_vars self.definition = definition self.qualifiers = qualifiers diff --git a/src/icclim/generic_indices/typing.py b/src/icclim/generic_indices/typing.py index 1cdeb811..dcd4ea65 100644 --- a/src/icclim/generic_indices/typing.py +++ b/src/icclim/generic_indices/typing.py @@ -48,3 +48,12 @@ def preprocess(self, *args, **kwargs) -> list[DataArray]: @abc.abstractmethod def postprocess(self, *args, **kwargs) -> DataArray: ... + + @abc.abstractmethod + def __eq__(self, __value: object) -> bool: + ... + + def clone(self) -> Indicator: + from copy import deepcopy + + return deepcopy(self) diff --git a/src/icclim/main.py b/src/icclim/main.py index 9000fe14..691a613c 100644 --- a/src/icclim/main.py +++ b/src/icclim/main.py @@ -22,7 +22,10 @@ from icclim.dcsc.dcsc_indices import DcscIndexRegistry from icclim.ecad.ecad_indices import EcadIndexRegistry -from icclim.ecad.xclim_binding import XclimBinding +from icclim.ecad.xclim_binding import ( + StandardizedPrecipitationIndex3, + StandardizedPrecipitationIndex6, +) from icclim.generic_indices.indicator import GenericIndicator from icclim.generic_indices.registry import GenericIndicatorRegistry from icclim.generic_indices.thresholds.factory import build_threshold @@ -626,8 +629,8 @@ def _build_standard_index_config( coef = None index = _parse_index_kind(index_name) if isinstance(index, StandardIndex): - standard_index = index - indicator = standard_index.indicator + standard_index = index.clone() + indicator = standard_index.indicator.clone() threshold = standard_index.threshold rename = standard_index.short_name output_unit = out_unit or standard_index.output_unit @@ -637,7 +640,7 @@ def _build_standard_index_config( rename = None output_unit = out_unit standard_index = None - indicator = GenericIndicatorRegistry.lookup(index) + indicator = index.clone() reference = ICCLIM_REFERENCE indicator_name = indicator.name else: @@ -800,8 +803,8 @@ def _compute_climate_index( and not isinstance( climate_index, ( - XclimBinding.StandardizedPrecipitationIndex6, - XclimBinding.StandardizedPrecipitationIndex3, + StandardizedPrecipitationIndex6, + StandardizedPrecipitationIndex3, ), ) ): diff --git a/src/icclim/models/registry.py b/src/icclim/models/registry.py index 85e6a0f0..49b5cd96 100644 --- a/src/icclim/models/registry.py +++ b/src/icclim/models/registry.py @@ -26,7 +26,7 @@ class Registry(Generic[T], ABC): @classmethod def lookup(cls, query: T | str) -> T: if isinstance(query, cls._item_class): - return query + return deepcopy(query) if isinstance(query, str): q = query.upper() for key, item in cls.catalog().items(): diff --git a/src/icclim/models/standard_index.py b/src/icclim/models/standard_index.py index 7a92b910..ae36391e 100644 --- a/src/icclim/models/standard_index.py +++ b/src/icclim/models/standard_index.py @@ -74,3 +74,8 @@ def __eq__(self, other: object) -> bool: and self.doy_window_width == other.doy_window_width and self.min_spell_length == other.min_spell_length ) + + def clone(self) -> StandardIndex: + from copy import deepcopy + + return deepcopy(self)