Skip to content

Timeseries .tim files

.tim files contain time series data of a D-Flow FM model. The support of .tim files for boundary time series will be discontinued. Instead, these will be replaced by the *.bc file.

They are represented by the classes below.

Model

TimModel

Bases: ParsableFileModel

Class representing a tim (*.tim) file.

Attributes:

Name Type Description
serializer_config TimSerializerConfig

Configuration for serialization of the .tim file.

comments List[str]

Header comments from the .tim file.

timeseries List[TimRecord]

A list of TimRecord objects, each containing a time value and associated data.

quantities_names Optional[List[str]]

List of names for the quantities in the timeseries.

Methods:

Name Description
_ext

Returns the file extension for .tim files.

_filename

Returns the default filename for .tim files.

_get_serializer

Returns the serializer callable for .tim files.

_get_parser

Returns the parser callable for .tim files.

_validate_timeseries_values

List[TimRecord]) -> List[TimRecord]: Validates the timeseries data.

as_dataframe

List[Any] = None) -> DataFrame: Returns the timeseries as a pandas DataFrame.

_validate_quantities_names

Validates that quantities_names match the values or each record.

Parameters:

Name Type Description Default
filepath Path

Path to the .tim file.

None
data Dict

Parsed data containing comments and timeseries.

required
serializer_config TimSerializerConfig

Configuration for serializing the .tim file.

required

Returns:

Type Description

List[TimRecord]: Validated list of TimRecord objects.

Raises:

Type Description
ValueError

If the timeseries has inconsistent column counts or duplicate time values.

Examples:

Create a TimModel object from a .tim file:

```python
>>> from hydrolib.core.dflowfm.tim.models import TimModel, TimRecord
>>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim")
>>> print(tim_model.timeseries)
[TimRecord(time=10.0, data=[1.232, 2.343, 3.454]), TimRecord(time=20.0, data=[4.565, 5.676, 6.787]), TimRecord(time=30.0, data=[1.5, 2.6, 3.7])]

```

Provide names for the quantities in the timeseries:

>>> quantities_names = ["discharge", "waterlevel", "salinity", "temperature", "initialtracer"]
>>> tim_model = TimModel(filepath="tests/data/input/source-sink/tim-5-columns.tim", quantities_names=quantities_names)
>>> print(tim_model.quantities_names)
['discharge', 'waterlevel', 'salinity', 'temperature', 'initialtracer']
>>> print(tim_model.as_dataframe())
       discharge  waterlevel  salinity  temperature  initialtracer
0.0          1.0         2.0       3.0          4.0            5.0
100.0        1.0         2.0       3.0          4.0            5.0
200.0        1.0         2.0       3.0          4.0            5.0
300.0        1.0         2.0       3.0          4.0            5.0
400.0        1.0         2.0       3.0          4.0            5.0

Create a TimModel object from a dictionary:

>>> data = {
...     "comments": ["# Example comment"],
...     "timeseries": [TimRecord(time=0.0, data=[1.0, 2.0])]
... }
>>> tim_model = TimModel(**data)
>>> print(tim_model.timeseries)
[TimRecord(time=0.0, data=[1.0, 2.0])]

Create TimModel from TimRecord objects:

>>> new_tim = TimModel()
>>> new_tim.comments = ["# Example comment"]
>>> new_tim.timeseries = [TimRecord(time=0.0, data=[1.0, 2.0])]

Serialize the TimModel to a .tim file:

>>> new_tim.save(filepath=Path("output.tim")) # doctest: +SKIP

See Also

TimParser: Used for parsing .tim files. TimSerializer: Used for serializing .tim files. TimRecord: Represents individual time and data entries in the timeseries.

Notes

This class ensures the integrity of the timeseries by validating data consistency and detecting duplicate time entries.

References
  • TIM file format <https://content.oss.deltares.nl/delft3dfm1d2d/D-Flow_FM_User_Manual_1D2D.pdf#C4>_
