Add unit and integration tests, update code base

Author: peanutfun

climada/util/calibrate/impact_func.py

@@ -20,8 +20,12 @@
 from climada.engine import Impact, ImpactCalc
 
 
-def cost_func_rmse(impact: Impact, data: pd.DataFrame) -> Number:
-    return np.sqrt(((impact - data) ** 2).mean(axis=None))
+def cost_func_rmse(
+    impact: Impact,
+    data: pd.DataFrame,
+    impact_proc: Callable[[Impact], pd.DataFrame] = lambda x: x.impact_at_reg(),
+) -> Number:
+    return np.sqrt(np.mean(((impact_proc(impact) - data) ** 2).to_numpy()))
 
 
 def impf_step_generator(threshold: Number, paa: Number) -> ImpactFuncSet:
@@ -34,6 +38,9 @@ def impf_step_generator(threshold: Number, paa: Number) -> ImpactFuncSet:
     )
 
 
+ConstraintType = Union[LinearConstraint, NonlinearConstraint, Mapping]
+
+
 @dataclass
 class Input:
     """Define the static input for a calibration task"""
@@ -44,17 +51,23 @@ class Input:
     cost_func: Callable[[Impact, pd.DataFrame], Number]
     impact_func_gen: Callable[..., ImpactFuncSet]
     bounds: Optional[Mapping[str, Union[Bounds, Tuple[Number, Number]]]] = None
-    constraints: Optional[
-        Mapping[str, Union[LinearConstraint, NonlinearConstraint, Mapping]]
-    ] = None
+    constraints: Optional[Union[ConstraintType, list[ConstraintType]]] = None
     impact_calc_kwds: Mapping[str, Any] = field(
-        default_factory=lambda: dict(assign_centroids=False)
+        default_factory=lambda: {"assign_centroids": False}
     )
+    align: InitVar[bool] = True
 
-    def __post_init__(self):
+    def __post_init__(self, align):
         """Prepare input data"""
-        self.hazard = self.hazard.select(event_id=self.data.index.tolist())
-        self.exposure.assign_centroids(self.hazard)
+        if align:
+            event_diff = np.setdiff1d(self.data.index, self.hazard.event_id)
+            if event_diff.size > 0:
+                raise RuntimeError(
+                    "Event IDs in 'data' do not match event IDs in 'hazard': \n"
+                    f"{event_diff}"
+                )
+            self.hazard = self.hazard.select(event_id=self.data.index.tolist())
+            self.exposure.assign_centroids(self.hazard)
 
 
 @dataclass
@@ -76,7 +89,7 @@ class Optimizer(ABC):
     def _target_func(self, impact: Impact, data: pd.DataFrame):
         return self.input.cost_func(impact, data)
 
-    def _kwargs_to_impact_func_gen(self, *args, **kwargs) -> Dict[str, Any]:
+    def _kwargs_to_impact_func_gen(self, *_, **kwargs) -> Dict[str, Any]:
         """Define how the parameters to 'opt_func' must be transformed"""
         return kwargs
 
@@ -88,7 +101,7 @@ def _opt_func(self, *args, **kwargs):
             exposures=self.input.exposure,
             impfset=impf_set,
             hazard=self.input.hazard,
-        ).impact(assign_centroids=False, **self.input.impact_calc_kwds)
+        ).impact(**self.input.impact_calc_kwds)
         return self._target_func(impact, self.input.data)
 
     @abstractmethod
@@ -105,35 +118,36 @@ def __post_init__(self):
         """Create a private attribute for storing the parameter names"""
         self._param_names: List[str] = list()
 
-    def _kwargs_to_impact_func_gen(self, *args, **kwargs) -> Dict[str, Any]:
+    def _kwargs_to_impact_func_gen(self, *args, **_) -> Dict[str, Any]:
         return dict(zip(self._param_names, args[0].flat))
 
     def _select_by_param_names(self, mapping: Mapping[str, Any]) -> List[Any]:
         """Return a list of entries from a map with matching keys or ``None``"""
         return [mapping.get(key) for key in self._param_names]
 
