feat: sources sinks converter (UNST-8528)

hydrolib/core/dflowfm/ext/models.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Dict, List, Literal, Optional, Union
+from typing import Dict, List, Literal, Optional, Set, Union
 
 from pydantic.v1 import Field, root_validator, validator
 from strenum import StrEnum
@@ -24,6 +24,13 @@
 from hydrolib.core.dflowfm.tim.models import TimModel
 from hydrolib.core.utils import str_is_empty_or_none
 
+SOURCE_SINKS_QUANTITIES_VALID_PREFIXES = (
+    "initialtracer",
+    "tracerbnd",
+    "sedfracbnd",
+    "initialsedfrac",
+)
+
 
 class Boundary(INIBasedModel):
     """
@@ -181,6 +188,65 @@ def validate_location_type(cls, v: str) -> str:
         return v
 
 
+class SourceSink(INIBasedModel):
+    """
+    A `[SourceSink]` block for use inside an external forcings file,
+    i.e., a [ExtModel][hydrolib.core.dflowfm.ext.models.SourceSink].
+
+    All lowercased attributes match with the source-sink input as described in
+    [UM Sec.C.5.2.4](https://content.oss.deltares.nl/delft3dfm1d2d/D-Flow_FM_User_Manual_1D2D.pdf#subsection.C.5.2.4).
+    """
+
+    _header: Literal["SourceSink"] = "SourceSink"
+    id: str = Field(alias="id")
+    name: str = Field("", alias="name")
+    locationfile: DiskOnlyFileModel = Field(
+        default_factory=lambda: DiskOnlyFileModel(None), alias="locationFile"
+    )
+
+    numcoordinates: Optional[int] = Field(alias="numCoordinates")
+    xcoordinates: Optional[List[float]] = Field(alias="xCoordinates")
+    ycoordinates: Optional[List[float]] = Field(alias="yCoordinates")
+
+    zsource: Optional[Union[float, List[float]]] = Field(alias="zSource")
+    zsink: Optional[Union[float, List[float]]] = Field(alias="zSink")
+    discharge: Union[float, List[float]] = Field(alias="discharge")
+    area: Optional[float] = Field(alias="Area")
+
+    salinitydelta: Optional[Union[List[float], float]] = Field(alias="SalinityDelta")
+    temperaturedelta: Optional[Union[List[float], float]] = Field(
+        alias="TemperatureDelta"
+    )
+
+    @classmethod
+    def _exclude_from_validation(cls, input_data: Optional[dict] = None) -> Set:
+        fields = cls.__fields__
+        unknown_keywords = [
+            key
+            for key in input_data.keys()
+            if key not in fields
+            and key.startswith(SOURCE_SINKS_QUANTITIES_VALID_PREFIXES)
+        ]
+        return set(unknown_keywords)
+
+    class Config:
+        """
+        Config class to tell Pydantic to accept fields not explicitly declared in the model.
+        """
+
+        # Allow dynamic fields
+        extra = "allow"
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Add dynamic attributes for fields starting with 'tracer'
+        for key, value in data.items():
+            if isinstance(key, str) and key.startswith(
+                SOURCE_SINKS_QUANTITIES_VALID_PREFIXES
+            ):
+                setattr(self, key, value)
+
+
 class MeteoForcingFileType(StrEnum):
     """
     Enum class containing the valid values for the forcingFileType
@@ -318,7 +384,7 @@ def is_intermediate_link(self) -> bool:
 
 
 class ExtGeneral(INIGeneral):
-    """The external forcing file's `[General]` section with file meta data."""
+    """The external forcing file's `[General]` section with file meta-data."""
 
     _header: Literal["General"] = "General"
     fileversion: str = Field("2.01", alias="fileVersion")
@@ -335,12 +401,14 @@ class ExtModel(INIModel):
         general (ExtGeneral): `[General]` block with file metadata.
         boundary (List[Boundary]): List of `[Boundary]` blocks for all boundary conditions.
         lateral (List[Lateral]): List of `[Lateral]` blocks for all lateral discharges.
+        source_sink (List[SourceSink]): List of `[SourceSink]` blocks for all source/sink terms.
         meteo (List[Meteo]): List of `[Meteo]` blocks for all meteorological forcings.
     """
 
     general: ExtGeneral = ExtGeneral()
     boundary: List[Boundary] = Field(default_factory=list)
     lateral: List[Lateral] = Field(default_factory=list)
+    source_sink: List[SourceSink] = Field(default_factory=list)
     meteo: List[Meteo] = Field(default_factory=list)
     serializer_config: INISerializerConfig = INISerializerConfig(
         section_indent=0, property_indent=0

hydrolib/core/dflowfm/extold/models.py

@@ -210,8 +210,6 @@ class ExtOldMeteoQuantity(StrEnum):
     """Long wave radiation"""
     SolarRadiation = "solarradiation"
     """Solar radiation"""
-    DischargeSalinityTemperatureSorSin = "discharge_salinity_temperature_sorsin"
-    """Discharge, salinity temperature source-sinks"""
     NudgeSalinityTemperature = "nudge_salinity_temperature"
     """Nudging salinity and temperature"""
     AirPressure = "airpressure"
@@ -302,6 +300,12 @@ def _missing_(cls, value):
             )
 
 
+class ExtOldSourcesSinks(StrEnum):
+    """Source and sink quantities"""
+
+    DischargeSalinityTemperatureSorSin = "discharge_salinity_temperature_sorsin"
+
+
 class ExtOldQuantity(StrEnum):
     """Enum class containing the valid values for the boundary conditions category
     of the external forcings.
@@ -789,3 +793,8 @@ def _get_serializer(
     @classmethod
     def _get_parser(cls) -> Callable[[Path], Dict]:
         return Parser.parse
+
+    @property
+    def quantities(self) -> List[str]:
+        """List all the quantities in the external forcings file."""
+        return [forcing.quantity for forcing in self.forcing]

hydrolib/core/dflowfm/ini/models.py

@@ -28,15 +28,22 @@
     ModelSaveSettings,
     ParsableFileModel,
 )
-
-from ..ini.io_models import CommentBlock, Document, Property, Section
-from .parser import Parser
-from .serializer import (
+from hydrolib.core.dflowfm.ini.io_models import (
+    CommentBlock,
+    Document,
+    Property,
+    Section,
+)
+from hydrolib.core.dflowfm.ini.parser import Parser
+from hydrolib.core.dflowfm.ini.serializer import (
     DataBlockINIBasedSerializerConfig,
     INISerializerConfig,
     write_ini,
 )
-from .util import UnknownKeywordErrorManager, make_list_validator
+from hydrolib.core.dflowfm.ini.util import (
+    UnknownKeywordErrorManager,
+    make_list_validator,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -121,12 +128,13 @@ class Config:
     @root_validator(pre=True)
     def _validate_unknown_keywords(cls, values):
         unknown_keyword_error_manager = cls._get_unknown_keyword_error_manager()
+        do_not_validate = cls._exclude_from_validation(values)
         if unknown_keyword_error_manager:
             unknown_keyword_error_manager.raise_error_for_unknown_keywords(
                 values,
                 cls._header,
                 cls.__fields__,
-                cls._exclude_fields(),
+                cls._exclude_fields() | do_not_validate,
             )
         return values
 
@@ -181,6 +189,11 @@ def validate(cls: Type["INIBasedModel"], value: Any) -> "INIBasedModel":
 
         return super().validate(value)
 
+    @classmethod
+    def _exclude_from_validation(cls, input_data: Optional = None) -> Set:
+        """Fields that should not be checked when validating existing fields as they will be dynamically added."""
+        return set()
+
     @classmethod
     def _exclude_fields(cls) -> Set:
         return {"comments", "datablock", "_header"}

hydrolib/core/dflowfm/ini/util.py

@@ -189,7 +189,7 @@ def validate_forbidden_fields(
         return values
 
     for field in field_names:
-        if values.get(field) != None:
+        if values.get(field) is not None:
             raise ValueError(
                 f"{field} is forbidden when {conditional_field_name} {operator_str(comparison_func)} {conditional_value}"
             )
@@ -228,7 +228,7 @@ def validate_required_fields(
         return values
 
     for field in field_names:
-        if values.get(field) == None:
+        if values.get(field) is None:
             raise ValueError(
                 f"{field} should be provided when {conditional_field_name} {operator_str(comparison_func)} {conditional_value}"
             )
@@ -654,28 +654,54 @@ def raise_error_for_unknown_keywords(
         """
         unknown_keywords = self._get_all_unknown_keywords(data, fields, excluded_fields)
 
-        if len(unknown_keywords) == 0:
-            return
-
-        raise ValueError(
-            f"Unknown keywords are detected in section: '{section_header}', '{unknown_keywords}'"
-        )
+        if len(unknown_keywords) > 0:
+            raise ValueError(
+                f"Unknown keywords are detected in section: '{section_header}', '{unknown_keywords}'"
+            )
 
     def _get_all_unknown_keywords(
         self, data: Dict[str, Any], fields: Dict[str, ModelField], excluded_fields: Set
     ) -> List[str]:
+        """
+        Get all unknown keywords in the data.
+
+        Args:
+            data: Dict[str, Any]: Input data containing all properties which are checked on unknown keywords.
+            fields: Dict[str, ModelField]: Known fields of the Model.
+            excluded_fields: Set[str]: Fields which should be excluded from the check for unknown keywords.
+
+        Returns:
+            List[str]: List of unknown keywords.
+        """
         list_of_unknown_keywords = []
-        for name in data:
-            if self._is_unknown_keyword(name, fields, excluded_fields):
-                list_of_unknown_keywords.append(name)
+        for keyword in data:
+            if self._is_unknown_keyword(keyword, fields, excluded_fields):
+                list_of_unknown_keywords.append(keyword)
 
         return list_of_unknown_keywords
 
+    @staticmethod
     def _is_unknown_keyword(
-        self, name: str, fields: Dict[str, ModelField], excluded_fields: Set
+        keyword: str, fields: Dict[str, ModelField], excluded_fields: Set
     ):
-        for model_field in fields.values():
-            if name == model_field.name or name == model_field.alias:
-                return False
+        """
+        Check if the given field name equals to any of the model field names or aliases, if not, the function checks if
+        the field is not in the excluded_fields parameter.
+
+        Args:
+            keyword: str: Name of the field.
+            fields: Dict[str, ModelField]: Known fields of the Model.
+            excluded_fields: Set[str]: Fields which should be excluded from the check for unknown keywords.
+
+        Returns:
+            bool: True if the field is unknown (not a field name or alias and and not in the exclude list),
+            False otherwise
+        """
+        exists = any(
+            keyword == model_field.name or keyword == model_field.alias
+            for model_field in fields.values()
+        )
+        # the field is not in the known fields, check if it should be excluded
+        unknown = not exists and keyword not in excluded_fields
 
-        return name not in excluded_fields
+        return unknown

hydrolib/core/dflowfm/polyfile/models.py

@@ -1,7 +1,7 @@
 """models.py defines all classes and functions related to representing pol/pli(z) files.
 """
 
-from typing import Callable, List, Optional, Sequence
+from typing import Callable, List, Optional, Sequence, Tuple
 
 from pydantic.v1 import Field
 
@@ -80,7 +80,30 @@ class PolyObject(BaseModel):
 
 
 class PolyFile(ParsableFileModel):
-    """Poly-file (.pol/.pli/.pliz) representation."""
+    """
+    Poly-file (.pol/.pli/.pliz) representation.
+
+    Notes:
+        - The `has_z_values` attribute is used to determine if the PolyFile contains z-values.
+        - The `has_z_values` is false by default and should be set to true if the PolyFile path ends with `.pliz`.
+        - The `***.pliz` file should have a 2*3 structure, where the third column contains the z-values, otherwise
+        (the parser will give an error).
+        - If there is a label in the file, the parser will ignore the label and read the file as a normal polyline file.
+        ```
+        tfl_01
+            2 2
+            0.00 1.00 #zee
+            0.00 2.00 #zee
+        ```
+        - if the file is .pliz, and the dimensions are 2*5 the first three columns will be considered as x, y, z values
+        and the last two columns will be considered as data values.
+        ```
+        L1
+            2 5
+            63.35 12.95 -4.20 -5.35 0
+            45.20 6.35 -3.00 -2.90 0
+        ```
+    """
 
     has_z_values: bool = False
     objects: Sequence[PolyObject] = Field(default_factory=list)
@@ -106,7 +129,63 @@ def _get_serializer(cls) -> Callable:
 
     @classmethod
     def _get_parser(cls) -> Callable:
-        # TODO Prevent circular dependency in Parser
+        # Prevent circular dependency in Parser
         from .parser import read_polyfile
 
         return read_polyfile
+
+    @property
+    def x(self) -> List[float]:
+        """X-coordinates of all points in the PolyFile."""
+        return [point.x for obj in self.objects for point in obj.points]
+
+    @property
+    def y(self) -> List[float]:
+        """Y-coordinates of all points in the PolyFile."""
+        return [point.y for obj in self.objects for point in obj.points]
+
+    def get_z_sources_sinks(self) -> Tuple[List[float], List[float]]:
+        """
+        Get the z values of the source and sink points from the polyline file.
+
+        Returns:
+            z_source, z_sinkA: Tuple[List[float]]:
+            If the polyline has data (more than 3 columns), then both the z_source and z_sink will be a list of two values.
+            Otherwise, the z_source and the z_sink will be a single value each.
+
+        Note:
+             - calling this method on a polyline file that does not have z-values will return a list of None.
+
+        Examples:
+        in case the polyline has 3 columns:
+            >>> polyline = PolyFile("tests/data/input/source-sink/leftsor.pliz")
+            >>> z_source, z_sink = polyline.get_z_sources_sinks()
+            >>> print(z_source, z_sink)
+            [-3] [-4.2]
+
+        in case the polyline has more than 3 columns:
+            >>> polyline = PolyFile("tests/data/input/source-sink/leftsor-5-columns.pliz") #Doctest: +SKIP
+            >>> z_source, z_sink = polyline.get_z_sources_sinks()
+            >>> print(z_source, z_sink)
+            [-3, -2.9] [-4.2, -5.35]
+
+        in case the polyline does not have z-values:
+            >>> root_dir = "tests/data/input/dflowfm_individual_files/polylines"
+            >>> polyline = PolyFile(f"{root_dir}/boundary-polyline-no-z-no-label.pli")
+            >>> z_source, z_sink = polyline.get_z_sources_sinks()
+            >>> print(z_source, z_sink)
+            [None] [None]
+        """
+        has_data = True if self.objects[0].points[0].data else False
+
+        z_source_sink = []
+        for elem in [0, -1]:
+            point = self.objects[0].points[elem]
+            if has_data:
+                z_source_sink.append([point.z, point.data[0]])
+            else:
+                z_source_sink.append([point.z])
+
+        z_sink: list[float | None] = z_source_sink[0]
+        z_source: list[float | None] = z_source_sink[1]
+        return z_source, z_sink