Source code in hydrolib\core\dflowfm\tim\models.py
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
class TimModel(ParsableFileModel):
    """Class representing a tim (*.tim) file.

    Attributes:
        serializer_config (TimSerializerConfig):
            Configuration for serialization of the .tim file.
        comments (List[str]):
            Header comments from the .tim file.
        timeseries (List[TimRecord]):
            A list of TimRecord objects, each containing a time value and associated data.
        quantities_names (Optional[List[str]]):
            List of names for the quantities in the timeseries.


    Methods:
        _ext() -> str:
            Returns the file extension for .tim files.
        _filename() -> str:
            Returns the default filename for .tim files.
        _get_serializer() -> Callable:
            Returns the serializer callable for .tim files.
        _get_parser() -> Callable:
            Returns the parser callable for .tim files.
        _validate_timeseries_values(cls, v: List[TimRecord]) -> List[TimRecord]:
            Validates the timeseries data.
        as_dataframe(columns: List[Any] = None) -> DataFrame:
            Returns the timeseries as a pandas DataFrame.
        _validate_quantities_names(cls, v, values) -> List[str]:
            Validates that quantities_names match the values or each record.

    Args:
        filepath (Path):
            Path to the .tim file.
        data (Dict):
            Parsed data containing comments and timeseries.
        serializer_config (TimSerializerConfig):
            Configuration for serializing the .tim file.

    Returns:
        List[TimRecord]:
            Validated list of TimRecord objects.

    Raises:
        ValueError:
            If the timeseries has inconsistent column counts or duplicate time values.

    Examples:
        Create a TimModel object from a .tim file:

            ```python
            >>> from hydrolib.core.dflowfm.tim.models import TimModel, TimRecord
            >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim")
            >>> print(tim_model.timeseries)
            [TimRecord(time=10.0, data=[1.232, 2.343, 3.454]), TimRecord(time=20.0, data=[4.565, 5.676, 6.787]), TimRecord(time=30.0, data=[1.5, 2.6, 3.7])]

            ```

        Provide names for the quantities in the timeseries:
            ```python
            >>> quantities_names = ["discharge", "waterlevel", "salinity", "temperature", "initialtracer"]
            >>> tim_model = TimModel(filepath="tests/data/input/source-sink/tim-5-columns.tim", quantities_names=quantities_names)
            >>> print(tim_model.quantities_names)
            ['discharge', 'waterlevel', 'salinity', 'temperature', 'initialtracer']
            >>> print(tim_model.as_dataframe())
                   discharge  waterlevel  salinity  temperature  initialtracer
            0.0          1.0         2.0       3.0          4.0            5.0
            100.0        1.0         2.0       3.0          4.0            5.0
            200.0        1.0         2.0       3.0          4.0            5.0
            300.0        1.0         2.0       3.0          4.0            5.0
            400.0        1.0         2.0       3.0          4.0            5.0

            ```

        Create a `TimModel` object from a dictionary:
            ```python
            >>> data = {
            ...     "comments": ["# Example comment"],
            ...     "timeseries": [TimRecord(time=0.0, data=[1.0, 2.0])]
            ... }
            >>> tim_model = TimModel(**data)
            >>> print(tim_model.timeseries)
            [TimRecord(time=0.0, data=[1.0, 2.0])]

            ```

        Create `TimModel` from `TimRecord` objects:
            ```python
            >>> new_tim = TimModel()
            >>> new_tim.comments = ["# Example comment"]
            >>> new_tim.timeseries = [TimRecord(time=0.0, data=[1.0, 2.0])]

            ```

        Serialize the `TimModel` to a .tim file:
            ```python
            >>> new_tim.save(filepath=Path("output.tim")) # doctest: +SKIP

            ```

    See Also:
        TimParser: Used for parsing .tim files.
        TimSerializer: Used for serializing .tim files.
        TimRecord: Represents individual time and data entries in the timeseries.

    Notes:
        This class ensures the integrity of the timeseries by validating data consistency and detecting duplicate time entries.

    References:
        - `TIM file format <https://content.oss.deltares.nl/delft3dfm1d2d/D-Flow_FM_User_Manual_1D2D.pdf#C4>`_
    """

    serializer_config = TimSerializerConfig()
    """TimSerializerConfig: The serialization configuration for the tim file."""

    comments: List[str] = Field(default_factory=list)
    """List[str]: A list with the header comment of the tim file."""

    timeseries: List[TimRecord] = Field(default_factory=list)
    """List[TimRecord]: A list containing the timeseries."""

    quantities_names: Optional[List[str]] = Field(default=None)

    def __init__(
        self,
        filepath: Optional[Union[str, Path]] = None,
        quantities_names: Optional[List[str]] = None,
        **parsable_file_kwargs,
    ):
        """
        Custom initializer to handle extra parameters specific to TimModel.

        Args:
            quantities_names (Optional[List[str]]): Names for the quantities in the timeseries.
            *args, **kwargs: Other arguments for the superclass.
        """
        super().__init__(filepath=filepath, **parsable_file_kwargs)
        self.quantities_names = quantities_names

    @classmethod
    def _ext(cls) -> str:
        return ".tim"

    @classmethod
    def _filename(cls) -> str:
        return "timeseries"

    @classmethod
    def _get_serializer(
        cls,
    ) -> Callable[[Path, Dict, TimSerializerConfig, ModelSaveSettings], None]:
        return TimSerializer.serialize

    @classmethod
    def _get_parser(cls) -> Callable[[Path], Dict]:
        return TimParser.parse

    @validator("timeseries", pre=True, check_fields=True, allow_reuse=True)
    def replace_fortran_scientific_notation_for_floats(cls, value, field):
        for record in value:
            if isinstance(record, dict):
                record["time"] = FortranUtils.replace_fortran_scientific_notation(
                    record["time"]
                )
                record["data"] = FortranUtils.replace_fortran_scientific_notation(
                    record["data"]
                )
            elif isinstance(record, TimRecord):
                record.time = FortranUtils.replace_fortran_scientific_notation(
                    record.time
                )
                record.data = FortranUtils.replace_fortran_scientific_notation(
                    record.data
                )

        return value

    @validator("timeseries")
    @classmethod
    def _validate_timeseries_values(cls, v: List[TimRecord]) -> List[TimRecord]:
        """Validate if the number of columns per timeseries matches and if the timeseries have no duplicate times.

        Args:
            v (List[TimRecord]): Timeseries to validate.

        Raises:
            ValueError: When the number of columns for timeseries is zero.
            ValueError: When the number of columns differs per timeseries.
            ValueError: When the timeseries has a duplicate time.

        Returns:
            List[TimRecord]: Validated timeseries.
        """
        if len(v) == 0:
            return v

        cls._raise_error_if_amount_of_columns_differ(v)
        cls._raise_error_if_duplicate_time(v)

        return v

    @staticmethod
    def _raise_error_if_amount_of_columns_differ(timeseries: List[TimRecord]) -> None:
        n_columns = len(timeseries[0].data)

        if n_columns == 0:
            raise ValueError("Time series cannot be empty.")

        for timrecord in timeseries:
            if len(timrecord.data) != n_columns:
                raise ValueError(
                    f"Time {timrecord.time}: Expected {n_columns} columns, but was {len(timrecord.data)}"
                )

    @staticmethod
    def _raise_error_if_duplicate_time(timeseries: List[TimRecord]) -> None:
        seen_times = set()
        for timrecord in timeseries:
            if timrecord.time in seen_times:
                raise ValueError(
                    f"Timeseries cannot contain duplicate times. Time: {timrecord.time} is duplicate."
                )
            seen_times.add(timrecord.time)

    @validator("quantities_names")
    def _validate_quantities_names(cls, v, values):
        """Validate if the number of quantities_names matches the number of columns in the timeseries.

        The validator compared the amount of quantities_names with the number of columns in the first record of
        the timeseries.
        """
        if v is not None:
            first_records_data = values["timeseries"][0].data
            if len(v) != len(first_records_data):
                raise ValueError(
                    f"The number of quantities_names ({len(v)}) must match the number of columns in the Tim file ({len(first_records_data)})."
                )
        return v

    def add_column(self, new_values: List[float], column_name: str = None) -> None:
        """
        Add new values to each TimRecord in the timeseries, representing a new location.

        Args:
            new_values (List[float]): A list of new values to add, one for each TimRecord.
            column_name (str, optional): The name of the new column. Defaults to None.
                if None, the column is named as "quantity-{len(quantities_names) + 1}".

        Raises:
            ValueError: If the number of new values does not match the number of TimRecords.

        Examples:
            ```python
            >>> tim_model = TimModel(
            ...     timeseries=[
            ...         TimRecord(time=0.0, data=[1.0, 2.0]),
            ...         TimRecord(time=1.0, data=[3.0, 4.0]),
            ...     ]
            ... )
            >>> tim_model.add_column([5.0, 6.0])
            >>> print(tim_model.timeseries)
            [TimRecord(time=0.0, data=[1.0, 2.0, 5.0]), TimRecord(time=1.0, data=[3.0, 4.0, 6.0])]

            ```
        """
        if len(new_values) != len(self.timeseries):
            raise ValueError(
                f"Expected {len(self.timeseries)} values, but got {len(new_values)}."
            )

        for record, value in zip(self.timeseries, new_values):
            record.data.append(value)

        if self.quantities_names:
            if column_name is None:
                column_name = f"quantity-{len(self.quantities_names) + 1}"
            self.quantities_names.append(column_name)

    def as_dataframe(self, columns: List[Any] = None) -> DataFrame:
        """Return the timeseries as a pandas DataFrame.

        Args:
            columns (List[Any, str], optional, Defaults to None):
                The column names for the DataFrame.

        Returns:
            DataFrame: The timeseries as a pandas DataFrame.

        Notes:
            - If the columns are not provided, the quantities_names will be used as column names.
            - If the quantities_names are not provided, the columns will be named as 0, 1, 2, etc.

        Examples:
            Create a `TimModel` object from a .tim file:
                >>> from hydrolib.core.dflowfm.tim.models import TimModel
                >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim")
                >>> df = tim_model.as_dataframe()
                >>> print(df)
                          0      1      2
                10.0  1.232  2.343  3.454
                20.0  4.565  5.676  6.787
                30.0  1.500  2.600  3.700

            If the `TimModel` object was created with quantities names:
                >>> quantities_names = ["Column1", "Column2", "Column3"]
                >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim", quantities_names=quantities_names)
                >>> df = tim_model.as_dataframe()
                >>> print(df)
                      Column1  Column2  Column3
                10.0    1.232    2.343    3.454
                20.0    4.565    5.676    6.787
                30.0    1.500    2.600    3.700

            To add column names to the DataFrame after you have created the `TimModel` object:
                >>> df = tim_model.as_dataframe(columns=["Column1", "Column2", "Column3"])
                >>> print(df)
                      Column1  Column2  Column3
                10.0    1.232    2.343    3.454
                20.0    4.565    5.676    6.787
                30.0    1.500    2.600    3.700
        """
        time_series = [record.data for record in self.timeseries]
        index = [record.time for record in self.timeseries]
        if not columns:
            columns = self.quantities_names
        return DataFrame(time_series, index=index, columns=columns)

    def as_dict(self) -> Dict[str, List[float]]:
        """Extract time series data from a TIM model.

        Extract the time series data (each column) from the TimModel object

        Returns:
            Dict[str, List[float]]: A dictionary containing the time series data form each column.
            the keys of the dictionary will be index starting from 1 to the number of columns in the tim file
            (excluding the first column(time)).

        Examples:
            ```python
            >>> tim_file = Path("tests/data/input/source-sink/leftsor.tim")
            >>> time_file = TimParser.parse(tim_file)
            >>> tim_model = TimModel(**time_file)
            >>> time_series = tim_model.as_dict()
            >>> 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],
                3: [3.0, 5.0, 12.0, 9.0, 23.0],
                4: [4.0, 4.0, 4.0, 4.0, 4.0]
            }
            ```
        """
        data = self.as_dataframe().to_dict(orient="list")
        return data

    def get_units(self):
        """Return the units for each quantity in the timeseries.

        Returns:
            List[str]: A list of units for each quantity in the timeseries.

        Examples:
            Create a `TimModel` object from a .tim file:
                ```python
                >>> from hydrolib.core.dflowfm.tim.models import TimModel
                >>> tim_model = TimModel(filepath="tests/data/input/source-sink/tim-5-columns.tim")
                >>> tim_model.quantities_names = ["discharge", "waterlevel", "temperature", "salinity", "initialtracer"]
                >>> print(tim_model.get_units())
                ['m3/s', 'm', 'degC', '1e-3', '-']

                ```
        """
        if self.quantities_names is None:
            return None
        return TimModel._get_quantity_unit(self.quantities_names)

