Demonstrate hazard set selection; logging pytest. (#333)

* Demonstrate hazard set selection; logging pytest.

Signed-off-by: Joe Moorhouse <[email protected]>

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Signed-off-by: Joe Moorhouse <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

pyproject.toml

@@ -97,6 +97,10 @@ complete = { call = "tasks.complete:main", help = "Create autocomplete files for
 [tool.pytest.ini_options]
 testpaths = "tests"
 addopts = "-v"
+log_cli = true
+log_cli_level = "INFO"
+log_cli_format = "%(asctime)s [%(levelname)s] %(message)s (%(filename)s:%(lineno)s)"
+log_cli_date_format = "%Y-%m-%d %H:%M:%S"
 #addopts = "--cov --cov-report html --cov-report term-missing --cov-fail-under 95"
 
 [tool.mypy]

src/physrisk/hazard_models/core_hazards.py

@@ -1,3 +1,4 @@
+from enum import Enum
 from typing import Dict, Iterable, NamedTuple, Optional, Protocol
 
 from physrisk.api.v1.hazard_data import HazardResource
@@ -163,8 +164,15 @@ def _no_selector(
         return candidates.first()
 
 
+class CoreFloodModels(Enum):
+    WRI = 1
+    TUDelft = 2
+
+
 class CoreInventorySourcePaths(InventorySourcePaths):
-    def __init__(self, inventory: Inventory):
+    def __init__(
+        self, inventory: Inventory, flood_model: CoreFloodModels = CoreFloodModels.WRI
+    ):
         super().__init__(inventory)
         for indicator_id in [
             "mean_work_loss/low",
@@ -176,7 +184,11 @@ def __init__(self, inventory: Inventory):
             ChronicHeat, "mean/degree/days/above/32c", self._select_chronic_heat
         )
         self.add_selector(
-            RiverineInundation, "flood_depth", self._select_riverine_inundation
+            RiverineInundation,
+            "flood_depth",
+            self._select_riverine_inundation
+            if flood_model == CoreFloodModels.WRI
+            else self._select_riverine_inundation_tudelft,
         )
         self.add_selector(
             CoastalInundation, "flood_depth", self._select_coastal_inundation
@@ -223,6 +235,15 @@ def _select_riverine_inundation(
             else candidates.with_model_gcm("MIROC-ESM-CHEM").first()
         )
 
+    @staticmethod
+    def _select_riverine_inundation_tudelft(
+        candidates: ResourceSubset,
+        scenario: str,
+        year: int,
+        hint: Optional[HazardDataHint] = None,
+    ):
+        return candidates.with_model_id("tudelft").first()
+
     @staticmethod
     def _select_wind(
         candidates: ResourceSubset,

tests/models/hazard_selection_test.py

@@ -0,0 +1,95 @@
+from typing import Optional
+import numpy as np
+
+from physrisk.data.hazard_data_provider import HazardDataHint
+from physrisk.data.inventory import EmbeddedInventory, Inventory
+from physrisk.data.pregenerated_hazard_model import ZarrHazardModel
+from physrisk.hazard_models.core_hazards import (
+    CoreFloodModels,
+    CoreInventorySourcePaths,
+    ResourceSubset,
+)
+from physrisk.kernel.hazard_model import HazardDataRequest, HazardEventDataResponse
+from physrisk.kernel.hazards import RiverineInundation
+
+from tests.data.hazard_model_store_test import (
+    TestData,
+    ZarrStoreMocker,
+    inundation_return_periods,
+)
+
+
+def test_tudelft_selection():
+    inventory: Inventory = EmbeddedInventory()
+    source_paths = CoreInventorySourcePaths(
+        inventory, flood_model=CoreFloodModels.TUDelft
+    ).source_paths()
+    assert (
+        source_paths[RiverineInundation](
+            indicator_id="flood_depth", scenario="rcp8p5", year=2050
+        )
+        == "inundation/river_tudelft/v2/flood_depth_unprot_rcp8p5_2050"
+    )
+    assert (
+        source_paths[RiverineInundation](
+            indicator_id="flood_depth", scenario="historical", year=-1
+        )
+        == "inundation/river_tudelft/v2/flood_depth_unprot_historical_1985"
+    )
+
+
+def test_customize_hazard_selection():
+    inventory: Inventory = EmbeddedInventory()
+
+    source_path_selectors = CoreInventorySourcePaths(inventory)
+
+    def select_riverine_inundation_tudelft(
+        candidates: ResourceSubset,
+        scenario: str,
+        year: int,
+        hint: Optional[HazardDataHint] = None,
+    ):
+        return candidates.with_model_id("tudelft").first()
+
+    # we can add selectors programmatically
+    # test_tudelft_selection shows an example of using the options in CoreInventorySourcePaths
+    source_path_selectors.add_selector(
+        RiverineInundation, "flood_depth", select_riverine_inundation_tudelft
+    )
+
+    custom_source_paths = source_path_selectors.source_paths()
+
+    def sp_riverine(scenario, year):
+        return custom_source_paths[RiverineInundation](
+            indicator_id="flood_depth", scenario=scenario, year=year
+        )
+
+    mocker = ZarrStoreMocker()
+    return_periods = inundation_return_periods()
+    flood_histo_curve = np.array(
+        [0.0596, 0.333, 0.505, 0.715, 0.864, 1.003, 1.149, 1.163, 1.163]
+    )
+
+    for path in [sp_riverine("historical", 1980)]:
+        mocker.add_curves_global(
+            path,
+            TestData.longitudes,
+            TestData.latitudes,
+            return_periods,
+            flood_histo_curve,
+        )
+
+    hazard_model = ZarrHazardModel(source_paths=custom_source_paths, store=mocker.store)
+
+    req = HazardDataRequest(
+        RiverineInundation,
+        TestData.longitudes[0],
+        TestData.latitudes[0],
+        indicator_id="flood_depth",
+        scenario="historical",
+        year=-1,
+    )
+    responses = hazard_model.get_hazard_events([req])
+    resp = responses[req]
+    assert isinstance(resp, HazardEventDataResponse)
+    np.testing.assert_almost_equal(resp.intensities, flood_histo_curve)