Physrisk assets

src/physrisk/api/v1/common.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 import numpy as np
 from pydantic import BaseModel, Field
@@ -33,10 +33,16 @@ class Asset(BaseModel):
     asset_class: str = Field(
         description="name of asset class; corresponds to physrisk class names, e.g. PowerGeneratingAsset"
     )
-    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")
+    type: Optional[str] = Field(None, description="Type of the asset <level_1>/<level_2>/<level_3>")
+    location: Optional[str] = Field(
+        None, description="Location (e.g. Africa, Asia, Europe, Global, Oceania, North America, South America)"
+    )
+    capacity: Optional[float] = Field(None, description="Power generation capacity")
+    attributes: Optional[Dict[str, str]] = Field(
+        None, description="Bespoke attributes (e.g. number of storeys, structure type, occupancy type)"
+    )
 
 
 class Assets(BaseModel):

src/physrisk/kernel/assets.py

@@ -113,13 +113,28 @@ def get_inundation_protection_return_period(self):
         return 250.0
 
 
+class Generic(Asset):
+    def __init__(self, latitude: float, longitude: float, *, location: Optional[str] = None):
+        super().__init__(latitude, longitude)
+        self.location = location
+
+
 class RealEstateAsset(Asset):
     def __init__(self, latitude: float, longitude: float, *, location: str, type: str):
         super().__init__(latitude, longitude)
         self.location = location
         self.type = type
 
 
+class ManufacturingAsset(Asset):
+    def __init__(
+        self, latitude: float, longitude: float, *, location: Optional[str] = None, type: Optional[str] = None
+    ):
+        super().__init__(latitude, longitude)
+        self.location = location
+        self.type = type
+
+
 class IndustrialActivity(Asset):
     def __init__(self, latitude: float, longitude: float, *, location: Optional[str] = None, type: str):
         super().__init__(latitude, longitude)

src/physrisk/requests.py

@@ -266,7 +266,7 @@ def _get_hazard_data(request: HazardDataRequest, hazard_model: HazardModel):
     return response
 
 
-def create_assets(asset: Assets, assets: Optional[List[Asset]]):
+def create_assets(asset: Assets, assets: Optional[List[Asset]]):  # noqa: max-complexity=11
     """Create list of Asset objects from the Assets API object:"""
     if assets is not None:
         if len(asset.items) != 0:
@@ -277,12 +277,28 @@ def create_assets(asset: Assets, assets: Optional[List[Asset]]):
         asset_objs = []
         for item in asset.items:
             if hasattr(module, item.asset_class):
+                kwargs: Dict[str, Any] = {}
+                if item.type is not None:
+                    kwargs["type"] = item.type
+                if item.location is not None:
+                    kwargs["location"] = item.location
+                if item.capacity is not None:
+                    kwargs["capacity"] = item.capacity
                 asset_obj = cast(
                     Asset,
-                    getattr(module, item.asset_class)(
-                        item.latitude, item.longitude, type=item.type, location=item.location
-                    ),
+                    getattr(module, item.asset_class)(item.latitude, item.longitude, **kwargs),
                 )
+                if item.attributes is not None:
+                    for key, value in item.attributes.items():
+                        if value.isdigit():
+                            value_as_double = float(value)
+                            setattr(
+                                asset_obj,
+                                key,
+                                int(value_as_double) if value_as_double.is_integer() else value_as_double,
+                            )
+                        else:
+                            setattr(asset_obj, key, value)
                 asset_objs.append(asset_obj)
             else:
                 raise ValueError(f"asset type '{item.asset_class}' not found")

tests/api/impact_requests_test.py

@@ -5,7 +5,7 @@
 from pydantic import TypeAdapter
 
 from physrisk import requests
-from physrisk.api.v1.common import Assets
+from physrisk.api.v1.common import Asset, Assets
 from physrisk.api.v1.impact_req_resp import RiskMeasures, RiskMeasuresHelper
 from physrisk.container import Container
 from physrisk.data.inventory import EmbeddedInventory
@@ -189,8 +189,26 @@ def test_thermal_power_generation(self):
             ]
         ]
 
+        assets_provided_in_the_request = False
+
         request_dict = {
-            "assets": Assets(items=[]),
+            "assets": Assets(
+                items=(
+                    [
+                        Asset(
+                            asset_class=asset.__class__.__name__,
+                            latitude=asset.latitude,
+                            longitude=asset.longitude,
+                            type=asset.type,
+                            capacity=asset.capacity,
+                            location=asset.location,
+                        )
+                        for asset in assets
+                    ]
+                    if assets_provided_in_the_request
+                    else []
+                )
+            ),
             "include_asset_level": True,
             "include_calc_details": True,
             "years": [2050],
@@ -550,7 +568,7 @@ def test_thermal_power_generation(self):
             request,
             ZarrHazardModel(source_paths=source_paths, reader=ZarrReader(store)),
             vulnerability_models=vulnerability_models,
-            assets=assets,
+            assets=None if assets_provided_in_the_request else assets,
         )
 
         # Air Temperature

tests/risk_models/risk_models_test.py

@@ -78,6 +78,10 @@ def _create_assets_json(self, assets: Sequence[RealEstateAsset]):
                     "location": asset.location,
                     "longitude": asset.longitude,
                     "latitude": asset.latitude,
+                    "attributes": {
+                        "number_of_storeys": "2",
+                        "structure_type": "concrete",
+                    },
                 }
                 for asset in assets
             ],