comments = Field(default_factory=list) class-attribute instance-attribute

List[str]: A list with the header comment of the tim file.

serializer_config = TimSerializerConfig() class-attribute instance-attribute

TimSerializerConfig: The serialization configuration for the tim file.

timeseries = Field(default_factory=list) class-attribute instance-attribute

List[TimRecord]: A list containing the timeseries.

__init__(filepath=None, quantities_names=None, **parsable_file_kwargs)

Custom initializer to handle extra parameters specific to TimModel.

Parameters:

Name Type Description Default
quantities_names Optional[List[str]]

Names for the quantities in the timeseries.

None
*args, **kwargs

Other arguments for the superclass.

required
Source code in hydrolib\core\dflowfm\tim\models.py
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def __init__(
    self,
    filepath: Optional[Union[str, Path]] = None,
    quantities_names: Optional[List[str]] = None,
    **parsable_file_kwargs,
):
    """
    Custom initializer to handle extra parameters specific to TimModel.

    Args:
        quantities_names (Optional[List[str]]): Names for the quantities in the timeseries.
        *args, **kwargs: Other arguments for the superclass.
    """
    super().__init__(filepath=filepath, **parsable_file_kwargs)
    self.quantities_names = quantities_names

add_column(new_values, column_name=None)

Add new values to each TimRecord in the timeseries, representing a new location.

