Skip to content

Commit

Permalink
Fix pydantic v2 validators
Browse files Browse the repository at this point in the history
  • Loading branch information
dladrichem committed Feb 7, 2024
1 parent 4cec458 commit 8bf55c8
Show file tree
Hide file tree
Showing 5 changed files with 387 additions and 189 deletions.
11 changes: 7 additions & 4 deletions flood_adapt/object_model/hazard/measure/green_infrastructure.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,10 @@
IGreenInfrastructure,
)
from flood_adapt.object_model.interface.site import ISite
from flood_adapt.object_model.io.unitfulvalue import UnitfulArea, UnitfulLength
from flood_adapt.object_model.io.unitfulvalue import (
UnitfulArea,
UnitfulHeight,
)


class GreenInfrastructure(HazardMeasure, IGreenInfrastructure):
Expand Down Expand Up @@ -55,7 +58,7 @@ def save(self, filepath: Union[str, os.PathLike]):
@staticmethod
def calculate_volume(
area: UnitfulArea,
height: UnitfulLength = UnitfulLength(value=0.0, units="meters"),
height: UnitfulHeight,
percent_area: float = 100.0,
) -> float:
"""Determine volume from area of the polygon and infiltration height
Expand All @@ -64,8 +67,8 @@ def calculate_volume(
----------
area : UnitfulArea
Area of polygon with units (calculated using calculate_polygon_area)
height : UnitfulLength, optional
Water height with units, by default 0.0
height : UnitfulHeight
Water height with units
percent_area : float, optional
Percentage area covered by green infrastructure [%], by default 100.0
Expand Down
132 changes: 53 additions & 79 deletions flood_adapt/object_model/interface/measures.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,14 @@
from enum import Enum
from typing import Any, Optional, Union

from pydantic import BaseModel, validator
from pydantic import BaseModel, Field, field_validator, model_validator, validator

from flood_adapt.object_model.io.unitfulvalue import (
UnitfulDischarge,
UnitfulHeight,
UnitfulLength,
UnitfulLengthRefValue,
UnitfulVolume,
UnitTypesVolume,
)


Expand Down Expand Up @@ -50,16 +50,10 @@ class SelectionType(str, Enum):
class MeasureModel(BaseModel):
"""BaseModel describing the expected variables and data types of attributes common to all measures"""

name: str
name: str = Field(..., min_length=1)
description: Optional[str] = ""
type: Union[HazardType, ImpactType]

@validator("name")
def validate_name(cls, name: str) -> str:
if len(name) < 1:
raise ValueError("Name cannot be empty")
return name


class HazardMeasureModel(MeasureModel):
"""BaseModel describing the expected variables and data types of attributes common to all impact measures"""
Expand All @@ -68,19 +62,24 @@ class HazardMeasureModel(MeasureModel):
selection_type: SelectionType
polygon_file: Optional[str] = None

@validator("polygon_file", always=True)
def validate_polygon_file(
cls, polygon_file: Optional[str], values: Any
) -> Optional[str]:
@field_validator("polygon_file")
@classmethod
def validate_polygon_file(cls, v: Optional[str]) -> Optional[str]:
if len(v) == 0:
raise ValueError("Polygon file path cannot be empty")
return v

@model_validator(mode="after")
def validate_selection_type(self) -> "HazardMeasureModel":
if (
values.get("selection_type")
self.selection_type
not in [SelectionType.aggregation_area, SelectionType.all]
and polygon_file is None
and self.polygon_file is None
):
raise ValueError(
"If `selection_type` is not 'aggregation_area' or 'all', then `polygon_file` needs to be set."
)
return polygon_file
return self


class ImpactMeasureModel(MeasureModel):
Expand Down Expand Up @@ -156,14 +155,15 @@ class PumpModel(HazardMeasureModel):
class GreenInfrastructureModel(HazardMeasureModel):
"""BaseModel describing the expected variables and data types of the "green infrastructure" hazard measure"""

volume: UnitfulVolume = UnitfulVolume(value=0.0, units=UnitTypesVolume.m3)
height: Optional[UnitfulLength] = None
volume: UnitfulVolume
height: Optional[UnitfulHeight] = None
aggregation_area_type: Optional[str] = None
aggregation_area_name: Optional[str] = None
percent_area: Optional[float] = None
percent_area: Optional[float] = Field(None, ge=0, le=100)

@validator("type", always=True)
def validate_type(cls, type: HazardType, values: Any) -> HazardType:
@field_validator("type")
@classmethod
def validate_type(cls, type: HazardType) -> HazardType:
if type not in [
HazardType.water_square,
HazardType.greening,
Expand All @@ -174,70 +174,44 @@ def validate_type(cls, type: HazardType, values: Any) -> HazardType:
)
return type

@validator("volume")
def validate_volume(cls, volume: UnitfulVolume, values: Any) -> UnitfulVolume:
if volume.value <= 0:
raise ValueError("Volume cannot be zero or negative")
return volume

@validator("height", always=True)
def validate_height(
cls, height: Optional[UnitfulLength], values: Any
) -> Optional[UnitfulLength]:
if values.get("type", "") == HazardType.total_storage:
if height is not None:
raise ValueError("Height cannot be set for total storage type measures")
return None # Height is not required for total storage type measures
elif not isinstance(height, UnitfulLength):
raise ValueError("Height must be a UnitfulLength")
elif height.value <= 0:
raise ValueError("Height cannot be zero or negative")
return height

@validator("percent_area", always=True)
def validate_percent_area(
cls, percent_area: Optional[float], values: Any
) -> Optional[float]:
if values.get("type", "") in [
HazardType.total_storage,
HazardType.water_square,
]:
if percent_area is not None:
@model_validator(mode="after")
def validate_hazard_type_values(self) -> "GreenInfrastructureModel":
if self.type == HazardType.total_storage:
if self.height is not None or self.percent_area is not None:
raise ValueError(
"Percent area cannot be set for total storage or water square type measures"
"Height and percent_area cannot be set for total storage type measures"
)
return None # Percent area is not required for total storage type measures
elif not isinstance(percent_area, float):
raise ValueError("Percent area must be a float")
elif percent_area < 0 or percent_area > 100:
raise ValueError("Percent area must be between 0 and 100")
return percent_area

@validator("aggregation_area_name", always=True)
def validate_aggregation_area_name(
cls, aggregation_area_name: Optional[str], values: Any
) -> Optional[str]:
if (
values.get("selection_type", "") == SelectionType.aggregation_area
and aggregation_area_name is None
return self
elif self.type == HazardType.water_square:
if self.percent_area is not None:
raise ValueError(
"Percentage_area cannot be set for water square type measures"
)
elif not isinstance(self.height, UnitfulHeight):
raise ValueError(
"Height needs to be set for water square type measures"
)
return self
elif not isinstance(self.height, UnitfulHeight) or not isinstance(
self.percent_area, float
):
raise ValueError(
"If `selection_type` is 'aggregation_area', then `aggregation_area_name` needs to be set."
"Height and percent_area needs to be set for greening type measures"
)
return aggregation_area_name
return self

@validator("aggregation_area_type", always=True)
def validate_aggregation_area_type(
cls, aggregation_area_type: Optional[str], values: Any
) -> Optional[str]:
if (
values.get("selection_type", "") == SelectionType.aggregation_area
and aggregation_area_type is None
):
raise ValueError(
"If `selection_type` is 'aggregation_area', then `aggregation_area_type` needs to be set."
)
return aggregation_area_type
@model_validator(mode="after")
def validate_selection_type_values(self) -> "GreenInfrastructureModel":
if self.selection_type == SelectionType.aggregation_area:
if self.aggregation_area_name is None:
raise ValueError(
"If `selection_type` is 'aggregation_area', then `aggregation_area_name` needs to be set."
)
if self.aggregation_area_type is None:
raise ValueError(
"If `selection_type` is 'aggregation_area', then `aggregation_area_type` needs to be set."
)
return self


class IMeasure(ABC):
Expand Down
23 changes: 17 additions & 6 deletions flood_adapt/object_model/io/unitfulvalue.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from enum import Enum

from pydantic import BaseModel
from pydantic import BaseModel, Field, root_validator


class UnitTypesLength(str, Enum):
Expand Down Expand Up @@ -100,15 +100,26 @@ def convert(self, new_units: UnitTypesLength) -> float:
new_conversion = 3.28084
elif new_units == "inch":
new_conversion = 1.0 / 0.0254
elif self.units == "miles":
elif new_units == "miles":
new_conversion = 1.0 / 1609.344
else:
ValueError("Invalid length units")
return conversion * new_conversion * self.value


class UnitfulHeight(UnitfulLength):
"""A special type of length that is always positive and non-zero. Used for heights."""
value: float = Field(..., gt=0)

@root_validator(pre=True)
def convert_length_to_height(cls, obj):
if isinstance(obj, UnitfulLength):
return UnitfulHeight(value=obj.value, units=obj.units)
return obj


class UnitfulArea(ValueUnitPair):
value: float
value: float = Field(..., gt=0)
units: UnitTypesArea

def convert(self, new_units: UnitTypesArea) -> float:
Expand All @@ -127,7 +138,7 @@ def convert(self, new_units: UnitTypesArea) -> float:
# first, convert to meters
if self.units == "cm2":
conversion = 1.0 / 10000 # meters
if self.units == "mm2":
elif self.units == "mm2":
conversion = 1.0 / 1000000 # meters
elif self.units == "m2":
conversion = 1.0 # meters
Expand All @@ -139,7 +150,7 @@ def convert(self, new_units: UnitTypesArea) -> float:
# second, convert to new units
if new_units == "cm2":
new_conversion = 10000.0
if new_units == "mm2":
elif new_units == "mm2":
new_conversion = 1000000.0
elif new_units == "m2":
new_conversion = 1.0
Expand Down Expand Up @@ -267,7 +278,7 @@ def convert(self, new_units: UnitTypesIntensity) -> float:


class UnitfulVolume(ValueUnitPair):
value: float
value: float = Field(..., gt=0)
units: UnitTypesVolume

def convert(self, new_units: UnitTypesVolume) -> float:
Expand Down
Loading

0 comments on commit 8bf55c8

Please sign in to comment.