Add generic wind vulnerability model; overhaul score-based risk measu…

…res and update Pydantic version

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

Pipfile

@@ -6,7 +6,7 @@ name = "pypi"
 [packages]
 affine = "==2.3.0"
 numpy = "==1.22.0"
-pydantic = "==1.9.0"
+pydantic = "==2.4.2"
 python-dotenv = "==0.19.2"
 requests = "==2.27.1"
 scipy = "==1.7.3"

notebooks/vulnerability_onboarding/Wind/Table_A2_Impact_Function_Slope.csv

@@ -0,0 +1,12 @@
+region,nb_countries,nb_events,vhalf_a,vhalf_b,vhalf_c,rmsf_a,rmsf_b,rmsf_c,tdr_a,tdr_b,tdr_c
+NA1,21,73,74.7,59.6,66.3,11.8,9.8,10.3,0.68,1.44,1
+NA2,2,43,74.7,86,89.2,9.5,8.7,8.7,2.11,1.16,1
+NI,6,31,74.7,58.7,70.8,7.8,6,7.2,0.85,2.03,1
+OC,11,48,74.7,49.7,64.1,22.5,14.7,17.7,0.6,2.31,1
+SI,2,19,74.7,46.8,52.4,20.1,8.6,9.1,0.2,1.8,1
+WP1,4,43,74.7,56.7,66.4,15.2,11.3,12.6,0.62,2.05,1
+WP2,1,83,74.7,84.7,188.4,38.2,36.7,104.9,25.89,16.44,1
+WP3,1,69,74.7,80.2,112.8,15.2,14.8,20.5,5.32,3.83,1
+WP4,5,64,74.7,135.6,190.5,73.8,35.9,43.8,35.56,3.35,1
+Combined,53,473,74.7,–,–,22.2,16.8,24.4,4.69,2.15,1
+Global calibration,53,473,74.7,73.4,110.1,22.2,22.2,33.1,4.69,4.84,1

setup.cfg

@@ -24,7 +24,7 @@ install_requires =
     numba==0.56.4
     numpy==1.22.0
     pillow==9.4.0
-    pydantic==1.9.0
+    pydantic==2.4.2
     python-dotenv==0.19.2
     requests==2.27.1
     scipy==1.7.3

src/physrisk/api/v1/common.py

@@ -33,8 +33,8 @@ class Asset(BaseModel):
     asset_class: str = Field(
         description="name of asset class; corresponds to physrisk class names, e.g. PowerGeneratingAsset"
     )
-    type: Optional[str] = Field(description="Type of the asset <level_1>/<level_2>/<level_3>")
-    location: Optional[str]
+    type: Optional[str] = Field(None, description="Type of the asset <level_1>/<level_2>/<level_3>")
+    location: Optional[str] = None
     latitude: float = Field(description="Latitude in degrees")
     longitude: float = Field(description="Longitude in degrees")
 

src/physrisk/api/v1/exposure_req_resp.py

@@ -10,7 +10,9 @@ class AssetExposureRequest(BaseModel):
     """Impact calculation request."""
 
     assets: Assets