Parameters:

Name Type Description Default
new_values List[float]

A list of new values to add, one for each TimRecord.

required
column_name str

The name of the new column. Defaults to None. if None, the column is named as "quantity-{len(quantities_names) + 1}".

None

Raises:

Type Description
ValueError

If the number of new values does not match the number of TimRecords.

Examples:

>>> tim_model = TimModel(
...     timeseries=[
...         TimRecord(time=0.0, data=[1.0, 2.0]),
...         TimRecord(time=1.0, data=[3.0, 4.0]),
...     ]
... )
>>> tim_model.add_column([5.0, 6.0])
>>> print(tim_model.timeseries)
[TimRecord(time=0.0, data=[1.0, 2.0, 5.0]), TimRecord(time=1.0, data=[3.0, 4.0, 6.0])]
Source code in hydrolib\core\dflowfm\tim\models.py
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def add_column(self, new_values: List[float], column_name: str = None) -> None:
    """
    Add new values to each TimRecord in the timeseries, representing a new location.

    Args:
        new_values (List[float]): A list of new values to add, one for each TimRecord.
        column_name (str, optional): The name of the new column. Defaults to None.
            if None, the column is named as "quantity-{len(quantities_names) + 1}".

    Raises:
        ValueError: If the number of new values does not match the number of TimRecords.

    Examples:
        ```python
        >>> tim_model = TimModel(
        ...     timeseries=[
        ...         TimRecord(time=0.0, data=[1.0, 2.0]),
        ...         TimRecord(time=1.0, data=[3.0, 4.0]),
        ...     ]
        ... )
        >>> tim_model.add_column([5.0, 6.0])
        >>> print(tim_model.timeseries)
        [TimRecord(time=0.0, data=[1.0, 2.0, 5.0]), TimRecord(time=1.0, data=[3.0, 4.0, 6.0])]

        ```
    """
    if len(new_values) != len(self.timeseries):
        raise ValueError(
            f"Expected {len(self.timeseries)} values, but got {len(new_values)}."
        )

    for record, value in zip(self.timeseries, new_values):
        record.data.append(value)

    if self.quantities_names:
        if column_name is None:
            column_name = f"quantity-{len(self.quantities_names) + 1}"
        self.quantities_names.append(column_name)

