diff --git a/src/physrisk/kernel/assets.py b/src/physrisk/kernel/assets.py index 14e79111..1ffcefbe 100644 --- a/src/physrisk/kernel/assets.py +++ b/src/physrisk/kernel/assets.py @@ -366,6 +366,9 @@ 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__()) @@ -373,7 +376,26 @@ def all_subclasses(cls: type) -> set[type]: 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 diff --git a/src/physrisk/kernel/hazards.py b/src/physrisk/kernel/hazards.py index 74833640..24b43583 100644 --- a/src/physrisk/kernel/hazards.py +++ b/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): diff --git a/src/physrisk/vulnerability_models/configuration/asset_factory.py b/src/physrisk/vulnerability_models/configuration/asset_factory.py index 686304e5..4f93e79d 100644 --- a/src/physrisk/vulnerability_models/configuration/asset_factory.py +++ b/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) diff --git a/src/physrisk/vulnerability_models/downtime.py b/src/physrisk/vulnerability_models/downtime.py index a34cc542..837b248a 100644 --- a/src/physrisk/vulnerability_models/downtime.py +++ b/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 diff --git a/src/physrisk/vulnerability_models/impact_function_selector.py b/src/physrisk/vulnerability_models/impact_function_selector.py index 18bd903d..737ef3e7 100644 --- a/src/physrisk/vulnerability_models/impact_function_selector.py +++ b/src/physrisk/vulnerability_models/impact_function_selector.py @@ -2,7 +2,6 @@ import collections from dataclasses import dataclass -from importlib import import_module import copy from importlib.resources import files @@ -10,14 +9,14 @@ 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: diff --git a/src/physrisk/vulnerability_models/vulnerability.py b/src/physrisk/vulnerability_models/vulnerability.py index d985a00e..c8f5e920 100644 --- a/src/physrisk/vulnerability_models/vulnerability.py +++ b/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__( diff --git a/tests/kernel/test_assets.py b/tests/kernel/test_assets.py index 1b56e899..ffe99141 100644 --- a/tests/kernel/test_assets.py +++ b/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(): diff --git a/tests/kernel/test_hazards.py b/tests/kernel/test_hazards.py new file mode 100644 index 00000000..a93f0387 --- /dev/null +++ b/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")