Add vulnerability functions.

We add vulnerability functions with the aim of replicating the paper:
'Climate change-related statistical indicators'
Statistics Paper Series No 48 from the European Central Bank.
https://www.ecb.europa.eu/pub/pdf/scpsps/ecb.sps48~e3fd21dd5a.en.pdf?4826f6ee77d3ac7a681916b6d419b751

Co-authored-by: Víctor de Luna <vdeluna@arfima.com>
Co-authored-by: Arfima Dev <dev@arfima.com>

Signed-off-by: Virginia Morales <vmorales@arfima.com>

Author: VMarfima

pyproject.toml

@@ -181,5 +181,5 @@ quote-style = "double"
 indent-style = "space"
 skip-magic-trailing-comma = false
 line-ending = "auto"
-#docstring-code-format = false
+docstring-code-format = true
 docstring-code-line-length = "dynamic"

src/physrisk/api/v1/common.py

@@ -1,7 +1,8 @@
+from enum import Enum
 from typing import Dict, List, Optional, Union
 
 import numpy as np
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class TypedArray(np.ndarray):
@@ -23,13 +24,25 @@ class Array(np.ndarray, metaclass=ArrayMeta):
     pass
 
 
-class Asset(BaseModel, extra="allow"):
-    """Defines an asset. An asset is identified first by its asset_class and then by its type within the class.
+class UseCaseId(Enum):
+    """Enumeration to select the 'use case' identifier used in vulnerability models and risk measures calculations."""
+
+    DEFAULT = 0
+    STRESS_TEST = 1
+    GENERIC = 2
+
+
+class Asset(BaseModel):
+    """Defines an asset.
+
+    An asset is identified first by its asset_class and then by its type within the class.
     An asset's value may be impacted through damage or through disruption
     disruption being reduction of an asset's ability to generate cashflows
     (or equivalent value, e.g. by reducing expenses or increasing sales).
     """
 
+    model_config = ConfigDict(extra="allow")
+
     asset_class: str = Field(
         description="name of asset class; corresponds to physrisk class names, e.g. PowerGeneratingAsset"
     )
@@ -104,30 +117,27 @@ class IntensityCurve(BaseModel):
 class ExceedanceCurve(BaseModel):
     """General exceedance curve (e.g. hazazrd, impact)."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
     values: np.ndarray = Field(default_factory=lambda: np.zeros(10), description="")
     exceed_probabilities: np.ndarray = Field(
         default_factory=lambda: np.zeros(10), description=""
     )
 
-    class Config:
-        arbitrary_types_allowed = True
-
 
 class Distribution(BaseModel):
     """General exceedance curve (e.g. hazazrd, impact)."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
     bin_edges: np.ndarray = Field(default_factory=lambda: np.zeros(11), description="")
     probabilities: np.ndarray = Field(
         default_factory=lambda: np.zeros(10), description=""
     )
 
-    class Config:
-        arbitrary_types_allowed = True
-
 
 class HazardEventDistrib(BaseModel):
     """Intensity curve of an acute hazard."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
     intensity_bin_edges: np.ndarray = Field(
         default_factory=lambda: np.zeros(10), description=""
     )
@@ -136,13 +146,11 @@ class HazardEventDistrib(BaseModel):
     )
     path: List[str] = Field([], description="Path to the hazard indicator data source.")
 
-    class Config:
-        arbitrary_types_allowed = True
-
 
 class VulnerabilityCurve(BaseModel):
     """Defines a damage or disruption curve."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
     asset_type: str = Field(...)
     location: str = Field(...)
     event_type: str = Field(description="hazard event type, e.g. RiverineInundation")
@@ -156,9 +164,6 @@ class VulnerabilityCurve(BaseModel):
         description="standard deviation of impact (damage or disruption)"
     )
 
-    class Config:
-        arbitrary_types_allowed = True
-
 
 class VulnerabilityCurves(BaseModel):
     """List of VulnerabilityCurve."""
@@ -169,6 +174,7 @@ class VulnerabilityCurves(BaseModel):
 class VulnerabilityDistrib(BaseModel):
     """Defines a vulnerability matrix."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
     intensity_bin_edges: np.ndarray = Field(
         default_factory=lambda: np.zeros(10), description=""
     )
@@ -178,6 +184,3 @@ class VulnerabilityDistrib(BaseModel):
     prob_matrix: np.ndarray = Field(
         default_factory=lambda: np.zeros(10), description=""
     )
-
-    class Config:
-        arbitrary_types_allowed = True

src/physrisk/api/v1/impact_req_resp.py

@@ -1,7 +1,7 @@
 from enum import Enum
 from typing import Dict, List, NamedTuple, Optional, Sequence
 