as_dataframe(columns=None)

Return the timeseries as a pandas DataFrame.

Parameters:

Name Type Description Default
columns List[Any, str], optional, Defaults to None

The column names for the DataFrame.

None

Returns:

Name Type Description
DataFrame DataFrame

The timeseries as a pandas DataFrame.

Notes
  • If the columns are not provided, the quantities_names will be used as column names.
  • If the quantities_names are not provided, the columns will be named as 0, 1, 2, etc.

Examples:

Create a TimModel object from a .tim file: >>> from hydrolib.core.dflowfm.tim.models import TimModel >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim") >>> df = tim_model.as_dataframe() >>> print(df) 0 1 2 10.0 1.232 2.343 3.454 20.0 4.565 5.676 6.787 30.0 1.500 2.600 3.700

If the TimModel object was created with quantities names: >>> quantities_names = ["Column1", "Column2", "Column3"] >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim", quantities_names=quantities_names) >>> df = tim_model.as_dataframe() >>> print(df) Column1 Column2 Column3 10.0 1.232 2.343 3.454 20.0 4.565 5.676 6.787 30.0 1.500 2.600 3.700

To add column names to the DataFrame after you have created the TimModel object: >>> df = tim_model.as_dataframe(columns=["Column1", "Column2", "Column3"]) >>> print(df) Column1 Column2 Column3 10.0 1.232 2.343 3.454 20.0 4.565 5.676 6.787 30.0 1.500 2.600 3.700

Source code in hydrolib\core\dflowfm\tim\models.py
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def as_dataframe(self, columns: List[Any] = None) -> DataFrame:
    """Return the timeseries as a pandas DataFrame.

    Args:
        columns (List[Any, str], optional, Defaults to None):
            The column names for the DataFrame.

    Returns:
        DataFrame: The timeseries as a pandas DataFrame.

    Notes:
        - If the columns are not provided, the quantities_names will be used as column names.
        - If the quantities_names are not provided, the columns will be named as 0, 1, 2, etc.

    Examples:
        Create a `TimModel` object from a .tim file:
            >>> from hydrolib.core.dflowfm.tim.models import TimModel
            >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim")
            >>> df = tim_model.as_dataframe()
            >>> print(df)
                      0      1      2
            10.0  1.232  2.343  3.454
            20.0  4.565  5.676  6.787
            30.0  1.500  2.600  3.700

        If the `TimModel` object was created with quantities names:
            >>> quantities_names = ["Column1", "Column2", "Column3"]
            >>> tim_model = TimModel(filepath="tests/data/input/tim/triple_data_for_timeseries.tim", quantities_names=quantities_names)
            >>> df = tim_model.as_dataframe()
            >>> print(df)
                  Column1  Column2  Column3
            10.0    1.232    2.343    3.454
            20.0    4.565    5.676    6.787
            30.0    1.500    2.600    3.700

        To add column names to the DataFrame after you have created the `TimModel` object:
            >>> df = tim_model.as_dataframe(columns=["Column1", "Column2", "Column3"])
            >>> print(df)
                  Column1  Column2  Column3
            10.0    1.232    2.343    3.454
            20.0    4.565    5.676    6.787
            30.0    1.500    2.600    3.700
    """
    time_series = [record.data for record in self.timeseries]
    index = [record.time for record in self.timeseries]
    if not columns:
        columns = self.quantities_names
    return DataFrame(time_series, index=index, columns=columns)