-    def run(self, params_init: Mapping[str, Number], **opt_kwargs):
+    def run(self, **opt_kwargs):
         """Execute the optimization"""
+        # Parse kwargs
+        params_init = opt_kwargs.pop("params_init")
+        method = opt_kwargs.pop("method", "trust-constr")
+
+        # Store names to rebuild dict when the minimize iterator returns an array
         self._param_names = list(params_init.keys())
 
-        # Transform data to match minimize input
+        # Transform bounds to match minimize input
         bounds = (
             self._select_by_param_names(self.input.bounds)
             if self.input.bounds is not None
             else None
         )
-        constraints = (
-            self._select_by_param_names(self.input.constraints)
-            if self.input.constraints is not None
-            else None
-        )
 
         x0 = np.array(list(params_init.values()))
         res = minimize(
-            fun=lambda x: self._opt_func(x),
+            fun=self._opt_func,
             x0=x0,
             bounds=bounds,
-            constraints=constraints,
+            constraints=self.input.constraints,
+            method=method,
             **opt_kwargs,
         )
 

climada/util/calibrate/test/test_calibrate.py

@@ -1,18 +1,19 @@
 """Tests for calibration module"""
 
 import unittest
-from unittest.mock import create_autospec
+from unittest.mock import create_autospec, patch, MagicMock
+from typing import Optional, List
 
 import numpy as np
 import numpy.testing as npt
 import pandas as pd
 from scipy.sparse import csr_matrix
-from shapely.geometry import Point
+from scipy.optimize import OptimizeResult
 
 from climada.entity import Exposures, ImpactFuncSet
 from climada.hazard import Hazard, Centroids
 
-from ..impact_func import Input, ScipyMinimizeOptimizer
+from climada.util.calibrate.impact_func import Input, ScipyMinimizeOptimizer
 
 
 def hazard():
@@ -21,13 +22,20 @@ def hazard():
     lon = [0, 1]
     centroids = Centroids.from_lat_lon(lat=lat, lon=lon)
     event_id = np.array([1, 3, 10])
-    intensity = csr_matrix([[1, 1], [2, 2], [3, 3]])
+    intensity = csr_matrix([[1, 0.1], [2, 0.2], [3, 2]])
     return Hazard(event_id=event_id, centroids=centroids, intensity=intensity)
 
 
 def exposure():
     """Create a dummy exposure instance"""
-    return Exposures(data=dict(longitude=[0, 1, 100], latitude=[1, 2, 50]))
+    return Exposures(
+        data=dict(
+            longitude=[0, 1, 100],
+            latitude=[1, 2, 50],
+            value=[1, 0.1, 1e6],
+            impf_=[1, 1, 1],
+        )
+    )
 
 
 class TestInputPostInit(unittest.TestCase):
@@ -49,12 +57,16 @@ def setUp(self):
         self.cost_func = lambda impact, data: 1.0
         self.impact_func_gen = lambda **kwargs: ImpactFuncSet()
 
-    def test_post_init_calls(self):
+    @patch("climada.util.calibrate.impact_func.np.setdiff1d")
+    def test_post_init_calls(self, setdiff1d_mock):
         """Test if post_init calls stuff correctly using mocks"""
         # Create mocks
-        hazard_mock_1 = create_autospec(Hazard, instance=True)
-        hazard_mock_2 = create_autospec(Hazard, instance=True)
-        exposure_mock = create_autospec(Exposures, instance=True)
+        hazard_mock_1 = create_autospec(Hazard())
+        event_id = [10]
+        hazard_mock_1.event_id = event_id
+        hazard_mock_2 = create_autospec(Hazard())
+        exposure_mock = create_autospec(Exposures())
+        setdiff1d_mock.return_value = np.array([])
 
         # Make first hazard mock return another instance
         hazard_mock_1.select.return_value = hazard_mock_2
