test: regression tests for the pandas-2 fixes in #1325

CHANGELOG.md

@@ -31,6 +31,7 @@ Code freeze date: YYYY-MM-DD
 - `Hazard.from_raster_xarray` now returns a sparse matrix instead of a sparse array [#1261](https://github.com/CLIMADA-project/climada_python/pull/1261).
 - `ImpactCalc.impact` now raises a clear `ValueError` when the supplied `Hazard` contains no events, instead of failing later inside `np.array_split` with an obscure message [#814](https://github.com/CLIMADA-project/climada_python/issues/814).
 - Fix TCTracks.from_FAST duplicate loading from year loop [#1269](github.com/CLIMADA-project/climada_python/pull/1269)
+- Replaced the calls to `DataFrame.append` and `Series.iteritems`, removed in pandas 2.0, that made `LitPop.from_shape_and_countries` (with a `GeoSeries` or `list` shape), `impact_data.hit_country_per_hazard`, `calibration_opt.calib_all` and `calibration_opt.calib_instance` (with a multi-row `df_out` and `yearly_impact=True`) raise `AttributeError`; `LitPop.from_shape_and_countries` no longer duplicates points that lie in overlapping shapes. [#1325](https://github.com/CLIMADA-project/climada_python/pull/1325), [#1320](https://github.com/CLIMADA-project/climada_python/pull/1320)
 
 ### Deprecated
 - `Impact.calc_freq_curve()` should not be given the parameter `return_per`. Use the parameter `return_periods` in `Impact.calc_freq_curve().interpolate()` instead.

climada/engine/test/test_impact_data.py

@@ -19,10 +19,14 @@
 Test Impact class.
 """
 
+import pickle
+import tempfile
 import unittest
 import warnings
+from pathlib import Path
 
 import numpy as np
+from scipy import sparse
 
 import climada.engine.impact_data as im_d
 from climada import CONFIG
@@ -300,9 +304,45 @@ def test_emdat_to_impact_2020format(self):
         self.assertAlmostEqual(1.610687e07, impact_emdat.at_event[4], places=0)
 
 
+class TestHitCountryPerHazard(unittest.TestCase):
+    """Test hit_country_per_hazard, which reads its inputs from pickle files."""
+
+    def test_hit_country_per_hazard_pass(self):
+        """One row per (track, hit country) pair."""
+        # two tracks over three centroids in USA (840) and CAN (124):
+        # track 0 hits centroids 0 and 2, track 1 hits centroid 1
+        inputs = {
+            "intensity": sparse.csr_matrix(
+                np.array([[10.0, 0.0, 20.0], [0.0, 30.0, 0.0]])
+            ),
+            "names": ["2017001N10W", "2017002N10W"],
+            "reg_id": np.array([840, 840, 124]),
+            "date": [736330, 736331],
+        }
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            paths = {}
+            for name, obj in inputs.items():
+                paths[name] = Path(tmp_dir, f"{name}.p")
+                with open(paths[name], "wb") as filef:
+                    pickle.dump(obj, filef)
+            hit_countries = im_d.hit_country_per_hazard(
+                paths["intensity"], paths["names"], paths["reg_id"], paths["date"]
+            )
+
+        self.assertListEqual(
+            list(hit_countries.columns), ["hit_country", "Date_start", "ibtracsID"]
+        )
+        self.assertEqual(hit_countries.shape[0], 3)
+        self.assertSetEqual(set(hit_countries["hit_country"]), {"USA", "CAN"})
+        self.assertListEqual(
+            list(hit_countries["Date_start"]), [736330, 736330, 736331]
+        )
+
+
 # Execute Tests
 if __name__ == "__main__":
     TESTS = unittest.TestLoader().loadTestsFromTestCase(TestEmdatImport)
     TESTS.addTests(unittest.TestLoader().loadTestsFromTestCase(TestEmdatProcessing))
     TESTS.addTests(unittest.TestLoader().loadTestsFromTestCase(TestEmdatToImpact))
+    TESTS.addTests(unittest.TestLoader().loadTestsFromTestCase(TestHitCountryPerHazard))
     unittest.TextTestRunner(verbosity=2).run(TESTS)

climada/entity/exposures/test/test_litpop.py

@@ -20,12 +20,16 @@
 """
 
 import unittest
+from unittest.mock import patch
 
+import geopandas
 import numpy as np
 import rasterio
+import shapely
 from rasterio import Affine
 from rasterio.crs import CRS
 
+from climada.entity.exposures.base import Exposures
 from climada.entity.exposures.litpop import litpop as lp
 
 
@@ -427,6 +431,50 @@ def test_get_value_unit_pass(self):
         self.assertEqual(lp.get_value_unit("none"), "")
 
 
+class TestFromShapeAndCountries(unittest.TestCase):
+    """Test from_shape_and_countries with a GeoSeries or list `shape`.
+
+    `from_countries` is patched out because the real path needs the GPW
+    population raster, which requires a manual NASA Earthdata download.
+    """
+
+    @staticmethod
+    def _exposures():
+        return Exposures(
+            geopandas.GeoDataFrame(
+                {"value": [10.0, 20.0, 30.0]},
+                geometry=geopandas.points_from_xy([0.5, 1.5, 9.5], [0.5, 0.5, 0.5]),
+                crs="EPSG:4326",
+            )
+        )
+
+    def test_geoseries_shape(self):
+        """A GeoSeries of polygons selects the points inside them."""
+        shapes = geopandas.GeoSeries(
+            [shapely.geometry.box(0, 0, 1, 1), shapely.geometry.box(1, 0, 2, 1)],
+            crs="EPSG:4326",
+        )
+
+        with patch.object(lp.LitPop, "from_countries", return_value=self._exposures()):
+            exp = lp.LitPop.from_shape_and_countries(shapes, "CHE", res_arcsec=30)
+
+        # the point at x=9.5 lies outside both boxes
+        self.assertCountEqual(exp.gdf["value"].tolist(), [10.0, 20.0])
+
+    def test_overlapping_shapes(self):
+        """A point inside several overlapping shapes is selected only once."""
+        shapes = [shapely.geometry.box(0, 0, 2, 1), shapely.geometry.box(0, 0, 1, 1)]
+
+        with patch.object(lp.LitPop, "from_countries", return_value=self._exposures()):
+            exp = lp.LitPop.from_shape_and_countries(shapes, "CHE", res_arcsec=30)
+
+        # the point at x=0.5 lies inside both boxes
+        self.assertCountEqual(exp.gdf["value"].tolist(), [10.0, 20.0])
+
+
 if __name__ == "__main__":
     TESTS = unittest.TestLoader().loadTestsFromTestCase(TestLitPop)
+    TESTS.addTests(
+        unittest.TestLoader().loadTestsFromTestCase(TestFromShapeAndCountries)
+    )
     unittest.TextTestRunner(verbosity=2).run(TESTS)

climada/test/test_calibration.py

@@ -27,7 +27,8 @@
 import climada.hazard.test as hazard_test
 from climada import CONFIG
 from climada.engine import ImpactCalc
-from climada.engine.calibration_opt import calib_instance
+from climada.engine.calibration_opt import calib_all, calib_instance
+from climada.entity import ImpactFuncSet
 from climada.entity.entity_def import Entity
 from climada.hazard.base import Hazard
 from climada.test import get_test_file
@@ -41,6 +42,17 @@
 class TestCalib(unittest.TestCase):
     """Test engine calibration method."""
 
+    @classmethod
+    def setUpClass(cls):
+        cls.hazard = Hazard.from_hdf5(HAZ_TEST_TC)
+        entity = Entity.from_excel(ENT_DEMO_TODAY)
+        entity.check()
+        cls.exposures = entity.exposures
+        cls.exposures.assign_centroids(cls.hazard)
+        cls.impf = entity.impact_funcs.get_func(
+            cls.hazard.haz_type, entity.exposures.gdf["impf_TC"].median()
+        )
+
     def test_calib_instance(self):
         """Test save calib instance"""
         # Read default entity values
@@ -93,6 +105,47 @@ def test_calib_instance(self):
         self.assertTrue(all(df_out["impact_CLIMADA"].values == impact.at_event))
         self.assertTrue(all(df_out_yearly["impact_CLIMADA"].values == [*IYS.values()]))
 
+    def test_calib_instance_yearly_multirow(self):
+        """Test calib_instance with yearly_impact and a multi-row df_out"""
+        impact = ImpactCalc(
+            self.exposures, ImpactFuncSet([self.impf]), self.hazard
+        ).impact(assign_centroids=False)
+        iys = impact.impact_per_year(all_years=True)
+        sel_years = sorted(iys.keys())[:3]
+
+        df_out = calib_instance(
+            self.hazard,
+            self.exposures,
+            self.impf,
+            pd.DataFrame({"year": sel_years}),
+            yearly_impact=True,
+        )
+
+        self.assertEqual(df_out.shape[0], len(sel_years))
+        for year in sel_years:
+            self.assertAlmostEqual(
+                df_out.loc[df_out["year"] == year, "impact_CLIMADA"].iloc[0], iys[year]
+            )
+
+    def test_calib_all(self):
+        """Test calib_all"""
+        df_result = calib_all(
+            hazard=self.hazard,
+            exposure=self.exposures,
+            impf_name_or_instance="emanuel",
+            param_full_dict={
+                "v_thresh": [25.7, 20.0],
+                "v_half": [70.0],
+                "scale": [1.0],
+            },
+            impact_data_source=pd.DataFrame({"impact": [1.0e9], "region_id": [840]}),
+            year_range=[2004, 2005],
+            yearly_impact=True,
+        )
+
+        self.assertSetEqual(set(df_result["v_thresh"]), {25.7, 20.0})
+        self.assertIn("impact_CLIMADA", df_result.columns)
+
 
 # Execute Tests
 if __name__ == "__main__":