as_dict()

Extract time series data from a TIM model.

Extract the time series data (each column) from the TimModel object

Returns:

Type Description
Dict[str, List[float]]

Dict[str, List[float]]: A dictionary containing the time series data form each column.

Dict[str, List[float]]

the keys of the dictionary will be index starting from 1 to the number of columns in the tim file

Dict[str, List[float]]

(excluding the first column(time)).

Examples:

>>> tim_file = Path("tests/data/input/source-sink/leftsor.tim")
>>> time_file = TimParser.parse(tim_file)
>>> tim_model = TimModel(**time_file)
>>> time_series = tim_model.as_dict()
>>> 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],
    3: [3.0, 5.0, 12.0, 9.0, 23.0],
    4: [4.0, 4.0, 4.0, 4.0, 4.0]
}
Source code in hydrolib\core\dflowfm\tim\models.py
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def as_dict(self) -> Dict[str, List[float]]:
    """Extract time series data from a TIM model.

    Extract the time series data (each column) from the TimModel object

    Returns:
        Dict[str, List[float]]: A dictionary containing the time series data form each column.
        the keys of the dictionary will be index starting from 1 to the number of columns in the tim file
        (excluding the first column(time)).

    Examples:
        ```python
        >>> tim_file = Path("tests/data/input/source-sink/leftsor.tim")
        >>> time_file = TimParser.parse(tim_file)
        >>> tim_model = TimModel(**time_file)
        >>> time_series = tim_model.as_dict()
        >>> 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],
            3: [3.0, 5.0, 12.0, 9.0, 23.0],
            4: [4.0, 4.0, 4.0, 4.0, 4.0]
        }
        ```
    """
    data = self.as_dataframe().to_dict(orient="list")
    return data

get_units()

Return the units for each quantity in the timeseries.

Returns:

Type Description

List[str]: A list of units for each quantity in the timeseries.

Examples:

Create a TimModel object from a .tim file:

>>> from hydrolib.core.dflowfm.tim.models import TimModel
>>> tim_model = TimModel(filepath="tests/data/input/source-sink/tim-5-columns.tim")
>>> tim_model.quantities_names = ["discharge", "waterlevel", "temperature", "salinity", "initialtracer"]
>>> print(tim_model.get_units())
['m3/s', 'm', 'degC', '1e-3', '-']

Source code in hydrolib\core\dflowfm\tim\models.py
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
def get_units(self):
    """Return the units for each quantity in the timeseries.

    Returns:
        List[str]: A list of units for each quantity in the timeseries.

    Examples:
        Create a `TimModel` object from a .tim file:
            ```python
            >>> from hydrolib.core.dflowfm.tim.models import TimModel
            >>> tim_model = TimModel(filepath="tests/data/input/source-sink/tim-5-columns.tim")
            >>> tim_model.quantities_names = ["discharge", "waterlevel", "temperature", "salinity", "initialtracer"]
            >>> print(tim_model.get_units())
            ['m3/s', 'm', 'degC', '1e-3', '-']

            ```
    """
    if self.quantities_names is None:
        return None
    return TimModel._get_quantity_unit(self.quantities_names)

TimRecord

Bases: BaseModel

Single tim record, representing a time and a list of data.

Source code in hydrolib\core\dflowfm\tim\models.py
14
15
16
17
18
19
20
21
class TimRecord(BaseModel):
    """Single tim record, representing a time and a list of data."""

    time: float
    """float: Time of the time record."""

    data: List[float] = Field(default_factory=list)
    """List[float]: Record of the time record."""

data = Field(default_factory=list) class-attribute instance-attribute

List[float]: Record of the time record.

time instance-attribute

float: Time of the time record.

Parser

TimParser

Bases: BaseParser

A parser for .tim files. Full line comments at the start of the file are supported. Comment lines start with either a * or a #. No other comments are supported.

