From ecae36e8b016a82c6447060f0684e89b28a99320 Mon Sep 17 00:00:00 2001 From: spjuhel Date: Wed, 17 Dec 2025 17:30:38 +0100 Subject: [PATCH 01/10] Cherrypicks from risk traj --- climada/trajectories/interpolation.py | 439 ++++++++++++++++++ .../trajectories/test/test_interpolation.py | 352 ++++++++++++++ 2 files changed, 791 insertions(+) create mode 100644 climada/trajectories/interpolation.py create mode 100644 climada/trajectories/test/test_interpolation.py diff --git a/climada/trajectories/interpolation.py b/climada/trajectories/interpolation.py new file mode 100644 index 0000000000..9f6687e449 --- /dev/null +++ b/climada/trajectories/interpolation.py @@ -0,0 +1,439 @@ +""" +This file is part of CLIMADA. + +Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS. + +CLIMADA is free software: you can redistribute it and/or modify it under the +terms of the GNU General Public License as published by the Free +Software Foundation, version 3. + +CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with CLIMADA. If not, see . + +--- + +This modules implements different sparce matrices and numpy arrays +interpolation approaches. + +""" + +import logging +from abc import ABC +from collections.abc import Callable +from typing import Any, Dict, List, Optional + +import numpy as np +from scipy import sparse + +LOGGER = logging.getLogger(__name__) + +__all__ = [ + "AllLinearStrategy", + "ExponentialExposureStrategy", + "linear_interp_arrays", + "linear_interp_imp_mat", + "exponential_interp_arrays", + "exponential_interp_imp_mat", +] + + +def linear_interp_imp_mat( + mat_start: sparse.csr_matrix, + mat_end: sparse.csr_matrix, + number_of_interpolation_points: int, +) -> List[sparse.csr_matrix]: + r""" + Linearly interpolates between two sparse impact matrices. + + Creates a sequence of matrices representing a linear transition from a starting + matrix to an ending matrix. The interpolation includes both the start and end + points. + + Parameters + ---------- + mat_start : scipy.sparse.csr_matrix + The starting impact matrix. Must have a shape compatible with `mat_end` + for arithmetic operations. + mat_end : scipy.sparse.csr_matrix + The ending impact matrix. Must have a shape compatible with `mat_start` + for arithmetic operations. + number_of_interpolation_points : int + The total number of matrices to return, including the start and end points. + Must be $\ge 2$. + + Returns + ------- + list of scipy.sparse.csr_matrix + A list of matrices, where the first element is `mat_start` and the last + element is `mat_end`. The total length of the list is + `number_of_interpolation_points`. + + Notes + ----- + The formula used for interpolation at proportion $p$ is: + $$M_p = M_{start} \cdot (1 - p) + M_{end} \cdot p$$ + The proportions $p$ range from 0 to 1, inclusive. + """ + + return [ + mat_start + prop * (mat_end - mat_start) + for prop in np.linspace(0, 1, number_of_interpolation_points) + ] + + +def exponential_interp_imp_mat( + mat_start: sparse.csr_matrix, + mat_end: sparse.csr_matrix, + number_of_interpolation_points: int, +) -> List[sparse.csr_matrix]: + r""" + Exponentially interpolates between two "impact matrices". + + This function performs interpolation in a logarithmic space, effectively + achieving an exponential-like transition between `mat_start` and `mat_end`. + It is designed for objects that wrap NumPy arrays and expose them via a + `.data` attribute. + + Parameters + ---------- + mat_start : object + The starting matrix object. Must have a `.data` attribute that is a + NumPy array of positive values. + mat_end : object + The ending matrix object. Must have a `.data` attribute that is a + NumPy array of positive values and have a compatible shape with `mat_start`. + number_of_interpolation_points : int + The total number of matrix objects to return, including the start and + end points. Must be $\ge 2$. + + Returns + ------- + list of object + A list of interpolated matrix objects. The first element corresponds to + `mat_start` and the last to `mat_end` (after the conversion/reversion). + The list length is `number_of_interpolation_points`. + + Notes + ----- + The interpolation is achieved by: + + 1. Mapping the matrix data to a transformed logarithmic space: + $$M'_{i} = \ln(M_{i})}$$ + (where $\ln$ is the natural logarithm, and $\epsilon$ is added to $M_{i}$ + to prevent $\ln(0)$). + 2. Performing standard linear interpolation on the transformed matrices + $M'_{start}$ and $M'_{end}$ to get $M'_{interp}$: + $$M'_{interp} = M'_{start} \cdot (1 - \text{ratio}) + M'_{end} \cdot \text{ratio}$$ + 3. Mapping the result back to the original domain: + $$M_{interp} = \exp(M'_{interp}$$ + """ + + mat_start = mat_start.copy() + mat_end = mat_end.copy() + mat_start.data = np.log(mat_start.data + np.finfo(float).eps) + mat_end.data = np.log(mat_end.data + np.finfo(float).eps) + + # Perform linear interpolation in the logarithmic domain + res = [] + num_points = number_of_interpolation_points + for point in range(num_points): + ratio = point / (num_points - 1) + mat_interpolated = mat_start * (1 - ratio) + ratio * mat_end + mat_interpolated.data = np.exp(mat_interpolated.data) + res.append(mat_interpolated) + return res + + +def linear_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarray: + r""" + Performs linear interpolation between two NumPy arrays over their first dimension. + + This function interpolates each metric (column) linearly across the time steps + (rows), including both the start and end states. + + Parameters + ---------- + arr_start : numpy.ndarray + The starting array of metrics. The first dimension (rows) is assumed to + represent the interpolation steps (e.g., dates/time points). + arr_end : numpy.ndarray + The ending array of metrics. Must have the exact same shape as `arr_start`. + + Returns + ------- + numpy.ndarray + An array with the same shape as `arr_start` and `arr_end`. The values + in the first dimension transition linearly from those in `arr_start` + to those in `arr_end`. + + Raises + ------ + ValueError + If `arr_start` and `arr_end` do not have the same shape. + + Notes + ----- + The interpolation is performed element-wise along the first dimension + (axis 0). For each row $i$ and proportion $p_i$, the result $R_i$ is calculated as: + + $$R_i = arr\_start_i \cdot (1 - p_i) + arr\_end_i \cdot p_i$$ + + where $p_i$ is generated by $\text{np.linspace}(0, 1, n)$ and $n$ is the + size of the first dimension ($\text{arr\_start.shape}[0]$). + """ + if arr_start.shape != arr_end.shape: + raise ValueError( + f"Cannot interpolate arrays of different shapes: {arr_start.shape} and {arr_end.shape}." + ) + interpolation_range = arr_start.shape[0] + prop1 = np.linspace(0, 1, interpolation_range) + prop0 = 1 - prop1 + if arr_start.ndim > 1: + prop0, prop1 = prop0.reshape(-1, 1), prop1.reshape(-1, 1) + + return np.multiply(arr_start, prop0) + np.multiply(arr_end, prop1) + + +def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarray: + r""" + Performs exponential interpolation between two NumPy arrays over their first dimension. + + This function achieves an exponential-like transition by performing linear + interpolation in the logarithmic space, suitable to interpolate over a dimension which has + a growth factor. + + Parameters + ---------- + arr_start : numpy.ndarray + The starting array of metrics. Values must be positive. + arr_end : numpy.ndarray + The ending array of metrics. Must have the exact same shape as `arr_start`. + + Returns + ------- + numpy.ndarray + An array with the same shape as `arr_start` and `arr_end`. The values + in the first dimension transition exponentially from those in `arr_start` + to those in `arr_end`. + + Raises + ------ + ValueError + If `arr_start` and `arr_end` do not have the same shape. + + Notes + ----- + The interpolation is performed by transforming the arrays to a logarithmic + domain, linearly interpolating, and then transforming back. + + The formula for the interpolated result $R$ at proportion $\text{prop}$ is: + $$ + R = \exp \left( + \ln(A_{start}) \cdot (1 - \text{prop}) + + \ln(A_{end}) \cdot \text{prop} + \right) + $$ + where $A_{start}$ and $A_{end}$ are the input arrays (with $\epsilon$ added + to prevent $\ln(0)$) and $\text{prop}$ ranges from 0 to 1. + """ + if arr_start.shape != arr_end.shape: + raise ValueError( + f"Cannot interpolate arrays of different shapes: {arr_start.shape} and {arr_end.shape}." + ) + interpolation_range = arr_start.shape[0] + + prop1 = np.linspace(0, 1, interpolation_range) + prop0 = 1 - prop1 + if arr_start.ndim > 1: + prop0, prop1 = prop0.reshape(-1, 1), prop1.reshape(-1, 1) + + # Perform log transformation, linear interpolation, and exponential back-transformation + log_arr_start = np.log(arr_start + np.finfo(float).eps) + log_arr_end = np.log(arr_end + np.finfo(float).eps) + + interpolated_log_arr = np.multiply(log_arr_start, prop0) + np.multiply( + log_arr_end, prop1 + ) + + return np.exp(interpolated_log_arr) + + +class InterpolationStrategyBase(ABC): + r""" + Base abstract class for defining a set of interpolation strategies. + + This class serves as a blueprint for implementing specific interpolation + methods (e.g., 'Linear', 'Exponential') across different impact dimensions: + Exposure (matrices), Hazard, and Vulnerability (arrays/metrics). + + Attributes + ---------- + exposure_interp : Callable + The function used to interpolate sparse impact matrices over the + exposure dimension. + Signature: (mat_start, mat_end, num_points, **kwargs) -> list[sparse.csr_matrix]. + hazard_interp : Callable + The function used to interpolate NumPy arrays of metrics over the + hazard dimension. + Signature: (arr_start, arr_end, **kwargs) -> np.ndarray. + vulnerability_interp : Callable + The function used to interpolate NumPy arrays of metrics over the + vulnerability dimension. + Signature: (arr_start, arr_end, **kwargs) -> np.ndarray. + """ + + exposure_interp: Callable + hazard_interp: Callable + vulnerability_interp: Callable + + def interp_over_exposure_dim( + self, + imp_E0: sparse.csr_matrix, + imp_E1: sparse.csr_matrix, + interpolation_range: int, + /, + **kwargs: Optional[Dict[str, Any]], + ) -> List[sparse.csr_matrix]: + """ + Interpolates between two impact matrices using the defined exposure strategy. + + This method calls the function assigned to :attr:`exposure_interp` to generate + a sequence of matrices. + + Parameters + ---------- + imp_E0 : scipy.sparse.csr_matrix + A sparse matrix of the impacts at the start of the range. + imp_E1 : scipy.sparse.csr_matrix + A sparse matrix of the impacts at the end of the range. + interpolation_range : int + The total number of time points to interpolate, including the start and end. + **kwargs : Optional[Dict[str, Any]] + Keyword arguments to pass to the underlying :attr:`exposure_interp` function. + + Returns + ------- + list of scipy.sparse.csr_matrix + A list of ``interpolation_range`` interpolated impact matrices. + + Raises + ------ + ValueError + If the underlying interpolation function raises a ``ValueError`` + indicating incompatible matrix shapes. + """ + try: + res = self.exposure_interp(imp_E0, imp_E1, interpolation_range, **kwargs) + except ValueError as err: + if str(err) == "inconsistent shapes": + raise ValueError( + "Tried to interpolate impact matrices of different shapes. " + "A possible reason could be Exposures of different shapes." + ) from err + + raise err + + return res + + def interp_over_hazard_dim( + self, + metric_0: np.ndarray, + metric_1: np.ndarray, + /, + **kwargs: Optional[Dict[str, Any]], + ) -> np.ndarray: + """ + Interpolates between two metric arrays using the defined hazard strategy. + + This method calls the function assigned to :attr:`hazard_interp`. + + Parameters + ---------- + metric_0 : numpy.ndarray + The starting array of metrics. + metric_1 : numpy.ndarray + The ending array of metrics. Must have the same shape as ``metric_0``. + **kwargs : Optional [Dict[str, Any]] + Keyword arguments to pass to the underlying :attr:`hazard_interp` function. + + Returns + ------- + numpy.ndarray + The resulting interpolated array. + """ + return self.hazard_interp(metric_0, metric_1, **kwargs) + + def interp_over_vulnerability_dim( + self, + metric_0: np.ndarray, + metric_1: np.ndarray, + /, + **kwargs: Optional[Dict[str, Any]], + ) -> np.ndarray: + """ + Interpolates between two metric arrays using the defined vulnerability strategy. + + This method calls the function assigned to :attr:`vulnerability_interp`. + + Parameters + ---------- + metric_0 : numpy.ndarray + The starting array of metrics. + metric_1 : numpy.ndarray + The ending array of metrics. Must have the same shape as ``metric_0``. + **kwargs : Optional[Dict[str, Any]] + Keyword arguments to pass to the underlying :attr:`vulnerability_interp` function. + + Returns + ------- + numpy.ndarray + The resulting interpolated array. + """ + # Note: Assuming the Callable takes the exact positional arguments + return self.vulnerability_interp(metric_0, metric_1, **kwargs) + + +class InterpolationStrategy(InterpolationStrategyBase): + r"""Interface for interpolation strategies. + + This is the class to use to define your own custom interpolation strategy. + """ + + def __init__( + self, + exposure_interp: Callable, + hazard_interp: Callable, + vulnerability_interp: Callable, + ) -> None: + super().__init__() + self.exposure_interp = exposure_interp + self.hazard_interp = hazard_interp + self.vulnerability_interp = vulnerability_interp + + +class AllLinearStrategy(InterpolationStrategyBase): + r"""Linear interpolation strategy over all dimensions.""" + + def __init__(self) -> None: + super().__init__() + self.exposure_interp = linear_interp_imp_mat + self.hazard_interp = linear_interp_arrays + self.vulnerability_interp = linear_interp_arrays + + +class ExponentialExposureStrategy(InterpolationStrategyBase): + r"""Exponential interpolation strategy for exposure and linear for Hazard and Vulnerability.""" + + def __init__(self) -> None: + super().__init__() + self.exposure_interp = ( + lambda mat_start, mat_end, points: exponential_interp_imp_mat( + mat_start, mat_end, points + ) + ) + self.hazard_interp = linear_interp_arrays + self.vulnerability_interp = linear_interp_arrays diff --git a/climada/trajectories/test/test_interpolation.py b/climada/trajectories/test/test_interpolation.py new file mode 100644 index 0000000000..693c9b9c33 --- /dev/null +++ b/climada/trajectories/test/test_interpolation.py @@ -0,0 +1,352 @@ +""" +This file is part of CLIMADA. + +Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS. + +CLIMADA is free software: you can redistribute it and/or modify it under the +terms of the GNU General Public License as published by the Free +Software Foundation, version 3. + +CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with CLIMADA. If not, see . + +--- + +Tests for interpolation + +""" + +import unittest +from unittest.mock import MagicMock + +import numpy as np +from scipy.sparse import csr_matrix + +from climada.trajectories.interpolation import ( + AllLinearStrategy, + ExponentialExposureStrategy, + InterpolationStrategy, + exponential_interp_arrays, + exponential_interp_imp_mat, + linear_interp_arrays, + linear_interp_imp_mat, +) + + +class TestInterpolationFuncs(unittest.TestCase): + def setUp(self): + # Create mock impact matrices for testing + self.imp_mat0 = csr_matrix(np.array([[1, 2], [3, 4]])) + self.imp_mat1 = csr_matrix(np.array([[5, 6], [7, 8]])) + self.imp_mat2 = csr_matrix(np.array([[5, 6, 7], [8, 9, 10]])) # Different shape + self.time_points = 5 + self.interpolation_range_5 = 5 + self.interpolation_range_1 = 1 + self.interpolation_range_2 = 2 + self.rtol = 1e-5 + self.atol = 1e-8 + + def test_linear_interp_arrays(self): + arr_start = np.array([10, 100]) + arr_end = np.array([20, 200]) + expected = np.array([10.0, 200.0]) + result = linear_interp_arrays(arr_start, arr_end) + np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) + + def test_linear_interp_arrays2D(self): + arr_start = np.array([[10, 100], [10, 100]]) + arr_end = np.array([[20, 200], [20, 200]]) + expected = np.array([[10.0, 100.0], [20, 200]]) + result = linear_interp_arrays(arr_start, arr_end) + np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) + + def test_linear_interp_arrays_shape(self): + arr_start = np.array([10, 100, 5]) + arr_end = np.array([20, 200]) + with self.assertRaises(ValueError): + linear_interp_arrays(arr_start, arr_end) + + def test_linear_interp_arrays_start_equals_end(self): + arr_start = np.array([5, 5]) + arr_end = np.array([5, 5]) + expected = np.array([5.0, 5.0]) + result = linear_interp_arrays(arr_start, arr_end) + np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) + + def test_exponential_interp_arrays_1d(self): + arr_start = np.array([1, 10, 100]) + arr_end = np.array([2, 20, 200]) + expected = np.array([1.0, 14.142136, 200.0]) + result = exponential_interp_arrays(arr_start, arr_end) + np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) + + def test_exponential_interp_arrays_shape(self): + arr_start = np.array([10, 100, 5]) + arr_end = np.array([20, 200]) + with self.assertRaises(ValueError): + exponential_interp_arrays(arr_start, arr_end) + + def test_exponential_interp_arrays_2d(self): + arr_start = np.array( + [ + [1, 10, 100], # date 1 metric a,b,c + [1, 10, 100], # date 2 metric a,b,c + [1, 10, 100], + ] + ) # date 3 metric a,b,c + arr_end = np.array([[2, 20, 200], [2, 20, 200], [2, 20, 200]]) + expected = np.array( + [[1.0, 10.0, 100.0], [1.4142136, 14.142136, 141.42136], [2, 20, 200]] + ) + result = exponential_interp_arrays(arr_start, arr_end) + np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) + + def test_exponential_interp_arrays_start_equals_end(self): + arr_start = np.array([5, 5]) + arr_end = np.array([5, 5]) + expected = np.array([5.0, 5.0]) + result = exponential_interp_arrays(arr_start, arr_end) + np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) + + def test_linear_impmat_interpolate(self): + result = linear_interp_imp_mat(self.imp_mat0, self.imp_mat1, self.time_points) + self.assertEqual(len(result), self.time_points) + for mat in result: + self.assertIsInstance(mat, csr_matrix) + + dense = np.array([r.todense() for r in result]) + expected = np.array( + [ + [[1.0, 2.0], [3.0, 4.0]], + [[2.0, 3.0], [4.0, 5.0]], + [[3.0, 4.0], [5.0, 6.0]], + [[4.0, 5.0], [6.0, 7.0]], + [[5.0, 6.0], [7.0, 8.0]], + ] + ) + np.testing.assert_array_equal(dense, expected) + + def test_linear_impmat_interpolate_inconsistent_shape(self): + with self.assertRaises(ValueError): + linear_interp_imp_mat(self.imp_mat0, self.imp_mat2, self.time_points) + + def test_exp_impmat_interpolate(self): + result = exponential_interp_imp_mat( + self.imp_mat0, self.imp_mat1, self.time_points + ) + self.assertEqual(len(result), self.time_points) + for mat in result: + self.assertIsInstance(mat, csr_matrix) + + dense = np.array([r.todense() for r in result]) + expected = np.array( + [ + [[1.0, 2.0], [3.0, 4.0]], + [[1.49534878, 2.63214803], [3.70779275, 4.75682846]], + [[2.23606798, 3.46410162], [4.58257569, 5.65685425]], + [[3.34370152, 4.55901411], [5.66374698, 6.72717132]], + [[5.0, 6.0], [7.0, 8.0]], + ] + ) + np.testing.assert_array_almost_equal(dense, expected) + + def test_exp_impmat_interpolate_inconsistent_shape(self): + with self.assertRaises(ValueError): + exponential_interp_imp_mat(self.imp_mat0, self.imp_mat2, self.time_points) + + +class TestInterpolationStrategies(unittest.TestCase): + + def setUp(self): + self.interpolation_range = 3 + self.dummy_metric_0 = np.array([10, 20]) + self.dummy_metric_1 = np.array([100, 200]) + self.dummy_matrix_0 = csr_matrix(np.array([[1, 2], [3, 4]])) + self.dummy_matrix_1 = csr_matrix(np.array([[10, 20], [30, 40]])) + + def test_InterpolationStrategy_init(self): + def mock_exposure(a, b, r): + return a + b + + def mock_hazard(a, b, r): + return a * b + + def mock_vulnerability(a, b, r): + return a / b + + strategy = InterpolationStrategy(mock_exposure, mock_hazard, mock_vulnerability) + self.assertEqual(strategy.exposure_interp, mock_exposure) + self.assertEqual(strategy.hazard_interp, mock_hazard) + self.assertEqual(strategy.vulnerability_interp, mock_vulnerability) + + def test_InterpolationStrategy_interp_exposure_dim(self): + mock_exposure = MagicMock(return_value=["mock_result"]) + strategy = InterpolationStrategy( + mock_exposure, linear_interp_arrays, linear_interp_arrays + ) + + result = strategy.interp_over_exposure_dim( + self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range + ) + mock_exposure.assert_called_once_with( + self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range + ) + self.assertEqual(result, ["mock_result"]) + + def test_InterpolationStrategy_interp_exposure_dim_inconsistent_shapes(self): + mock_exposure = MagicMock(side_effect=ValueError("inconsistent shapes")) + strategy = InterpolationStrategy( + mock_exposure, linear_interp_arrays, linear_interp_arrays + ) + + with self.assertRaisesRegex( + ValueError, "Tried to interpolate impact matrices of different shape" + ): + strategy.interp_over_exposure_dim( + self.dummy_matrix_0, + csr_matrix(np.array([[1]])), + self.interpolation_range, + ) + mock_exposure.assert_called_once() # Ensure it was called + + def test_InterpolationStrategy_interp_hazard_dim(self): + mock_hazard = MagicMock(return_value=np.array([1, 2, 3])) + strategy = InterpolationStrategy( + linear_interp_imp_mat, mock_hazard, linear_interp_arrays + ) + + result = strategy.interp_over_hazard_dim( + self.dummy_metric_0, self.dummy_metric_1 + ) + mock_hazard.assert_called_once_with(self.dummy_metric_0, self.dummy_metric_1) + np.testing.assert_array_equal(result, np.array([1, 2, 3])) + + def test_InterpolationStrategy_interp_vulnerability_dim(self): + mock_vulnerability = MagicMock(return_value=np.array([4, 5, 6])) + strategy = InterpolationStrategy( + linear_interp_imp_mat, linear_interp_arrays, mock_vulnerability + ) + + result = strategy.interp_over_vulnerability_dim( + self.dummy_metric_0, self.dummy_metric_1 + ) + mock_vulnerability.assert_called_once_with( + self.dummy_metric_0, self.dummy_metric_1 + ) + np.testing.assert_array_equal(result, np.array([4, 5, 6])) + + +class TestConcreteInterpolationStrategies(unittest.TestCase): + + def setUp(self): + self.interpolation_range = 3 + self.dummy_metric_0 = np.array([10, 20, 30]) + self.dummy_metric_1 = np.array([100, 200, 300]) + self.dummy_matrix_0 = csr_matrix([[1, 2], [3, 4]]) + self.dummy_matrix_1 = csr_matrix([[10, 20], [30, 40]]) + self.dummy_matrix_0_1_lin = csr_matrix([[5.5, 11], [16.5, 22]]) + self.dummy_matrix_0_1_exp = csr_matrix( + [[3.162278, 6.324555], [9.486833, 12.649111]] + ) + self.rtol = 1e-5 + self.atol = 1e-8 + + def test_AllLinearStrategy_init_and_methods(self): + strategy = AllLinearStrategy() + self.assertEqual(strategy.exposure_interp, linear_interp_imp_mat) + self.assertEqual(strategy.hazard_interp, linear_interp_arrays) + self.assertEqual(strategy.vulnerability_interp, linear_interp_arrays) + + # Test hazard interpolation + expected_hazard_interp = linear_interp_arrays( + self.dummy_metric_0, self.dummy_metric_1 + ) + result_hazard = strategy.interp_over_hazard_dim( + self.dummy_metric_0, self.dummy_metric_1 + ) + np.testing.assert_allclose( + result_hazard, expected_hazard_interp, rtol=self.rtol, atol=self.atol + ) + + # Test vulnerability interpolation + expected_vulnerability_interp = linear_interp_arrays( + self.dummy_metric_0, self.dummy_metric_1 + ) + result_vulnerability = strategy.interp_over_vulnerability_dim( + self.dummy_metric_0, self.dummy_metric_1 + ) + np.testing.assert_allclose( + result_vulnerability, + expected_vulnerability_interp, + rtol=self.rtol, + atol=self.atol, + ) + + # Test exposure interpolation (using mock for linear_interp_imp_mat) + result_exposure = strategy.interp_over_exposure_dim( + self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range + ) + # Verify the structure/first/last elements of the mock output + self.assertEqual(len(result_exposure), self.interpolation_range) + np.testing.assert_allclose(result_exposure[0].data, self.dummy_matrix_0.data) + np.testing.assert_allclose( + result_exposure[1].data, self.dummy_matrix_0_1_lin.data + ) + np.testing.assert_allclose(result_exposure[2].data, self.dummy_matrix_1.data) + + def test_ExponentialExposureInterpolation_init_and_methods(self): + strategy = ExponentialExposureStrategy() + # Test hazard interpolation (should be linear) + expected_hazard_interp = linear_interp_arrays( + self.dummy_metric_0, self.dummy_metric_1 + ) + result_hazard = strategy.interp_over_hazard_dim( + self.dummy_metric_0, self.dummy_metric_1 + ) + np.testing.assert_allclose( + result_hazard, expected_hazard_interp, rtol=self.rtol, atol=self.atol + ) + + # Test vulnerability interpolation (should be linear) + expected_vulnerability_interp = linear_interp_arrays( + self.dummy_metric_0, self.dummy_metric_1 + ) + result_vulnerability = strategy.interp_over_vulnerability_dim( + self.dummy_metric_0, self.dummy_metric_1 + ) + np.testing.assert_allclose( + result_vulnerability, + expected_vulnerability_interp, + rtol=self.rtol, + atol=self.atol, + ) + + # Test exposure interpolation (using mock for exponential_interp_imp_mat) + result_exposure = strategy.interp_over_exposure_dim( + self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range + ) + # Verify the structure/first/last elements of the mock output + self.assertEqual(len(result_exposure), self.interpolation_range) + np.testing.assert_allclose(result_exposure[0].data, self.dummy_matrix_0.data) + np.testing.assert_allclose( + result_exposure[1].data, + self.dummy_matrix_0_1_exp.data, + rtol=self.rtol, + atol=self.atol, + ) + np.testing.assert_allclose(result_exposure[-1].data, self.dummy_matrix_1.data) + + +if __name__ == "__main__": + TESTS = unittest.TestLoader().loadTestsFromTestCase( + TestConcreteInterpolationStrategies + ) + TESTS.addTests(unittest.TestLoader().loadTestsFromTestCase(TestInterpolationFuncs)) + TESTS.addTests( + unittest.TestLoader().loadTestsFromTestCase(TestInterpolationStrategies) + ) + unittest.TextTestRunner(verbosity=2).run(TESTS) From 6eee8c5798a2296ce81370f912324c5b2c8c1aec Mon Sep 17 00:00:00 2001 From: spjuhel Date: Thu, 18 Dec 2025 11:09:03 +0100 Subject: [PATCH 02/10] cherry picks __init__ --- climada/trajectories/__init__.py | 35 ++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 climada/trajectories/__init__.py diff --git a/climada/trajectories/__init__.py b/climada/trajectories/__init__.py new file mode 100644 index 0000000000..db58a711ca --- /dev/null +++ b/climada/trajectories/__init__.py @@ -0,0 +1,35 @@ +""" +This file is part of CLIMADA. + +Copyright (C) 2017 ETH Zurich, CLIMADA contributors listed in AUTHORS. + +CLIMADA is free software: you can redistribute it and/or modify it under the +terms of the GNU General Public License as published by the Free +Software Foundation, version 3. + +CLIMADA is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU General Public License for more details. + +You should have received a copy of the GNU General Public License along +with CLIMADA. If not, see . + +--- + +This module implements risk trajectory objects which enable computation and +possibly interpolation of risk metric over multiple dates. + +""" + +from .interpolated_trajectory import InterpolatedRiskTrajectory +from .interpolation import AllLinearStrategy, ExponentialExposureStrategy +from .snapshot import Snapshot +from .static_trajectory import StaticRiskTrajectory + +__all__ = [ + "InterpolatedRiskTrajectory", + "AllLinearStrategy", + "ExponentialExposureStrategy", + "Snapshot", + "StaticRiskTrajectory", +] From d11871fbd1cd5b0d0cb6c32e74ef1594075131d9 Mon Sep 17 00:00:00 2001 From: spjuhel Date: Thu, 18 Dec 2025 11:09:34 +0100 Subject: [PATCH 03/10] cherry pick __init__ (for real) --- climada/trajectories/__init__.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/climada/trajectories/__init__.py b/climada/trajectories/__init__.py index db58a711ca..5e0016c39d 100644 --- a/climada/trajectories/__init__.py +++ b/climada/trajectories/__init__.py @@ -21,15 +21,9 @@ """ -from .interpolated_trajectory import InterpolatedRiskTrajectory from .interpolation import AllLinearStrategy, ExponentialExposureStrategy -from .snapshot import Snapshot -from .static_trajectory import StaticRiskTrajectory __all__ = [ - "InterpolatedRiskTrajectory", "AllLinearStrategy", "ExponentialExposureStrategy", - "Snapshot", - "StaticRiskTrajectory", ] From 463568cb6076ebc59124baa5f80a949f66bca03b Mon Sep 17 00:00:00 2001 From: spjuhel Date: Mon, 5 Jan 2026 15:25:15 +0100 Subject: [PATCH 04/10] complies with pylint --- climada/trajectories/interpolation.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/climada/trajectories/interpolation.py b/climada/trajectories/interpolation.py index 9f6687e449..c07c53883c 100644 --- a/climada/trajectories/interpolation.py +++ b/climada/trajectories/interpolation.py @@ -430,10 +430,6 @@ class ExponentialExposureStrategy(InterpolationStrategyBase): def __init__(self) -> None: super().__init__() - self.exposure_interp = ( - lambda mat_start, mat_end, points: exponential_interp_imp_mat( - mat_start, mat_end, points - ) - ) + self.exposure_interp = exponential_interp_imp_mat self.hazard_interp = linear_interp_arrays self.vulnerability_interp = linear_interp_arrays From 76e237f2d22e28eef32f548cf6812f28e727a0eb Mon Sep 17 00:00:00 2001 From: spjuhel Date: Tue, 6 Jan 2026 10:44:03 +0100 Subject: [PATCH 05/10] API reference files --- doc/api/climada/climada.trajectories.interpolation.rst | 7 +++++++ doc/api/climada/climada.trajectories.rst | 7 +++++++ 2 files changed, 14 insertions(+) create mode 100644 doc/api/climada/climada.trajectories.interpolation.rst create mode 100644 doc/api/climada/climada.trajectories.rst diff --git a/doc/api/climada/climada.trajectories.interpolation.rst b/doc/api/climada/climada.trajectories.interpolation.rst new file mode 100644 index 0000000000..98e1ec7b32 --- /dev/null +++ b/doc/api/climada/climada.trajectories.interpolation.rst @@ -0,0 +1,7 @@ +climada\.trajectories\.interpolation module +---------------------------------------- + +.. automodule:: climada.trajectories.interpolation + :members: + :undoc-members: + :show-inheritance: diff --git a/doc/api/climada/climada.trajectories.rst b/doc/api/climada/climada.trajectories.rst new file mode 100644 index 0000000000..ca449199d5 --- /dev/null +++ b/doc/api/climada/climada.trajectories.rst @@ -0,0 +1,7 @@ + +climada\.trajectories module +============================ + +.. toctree:: + + climada.trajectories.interpolation From 984215348ee6e1c803ad894a42dfa67ff94746e4 Mon Sep 17 00:00:00 2001 From: Samuel Juhel <10011382+spjuhel@users.noreply.github.com> Date: Tue, 6 Jan 2026 13:17:32 +0100 Subject: [PATCH 06/10] Apply suggestions from code review Co-authored-by: Chahan M. Kropf --- climada/trajectories/interpolation.py | 22 ++++++++++++++----- .../trajectories/test/test_interpolation.py | 4 ++-- 2 files changed, 18 insertions(+), 8 deletions(-) diff --git a/climada/trajectories/interpolation.py b/climada/trajectories/interpolation.py index c07c53883c..9905bb24de 100644 --- a/climada/trajectories/interpolation.py +++ b/climada/trajectories/interpolation.py @@ -150,7 +150,8 @@ def exponential_interp_imp_mat( def linear_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarray: r""" - Performs linear interpolation between two NumPy arrays over their first dimension. + Performs linear interpolation between two n x m NumPy arrays over their + first dimension (n rows). This function interpolates each metric (column) linearly across the time steps (rows), including both the start and end states. @@ -175,6 +176,11 @@ def linear_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarr ValueError If `arr_start` and `arr_end` do not have the same shape. + Example + -------- + >>> arr_start = [ [ 1, 1], [1, 2], [10, 20] ] + >>> arr_end = [ [2, 2], [5, 6], [10, 30] ] + >>> linear_interp_arrays(arr_start, arr_end) = [ [1, 1], [3, 4], [10, 30] ] Notes ----- The interpolation is performed element-wise along the first dimension @@ -203,8 +209,7 @@ def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np. Performs exponential interpolation between two NumPy arrays over their first dimension. This function achieves an exponential-like transition by performing linear - interpolation in the logarithmic space, suitable to interpolate over a dimension which has - a growth factor. + interpolation in the logarithmic space. Parameters ---------- @@ -224,6 +229,10 @@ def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np. ------ ValueError If `arr_start` and `arr_end` do not have the same shape. + + See Also + --------- + linear_interp_arrays: linear version of the interpolation. Notes ----- @@ -264,11 +273,12 @@ def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np. class InterpolationStrategyBase(ABC): r""" - Base abstract class for defining a set of interpolation strategies. + Base abstract class for defining a set of interpolation strategies for impact outputs. This class serves as a blueprint for implementing specific interpolation - methods (e.g., 'Linear', 'Exponential') across different impact dimensions: - Exposure (matrices), Hazard, and Vulnerability (arrays/metrics). + methods (e.g., 'Linear', 'Exponential') for impact outputs (impact matrices + or metrics such as aai_agg, eai_exp or at_event) across the input dimentions + Exposure (impact matrices only), Hazard, and Vulnerability (metrics only). Attributes ---------- diff --git a/climada/trajectories/test/test_interpolation.py b/climada/trajectories/test/test_interpolation.py index 693c9b9c33..58c06dc917 100644 --- a/climada/trajectories/test/test_interpolation.py +++ b/climada/trajectories/test/test_interpolation.py @@ -95,9 +95,9 @@ def test_exponential_interp_arrays_2d(self): [ [1, 10, 100], # date 1 metric a,b,c [1, 10, 100], # date 2 metric a,b,c - [1, 10, 100], + [1, 10, 100], # date 3 metric a,b,c ] - ) # date 3 metric a,b,c + ) arr_end = np.array([[2, 20, 200], [2, 20, 200], [2, 20, 200]]) expected = np.array( [[1.0, 10.0, 100.0], [1.4142136, 14.142136, 141.42136], [2, 20, 200]] From c42e8714fe31675246edb5bb5890a4e2b99bc171 Mon Sep 17 00:00:00 2001 From: spjuhel Date: Tue, 6 Jan 2026 13:17:49 +0100 Subject: [PATCH 07/10] Renames base functions --- climada/trajectories/interpolation.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/climada/trajectories/interpolation.py b/climada/trajectories/interpolation.py index c07c53883c..4921f20adb 100644 --- a/climada/trajectories/interpolation.py +++ b/climada/trajectories/interpolation.py @@ -35,13 +35,13 @@ "AllLinearStrategy", "ExponentialExposureStrategy", "linear_interp_arrays", - "linear_interp_imp_mat", + "linear_interp_matrix_elemwise", "exponential_interp_arrays", - "exponential_interp_imp_mat", + "exponential_interp_matrix_elemwise", ] -def linear_interp_imp_mat( +def linear_interp_matrix_elemwise( mat_start: sparse.csr_matrix, mat_end: sparse.csr_matrix, number_of_interpolation_points: int, @@ -85,7 +85,7 @@ def linear_interp_imp_mat( ] -def exponential_interp_imp_mat( +def exponential_interp_matrix_elemwise( mat_start: sparse.csr_matrix, mat_end: sparse.csr_matrix, number_of_interpolation_points: int, @@ -420,7 +420,7 @@ class AllLinearStrategy(InterpolationStrategyBase): def __init__(self) -> None: super().__init__() - self.exposure_interp = linear_interp_imp_mat + self.exposure_interp = linear_interp_matrix_elemwise self.hazard_interp = linear_interp_arrays self.vulnerability_interp = linear_interp_arrays @@ -430,6 +430,6 @@ class ExponentialExposureStrategy(InterpolationStrategyBase): def __init__(self) -> None: super().__init__() - self.exposure_interp = exponential_interp_imp_mat + self.exposure_interp = exponential_interp_matrix_elemwise self.hazard_interp = linear_interp_arrays self.vulnerability_interp = linear_interp_arrays From 7482702451f9a513b2a5fa244a2acebd93c7cdb9 Mon Sep 17 00:00:00 2001 From: spjuhel Date: Tue, 6 Jan 2026 14:26:11 +0100 Subject: [PATCH 08/10] Improves vocables (hopefully) --- climada/trajectories/interpolation.py | 88 ++++++++++++++++++--------- 1 file changed, 58 insertions(+), 30 deletions(-) diff --git a/climada/trajectories/interpolation.py b/climada/trajectories/interpolation.py index 8d23aa4d5c..45fc18f575 100644 --- a/climada/trajectories/interpolation.py +++ b/climada/trajectories/interpolation.py @@ -34,9 +34,9 @@ __all__ = [ "AllLinearStrategy", "ExponentialExposureStrategy", - "linear_interp_arrays", + "linear_convex_combination", "linear_interp_matrix_elemwise", - "exponential_interp_arrays", + "exponential_convex_combination", "exponential_interp_matrix_elemwise", ] @@ -148,9 +148,9 @@ def exponential_interp_matrix_elemwise( return res -def linear_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarray: +def linear_convex_combination(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarray: r""" - Performs linear interpolation between two n x m NumPy arrays over their + Performs a linear convex combination between two n x m NumPy arrays over their first dimension (n rows). This function interpolates each metric (column) linearly across the time steps @@ -180,7 +180,9 @@ def linear_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarr -------- >>> arr_start = [ [ 1, 1], [1, 2], [10, 20] ] >>> arr_end = [ [2, 2], [5, 6], [10, 30] ] - >>> linear_interp_arrays(arr_start, arr_end) = [ [1, 1], [3, 4], [10, 30] ] + >>> linear_interp_arrays(arr_start, arr_end) + >>> [[1, 1], [3, 4], [10, 30]] + Notes ----- The interpolation is performed element-wise along the first dimension @@ -204,9 +206,11 @@ def linear_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarr return np.multiply(arr_start, prop0) + np.multiply(arr_end, prop1) -def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np.ndarray: +def exponential_convex_combination( + arr_start: np.ndarray, arr_end: np.ndarray +) -> np.ndarray: r""" - Performs exponential interpolation between two NumPy arrays over their first dimension. + Performs exponential convex combination between two NumPy arrays over their first dimension. This function achieves an exponential-like transition by performing linear interpolation in the logarithmic space. @@ -229,7 +233,7 @@ def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np. ------ ValueError If `arr_start` and `arr_end` do not have the same shape. - + See Also --------- linear_interp_arrays: linear version of the interpolation. @@ -271,28 +275,49 @@ def exponential_interp_arrays(arr_start: np.ndarray, arr_end: np.ndarray) -> np. return np.exp(interpolated_log_arr) -class InterpolationStrategyBase(ABC): +class ImpactInterpolationStrategy(ABC): r""" Base abstract class for defining a set of interpolation strategies for impact outputs. This class serves as a blueprint for implementing specific interpolation - methods (e.g., 'Linear', 'Exponential') for impact outputs (impact matrices - or metrics such as aai_agg, eai_exp or at_event) across the input dimentions - Exposure (impact matrices only), Hazard, and Vulnerability (metrics only). + methods (e.g., 'Linear', 'Exponential') describing how impact outputs + should evolve between two points in time. + + Impacts result from three dimensions—Exposure, Hazard, and Vulnerability— + each of which may change differently over time. Consequently, a distinct + interpolation strategy is defined for each dimension. + + Exposure interpolation differs from Hazard and Vulnerability interpolation. + Changes in exposure do not alter the shape of the impact matrices, which + allows direct interpolation of the matrices themselves. For the Exposure + dimension, interpolation therefore consists of generating intermediate + impact matrices between the two time points, with exposure evolving while + hazard and vulnerability remain fixed (to either the first or second point). + + In contrast, changes in Hazard may alter the + set of events between the two time points, making direct interpolation of + impact matrices impossible. Instead, impacts are first aggregated over the + event dimension (i.e. the EAI metric). The evolution of impacts is then + interpolated as a convex combination of metric sequences computed from two + scenarios: one with hazard fixed at the initial time point and one with + hazard fixed at the final time point. + + The same aggregation-based interpolation approach is applied to the + Vulnerability dimension. Attributes ---------- exposure_interp : Callable - The function used to interpolate sparse impact matrices over the - exposure dimension. + The function used to interpolate sparse impact matrices over time + with changing exposure dimension. Signature: (mat_start, mat_end, num_points, **kwargs) -> list[sparse.csr_matrix]. hazard_interp : Callable - The function used to interpolate NumPy arrays of metrics over the - hazard dimension. + The function used to interpolate NumPy arrays of metrics over time + with changing hazard dimension. Signature: (arr_start, arr_end, **kwargs) -> np.ndarray. vulnerability_interp : Callable - The function used to interpolate NumPy arrays of metrics over the - vulnerability dimension. + The function used to interpolate NumPy arrays of metrics over time + with changing vulnerability dimension. Signature: (arr_start, arr_end, **kwargs) -> np.ndarray. """ @@ -309,10 +334,11 @@ def interp_over_exposure_dim( **kwargs: Optional[Dict[str, Any]], ) -> List[sparse.csr_matrix]: """ - Interpolates between two impact matrices using the defined exposure strategy. + Interpolates between two impact matrices using the defined strategy for the exposure + dimension. This method calls the function assigned to :attr:`exposure_interp` to generate - a sequence of matrices. + a sequence of impact matrices of length "interpolation_range". Parameters ---------- @@ -357,7 +383,8 @@ def interp_over_hazard_dim( **kwargs: Optional[Dict[str, Any]], ) -> np.ndarray: """ - Interpolates between two metric arrays using the defined hazard strategy. + Generates the convex combination between two arrays of metrics using + the defined interpolation strategy for the hazard dimension. This method calls the function assigned to :attr:`hazard_interp`. @@ -385,7 +412,8 @@ def interp_over_vulnerability_dim( **kwargs: Optional[Dict[str, Any]], ) -> np.ndarray: """ - Interpolates between two metric arrays using the defined vulnerability strategy. + Generates the convex combination between two arrays of metrics using + the defined interpolation strategy for the hazard dimension. This method calls the function assigned to :attr:`vulnerability_interp`. @@ -407,10 +435,10 @@ def interp_over_vulnerability_dim( return self.vulnerability_interp(metric_0, metric_1, **kwargs) -class InterpolationStrategy(InterpolationStrategyBase): +class CustomImpactInterpolationStrategy(ImpactInterpolationStrategy): r"""Interface for interpolation strategies. - This is the class to use to define your own custom interpolation strategy. + This is the class to use to define custom interpolation strategies. """ def __init__( @@ -425,21 +453,21 @@ def __init__( self.vulnerability_interp = vulnerability_interp -class AllLinearStrategy(InterpolationStrategyBase): +class AllLinearStrategy(ImpactInterpolationStrategy): r"""Linear interpolation strategy over all dimensions.""" def __init__(self) -> None: super().__init__() self.exposure_interp = linear_interp_matrix_elemwise - self.hazard_interp = linear_interp_arrays - self.vulnerability_interp = linear_interp_arrays + self.hazard_interp = linear_convex_combination + self.vulnerability_interp = linear_convex_combination -class ExponentialExposureStrategy(InterpolationStrategyBase): +class ExponentialExposureStrategy(ImpactInterpolationStrategy): r"""Exponential interpolation strategy for exposure and linear for Hazard and Vulnerability.""" def __init__(self) -> None: super().__init__() self.exposure_interp = exponential_interp_matrix_elemwise - self.hazard_interp = linear_interp_arrays - self.vulnerability_interp = linear_interp_arrays + self.hazard_interp = linear_convex_combination + self.vulnerability_interp = linear_convex_combination From d7c4a9b6b78d4f96b9ddfb6449530ae407a0c32f Mon Sep 17 00:00:00 2001 From: spjuhel Date: Tue, 6 Jan 2026 14:57:09 +0100 Subject: [PATCH 09/10] Shifts to pytest format --- .../trajectories/test/test_interpolation.py | 433 ++++++------------ 1 file changed, 140 insertions(+), 293 deletions(-) diff --git a/climada/trajectories/test/test_interpolation.py b/climada/trajectories/test/test_interpolation.py index 58c06dc917..f601fce8ff 100644 --- a/climada/trajectories/test/test_interpolation.py +++ b/climada/trajectories/test/test_interpolation.py @@ -20,333 +20,180 @@ """ -import unittest from unittest.mock import MagicMock import numpy as np +import pytest from scipy.sparse import csr_matrix from climada.trajectories.interpolation import ( AllLinearStrategy, + CustomImpactInterpolationStrategy, ExponentialExposureStrategy, - InterpolationStrategy, - exponential_interp_arrays, - exponential_interp_imp_mat, - linear_interp_arrays, - linear_interp_imp_mat, + exponential_convex_combination, + exponential_interp_matrix_elemwise, + linear_convex_combination, + linear_interp_matrix_elemwise, ) +# --- Fixtures --- + + +@pytest.fixture +def interpolation_data(): + """Provides common matrices and constants for interpolation tests.""" + return { + "imp_mat0": csr_matrix(np.array([[1, 2], [3, 4]])), + "imp_mat1": csr_matrix(np.array([[5, 6], [7, 8]])), + "imp_mat2": csr_matrix(np.array([[5, 6, 7], [8, 9, 10]])), + "time_points": 5, + "rtol": 1e-5, + "atol": 1e-8, + "dummy_metric_0": np.array([10, 20, 30]), + "dummy_metric_1": np.array([100, 200, 300]), + "dummy_matrix_0": csr_matrix([[1, 2], [3, 4]]), + "dummy_matrix_1": csr_matrix([[10, 20], [30, 40]]), + } + + +# --- Tests for Interpolation Functions --- + + +def test_linear_interp_arrays(interpolation_data): + arr_start = np.array([10, 50, 100]) + arr_end = np.array([20, 100, 200]) + expected = np.array([10.0, 75.0, 200.0]) + result = linear_convex_combination(arr_start, arr_end) + np.testing.assert_allclose( + result, + expected, + rtol=interpolation_data["rtol"], + atol=interpolation_data["atol"], + ) -class TestInterpolationFuncs(unittest.TestCase): - def setUp(self): - # Create mock impact matrices for testing - self.imp_mat0 = csr_matrix(np.array([[1, 2], [3, 4]])) - self.imp_mat1 = csr_matrix(np.array([[5, 6], [7, 8]])) - self.imp_mat2 = csr_matrix(np.array([[5, 6, 7], [8, 9, 10]])) # Different shape - self.time_points = 5 - self.interpolation_range_5 = 5 - self.interpolation_range_1 = 1 - self.interpolation_range_2 = 2 - self.rtol = 1e-5 - self.atol = 1e-8 - - def test_linear_interp_arrays(self): - arr_start = np.array([10, 100]) - arr_end = np.array([20, 200]) - expected = np.array([10.0, 200.0]) - result = linear_interp_arrays(arr_start, arr_end) - np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) - - def test_linear_interp_arrays2D(self): - arr_start = np.array([[10, 100], [10, 100]]) - arr_end = np.array([[20, 200], [20, 200]]) - expected = np.array([[10.0, 100.0], [20, 200]]) - result = linear_interp_arrays(arr_start, arr_end) - np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) - - def test_linear_interp_arrays_shape(self): - arr_start = np.array([10, 100, 5]) - arr_end = np.array([20, 200]) - with self.assertRaises(ValueError): - linear_interp_arrays(arr_start, arr_end) - - def test_linear_interp_arrays_start_equals_end(self): - arr_start = np.array([5, 5]) - arr_end = np.array([5, 5]) - expected = np.array([5.0, 5.0]) - result = linear_interp_arrays(arr_start, arr_end) - np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) - - def test_exponential_interp_arrays_1d(self): - arr_start = np.array([1, 10, 100]) - arr_end = np.array([2, 20, 200]) - expected = np.array([1.0, 14.142136, 200.0]) - result = exponential_interp_arrays(arr_start, arr_end) - np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) - - def test_exponential_interp_arrays_shape(self): - arr_start = np.array([10, 100, 5]) - arr_end = np.array([20, 200]) - with self.assertRaises(ValueError): - exponential_interp_arrays(arr_start, arr_end) - - def test_exponential_interp_arrays_2d(self): - arr_start = np.array( - [ - [1, 10, 100], # date 1 metric a,b,c - [1, 10, 100], # date 2 metric a,b,c - [1, 10, 100], # date 3 metric a,b,c - ] - ) - arr_end = np.array([[2, 20, 200], [2, 20, 200], [2, 20, 200]]) - expected = np.array( - [[1.0, 10.0, 100.0], [1.4142136, 14.142136, 141.42136], [2, 20, 200]] - ) - result = exponential_interp_arrays(arr_start, arr_end) - np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) - - def test_exponential_interp_arrays_start_equals_end(self): - arr_start = np.array([5, 5]) - arr_end = np.array([5, 5]) - expected = np.array([5.0, 5.0]) - result = exponential_interp_arrays(arr_start, arr_end) - np.testing.assert_allclose(result, expected, rtol=self.rtol, atol=self.atol) - - def test_linear_impmat_interpolate(self): - result = linear_interp_imp_mat(self.imp_mat0, self.imp_mat1, self.time_points) - self.assertEqual(len(result), self.time_points) - for mat in result: - self.assertIsInstance(mat, csr_matrix) - - dense = np.array([r.todense() for r in result]) - expected = np.array( - [ - [[1.0, 2.0], [3.0, 4.0]], - [[2.0, 3.0], [4.0, 5.0]], - [[3.0, 4.0], [5.0, 6.0]], - [[4.0, 5.0], [6.0, 7.0]], - [[5.0, 6.0], [7.0, 8.0]], - ] - ) - np.testing.assert_array_equal(dense, expected) - def test_linear_impmat_interpolate_inconsistent_shape(self): - with self.assertRaises(ValueError): - linear_interp_imp_mat(self.imp_mat0, self.imp_mat2, self.time_points) +@pytest.mark.parametrize( + "func", [linear_convex_combination, exponential_convex_combination] +) +def test_convex_combination_shape_error(func): + arr_start = np.array([10, 100, 5]) + arr_end = np.array([20, 200]) + with pytest.raises(ValueError, match="different shapes"): + func(arr_start, arr_end) + + +def test_exponential_convex_combination_2d(interpolation_data): + arr_start = np.array([[1, 10, 100]] * 3) + arr_end = np.array([[2, 20, 200]] * 3) + expected = np.array( + [[1.0, 10.0, 100.0], [1.4142136, 14.142136, 141.42136], [2, 20, 200]] + ) + result = exponential_convex_combination(arr_start, arr_end) + np.testing.assert_allclose( + result, + expected, + rtol=interpolation_data["rtol"], + atol=interpolation_data["atol"], + ) - def test_exp_impmat_interpolate(self): - result = exponential_interp_imp_mat( - self.imp_mat0, self.imp_mat1, self.time_points - ) - self.assertEqual(len(result), self.time_points) - for mat in result: - self.assertIsInstance(mat, csr_matrix) - - dense = np.array([r.todense() for r in result]) - expected = np.array( - [ - [[1.0, 2.0], [3.0, 4.0]], - [[1.49534878, 2.63214803], [3.70779275, 4.75682846]], - [[2.23606798, 3.46410162], [4.58257569, 5.65685425]], - [[3.34370152, 4.55901411], [5.66374698, 6.72717132]], - [[5.0, 6.0], [7.0, 8.0]], - ] - ) - np.testing.assert_array_almost_equal(dense, expected) - def test_exp_impmat_interpolate_inconsistent_shape(self): - with self.assertRaises(ValueError): - exponential_interp_imp_mat(self.imp_mat0, self.imp_mat2, self.time_points) +@pytest.mark.parametrize( + "func", [linear_convex_combination, exponential_convex_combination] +) +def test_convex_combinations_start_equals_end(interpolation_data, func): + """Test that if start and end are identical, the result is the same array.""" + arr = np.array([5.0, 5.0]) + result = func(arr, arr) + np.testing.assert_allclose(result, arr, rtol=interpolation_data["rtol"]) -class TestInterpolationStrategies(unittest.TestCase): +def test_linear_impmat_interpolate(interpolation_data): + data = interpolation_data + result = linear_interp_matrix_elemwise( + data["imp_mat0"], data["imp_mat1"], data["time_points"] + ) - def setUp(self): - self.interpolation_range = 3 - self.dummy_metric_0 = np.array([10, 20]) - self.dummy_metric_1 = np.array([100, 200]) - self.dummy_matrix_0 = csr_matrix(np.array([[1, 2], [3, 4]])) - self.dummy_matrix_1 = csr_matrix(np.array([[10, 20], [30, 40]])) + assert len(result) == data["time_points"] + assert all(isinstance(mat, csr_matrix) for mat in result) + + dense = np.array([r.todense() for r in result]) + expected = np.array( + [ + [[1.0, 2.0], [3.0, 4.0]], + [[2.0, 3.0], [4.0, 5.0]], + [[3.0, 4.0], [5.0, 6.0]], + [[4.0, 5.0], [6.0, 7.0]], + [[5.0, 6.0], [7.0, 8.0]], + ] + ) + np.testing.assert_array_equal(dense, expected) - def test_InterpolationStrategy_init(self): - def mock_exposure(a, b, r): - return a + b - def mock_hazard(a, b, r): - return a * b +# --- Tests for Interpolation Strategies --- - def mock_vulnerability(a, b, r): - return a / b - strategy = InterpolationStrategy(mock_exposure, mock_hazard, mock_vulnerability) - self.assertEqual(strategy.exposure_interp, mock_exposure) - self.assertEqual(strategy.hazard_interp, mock_hazard) - self.assertEqual(strategy.vulnerability_interp, mock_vulnerability) +def test_custom_strategy_init(): + mock_func = lambda a, b, r: a + b + strategy = CustomImpactInterpolationStrategy(mock_func, mock_func, mock_func) + assert strategy.exposure_interp == mock_func + assert strategy.hazard_interp == mock_func + assert strategy.vulnerability_interp == mock_func - def test_InterpolationStrategy_interp_exposure_dim(self): - mock_exposure = MagicMock(return_value=["mock_result"]) - strategy = InterpolationStrategy( - mock_exposure, linear_interp_arrays, linear_interp_arrays - ) - result = strategy.interp_over_exposure_dim( - self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range - ) - mock_exposure.assert_called_once_with( - self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range - ) - self.assertEqual(result, ["mock_result"]) +def test_custom_strategy_exposure_dim_error(interpolation_data): + mock_exposure = MagicMock(side_effect=ValueError("inconsistent shapes")) + strategy = CustomImpactInterpolationStrategy( + mock_exposure, linear_convex_combination, linear_convex_combination + ) - def test_InterpolationStrategy_interp_exposure_dim_inconsistent_shapes(self): - mock_exposure = MagicMock(side_effect=ValueError("inconsistent shapes")) - strategy = InterpolationStrategy( - mock_exposure, linear_interp_arrays, linear_interp_arrays + with pytest.raises( + ValueError, match="Tried to interpolate impact matrices of different shape" + ): + strategy.interp_over_exposure_dim( + interpolation_data["dummy_matrix_0"], csr_matrix(np.array([[1]])), 3 ) - with self.assertRaisesRegex( - ValueError, "Tried to interpolate impact matrices of different shape" - ): - strategy.interp_over_exposure_dim( - self.dummy_matrix_0, - csr_matrix(np.array([[1]])), - self.interpolation_range, - ) - mock_exposure.assert_called_once() # Ensure it was called - - def test_InterpolationStrategy_interp_hazard_dim(self): - mock_hazard = MagicMock(return_value=np.array([1, 2, 3])) - strategy = InterpolationStrategy( - linear_interp_imp_mat, mock_hazard, linear_interp_arrays - ) - result = strategy.interp_over_hazard_dim( - self.dummy_metric_0, self.dummy_metric_1 - ) - mock_hazard.assert_called_once_with(self.dummy_metric_0, self.dummy_metric_1) - np.testing.assert_array_equal(result, np.array([1, 2, 3])) +# --- Tests for Concrete Strategies --- - def test_InterpolationStrategy_interp_vulnerability_dim(self): - mock_vulnerability = MagicMock(return_value=np.array([4, 5, 6])) - strategy = InterpolationStrategy( - linear_interp_imp_mat, linear_interp_arrays, mock_vulnerability - ) - - result = strategy.interp_over_vulnerability_dim( - self.dummy_metric_0, self.dummy_metric_1 - ) - mock_vulnerability.assert_called_once_with( - self.dummy_metric_0, self.dummy_metric_1 - ) - np.testing.assert_array_equal(result, np.array([4, 5, 6])) +def test_all_linear_strategy(interpolation_data): + data = interpolation_data + strategy = AllLinearStrategy() -class TestConcreteInterpolationStrategies(unittest.TestCase): + # Test property assignment + assert strategy.exposure_interp == linear_interp_matrix_elemwise - def setUp(self): - self.interpolation_range = 3 - self.dummy_metric_0 = np.array([10, 20, 30]) - self.dummy_metric_1 = np.array([100, 200, 300]) - self.dummy_matrix_0 = csr_matrix([[1, 2], [3, 4]]) - self.dummy_matrix_1 = csr_matrix([[10, 20], [30, 40]]) - self.dummy_matrix_0_1_lin = csr_matrix([[5.5, 11], [16.5, 22]]) - self.dummy_matrix_0_1_exp = csr_matrix( - [[3.162278, 6.324555], [9.486833, 12.649111]] - ) - self.rtol = 1e-5 - self.atol = 1e-8 - - def test_AllLinearStrategy_init_and_methods(self): - strategy = AllLinearStrategy() - self.assertEqual(strategy.exposure_interp, linear_interp_imp_mat) - self.assertEqual(strategy.hazard_interp, linear_interp_arrays) - self.assertEqual(strategy.vulnerability_interp, linear_interp_arrays) - - # Test hazard interpolation - expected_hazard_interp = linear_interp_arrays( - self.dummy_metric_0, self.dummy_metric_1 - ) - result_hazard = strategy.interp_over_hazard_dim( - self.dummy_metric_0, self.dummy_metric_1 - ) - np.testing.assert_allclose( - result_hazard, expected_hazard_interp, rtol=self.rtol, atol=self.atol - ) - - # Test vulnerability interpolation - expected_vulnerability_interp = linear_interp_arrays( - self.dummy_metric_0, self.dummy_metric_1 - ) - result_vulnerability = strategy.interp_over_vulnerability_dim( - self.dummy_metric_0, self.dummy_metric_1 - ) - np.testing.assert_allclose( - result_vulnerability, - expected_vulnerability_interp, - rtol=self.rtol, - atol=self.atol, - ) - - # Test exposure interpolation (using mock for linear_interp_imp_mat) - result_exposure = strategy.interp_over_exposure_dim( - self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range - ) - # Verify the structure/first/last elements of the mock output - self.assertEqual(len(result_exposure), self.interpolation_range) - np.testing.assert_allclose(result_exposure[0].data, self.dummy_matrix_0.data) - np.testing.assert_allclose( - result_exposure[1].data, self.dummy_matrix_0_1_lin.data - ) - np.testing.assert_allclose(result_exposure[2].data, self.dummy_matrix_1.data) - - def test_ExponentialExposureInterpolation_init_and_methods(self): - strategy = ExponentialExposureStrategy() - # Test hazard interpolation (should be linear) - expected_hazard_interp = linear_interp_arrays( - self.dummy_metric_0, self.dummy_metric_1 - ) - result_hazard = strategy.interp_over_hazard_dim( - self.dummy_metric_0, self.dummy_metric_1 - ) - np.testing.assert_allclose( - result_hazard, expected_hazard_interp, rtol=self.rtol, atol=self.atol - ) + # Test Hazard dim + result_haz = strategy.interp_over_hazard_dim( + data["dummy_metric_0"], data["dummy_metric_1"] + ) + expected_haz = linear_convex_combination( + data["dummy_metric_0"], data["dummy_metric_1"] + ) + np.testing.assert_allclose(result_haz, expected_haz) - # Test vulnerability interpolation (should be linear) - expected_vulnerability_interp = linear_interp_arrays( - self.dummy_metric_0, self.dummy_metric_1 - ) - result_vulnerability = strategy.interp_over_vulnerability_dim( - self.dummy_metric_0, self.dummy_metric_1 - ) - np.testing.assert_allclose( - result_vulnerability, - expected_vulnerability_interp, - rtol=self.rtol, - atol=self.atol, - ) + # Test Exposure dim + result_exp = strategy.interp_over_exposure_dim( + data["dummy_matrix_0"], data["dummy_matrix_1"], 3 + ) + assert len(result_exp) == 3 + # Check midpoint (index 1) manually + expected_mid = csr_matrix([[5.5, 11], [16.5, 22]]) + np.testing.assert_allclose(result_exp[1].data, expected_mid.data) - # Test exposure interpolation (using mock for exponential_interp_imp_mat) - result_exposure = strategy.interp_over_exposure_dim( - self.dummy_matrix_0, self.dummy_matrix_1, self.interpolation_range - ) - # Verify the structure/first/last elements of the mock output - self.assertEqual(len(result_exposure), self.interpolation_range) - np.testing.assert_allclose(result_exposure[0].data, self.dummy_matrix_0.data) - np.testing.assert_allclose( - result_exposure[1].data, - self.dummy_matrix_0_1_exp.data, - rtol=self.rtol, - atol=self.atol, - ) - np.testing.assert_allclose(result_exposure[-1].data, self.dummy_matrix_1.data) +def test_exponential_exposure_strategy(interpolation_data): + data = interpolation_data + strategy = ExponentialExposureStrategy() -if __name__ == "__main__": - TESTS = unittest.TestLoader().loadTestsFromTestCase( - TestConcreteInterpolationStrategies + result_exp = strategy.interp_over_exposure_dim( + data["dummy_matrix_0"], data["dummy_matrix_1"], 3 ) - TESTS.addTests(unittest.TestLoader().loadTestsFromTestCase(TestInterpolationFuncs)) - TESTS.addTests( - unittest.TestLoader().loadTestsFromTestCase(TestInterpolationStrategies) + + # Midpoint should be geometric mean for exponential strategy + # sqrt(1*10) = 3.162278 + expected_mid_data = np.array([3.162278, 6.324555, 9.486833, 12.649111]) + np.testing.assert_allclose( + result_exp[1].data, expected_mid_data, rtol=data["rtol"], atol=data["atol"] ) - unittest.TextTestRunner(verbosity=2).run(TESTS) From 51430310359bc93c771ab8c734e502c3736e0149 Mon Sep 17 00:00:00 2001 From: spjuhel Date: Tue, 6 Jan 2026 15:17:56 +0100 Subject: [PATCH 10/10] missing test --- .../trajectories/test/test_interpolation.py | 46 +++++++++++++------ 1 file changed, 32 insertions(+), 14 deletions(-) diff --git a/climada/trajectories/test/test_interpolation.py b/climada/trajectories/test/test_interpolation.py index f601fce8ff..da6f614fe1 100644 --- a/climada/trajectories/test/test_interpolation.py +++ b/climada/trajectories/test/test_interpolation.py @@ -107,26 +107,44 @@ def test_convex_combinations_start_equals_end(interpolation_data, func): np.testing.assert_allclose(result, arr, rtol=interpolation_data["rtol"]) -def test_linear_impmat_interpolate(interpolation_data): +@pytest.mark.parametrize( + "func,expected", + [ + ( + linear_interp_matrix_elemwise, + np.array( + [ + [[1.0, 2.0], [3.0, 4.0]], + [[2.0, 3.0], [4.0, 5.0]], + [[3.0, 4.0], [5.0, 6.0]], + [[4.0, 5.0], [6.0, 7.0]], + [[5.0, 6.0], [7.0, 8.0]], + ] + ), + ), + ( + exponential_interp_matrix_elemwise, + np.array( + [ + [[1.0, 2.0], [3.0, 4.0]], + [[1.49534878, 2.63214803], [3.70779275, 4.75682846]], + [[2.23606798, 3.46410162], [4.58257569, 5.65685425]], + [[3.34370152, 4.55901411], [5.66374698, 6.72717132]], + [[5.0, 6.0], [7.0, 8.0]], + ] + ), + ), + ], +) +def test_impmat_interpolate(interpolation_data, func, expected): data = interpolation_data - result = linear_interp_matrix_elemwise( - data["imp_mat0"], data["imp_mat1"], data["time_points"] - ) + result = func(data["imp_mat0"], data["imp_mat1"], data["time_points"]) assert len(result) == data["time_points"] assert all(isinstance(mat, csr_matrix) for mat in result) dense = np.array([r.todense() for r in result]) - expected = np.array( - [ - [[1.0, 2.0], [3.0, 4.0]], - [[2.0, 3.0], [4.0, 5.0]], - [[3.0, 4.0], [5.0, 6.0]], - [[4.0, 5.0], [6.0, 7.0]], - [[5.0, 6.0], [7.0, 8.0]], - ] - ) - np.testing.assert_array_equal(dense, expected) + np.testing.assert_array_almost_equal(dense, expected) # --- Tests for Interpolation Strategies ---