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.
Source code in hydrolib/core/dflowfm/tim/models.py
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 | class TimModel(ParsableFileModel):
"""Class representing a tim (*.tim) file."""
serializer_config = TimSerializerConfig()
"""TimSerializerConfig: The serialization configuration for the tim file."""
comments: List[str] = []
"""List[str]: A list with the header comment of the tim file."""
timeseries: List[TimRecord] = []
"""List[TimRecord]: A list containing the timeseries."""
@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 amount of columns per timeseries match and if the timeseries have no duplicate times.
Args:
v (List[TimRecord]): Timeseries to validate.
Raises:
ValueError: When the amount of columns for timeseries is zero.
ValueError: When the amount 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)
|
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 = []
class-attribute
instance-attribute
List[TimRecord]: A list containing the timeseries.
TimRecord
Bases: BaseModel
Single tim record, representing a time and a list of data.
Source code in hydrolib/core/dflowfm/tim/models.py
| 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] = []
"""List[float]: Record of the time record."""
|
data = []
class-attribute
instance-attribute
List[float]: Record of the time record.
time
instance-attribute
float: Time of the time record.
Parser
TimParser
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
7
8
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
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 | class TimParser:
"""
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.
"""
comments: List[str] = []
timeseries: List[TimData] = []
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_header_comments(lines: List[str]) -> Tuple[List[str], int]:
"""Read the header comments of the lines from the .tim file.
The comments are only expected at the start of the .tim file.
When a non comment line is encountered, all comments from the header will be retuned together with the start index of the timeseries data.
Args:
lines (List[str]): Lines from the the .tim file which is read.
Returns:
Tuple of List[str] and int, the List[str] contains the commenst from the header, the int is the start index of the timeseries.
"""
comments: List[str] = []
start_timeseries_index = 0
for line_index in range(len(lines)):
line = lines[line_index].strip()
if len(line) == 0:
comments.append(line)
continue
if line.startswith("#") or line.startswith("*"):
comments.append(line[1:])
continue
start_timeseries_index = line_index
break
return comments, start_timeseries_index
@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_contains_comment(line: str, line_index: int) -> None:
if "#" in line or "*" in line:
raise ValueError(
f"Line {line_index}: comments are only supported at the start of the file, before the time series data."
)
@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:
Returns:
Raises:
Source code in hydrolib/core/dflowfm/tim/parser.py
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 | @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.
"""
comments: List[str] = []
timeseries: List[TimData] = []
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:
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
| 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.