diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b449f222..2b4c3fe28 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/climada/engine/test/test_impact_data.py b/climada/engine/test/test_impact_data.py index 05ad8ea41..6c3cd50e7 100644 --- a/climada/engine/test/test_impact_data.py +++ b/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) diff --git a/climada/entity/exposures/test/test_litpop.py b/climada/entity/exposures/test/test_litpop.py index 048819e5f..a1b8e4ee9 100644 --- a/climada/entity/exposures/test/test_litpop.py +++ b/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) diff --git a/climada/test/test_calibration.py b/climada/test/test_calibration.py index 5b83b3a3f..046812806 100644 --- a/climada/test/test_calibration.py +++ b/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__":