diff --git a/python/ngen_cal/src/ngen/cal/errors.py b/python/ngen_cal/src/ngen/cal/errors.py new file mode 100644 index 00000000..c3be7f0b --- /dev/null +++ b/python/ngen_cal/src/ngen/cal/errors.py @@ -0,0 +1 @@ +class UnsupportedFeatureError(ValueError): ... diff --git a/python/ngen_cal/src/ngen/cal/ngen.py b/python/ngen_cal/src/ngen/cal/ngen.py index d4f7a4c4..90fdabdb 100644 --- a/python/ngen_cal/src/ngen/cal/ngen.py +++ b/python/ngen_cal/src/ngen/cal/ngen.py @@ -28,6 +28,7 @@ from hypy.nexus import Nexus from hypy.catchment import Catchment + class NgenStrategy(str, Enum): """ """ @@ -83,7 +84,7 @@ class NgenBase(ModelExec): nexus: Optional[FilePath] crosswalk: Optional[FilePath] ngen_realization: Optional[NgenRealization] - routing_output: Optional[Path] = Field(default=Path("flowveldepth_Ngen.csv")) + routing_output: Path = Path("flowveldepth_Ngen.csv") #optional fields partitions: Optional[FilePath] parallel: Optional[PosInt] @@ -288,20 +289,27 @@ def check_for_partitions(cls, values: dict): raise ValueError("Must provide partitions if using parallel") return values - @root_validator() - def validate_hydrofabic(cls, values: dict) -> dict: + @root_validator + def _validate_model(cls, values: dict) -> dict: + NgenBase._verify_hydrofabric(values) + realization: Optional[NgenRealization] = values.get("ngen_realization") + NgenBase._verify_ngen_realization(realization) + return values + + @staticmethod + def _verify_hydrofabric(values: dict) -> None: """ - Validates hydrofabric information is provided either as (deprecated) GeoJSON + Verify hydrofabric information is provided either as (deprecated) GeoJSON or (preferred) GeoPackage files. Args: - values (dict): configuation values to check + values: root validator dictionary Raises: ValueError: If a geopackage hydrofabric or set of geojsons are not found Returns: - dict: Valid configuration elements for hydrofabric requirements + None """ hf: FilePath = values.get('hydrofabric') cats: FilePath = values.get("catchments") @@ -313,17 +321,32 @@ def validate_hydrofabic(cls, values: dict) -> dict: "or proide catchment, nexus, and crosswalk geojson files." raise ValueError(msg) - if hf is not None: - try: - p = Path(hf) - if not p.exists(): - raise - except: - raise TypeError("hydrofabric must be a valid file path") if cats is not None or nex is not None or x is not None: warnings.warn("GeoJSON support will be deprecated in a future release, use geopackage hydrofabric.", DeprecationWarning) - return values + @staticmethod + def _verify_ngen_realization(realization: Optional[NgenRealization]) -> None: + """ + Verify `ngen_realization` uses supported features. + + Args: + realization: maybe an `NgenRealization` instance + + Raises: + UnsupportedFeatureError: If `realization.output_root` is not None. + Feature not supported. + + Returns: + None + """ + if realization is None: + return None + + if realization.output_root is not None: + from .errors import UnsupportedFeatureError + raise UnsupportedFeatureError( + "ngen realization `output_root` field is not supported by ngen.cal. will be removed in future; see https://github.com/NOAA-OWP/ngen-cal/issues/150" + ) def update_config(self, i: int, params: 'pd.DataFrame', id: str = None, path=Path("./")): """_summary_ diff --git a/python/ngen_cal/tests/test_ngen.py b/python/ngen_cal/tests/test_ngen.py index ad7f826d..cab77249 100644 --- a/python/ngen_cal/tests/test_ngen.py +++ b/python/ngen_cal/tests/test_ngen.py @@ -1,5 +1,15 @@ +import pathlib + import pytest -from ngen.cal.ngen import NgenExplicit, NgenIndependent, NgenUniform, NgenStrategy +import pydantic +from ngen.cal.ngen import ( + Ngen, + NgenBase, + NgenExplicit, + NgenIndependent, + NgenStrategy, + NgenUniform, +) def test_NgenExplicit_strategy_default_value(): @@ -13,7 +23,20 @@ def test_NgenIndependent_strategy_default_value(): o = NgenIndependent.construct() assert o.strategy == NgenStrategy.independent + def test_NgenUniform_strategy_default_value(): # construct object without validation. o = NgenUniform.construct() assert o.strategy == NgenStrategy.uniform + + +def test_NgenBase_verify_realization(ngen_config: Ngen): + # session level pytest fixture. take deep copy to avoid pollution + config = ngen_config.__root__.copy(deep=True) + assert isinstance(config, NgenBase) + + assert config.ngen_realization is not None, "should have already raised if not None" + config.ngen_realization.output_root = pathlib.Path("./output_root") + + with pytest.raises(pydantic.ValidationError): + Ngen.parse_obj(dict(config))