@@ -69,6 +81,8 @@ def test_post_init_calls(self):
         )
 
         # Query checks
+        npt.assert_array_equal(setdiff1d_mock.call_args.args[0], self.data_events)
+        npt.assert_array_equal(setdiff1d_mock.call_args.args[1], event_id)
         hazard_mock_1.select.assert_called_once_with(event_id=self.data_events)
         self.assertNotEqual(input.hazard, hazard_mock_1)
         self.assertEqual(input.hazard, hazard_mock_2)
@@ -90,49 +104,137 @@ def test_post_init(self):
         self.assertIn("centr_", input.exposure.gdf)
         npt.assert_array_equal(input.exposure.gdf["centr_"], [0, 1, -1])
 
+    def test_non_matching_events(self):
+        """Test if non-matching events result in errors"""
+        data = pd.DataFrame(data={"a": [1, 2, 3]}, index=[9, 3, 12])
+        input_kwargs = {
+            "hazard": self.hazard,
+            "exposure": self.exposure,
+            "data": data,
+            "cost_func": self.cost_func,
+            "impact_func_gen": self.impact_func_gen,
+            "align": False,
+        }
+
+        # No error without alignment
+        Input(**input_kwargs)
+
+        # Error with alignment
+        input_kwargs.update(align=True)
+        with self.assertRaises(RuntimeError) as cm:
+            Input(**input_kwargs)
+
+        self.assertIn(
+            "Event IDs in 'data' do not match event IDs in 'hazard'", str(cm.exception)
+        )
+        self.assertIn("9", str(cm.exception))
+        self.assertIn("12", str(cm.exception))
+        self.assertNotIn("3", str(cm.exception))
+
 
 class TestScipyMinimizeOptimizer(unittest.TestCase):
     """Tests for the optimizer based on scipy.optimize.minimize"""
 
     def setUp(self):
         """Mock the input and create the optimizer"""