-from pydantic import BaseModel, Field, computed_field
+from pydantic import BaseModel, ConfigDict, Field, computed_field
 
 from physrisk.api.v1.common import (
     Assets,
@@ -10,6 +10,7 @@
     VulnerabilityDistrib,
 )
 from physrisk.api.v1.hazard_data import Scenario
+from physrisk.api.v1.common import UseCaseId
 
 
 class CalcSettings(BaseModel):
@@ -36,9 +37,13 @@ class AssetImpactRequest(BaseModel):
     include_calc_details: bool = Field(
         True, description="If true, include impact calculation details."
     )
-    use_case_id: str = Field(
-        "",
-        description="Identifier for 'use case' used in the risk measures calculation.",
+    use_case_id: UseCaseId = Field(
+        UseCaseId.DEFAULT,
+        description=(
+            "Enumeration to select the 'use case' identifier"
+            "used in vulnerability models and risk measures calculations."
+            "Options: DEFAULT, STRESSTEST, GENERIC"
+        ),
     )
     provider_max_requests: Dict[str, int] = Field(
         {},
@@ -70,6 +75,8 @@ class Category(int, Enum):
     HIGH = 3
     REDFLAG = 4
 
+    NORISK = -1
+
 
 class RiskMeasureDefinition(BaseModel):
     measure_id: str = Field(None, description="Identifier for the risk measure.")
@@ -95,7 +102,7 @@ class RiskScoreValue(BaseModel):
     )
 
 
-class ScoreBasedRiskMeasureDefinition(BaseModel, frozen=True):
+class ScoreBasedRiskMeasureDefinition(BaseModel):
     hazard_types: List[str] = Field(
         [], description="Defines the hazards that the measure is used for."
     )
@@ -111,7 +118,7 @@ class ScoreBasedRiskMeasureDefinition(BaseModel, frozen=True):
     #    [], description="The identifiers of the underlying risk measures from which the scores are inferred."
     # )
 
-    # should be sufficient to pass frozen=True, but does not seem to work (pydantic docs says feature in beta)
+    # It is not enough to pass frozen=True, since not all attributes are hashable
     def __hash__(self):
         return id(self)
 
@@ -166,6 +173,7 @@ class ImpactKey(BaseModel):
 class AssetSingleImpact(BaseModel):
     """Impact at level of single asset and single type of hazard."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
     key: ImpactKey
 
     @computed_field  # deprecated: use key instead
@@ -191,9 +199,6 @@ def year(self) -> str:
         description="""Details of impact calculation for acute hazard calculations.""",
     )
 
-    class Config:
-        arbitrary_types_allowed = True
-
 
 class AssetLevelImpact(BaseModel):
     """Impact at asset level. Each asset can have impacts for multiple hazard types."""

src/physrisk/container.py

@@ -2,18 +2,14 @@
 
 from dependency_injector import containers, providers
 
+from physrisk.kernel import calculation as calc
 from physrisk.data.hazard_data_provider import SourcePath
 from physrisk.data.inventory import EmbeddedInventory
 from physrisk.data.inventory_reader import InventoryReader
 from physrisk.data.pregenerated_hazard_model import ZarrHazardModel
 from physrisk.data.zarr_reader import ZarrReader
-from physrisk.kernel import calculation as calc
 from physrisk.kernel.hazard_model import HazardModelFactory
-from physrisk.kernel.vulnerability_model import (
-    DictBasedVulnerabilityModels,
-    VulnerabilityModels,
-    VulnerabilityModelsFactory,
-)
+from physrisk.kernel.vulnerability_model import DictBasedVulnerabilityModelsFactory
 from physrisk.requests import Requester, _create_inventory, create_source_paths
 
 
@@ -41,11 +37,6 @@ def hazard_model(
         )
 
 
-class DictBasedVulnerabilityModelsFactory(VulnerabilityModelsFactory):
-    def vulnerability_models(self) -> VulnerabilityModels:
-        return DictBasedVulnerabilityModels(calc.get_default_vulnerability_models())
-
-
 class Container(containers.DeclarativeContainer):
     config = providers.Configuration(default={"zarr_sources": ["embedded", "hazard"]})
 

src/physrisk/data/inventory.py

@@ -9,10 +9,9 @@
 
 import physrisk.data.colormap_provider as colormap_provider
 import physrisk.data.static.hazard
+from physrisk.api.v1.hazard_data import HazardResource, Period
 from physrisk.data.inventory_reader import HazardModels
 
-from ..api.v1.hazard_data import HazardResource, Period
-
 # from physrisk.kernel.hazards import ChronicHeat