Make mapping of unknown occupancy code to Hazus curves optional

src/physrisk/api/v1/impact_req_resp.py

@@ -29,6 +29,11 @@ class CalcSettings(BaseModel):
         default=None,
         description="Dictionary of hazards and corresponding indicator ids to include in analysis.",
     )
+    map_unknown_occ: bool = Field(
+        default=False,
+        description="If True, map unknown occupancy codes to an average of Hazus vulnerability curves."
+        "Only acts if the requester is configured to use Hazus vulerability curves.",
+    )
 
 
 class AssetMeasuresSpecification(BaseModel):

src/physrisk/container.py

@@ -101,14 +101,16 @@ def image_creator(self):
 
 class DictBasedVulnerabilityModelsFactory(PVulnerabilityModelsFactory):
     def vulnerability_models(
-        self, hazard_scope: dict[type[Hazard], set[str] | None] | None = None
+        self,
+        hazard_scope: dict[type[Hazard], set[str] | None] | None = None,
+        map_unknown_occ: bool = False,
     ) -> PVulnerabilityModels:
         return DictBasedVulnerabilityModels(
             calc.alternate_default_vulnerability_models_scores()
         )
 
 
-class DefaultVulnerabilityModelFactory(VulnerabilityModelsFactory):
+class DefaultVulnerabilityModelsFactory(VulnerabilityModelsFactory):
     """Default vulnerability approach. 'default_vulnerability_models'
     programmatic models are used, to which FEMA Hazus vulnerability-based
     models are added and finally configuration.
@@ -168,7 +170,7 @@ class Container(containers.DeclarativeContainer):
 
     measures_factory = providers.Factory(calc.DefaultMeasuresFactory)
 
-    vulnerability_models_factory = providers.Factory(DefaultVulnerabilityModelFactory)
+    vulnerability_models_factory = providers.Factory(DefaultVulnerabilityModelsFactory)
 
     requester = providers.Singleton(
         Requester,

src/physrisk/kernel/vulnerability_model.py

@@ -137,7 +137,9 @@ def vuln_model_for_asset_of_type(
 
 class VulnerabilityModelsFactory(Protocol):
     def vulnerability_models(
-        self, hazard_scope: dict[type[Hazard], set[str] | None] | None = None
+        self,
+        hazard_scope: dict[type[Hazard], set[str] | None] | None = ...,
+        map_unknown_occ: bool = ...,
     ) -> VulnerabilityModels:
         """Create a VulnerabilityModels instance, that can based on a number of options.
 

src/physrisk/requests.py

@@ -291,7 +291,8 @@ def get_asset_impacts(self, request: AssetImpactRequest) -> AssetImpactResponse:
             }
 
         vulnerability_models = self.vulnerability_models_factory.vulnerability_models(
-            hazard_scope=hazard_scope
+            hazard_scope=hazard_scope,
+            map_unknown_occ=request.calc_settings.map_unknown_occ,
         )
         asset_measure_calculators = self.measures_factory.asset_calculators(
             request.use_case_id

src/physrisk/vulnerability_models/impact_function_selector.py

@@ -3,6 +3,7 @@
 import collections
 from dataclasses import dataclass
 from importlib import import_module
+import copy
 
 from importlib.resources import files
 from typing import NamedTuple, Protocol, Sequence
@@ -304,6 +305,8 @@ def __init__(self, config_items: Sequence[VulnerabilityConfigItem]):
         )
         self.vulnerability_config_items = config_items
 
