create TimToForcingConverter class and tests

hydrolib/tools/ext_old_to_new/converters.py

@@ -3,7 +3,7 @@
 from typing import Any, Dict, List, Union
 
 from hydrolib.core.basemodel import DiskOnlyFileModel
-from hydrolib.core.dflowfm.bc.models import ForcingModel
+from hydrolib.core.dflowfm.bc.models import ForcingModel, TimeSeries
 from hydrolib.core.dflowfm.ext.models import (
     SOURCE_SINKS_QUANTITIES_VALID_PREFIXES,
     Boundary,
@@ -257,11 +257,11 @@ def get_time_series_data(tim_model: TimModel) -> Dict[str, List[float]]:
             (excluding the first column(time)).
 
         Examples:
-            >>> tim_file = Path("tests/data/external_forcings/initial_waterlevel.tim")
+            >>> tim_file = Path("tests/data/input/source-sink/leftsor.tim")
             >>> time_file = TimParser.parse(tim_file)
             >>> tim_model = TimModel(**time_file)
             >>> time_series = SourceSinkConverter().get_time_series_data(tim_model)
-            >>> print(time_series)
+            >>> print(time_series) # doctest: +SKIP
             {
                 1: [1.0, 1.0, 3.0, 5.0, 8.0],
                 2: [2.0, 2.0, 5.0, 8.0, 10.0],
@@ -376,7 +376,7 @@ def parse_tim_model(
 
             >>> converter = SourceSinkConverter()
             >>> time_series = converter.parse_tim_model(tim_file, ext_file_quantity_list)
-            >>> print(time_series)
+            >>> print(time_series) # doctest: +SKIP
             {
                 "discharge": [1.0, 1.0, 1.0, 1.0, 1.0],
                 "salinitydelta": [2.0, 2.0, 2.0, 2.0, 2.0],
@@ -466,7 +466,9 @@ def convert(
             **temp_salinity_mdu:
                 keyword arguments that will be provided if you want to provide the temperature and salinity details from
                 the mdu file, the dictionary will have two keys `temperature`, `salinity` and the values are only bool.
-                >>> {'salinity': True, 'temperature': True}
+                ```python
+                {'salinity': True, 'temperature': True}
+                ```
 
         Returns:
             SourceSink: A SourceSink object that represents the converted forcing
@@ -561,3 +563,76 @@ def contains(quantity_class, quantity) -> bool:
             return False
 
         return True
+
+
+class TimToForcingConverter:
+    """
+    A class to convert TimModel data into ForcingModel data for boundary condition definitions.
+
+    The class provides a static method `convert` to convert a TimModel object into a ForcingModel object.
+
+    The method requires the following arguments:
+    - `tim_model`: A TimModel object containing the time series data.
+    - `start_time`: The reference time for the forcing data.
+    - `time_interpolation`: The time interpolation method for the forcing data.
+    - `units`: A list of units corresponding to the forcing quantities.
+    - `user_defined_names`: A list of user-defined names for the forcing blocks.
+    """
+
+    @staticmethod
+    def convert(
+        tim_model: TimModel,
+        start_time: str,
+        time_interpolation: str = "linear",
+        units: List[str] = None,
+        user_defined_names: List[str] = None,
+    ) -> ForcingModel:
+        """
+        Convert a TimModel into a ForcingModel.
+
+        Args:
+            tim_model (TimModel):
+                The input TimModel to be converted.
+            start_time (str):
+                The reference time for the forcing data.
+            time_interpolation (str, optional):
+                The time interpolation method for the forcing data. Defaults to "linear".
+            units (List[str], optional):
+                A list of units corresponding to the forcing quantities.
+            user_defined_names (List[str], optional):
+                A list of user-defined names for the forcing blocks.
+
+        Returns:
+            ForcingModel: The converted ForcingModel.
+        """
+        if units is None or user_defined_names is None:
+            raise ValueError("Both 'units' and 'user_defined_names' must be provided.")
+
+        if len(units) != len(user_defined_names):
+            raise ValueError(
+                "The lengths of 'units' and 'user_defined_names' must match."
+            )
+
+        if len(tim_model.timeseries) != len(user_defined_names):
+            raise ValueError(
+                "The number of timeseries in TimModel must match the number of user-defined names."
+            )
+
+        df = tim_model.as_dataframe()
+        time_data = df.index.tolist()
+        forcings_list = []
+
+        for i, (column, vals) in enumerate(df.items()):
+            unit = units[i]
+            forcing = TimeSeries(
+                name=user_defined_names[i],
+                function="timeseries",
+                timeinterpolation=time_interpolation,
+                quantity=["time", column],
+                unit=[start_time, unit],
+                datablock=[time_data, vals.values.tolist()],
+            )
+            forcings_list.append(forcing)
+
+        forcing_model = ForcingModel(forcing=forcings_list)
+        return forcing_model

tests/tools/test_tim_to_bc.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+from hydrolib.core.dflowfm.tim.models import TimModel
+from hydrolib.tools.ext_old_to_new.converters import TimToForcingConverter
+
+
+def test_tim_to_bc_converter(input_files_dir: Path):
+    filepath = input_files_dir / "source-sink/tim-5-columns.tim"
+    user_defined_names = [
+        "any-name-1",
+        "any-name-2",
+        "any-name-3",
+        "any-name-4",
+        "any-name-5",
+    ]
+    tim_model = TimModel(filepath, user_defined_names)
+
+    units = ["m³/s", "m", "C", "ppt", "check-later"]
+    start_time = "minutes since 2015-01-01 00:00:00"
+    df = tim_model.as_dataframe()
+    converter = TimToForcingConverter()
+    forcing_model = converter.convert(
+        tim_model=tim_model,
+        start_time=start_time,
+        units=units,
+        user_defined_names=user_defined_names,
+    )
+
+    assert len(forcing_model.forcing) == 5
+    assert [forcing_model.forcing[i].name for i in range(5)] == user_defined_names
+    assert [
+        forcing_model.forcing[i].quantityunitpair[1].unit for i in range(5)
+    ] == units
+    forcing = forcing_model.forcing[0]
+    assert forcing.datablock[0] == df.index.tolist()
+    assert forcing.datablock[1] == df.iloc[:, 0].tolist()