Source code in hydrolib\core\dflowfm\tim\parser.py
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class TimParser(BaseParser):
    """
    A parser for .tim files.
    Full line comments at the start of the file are supported. Comment lines start with either a `*` or a `#`.
    No other comments are supported.
    """

    @staticmethod
    def parse(filepath: Path) -> Dict[str, List[Any]]:
        """Parse a .tim file into a dictionary with comments and time series data.

        Args:
            filepath (Path): Path to the .tim file to be parsed.

        Returns:
            Dict[str, List[Any]]: A dictionary with keys "comments" and "timeseries".
            - "comments" represents comments found at the start of the file.
            - "timeseries" is a list of dictionaries with the key as "time" and values as "data".
                - "time" is a time as a string.
                - "data" is data as a list of strings.

        Raises:
            ValueError: If the file contains a comment that is not at the start of the file.
            ValueError: If the data of the timeseries is empty.
        """
        with filepath.open(encoding="utf8") as file:
            lines = file.readlines()
            comments, start_timeseries_index = TimParser._read_header_comments(lines)
            timeseries = TimParser._read_time_series_data(lines, start_timeseries_index)

        return {"comments": comments, "timeseries": timeseries}

    @staticmethod
    def _read_time_series_data(
        lines: List[str], start_timeseries_index: int
    ) -> List[TimData]:
        timeseries: List[TimData] = []
        for line_index in range(start_timeseries_index, len(lines)):
            line = lines[line_index].strip()

            if len(line) == 0:
                continue

            TimParser._raise_error_if_contains_comment(line, line_index + 1)

            time, *values = line.split()

            TimParser._raise_error_if_values_empty(values, line_index)

            timrecord = {"time": time, "data": values}
            timeseries.append(timrecord)

        return timeseries

    @staticmethod
    def _raise_error_if_values_empty(values: List[str], line_index: int) -> None:
        if len(values) == 0:
            raise ValueError(f"Line {line_index}: Time series cannot be empty.")

parse(filepath) staticmethod

Parse a .tim file into a dictionary with comments and time series data.

Parameters:

Name Type Description Default
filepath Path

Path to the .tim file to be parsed.

required

Returns:

Type Description
Dict[str, List[Any]]

Dict[str, List[Any]]: A dictionary with keys "comments" and "timeseries".

Dict[str, List[Any]]
  • "comments" represents comments found at the start of the file.
Dict[str, List[Any]]
  • "timeseries" is a list of dictionaries with the key as "time" and values as "data".
  • "time" is a time as a string.
  • "data" is data as a list of strings.

Raises:

Type Description
ValueError

If the file contains a comment that is not at the start of the file.

ValueError

If the data of the timeseries is empty.

Source code in hydrolib\core\dflowfm\tim\parser.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@staticmethod
def parse(filepath: Path) -> Dict[str, List[Any]]:
    """Parse a .tim file into a dictionary with comments and time series data.

    Args:
        filepath (Path): Path to the .tim file to be parsed.

    Returns:
        Dict[str, List[Any]]: A dictionary with keys "comments" and "timeseries".
        - "comments" represents comments found at the start of the file.
        - "timeseries" is a list of dictionaries with the key as "time" and values as "data".
            - "time" is a time as a string.
            - "data" is data as a list of strings.

    Raises:
        ValueError: If the file contains a comment that is not at the start of the file.
        ValueError: If the data of the timeseries is empty.
    """
    with filepath.open(encoding="utf8") as file:
        lines = file.readlines()
        comments, start_timeseries_index = TimParser._read_header_comments(lines)
        timeseries = TimParser._read_time_series_data(lines, start_timeseries_index)

    return {"comments": comments, "timeseries": timeseries}

Serializer

TimSerializer