+        self._map_unknown_occ = False
+
     def select(
         self,
         asset: Asset,
@@ -339,10 +342,26 @@ def select(
             raise TypeError(
                 f"OED-Hazus mapping only supports assets of type 'Asset', got '{type(asset)}'"
             )
+        if (
+            not self.map_unknown_occ
+            and asset.occupancy_code == _UNKNOWN_SENTINELS["occupancy_code"]
+        ):
+            return None
         impact_function, _ = self.oed_mapper.map(asset, hazard_type)
         # consider logging the explanation (/ exposing via service)
         return impact_function
 
+    @property
+    def map_unknown_occ(self) -> bool:
+        return self._map_unknown_occ
+
+    def with_runtime_options(self, *, map_unknown_occ: bool):
+        """Return a new selector with the given runtime options."""
+        # shallow copy, so that the underlying OEDHazusMapper is shared
+        clone = copy.copy(self)
+        clone._map_unknown_occ = map_unknown_occ
+        return clone
+
 
 class CombinedImpactFunctionSelector(ImpactFunctionSelector):
     def __init__(

src/physrisk/vulnerability_models/vulnerability.py

@@ -58,23 +58,28 @@ def __init__(
         standard_of_protection: StandardOfProtection = StandardOfProtection.CONSTANT_DEPTH,
     ):  # default_vulnerability_models):
         self.config_items = config
-        oed_hazus_selector = OEDHazusImpactFunctionSelector(config)
+        self.oed_hazus_selector = OEDHazusImpactFunctionSelector(config)
         self.config_based_selector = ConfigBasedImpactFunctionSelector(config)
-        self.combined_selector = CombinedImpactFunctionSelector(
-            oed_hazus_selector, self.config_based_selector
-        )
         self.programmatic_models = programmatic_models
         self.use_oed_hazus_curves = use_oed_hazus_curves
         self.standard_of_protection = standard_of_protection
 
     def vulnerability_models(
         self,
         hazard_scope: dict[type[Hazard], set[str] | None] | None = None,
+        map_unknown_occ: bool = False,
         disable_api_calls=False,
     ) -> PVulnerabilityModels:
+        oed_hazus_selector = self.oed_hazus_selector.with_runtime_options(
+            map_unknown_occ=map_unknown_occ
+        )
+        combined_selector = CombinedImpactFunctionSelector(
+            oed_hazus_selector, self.config_based_selector
+        )
+
         return VulnerabilityModels(
             config_based_selector=self.config_based_selector,
-            combined_selector=self.combined_selector,
+            combined_selector=combined_selector,
             programmatic_models=self.programmatic_models,
             hazard_scope=hazard_scope,
             disable_api_calls=disable_api_calls,
@@ -123,7 +128,7 @@ def __init__(
 
         self.hazard_scope = hazard_scope
         self.disable_api_calls = disable_api_calls
-        self.stanard_of_protection = standard_of_protection
+        self.standard_of_protection = standard_of_protection
 
         all_models: dict[VulnModelKey, VulnerabilityModelBase] = {}
 
@@ -141,7 +146,7 @@ def __init__(
 
         # add config based models
         config_based_models_dict = VulnerabilityModels.config_based_models(
-            config_based_selector, self.stanard_of_protection
+            config_based_selector, self.standard_of_protection
         )
         for k, v in config_based_models_dict.items():
             if k not in all_models:

tests/risk_models/test_risk_models.py

@@ -527,6 +527,7 @@ def test_generic_model_via_requests_default_vulnerability():
         "include_calc_details": True,
         "years": years,
         "scenarios": scenarios,
+        "calc_settings": {"map_unknown_occ": True},
     }
 
     container = Container()
@@ -722,7 +723,9 @@ def hazard_model(
 
     class TestVulnerabilityModelsFactory(PVulnerabilityModelsFactory):
         def vulnerability_models(
-            self, hazard_scope: dict[type[Hazard], set[str] | None] | None = None
+            self,
+            hazard_scope: dict[type[Hazard], set[str] | None] | None = None,
+            map_unknown_occ: bool = False,
         ) -> VulnerabilityModels:
             return _vulnerability_models()
 

tests/vulnerability_models/test_oed_based_config.py

@@ -539,7 +539,9 @@ def test_combined_selector():
 
 def test_combine_curves():
     vulnerability_config = VulnerabilityModelsFactory.embedded_vulnerability_config()
-    oed_hazus_selector = OEDHazusImpactFunctionSelector(vulnerability_config)
+    oed_hazus_selector = OEDHazusImpactFunctionSelector(
+        vulnerability_config
+    ).with_runtime_options(map_unknown_occ=True)
     asset = OEDAsset(
         latitude=50.0,
         longitude=1.0,