-    calc_settings: CalcSettings = Field(default_factory=CalcSettings, description="Interpolation method.")
+    calc_settings: CalcSettings = Field(
+        default_factory=CalcSettings, description="Interpolation method."  # type:ignore
+    )
     scenario: str = Field("rcp8p5", description="Name of scenario ('rcp8p5')")
     year: int = Field(
         2050,

src/physrisk/api/v1/hazard_data.py

@@ -59,7 +59,7 @@ class Scenario(BaseModel):
 
     id: str
     years: List[int]
-    periods: Optional[List[Period]]
+    # periods: Optional[List[Period]]
 
 
 def expanded(item: str, key: str, param: str):
@@ -89,7 +89,7 @@ class HazardResource(BaseModel):
     description: str = Field(
         description="Brief description in mark down of the indicator and model that generated the indicator."
     )
-    map: Optional[MapInfo] = Field(description="Optional information used for display of the indicator in a map.")
+    map: Optional[MapInfo] = Field(None, description="Optional information used for display of the indicator in a map.")
     scenarios: List[Scenario] = Field(description="Climate change scenarios for which the indicator is available.")
     units: str = Field(description="Units of the hazard indicator.")
 
@@ -112,13 +112,13 @@ def expand_resource(
     resource: HazardResource, keys: List[str], params: Dict[str, List[str]]
 ) -> Iterable[HazardResource]:
     if len(keys) == 0:
-        yield resource.copy(deep=True, update={"params": {}})
+        yield resource.model_copy(deep=True, update={"params": {}})
     else:
         keys = keys.copy()
         key = keys.pop()
         for item in expand_resource(resource, keys, params):
             for param in params[key]:
-                yield item.copy(
+                yield item.model_copy(
                     deep=True,
                     update={
                         "indicator_id": expand(item.indicator_id, key, param),
@@ -128,7 +128,7 @@ def expand_resource(
                         "map": None
                         if item.map is None
                         else (
-                            item.map.copy(
+                            item.map.model_copy(
                                 deep=True,
                                 update={"path": expand(item.map.path if item.map.path is not None else "", key, param)},
                             )
@@ -148,9 +148,9 @@ class InventorySource(Flag):
 
 
 class HazardAvailabilityRequest(BaseModel):
-    types: Optional[List[str]]  # e.g. ["RiverineInundation"]
+    types: Optional[List[str]] = []  # e.g. ["RiverineInundation"]
     sources: Optional[List[str]] = Field(
-        description="Sources of inventory, can be 'embedded', 'hazard' or 'hazard_test'."
+        None, description="Sources of inventory, can be 'embedded', 'hazard' or 'hazard_test'."
     )
 
 
@@ -171,11 +171,11 @@ class HazardDataRequestItem(BaseModel):
     longitudes: List[float]
     latitudes: List[float]
     request_item_id: str
-    hazard_type: Optional[str]  # e.g. RiverineInundation
-    event_type: Optional[str]  # e.g. RiverineInundation; deprecated: use hazard_type
+    hazard_type: Optional[str] = None  # e.g. RiverineInundation
+    event_type: Optional[str] = None  # e.g. RiverineInundation; deprecated: use hazard_type
     indicator_id: str
-    indicator_model_gcm: Optional[str]
-    path: Optional[str]
+    indicator_model_gcm: Optional[str] = ""
+    path: Optional[str] = None
     scenario: str  # e.g. rcp8p5
     year: int
 
@@ -188,7 +188,7 @@ class HazardDataRequest(BaseHazardRequest):
 class HazardDataResponseItem(BaseModel):
     intensity_curve_set: List[IntensityCurve]
     request_item_id: str
-    event_type: str
+    event_type: Optional[str]
     model: str
     scenario: str
     year: int

src/physrisk/api/v1/impact_req_resp.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 from pydantic import BaseModel, Field
 
@@ -14,7 +14,9 @@ class AssetImpactRequest(BaseModel):
     """Impact calculation request."""
 
     assets: Assets
-    calc_settings: CalcSettings = Field(default_factory=CalcSettings, description="Interpolation method.")
+    calc_settings: CalcSettings = Field(
+        default_factory=CalcSettings, description="Interpolation method."  # type:ignore
+    )
     include_asset_level: bool = Field(True, description="If true, include asset-level impacts.")
     include_measures: bool = Field(True, description="If true, include measures.")
     include_calc_details: bool = Field(True, description="If true, include impact calculation details.")
@@ -25,32 +27,103 @@ class AssetImpactRequest(BaseModel):
     )
 
 
-# region Response
+class Category(int, Enum):
+    NODATA = 0
+    LOW = 1
+    MEDIUM = 2
+    HIGH = 3
+    REDFLAG = 4
 
 
-class Category(str, Enum):
-    NODATA = "NODATA"
-    LOW = "LOW"
-    MEDIUM = "MEDIUM"
-    HIGH = "HIGH"
-    REDFLAG = "REDFLAG"
+class RiskKey(BaseModel):
+    scenario_id: str
+    year: str
 
 
-class Indicator(BaseModel):
-    value: float
-    label: str
+class RiskMeasureKey(RiskKey):
+    risk_measure_id: str = Field("", description="Identifier of the risk measure.")
 
 
-class RiskMeasureResult(BaseModel):
-    """Provides a risk category based on one or more risk indicators.
-    A risk indicator is a quantity derived from one or more vulnerability models,
-    e.g. the change in 1-in-100 year damage or disruption.
+class AssetsRiskScores(BaseModel):
+    """Risk scores for a set of assets, with risk measures used to calculate the measures.
+    A single score may be derived from multiple risk measures in principle, the measures are identified
+    by the ScoreBasedMeasureDefinition corresponding to the asset.
+    In principle multiple measures may be used to compute the score, hence 'measures_0', 'measures_1' etc,
+    although no example yet.
     """
 
-    category: Category = Field(description="Result category.")
-    cat_defn: str = Field(description="Definition of the category for the particular indicator.")
-    indicators: List[Indicator]
-    summary: str = Field(description="Summary of the indicator.")
+    key: RiskKey
+    scores: List[int] = Field(None, description="Identifier for the risk measure.")
+    measures_0: List[float]
+    measures_1: Optional[List[float]]
+
+
+class AssetRiskMeasures(BaseModel):
+    """Risk measures for a set of assets."""
+
+    key: RiskMeasureKey
+    measures: List[float]
+
+
+class AssetScoreModel(BaseModel):
+    asset_model_id: List[str]
+
+
+class RiskMeasureDefinition(BaseModel):
+    measure_id: str = Field(None, description="Identifier for the risk measure.")
+    measure_index: int = Field(None, description="Identifier for the risk measure.")
+    label: str = Field(
+        "<short description of the measure, e.g. fractional loss for 1-in-100 year event.",
+        description="Value of the score.",
+    )
+    description: str
+
+
+class RiskScoreValue(BaseModel):
+    value: Category = Field("", description="Value of the score: red, amber, green, nodata.")
+    label: str = Field(
+        "", description="Short description of value, e.g. material increase in loss for 1-in-100 year event."
+    )
+    description: str = Field(
+        "",
+        description="Full description of value including criteria, \
+        e.g. change in fractional loss from 1-in-100 year event is greater than 10%.",
+    )
+
+
+class ScoreBasedRiskMeasureDefinition(BaseModel, frozen=True):
+    hazard_types: List[str] = Field([], description="Defines the hazards that the measure is used for.")
+    values: List[RiskScoreValue] = Field([], description="Defines the set of values that the score can take.")
+    child_measure_ids: List[str] = Field(
+        [], description="The identifiers of the risk measures used to calculate the score."
+    )
+
+    # should be sufficient to pass frozen=True, but does not seem to work (pydantic docs says feature in beta)
+    def __hash__(self):
+        return id(self)
+
+
+class HazardRiskMeasures(BaseModel):
+    """Risk measures for one particular type of hazard"""
+
+    hazard_type: str
+    scores_for_assets: Optional[List[AssetsRiskScores]] = Field(
+        [], description="Risk scores for the set of assets for different scenarios and years."
+    )
+    measures_for_assets: Optional[List[AssetRiskMeasures]] = Field(
+        [], description="Risk measures for the set of assets for different scenarios and years."
+    )
+    score_measure_ids_for_assets: Optional[List[str]] = Field(
+        None, description="Identifiers of the score-based risk measures used for each asset."
+    )
+
+
+class RiskMeasures(BaseModel):
+    """Risk measures"""
+
+    hazard_risk_measures: List[HazardRiskMeasures]
+    score_definitions: Optional[Dict[str, ScoreBasedRiskMeasureDefinition]]
+    measures_definitions: Optional[List[RiskMeasureDefinition]]
 
 
 class AcuteHazardCalculationDetails(BaseModel):
@@ -61,19 +134,19 @@ class AcuteHazardCalculationDetails(BaseModel):
     vulnerability_distribution: VulnerabilityDistrib
 
 
-class AssetSingleHazardImpact(BaseModel):
+class AssetSingleImpact(BaseModel):
     """Impact at level of single asset and single type of hazard."""
 
-    hazard_type: str = Field("", description="Type of the hazard.")
-    impact_type: str = Field(
+    hazard_type: Optional[str] = Field("", description="Type of the hazard.")
+    impact_type: Optional[str] = Field(
         "damage",
         description="""'damage' or 'disruption'. Whether the impact is fractional damage to the asset
-        ('damage') or disruption to the annual economic benefit obtained from the asset ('disruption'), expressed as
+        ('damage') or disruption to an operation, expressed as
         fractional decrease to an equivalent cash amount.""",
     )
-    risk_measure: Optional[RiskMeasureResult]
-    impact_distribution: Distribution
-    impact_exceedance: ExceedanceCurve
+    year: Optional[str] = None
+    impact_distribution: Optional[Distribution]
+    impact_exceedance: Optional[ExceedanceCurve]
     impact_mean: float
     impact_std_deviation: float
     calc_details: Optional[AcuteHazardCalculationDetails] = Field(
@@ -93,13 +166,11 @@ class AssetLevelImpact(BaseModel):
         description="""Asset identifier; will appear if provided in the request
         otherwise order of assets in response is identical to order of assets in request.""",
     )
-    impacts: List[AssetSingleHazardImpact] = Field([], description="Impacts for each hazard type.")
+    impacts: List[AssetSingleImpact] = Field([], description="Impacts for each hazard type.")
 
 
 class AssetImpactResponse(BaseModel):
     """Response to impact request."""
 
-    asset_impacts: List[AssetLevelImpact]
-
-
-# endregion
+    asset_impacts: Optional[List[AssetLevelImpact]] = None
+    risk_measures: Optional[RiskMeasures] = None

src/physrisk/data/inventory.py

@@ -6,7 +6,7 @@
 from collections import defaultdict
 from typing import DefaultDict, Dict, Iterable, List, Tuple
 
-from pydantic import parse_obj_as
+from pydantic import TypeAdapter, parse_obj_as
 
 import physrisk.data.colormap_provider as colormap_provider
 import physrisk.data.static.hazard
@@ -50,7 +50,7 @@ class EmbeddedInventory(Inventory):
 
     def __init__(self):
         with importlib.resources.open_text(physrisk.data.static.hazard, "inventory.json") as f:
-            models = parse_obj_as(HazardModels, json.load(f)).resources
+            models = TypeAdapter(HazardModels).validate_python(json.load(f)).resources
             expanded_models = expand(models)
             super().__init__(expanded_models)
 

src/physrisk/data/inventory_reader.py

@@ -4,7 +4,7 @@
 
 import s3fs
 from fsspec import AbstractFileSystem
-from pydantic import BaseModel, parse_obj_as
+from pydantic import BaseModel, TypeAdapter
 
 from physrisk.api.v1.hazard_data import HazardResource
 
@@ -48,7 +48,7 @@ def read(self, path: str) -> List[HazardResource]:
         if not self._fs.exists(self._full_path(path)):
             return []
         json_str = self.read_json(path)
-        models = parse_obj_as(HazardModels, json.loads(json_str)).resources
+        models = TypeAdapter(HazardModels).validate_python(json.loads(json_str)).resources
         return models
 
     def read_description_markdown(self, paths: List[str]) -> Dict[str, str]:
@@ -73,7 +73,7 @@ def append(self, path: str, hazard_models: Iterable[HazardResource]):
         for model in hazard_models:
             combined[model.key()] = model
         models = HazardModels(resources=list(combined.values()))
-        json_str = json.dumps(models.dict())
+        json_str = json.dumps(models.model_dump())
         with self._fs.open(self._full_path(path), "w") as f:
             f.write(json_str)