Clean up class resolution

src/physrisk/kernel/assets.py

@@ -366,14 +366,36 @@ class WindTurbine(Asset):
     rotor_diameter: Optional[float] = None
 
 
+_EXCLUDED_ASSET_TYPES = (OEDAsset, SimpleTypeLocationAsset, TestAsset)
+
+
 def all_asset_types():
     def all_subclasses(cls: type) -> set[type]:
         subclasses = set(cls.__subclasses__())
         for subclass in cls.__subclasses__():
             subclasses |= all_subclasses(subclass)
         return subclasses
 
-    all_asset_types = all_subclasses(Asset)
-    for a in [OEDAsset, SimpleTypeLocationAsset, TestAsset]:
-        all_asset_types.remove(a)
-    return all_asset_types
+    return all_subclasses(Asset).difference(_EXCLUDED_ASSET_TYPES)
+
+
+def asset_class(name: str) -> type[Asset]:
+    """Return the asset class with the supplied name.
+
+    Args:
+        name: Name of a class defined in this module.
+
+    Returns:
+        The matching ``Asset`` class or subclass.
+
+    Raises:
+        AttributeError: The name does not identify an ``Asset`` class.
+    """
+    candidate = globals().get(name)
+    if (
+        not isinstance(candidate, type)
+        or not issubclass(candidate, Asset)
+        or candidate in _EXCLUDED_ASSET_TYPES
+    ):
+        raise AttributeError(f"unknown asset class {name!r}")
+    return candidate

src/physrisk/kernel/hazards.py

@@ -1,5 +1,3 @@
-import inspect
-import sys
 from enum import Enum
 from typing import Dict, Type
 
@@ -126,16 +124,30 @@ class Subsidence(Hazard):
     pass
 
 
-def all_hazards():
+def all_hazards() -> list[type[Hazard]]:
     return [
         obj
-        for _, obj in inspect.getmembers(sys.modules[__name__])
-        if inspect.isclass(obj) and issubclass(obj, Hazard) and obj is not Hazard
+        for _, obj in sorted(globals().items())
+        if isinstance(obj, type) and issubclass(obj, Hazard) and obj is not Hazard
     ]
 
 
-def hazard_class(name: str) -> Type[Hazard]:
-    return getattr(sys.modules[__name__], name)
+def hazard_class(name: str) -> type[Hazard]:
+    """Return the hazard class with the supplied name.
+
+    Args:
+        name: Name of a class defined in this module.
+
+    Returns:
+        The matching ``Hazard`` class or subclass.
+
+    Raises:
+        AttributeError: The name does not identify a ``Hazard`` class.
+    """
+    candidate = globals().get(name)
+    if not isinstance(candidate, type) or not issubclass(candidate, Hazard):
+        raise AttributeError(f"unknown hazard class {name!r}")
+    return candidate
 
 
 class Landslide(Hazard):

src/physrisk/vulnerability_models/configuration/asset_factory.py

@@ -1,6 +1,5 @@
-from importlib import import_module
 import logging
-from typing import Any, Optional, Protocol, Sequence, cast
+from typing import Any, Optional, Protocol, Sequence
 
 import pandas as pd
 from physrisk.api.v1.common import Asset as APIAsset
@@ -14,6 +13,7 @@
     ThermalPowerGeneratingAsset,
     TransportationAsset,
     UtilityAsset,
+    asset_class,
 )
 from physrisk.kernel.financial_model import FinancialDataProvider
 from physrisk.risk_models.portfolio_risk_model import FinancialDataStore
@@ -114,7 +114,6 @@ def interval_df(mapping: dict, col: str) -> pd.DataFrame:
                 index=pd.IntervalIndex.from_tuples(mapping.keys(), closed="both"),
             )
 
-        self.module = import_module("physrisk.kernel.assets")
         self.occupancy_mapping = (
             occupancy_mapping or default_oed_occ_codes_to_asset_types
         )
@@ -180,13 +179,9 @@ def create_asset(self, api_asset: APIAsset) -> Asset:
             kwargs["buffer"] = 0.0
         return self._create_asset(asset_class, kwargs)
 