Source code in hydrolib\core\dflowfm\tim\serializer.py
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class TimSerializer:
    @staticmethod
    def serialize(
        path: Path,
        data: Dict[str, List[Any]],
        config: TimSerializerConfig,
        save_settings: ModelSaveSettings,
    ) -> None:
        """
        Serialize timeseries data to a file in .tim format.

        Args:
            path (Path): The path to the destination .tim file.
            data (Dict[str, List[Any]]): A dictionary with keys "comments" and "timeseries".
            - "comments" represents comments found at the start of the file.
            - "timeseries" is a list of dictionaries with the key as "time" and values as "data".
                - "time" is a time as a string.
                - "data" is data as a list of strings.

            config (TimSerializerConfig): The serialization configuration settings.
            save_settings (ModelSaveSettings): The save settings to be used.
        """
        path.parent.mkdir(parents=True, exist_ok=True)

        commentlines = TimSerializer._serialize_comment_lines(data)
        timeserieslines = TimSerializer._serialize_timeseries_lines(data, config)

        file_content = TimSerializer._serialize_file_content(
            timeserieslines, commentlines
        )
        with path.open("w", encoding="utf8") as file:
            file.write(file_content)

    @staticmethod
    def _serialize_comment_lines(data: Dict[str, List[Any]]) -> List[str]:
        commentlines = []
        for comment in data["comments"]:
            commentlines.append(f"#{comment}")
        return commentlines

    @staticmethod
    def _serialize_timeseries_lines(
        data: Dict[str, List[Any]], config: TimSerializerConfig
    ) -> List[str]:
        format_float = lambda v: f"{v:{config.float_format}}"
        timeseriesblock = TimSerializer._serialize_to_timeseries_block(
            data, format_float
        )
        timeserieslines = TimSerializer._serialize_timeseries_to_lines(
            timeseriesblock, config
        )
        return timeserieslines

    @staticmethod
    def _serialize_to_timeseries_block(
        data: Dict[str, List[Any]], format_float: Callable[[float], str]
    ) -> TimeSeriesBlock:
        timeseries_block: TimeSeriesBlock = []
        for timeseries in data["timeseries"]:
            time = timeseries["time"]
            row_elements = timeseries["data"]
            timeseries_row = [format_float(time)] + [
                format_float(value) for value in row_elements
            ]
            timeseries_block.append(timeseries_row)
        return timeseries_block

    @staticmethod
    def _serialize_timeseries_to_lines(
        timeseries_block: TimeSeriesBlock, config: TimSerializerConfig
    ) -> List[str]:
        # Make sure the columns are aligned and have the proper spacing
        column_space = " " * config.column_spacing
        column_lengths = TimSerializer._get_column_lengths(timeseries_block)

        timeserieslines = []
        for timeseries_row in timeseries_block:
            row_elements: List[str] = []
            for index, value in enumerate(timeseries_row):
                whitespace_offset = TimSerializer._get_offset_whitespace(
                    value, column_lengths[index]
                )
                row_elements.append(value + whitespace_offset)

            line = column_space.join(row_elements)
            timeserieslines.append(line)
        return timeserieslines

    @staticmethod
    def _get_offset_whitespace(value: Optional[str], max_length: int) -> str:
        value_length = len(value) if value is not None else 0
        return " " * max(max_length - value_length, 0)

    @staticmethod
    def _serialize_file_content(timeserieslines: List[str], commentlines: List[str]):
        lines = []
        lines.extend(commentlines)
        lines.extend(timeserieslines)
        file_content = "\n".join(lines)
        return file_content

    @staticmethod
    def _get_column_lengths(timeseries_block: TimeSeriesBlock) -> List[int]:
        if len(timeseries_block) == 0:
            return []

        n_columns = len(timeseries_block[0])
        column_lengths = [0] * n_columns

        for timeseries_row in timeseries_block:
            for index, row_element in enumerate(timeseries_row):
                if len(row_element) > column_lengths[index]:
                    column_lengths[index] = len(row_element)

        return column_lengths

serialize(path, data, config, save_settings) staticmethod

Serialize timeseries data to a file in .tim format.

Parameters:

Name Type Description Default
path Path

The path to the destination .tim file.

required
data Dict[str, List[Any]]

A dictionary with keys "comments" and "timeseries".

required
config TimSerializerConfig

The serialization configuration settings.

required
save_settings ModelSaveSettings

The save settings to be used.

required
Source code in hydrolib\core\dflowfm\tim\serializer.py
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def serialize(
    path: Path,
    data: Dict[str, List[Any]],
    config: TimSerializerConfig,
    save_settings: ModelSaveSettings,
) -> None:
    """
    Serialize timeseries data to a file in .tim format.

    Args:
        path (Path): The path to the destination .tim file.
        data (Dict[str, List[Any]]): A dictionary with keys "comments" and "timeseries".
        - "comments" represents comments found at the start of the file.
        - "timeseries" is a list of dictionaries with the key as "time" and values as "data".
            - "time" is a time as a string.
            - "data" is data as a list of strings.

        config (TimSerializerConfig): The serialization configuration settings.
        save_settings (ModelSaveSettings): The save settings to be used.
    """
    path.parent.mkdir(parents=True, exist_ok=True)

    commentlines = TimSerializer._serialize_comment_lines(data)
    timeserieslines = TimSerializer._serialize_timeseries_lines(data, config)

    file_content = TimSerializer._serialize_file_content(
        timeserieslines, commentlines
    )
    with path.open("w", encoding="utf8") as file:
        file.write(file_content)

TimSerializerConfig

Bases: SerializerConfig

Configuration settings for the TimSerializer.

Source code in hydrolib\core\dflowfm\tim\serializer.py
10
11
12
13
14
class TimSerializerConfig(SerializerConfig):
    """Configuration settings for the TimSerializer."""

    column_spacing: int = 1
    """(int): The number of spaces to include between columns in the serialized .tim file."""

column_spacing = 1 class-attribute instance-attribute

(int): The number of spaces to include between columns in the serialized .tim file.