-        self.input = create_autospec(Input, instance=True)
-        self.optimizer = ScipyMinimizeOptimizer(self.input)
-
-    def test_kwargs_to_impact_func_gen(self):
-        """Test the _kwargs_to_impact_func_gen method"""
-        # _param_names is empty in the beginning
-        x = np.array([1, 2, 3])
-        self.assertDictEqual(self.optimizer._kwargs_to_impact_func_gen(x), {})
-
-        # Now populate it and try again
-        self.optimizer._param_names = ["x_2", "x_1", "x_3"]
-        result = {"x_2": 1, "x_1": 2, "x_3": 3}
-        self.assertDictEqual(self.optimizer._kwargs_to_impact_func_gen(x), result)
-
-        # Other arguments are ignored
-        self.assertDictEqual(
-            self.optimizer._kwargs_to_impact_func_gen(x, x + 3), result
+        self.input = MagicMock(
+            spec_set=Input(
+                hazard=create_autospec(Hazard, instance=True),
+                exposure=create_autospec(Exposures, instance=True),
+                data=create_autospec(pd.DataFrame, instance=True),
+                cost_func=MagicMock(),
+                impact_func_gen=MagicMock(),
+                align=False,
+            )
         )
+        self.optimizer = ScipyMinimizeOptimizer(self.input)
 
-        # Array is flattened, iterator stops
-        self.assertDictEqual(
-            self.optimizer._kwargs_to_impact_func_gen(np.array([[1, 2], [3, 4]])),
-            result,
+    @patch("climada.util.calibrate.impact_func.ImpactCalc", autospec=True)
+    def test_kwargs_to_impact_func_gen(self, _):
+        """Test transform of minimize func arguments to impact_func_gen arguments
+
+        We test the method '_kwargs_to_impact_func_gen' through 'run' because it is
+        private.
+        """
+        # Create stubs
+        self.input.constraints = None
+        self.input.bounds = None
+        self.input.cost_func.return_value = 1.0
+
+        # Call 'run', make sure that 'minimize' is only with these parameters
+        params_init = {"x_2": 1, "x 1": 2, "x_3": 3}  # NOTE: Also works with whitespace
+        self.optimizer.run(params_init=params_init, options={"maxiter": 1})
+
+        # Check call to '_kwargs_to_impact_func_gen'
+        self.input.impact_func_gen.assert_any_call(**params_init)
+
+    def test_output(self):
+        """Check output reporting"""
+        params_init = {"x_2": 1, "x 1": 2, "x_3": 3}
+        target_func_value = 1.12
+        self.input.constraints = None
+        self.input.bounds = None
+
+        # Mock the optimization function and call 'run'
+        with patch.object(self.optimizer, "_opt_func") as opt_func_mock:
+            opt_func_mock.return_value = target_func_value
+            output = self.optimizer.run(params_init=params_init, options={"maxiter": 1})
+
+        # Assert output
+        self.assertListEqual(list(output.params.keys()), list(params_init.keys()))
+        npt.assert_allclose(list(output.params.values()), list(params_init.values()))
+        self.assertEqual(output.target, target_func_value)
+        self.assertTrue(output.success)  # NOTE: For scipy.optimize, this means no error
+        self.assertIsInstance(output.result, OptimizeResult)
+
+    @patch("climada.util.calibrate.impact_func.minimize", autospec=True)
+    def test_bounds_select(self, minimize_mock):
+        """Test the _select_by_param_names method
+
+        We test the method '_select_by_param_names' through 'run' because it is private.
+        """
+
+        def assert_bounds_in_call(bounds: Optional[List]):
+            """Check if scipy.optimize.minimize was called with the expected kwargs"""
+            call_kwargs = minimize_mock.call_args.kwargs
+            print(minimize_mock.call_args)
+
+            if bounds is None:
+                self.assertIsNone(call_kwargs["bounds"])
+            else:
+                self.assertListEqual(call_kwargs["bounds"], bounds)
+
+        # Initialize params and mock return value
+        params_init = {"x_2": 1, "x_1": 2, "x_3": 3}
+        minimize_mock.return_value = OptimizeResult(
+            x=np.array(list(params_init.values())), fun=0, success=True
         )
 
-    def test_select_by_keys(self):
-        """Test the _select_by_keys method"""
-        param_names = ["a", "b", "c", "d"]
-        mapping = dict(zip(param_names, [1, "2", (1, 2)]))
-
-        # _param_names is empty in the beginning
-        self.assertListEqual(self.optimizer._select_by_param_names(mapping), [])
-
-        # Set _param_names
-        self.optimizer._param_names = param_names
-
-        # Check result
-        self.assertListEqual(
-            self.optimizer._select_by_param_names(mapping), [1, "2", (1, 2), None]
-        )
+        # Set constraints and bounds to None (default)
+        self.input.bounds = None
+
+        # Call the optimizer (constraints and bounds are None)
+        self.optimizer.run(params_init=params_init)
+        self.assertListEqual(self.optimizer._param_names, list(params_init.keys()))
+        minimize_mock.assert_called_once()
+        assert_bounds_in_call(None)
+        minimize_mock.reset_mock()
+
+        # Set new bounds and constraints
+        self.input.bounds = {"x_1": "a", "x_4": "b", "x_3": (1, 2)}
+        self.input.constraints = {"x_5": [1], "x_2": 2}
+
+        # Call the optimizer
+        self.optimizer.run(params_init=params_init)
+        self.assertListEqual(self.optimizer._param_names, list(params_init.keys()))
+        minimize_mock.assert_called_once()
+        assert_bounds_in_call(bounds=[None, "a", (1, 2)])
+
+
+# Execute Tests
+if __name__ == "__main__":
+    TESTS = unittest.TestLoader().loadTestsFromTestCase(TestInputPostInit)
+    TESTS.addTests(
+        unittest.TestLoader().loadTestsFromTestCase(TestScipyMinimizeOptimizer)
+    )
+    unittest.TextTestRunner(verbosity=2).run(TESTS)