-    def _create_asset(self, asset_class: str, kwargs: dict[str, Any]) -> Asset:
-        if hasattr(self.module, asset_class):
-            init = getattr(self.module, asset_class)
-            asset_obj = cast(
-                Asset,
-                init(**kwargs),
-            )
-            return asset_obj
-        else:
-            raise ValueError(f"asset type '{asset_class}' not found")
+    def _create_asset(self, asset_class_name: str, kwargs: dict[str, Any]) -> Asset:
+        try:
+            init = asset_class(asset_class_name)
+        except AttributeError as error:
+            raise ValueError(f"asset type '{asset_class_name}' not found") from error
+        return init(**kwargs)

src/physrisk/vulnerability_models/downtime.py

@@ -4,13 +4,12 @@
 from typing import Dict, List, Optional, Sequence, Type
 
 import numpy as np
-from physrisk.kernel.assets import Asset
+from physrisk.kernel.assets import Asset, asset_class
 
 from physrisk.vulnerability_models.config_based_impact_curves import (
     DowntimeConfigItem,
     ImpactCurveKey,
 )
-from physrisk.vulnerability_models.vulnerability import get_asset_type
 
 
 class DowntimeModelBase:
@@ -64,7 +63,7 @@ def config_based_models(self, config: Sequence[DowntimeConfigItem]):
             list
         )
         for item in config:
-            items_by_asset_class[get_asset_type(item.asset_class)].append(item)
+            items_by_asset_class[asset_class(item.asset_class)].append(item)
 
         ancestors = dict(
             sorted(
@@ -74,33 +73,30 @@ def config_based_models(self, config: Sequence[DowntimeConfigItem]):
         )
         for k, v in ancestors.items():
             keys = [item.asset_identifier for item in items_by_asset_class[k]]
-            for asset_class in v:
-                if (
-                    asset_class not in [k, object]
-                    and asset_class in items_by_asset_class
-                ):
+            for asset_type in v:
+                if asset_type not in [k, object] and asset_type in items_by_asset_class:
                     items_by_asset_class[k] = add(
                         items_by_asset_class[k],
                         [
                             item
-                            for item in items_by_asset_class[asset_class]
+                            for item in items_by_asset_class[asset_type]
                             if item.asset_identifier not in keys
                         ],
                     )
                     keys = [item.asset_identifier for item in items_by_asset_class[k]]
 
         models: Dict[Type[Asset], List[DowntimeModelBase]] = defaultdict(list)
-        for asset_class, items in items_by_asset_class.items():
+        for asset_type, items in items_by_asset_class.items():
             try:
-                models[asset_class].append(
+                models[asset_type].append(
                     ConfigBasedDowntimeModel(
-                        asset_class=asset_class.__name__,
+                        asset_class=asset_type.__name__,
                         config_items=items,
                     )
                 )
             except Exception:
                 raise ValueError(
-                    f"Could not convert config item with asset {asset_class.__name__}."
+                    f"Could not convert config item with asset {asset_type.__name__}."
                 )
         return models
 

src/physrisk/vulnerability_models/impact_function_selector.py

@@ -2,22 +2,21 @@
 
 import collections
 from dataclasses import dataclass
-from importlib import import_module
 import copy
 
 from importlib.resources import files
 from typing import NamedTuple, Protocol, Sequence
 
 import pandas as pd
 import physrisk.data.static.vulnerability.oed_hazus
-import physrisk.kernel.assets
-from physrisk.kernel.assets import Asset, OEDAsset
+from physrisk.kernel.assets import Asset, OEDAsset, all_asset_types, asset_class
 from physrisk.kernel.hazards import (
     CoastalInundation,
     Hazard,
     PluvialInundation,
     RiverineInundation,
     Wind,
+    hazard_class,
 )
 from physrisk.kernel.impact_distrib import ImpactType
 from physrisk.vulnerability_models.config_based_impact_curves import (
@@ -95,8 +94,6 @@ class GroupLookup:
 
 class ConfigBasedImpactFunctionSelector(ImpactFunctionSelector):
     def __init__(self, config_items: Sequence[VulnerabilityConfigItem]):
-        physrisk_assets = import_module("physrisk.kernel.assets")
-        physrisk_hazards = import_module("physrisk.kernel.hazards")
         grouped_items: dict[VulnModelKey, list[VulnerabilityConfigItem]] = (
             collections.defaultdict(list)
         )
@@ -113,8 +110,8 @@ def __init__(self, config_items: Sequence[VulnerabilityConfigItem]):
                     f"occupancy_code-based config must use asset_class='Asset' because "
                     f"the occupancy code already encodes the asset type."
                 )
-            asset_type = getattr(physrisk_assets, item.asset_class)
-            hazard_type = getattr(physrisk_hazards, item.hazard_class)
+            asset_type = asset_class(item.asset_class)
+            hazard_type = hazard_class(item.hazard_class)
             impact_type = (
                 ImpactType.damage
                 if item.impact_id == "damage"
@@ -151,14 +148,14 @@ def __init__(self, config_items: Sequence[VulnerabilityConfigItem]):
 
         # deal with ancestors: the ancestors of each asset class are identified at this point, so that if
         # a match is not found with the class itself an ancestor match can be attempted.
-        all_asset_types = physrisk.kernel.assets.all_asset_types()
-        all_asset_types.add(OEDAsset)
-        all_asset_types.add(Asset)
+        asset_types = all_asset_types()
+        asset_types.add(OEDAsset)
+        asset_types.add(Asset)
         # set(k.asset_type for k in self._groups.keys())
 
         # consider excluding OEDAsset
         self._ancestors = {
-            t: [a for a in t.mro() if a not in [t, object]] for t in all_asset_types
+            t: [a for a in t.mro() if a not in [t, object]] for t in asset_types
         }
 
     def _get_indicator_id(self, indicator_id: str) -> str:

src/physrisk/vulnerability_models/vulnerability.py

@@ -1,8 +1,6 @@
 import importlib.resources
-from importlib import import_module
 from typing import Dict, Sequence
 
-import physrisk.kernel.assets
 from physrisk.kernel.assets import Asset
 from physrisk.kernel.hazards import (
     CoastalInundation,
@@ -40,14 +38,6 @@
     VulnModelKey,
 )
 
-physrisk_assets = import_module("physrisk.kernel.assets")
-
-
-def get_asset_type(asset_class: str):
-    return getattr(
-        physrisk_assets, asset_class, getattr(physrisk_assets, asset_class, None)
-    )
-
 
 class VulnerabilityModelsFactory(PVulnerabilityModelsFactory):
     def __init__(

tests/kernel/test_assets.py

@@ -1,8 +1,29 @@
 import numpy as np
+import pytest
 from shapely import Point
 from shapely.ops import transform
 
-from physrisk.kernel.assets import Asset, project_4326_to_3857
+from physrisk.kernel.assets import (
+    Asset,
+    RealEstateAsset,
+    asset_class,
+    project_4326_to_3857,
+)
+
+
+@pytest.mark.parametrize(
+    "excluded", ["OEDAsset", "SimpleTypeLocationAsset", "TestAsset"]
+)
+def test_asset_class_returns_only_public_asset_types(excluded):
+    assert asset_class("RealEstateAsset") is RealEstateAsset
+
+    with pytest.raises(AttributeError, match="unknown asset class"):
+        asset_class(excluded)
+
+
+def test_asset_class_rejects_non_asset_types():
+    with pytest.raises(AttributeError, match="unknown asset class"):
+        asset_class("FuelKind")
 
 
 def test_buffered_geometry_contains_origin():

tests/kernel/test_hazards.py

@@ -0,0 +1,10 @@
+import pytest
+
+from physrisk.kernel.hazards import Wind, hazard_class
+
+
+def test_hazard_class_returns_only_hazard_types():
+    assert hazard_class("Wind") is Wind
+
+    with pytest.raises(AttributeError, match="unknown hazard class"):
+        hazard_class("HazardKind")