Skip to content

Polyline/polygon files

Polyline .pli(z) and polygon .pol files are basic spatial input files for a D-Flow FM model to select particular locations, and are used in various other input files.

They are represented by the classes below.

Model

models.py defines all classes and functions related to representing pol/pli(z) files.

Description

Bases: BaseModel

Description of a single PolyObject.

The Description will be prepended to a block. Each line will start with a '*'.

Attributes:

Name Type Description
content str

The content of this Description.
Source code in hydrolib/core/dflowfm/polyfile/models.py 
 9
10
11
12
13
14
15
16
17
18
19
class Description(BaseModel):
    """Description of a single PolyObject.

    The Description will be prepended to a block. Each line will
    start with a '*'.

    Attributes:
        content (str): The content of this Description.
    """

    content: str

Metadata

Bases: BaseModel

Metadata of a single PolyObject.

Attributes:

Name Type Description
name str

The name of the PolyObject
n_rows int

The number of rows (i.e. Point instances) of the PolyObject
n_columns int

The total number of values in a Point, including x, y, and z.
Source code in hydrolib/core/dflowfm/polyfile/models.py 
22
23
24
25
26
27
28
29
30
31
32
33
class Metadata(BaseModel):
    """Metadata of a single PolyObject.

    Attributes:
        name (str): The name of the PolyObject
        n_rows (int): The number of rows (i.e. Point instances) of the PolyObject
        n_columns (int): The total number of values in a Point, including x, y, and z.
    """

    name: str
    n_rows: int
    n_columns: int

Point

Bases: BaseModel

Point consisting of a x and y coordinate, an optional z coordinate and data.

Attributes:

Name Type Description
x float

The x-coordinate of this Point
y float

The y-coordinate of this Point
z Optional[float]

An optional z-coordinate of this Point.
data Sequence[float]

The additional data variables of this Point.
Source code in hydrolib/core/dflowfm/polyfile/models.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class Point(BaseModel):
    """Point consisting of a x and y coordinate, an optional z coordinate and data.

    Attributes:
        x (float): The x-coordinate of this Point
        y (float): The y-coordinate of this Point
        z (Optional[float]): An optional z-coordinate of this Point.
        data (Sequence[float]): The additional data variables of this Point.
    """

    x: float
    y: float
    z: Optional[float]
    data: Sequence[float]

    def _get_identifier(self, data: dict) -> Optional[str]:
        x = data.get("x")
        y = data.get("y")
        z = data.get("z")
        return f"x:{x} y:{y} z:{z}"

PolyFile

Bases: ParsableFileModel

Poly-file (.pol/.pli/.pliz) representation.

Source code in hydrolib/core/dflowfm/polyfile/models.py 
 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
class PolyFile(ParsableFileModel):
    """Poly-file (.pol/.pli/.pliz) representation."""

    has_z_values: bool = False
    objects: Sequence[PolyObject] = []

    def _serialize(self, _: dict, save_settings: ModelSaveSettings) -> None:
        from .serializer import write_polyfile

        # We skip the passed dict for a better one.
        write_polyfile(self._resolved_filepath, self.objects, self.serializer_config)

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

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

    @classmethod
    def _get_serializer(cls) -> Callable:
        # Unused, but requires abstract implementation
        pass

    @classmethod
    def _get_parser(cls) -> Callable:
        # TODO Prevent circular dependency in Parser
        from .parser import read_polyfile

        return read_polyfile

PolyObject

Bases: BaseModel

PolyObject describing a single block in a poly file.

The metadata should be consistent with the points: - The number of points should be equal to number of rows defined in the metadata - The data of each point should be equal to the number of columns defined in the metadata.

Attributes:

Name Type Description
description Optional[Description]

An optional description of this PolyObject
metadata Metadata

The Metadata of this PolObject, describing the structure
points List[Point]

The points describing this PolyObject, structured according to the Metadata
Source code in hydrolib/core/dflowfm/polyfile/models.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class PolyObject(BaseModel):
    """PolyObject describing a single block in a poly file.

    The metadata should be consistent with the points:
    - The number of points should be equal to number of rows defined in the metadata
    - The data of each point should be equal to the number of columns defined in the
      metadata.

    Attributes:
        description (Optional[Description]):
            An optional description of this PolyObject
        metadata (Metadata):
            The Metadata of this PolObject, describing the structure
        points (List[Point]):
            The points describing this PolyObject, structured according to the Metadata
    """

    description: Optional[Description]
    metadata: Metadata
    points: List[Point]

Parser

parser.py defines all classes and functions related to parsing pol/pli(z) files.

Block

Bases: BaseModel

Block is a temporary object which will be converted into a PolyObject.

The fields are supposed to be set during the lifetime of this object. When all fields are set, finalize can be called.

Attributes:

Name Type Description
start_line int

The starting line of this current block.
name Optional[str]

The name of this block. Defaults to None.
dimensions Optional[Tuple[int, int]]

The dimensions (n_rows, n_columns) of this Block. Defaults to None.
points Optional[List[Point]]

The points of this block. Defaults to None.
ws_warnings List[ParseMsg]

The whitespace warnings associated with this block. Defaults to an empty list.
empty_lines List[int]

The line numbers of the empty lines. Defaults to an empty list.
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
 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
class Block(BaseModel):
    """Block is a temporary object which will be converted into a PolyObject.

    The fields are supposed to be set during the lifetime of this object.
    When all fields are set, finalize can be called.

    Attributes:
        start_line (int): The starting line of this current block.
        name (Optional[str]): The name of this block. Defaults to None.
        dimensions (Optional[Tuple[int, int]]):
            The dimensions (n_rows, n_columns) of this Block. Defaults to None.
        points (Optional[List[Point]]):
            The points of this block. Defaults to None.
        ws_warnings (List[ParseMsg]):
            The whitespace warnings associated with this block.
            Defaults to an empty list.
        empty_lines (List[int]):
            The line numbers of the empty lines. Defaults to an empty list.
    """

    start_line: int

    description: Optional[List[str]] = None
    name: Optional[str] = None
    dimensions: Optional[Tuple[int, int]] = None
    points: Optional[List[Point]] = None

    ws_warnings: List[ParseMsg] = []
    empty_lines: List[int] = []

    def finalize(self) -> Optional[Tuple[PolyObject, List[ParseMsg]]]:
        """Finalise this Block and return the constructed PolyObject and warnings

        If the metadata or the points are None, then None is returned.

        Returns:
            Optional[Tuple[PolyObject, List[ParseMsg]]]:
                The constructed PolyObject and warnings encountered while parsing it.
        """
        metadata = self._get_metadata()

        if metadata is None or self.points is None:
            return None

        obj = PolyObject(
            description=self._get_description(), metadata=metadata, points=self.points
        )

        return obj, self.ws_warnings + self._get_empty_line_warnings()

    def _get_description(self) -> Optional[Description]:
        if self.description is not None:
            return Description(content="\n".join(self.description))
        else:
            return None

    def _get_metadata(self) -> Optional[Metadata]:
        if self.name is None or self.dimensions is None:
            return None

        (n_rows, n_columns) = self.dimensions
        return Metadata(name=self.name, n_rows=n_rows, n_columns=n_columns)

    def _get_empty_line_warnings(self):
        if len(self.empty_lines) == 0:
            return []

        warnings = []
        empty_line = (self.empty_lines[0], self.empty_lines[0])

        for line in self.empty_lines[1:]:
            if line == empty_line[1] + 1:
                empty_line = (empty_line[0], line)
            else:
                warnings.append(Block._get_empty_line_msg(empty_line))
                empty_line = (line, line)
        warnings.append(Block._get_empty_line_msg(empty_line))

        return warnings

    @staticmethod
    def _get_empty_line_msg(line_range: Tuple[int, int]) -> ParseMsg:
        return ParseMsg(
            line_start=line_range[0],
            line_end=line_range[1],
            reason="Empty lines are ignored.",
        )

finalize()

Finalise this Block and return the constructed PolyObject and warnings

If the metadata or the points are None, then None is returned.

Returns:

Type Description
Optional[Tuple[PolyObject, List[ParseMsg]]]

Optional[Tuple[PolyObject, List[ParseMsg]]]: The constructed PolyObject and warnings encountered while parsing it.
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def finalize(self) -> Optional[Tuple[PolyObject, List[ParseMsg]]]:
    """Finalise this Block and return the constructed PolyObject and warnings

    If the metadata or the points are None, then None is returned.

    Returns:
        Optional[Tuple[PolyObject, List[ParseMsg]]]:
            The constructed PolyObject and warnings encountered while parsing it.
    """
    metadata = self._get_metadata()

    if metadata is None or self.points is None:
        return None

    obj = PolyObject(
        description=self._get_description(), metadata=metadata, points=self.points
    )

    return obj, self.ws_warnings + self._get_empty_line_warnings()

ErrorBuilder

ErrorBuilder provides the functionality to the Parser to keep track of errors.

Source code in hydrolib/core/dflowfm/polyfile/parser.py 
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
class ErrorBuilder:
    """ErrorBuilder provides the functionality to the Parser to keep track of errors."""

    def __init__(self) -> None:
        """Create a new ErorrorBuilder"""
        self._current_block: Optional[InvalidBlock] = None

    def start_invalid_block(
        self, block_start: int, invalid_line: int, reason: str
    ) -> None:
        """Start a new invalid block if none exists at the moment

        If we are already in an invalid block, or the previous one
        was never finalised, we will not log the reason, and assume
        it is one long invalid block.

        Args:
            block_start (int): The start of the invalid block.
            invalid_line (int): The actual offending line number.
            reason (str): The reason why this block is invalid.
        """
        if self._current_block is None:
            self._current_block = InvalidBlock(
                start_line=block_start, invalid_line=invalid_line, reason=reason
            )

    def end_invalid_block(self, line: int) -> None:
        """Store the end line of the current block

        If no invalid block currently exists, nothing will be done.

        Args:
            line (int): the final line of this invalid block
        """
        if self._current_block is not None:
            self._current_block.end_line = line

    def finalize_previous_error(self) -> Optional[ParseMsg]:
        """Finalize the current invalid block if it exists

        If no current invalid block exists, None will be returned, and nothing will
        change. If a current block exists, it will be converted into a ParseMsg and
        returned. The current invalid block will be reset.

        Returns:
            Optional[ParseMsg]: The corresponding ParseMsg if an InvalidBlock exists.
        """
        if self._current_block is not None:
            msg = self._current_block.to_msg()
            self._current_block = None

            return msg
        else:
            return None

__init__()

Create a new ErorrorBuilder

Source code in hydrolib/core/dflowfm/polyfile/parser.py 
180
181
182
def __init__(self) -> None:
    """Create a new ErorrorBuilder"""
    self._current_block: Optional[InvalidBlock] = None

end_invalid_block(line)

Store the end line of the current block

If no invalid block currently exists, nothing will be done.

Parameters:

Name Type Description Default
line int

the final line of this invalid block

 required
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
203
204
205
206
207
208
209
210
211
212
def end_invalid_block(self, line: int) -> None:
    """Store the end line of the current block

    If no invalid block currently exists, nothing will be done.

    Args:
        line (int): the final line of this invalid block
    """
    if self._current_block is not None:
        self._current_block.end_line = line

finalize_previous_error()

Finalize the current invalid block if it exists

If no current invalid block exists, None will be returned, and nothing will change. If a current block exists, it will be converted into a ParseMsg and returned. The current invalid block will be reset.

Returns:

Type Description
Optional[ParseMsg]

Optional[ParseMsg]: The corresponding ParseMsg if an InvalidBlock exists.
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def finalize_previous_error(self) -> Optional[ParseMsg]:
    """Finalize the current invalid block if it exists

    If no current invalid block exists, None will be returned, and nothing will
    change. If a current block exists, it will be converted into a ParseMsg and
    returned. The current invalid block will be reset.

    Returns:
        Optional[ParseMsg]: The corresponding ParseMsg if an InvalidBlock exists.
    """
    if self._current_block is not None:
        msg = self._current_block.to_msg()
        self._current_block = None

        return msg
    else:
        return None

start_invalid_block(block_start, invalid_line, reason)

Start a new invalid block if none exists at the moment

If we are already in an invalid block, or the previous one was never finalised, we will not log the reason, and assume it is one long invalid block.

Parameters:

Name Type Description Default
block_start int

The start of the invalid block.

 required
invalid_line int

The actual offending line number.

 required
reason str

The reason why this block is invalid.

 required
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def start_invalid_block(
    self, block_start: int, invalid_line: int, reason: str
) -> None:
    """Start a new invalid block if none exists at the moment

    If we are already in an invalid block, or the previous one
    was never finalised, we will not log the reason, and assume
    it is one long invalid block.

    Args:
        block_start (int): The start of the invalid block.
        invalid_line (int): The actual offending line number.
        reason (str): The reason why this block is invalid.
    """
    if self._current_block is None:
        self._current_block = InvalidBlock(
            start_line=block_start, invalid_line=invalid_line, reason=reason
        )

InvalidBlock

Bases: BaseModel

InvalidBlock is a temporary object which will be converted into a ParseMsg.

Attributes:

Name Type Description
start_line int

The start line of this InvalidBlock
end_line Optional[int]

The end line of this InvalidBlock if it is set. Defaults to None.
invalid_line int

The line which is causing this block to be invalid.
reason str

A human-readable string detailing the reason of the ParseMsg.
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
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
class InvalidBlock(BaseModel):
    """InvalidBlock is a temporary object which will be converted into a ParseMsg.

    Attributes:
        start_line (int): The start line of this InvalidBlock
        end_line (Optional[int]):
            The end line of this InvalidBlock if it is set. Defaults to None.
        invalid_line (int): The line which is causing this block to be invalid.
        reason (str): A human-readable string detailing the reason of the ParseMsg.
    """

    start_line: int
    end_line: Optional[int] = None
    invalid_line: int
    reason: str

    def to_msg(self) -> ParseMsg:
        """Convert this InvalidBlock to the corresponding ParseMsg

        Returns:
            ParseMsg: The ParseMsg corresponding with this InvalidBlock
        """
        return ParseMsg(
            line_start=self.start_line,
            line_end=self.end_line,
            reason=f"{self.reason} at line {self.invalid_line}.",
        )

to_msg()

Convert this InvalidBlock to the corresponding ParseMsg

Returns:

Name Type Description
ParseMsg ParseMsg

The ParseMsg corresponding with this InvalidBlock
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
164
165
166
167
168
169
170
171
172
173
174
def to_msg(self) -> ParseMsg:
    """Convert this InvalidBlock to the corresponding ParseMsg

    Returns:
        ParseMsg: The ParseMsg corresponding with this InvalidBlock
    """
    return ParseMsg(
        line_start=self.start_line,
        line_end=self.end_line,
        reason=f"{self.reason} at line {self.invalid_line}.",
    )

ParseMsg

Bases: BaseModel

ParseMsg defines a single message indicating a significant parse event.

Attributes:

Name Type Description
line_start int

The start line of the block to which this ParseMsg refers.
line_end int

The end line of the block to which this ParseMsg refers.
column Optional[Tuple[int, int]]

An optional begin and end column to which this ParseMsg refers.
reason str

A human-readable string detailing the reason of the ParseMsg.
Source code in hydrolib/core/dflowfm/polyfile/parser.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
49
50
51
52
53
54
55
56
class ParseMsg(BaseModel):
    """ParseMsg defines a single message indicating a significant parse event.

    Attributes:
        line_start (int): The start line of the block to which this ParseMsg refers.
        line_end (int): The end line of the block to which this ParseMsg refers.
        column (Optional[Tuple[int, int]]):
            An optional begin and end column to which this ParseMsg refers.
        reason (str): A human-readable string detailing the reason of the ParseMsg.
    """

    line_start: int
    line_end: int

    column: Optional[Tuple[int, int]]
    reason: str

    def format_parsemsg_to_string(self, file_path: Optional[Path] = None) -> str:
        """Format string describing this ParseMsg

        Args:
            file_path (Optional[Path], optional):
                The file path mentioned in the message if specified. Defaults to None.
        """
        if self.line_start != self.line_end:
            block_suffix = f"\nInvalid block {self.line_start}:{self.line_end}"
        else:
            block_suffix = f"\nInvalid line {self.line_start}"

        col_suffix = (
            f"\nColumns {self.column[0]}:{self.column[1]}"
            if self.column is not None
            else ""
        )
        file_suffix = f"\nFile: {file_path}" if file_path is not None else ""

        message = f"{self.reason}{block_suffix}{col_suffix}{file_suffix}"

        return message

format_parsemsg_to_string(file_path=None)

Format string describing this ParseMsg

Parameters:

Name Type Description Default
file_path Optional[Path]

The file path mentioned in the message if specified. Defaults to None.

 None
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def format_parsemsg_to_string(self, file_path: Optional[Path] = None) -> str:
    """Format string describing this ParseMsg

    Args:
        file_path (Optional[Path], optional):
            The file path mentioned in the message if specified. Defaults to None.
    """
    if self.line_start != self.line_end:
        block_suffix = f"\nInvalid block {self.line_start}:{self.line_end}"
    else:
        block_suffix = f"\nInvalid line {self.line_start}"

    col_suffix = (
        f"\nColumns {self.column[0]}:{self.column[1]}"
        if self.column is not None
        else ""
    )
    file_suffix = f"\nFile: {file_path}" if file_path is not None else ""

    message = f"{self.reason}{block_suffix}{col_suffix}{file_suffix}"

    return message

Parser

Parser provides the functionality to parse a polyfile line by line.

The Parser parses blocks describing PolyObject instances by relying on a rudimentary state machine. The states are encoded with the StateType. New lines are fed through the feed_line method. After each line the internal state will be updated. When a complete block is read, it will be converted into a PolyObject and stored internally. When finalise is called, the constructed objects, as well as any warnings and errors describing invalid blocks, will be returned.

Each state defines a feed_line method, stored in the _feed_line dict, which consumes a line and potentially transitions the state into the next. Each state further defines a finalise method, stored in the _finalise dict, which is called upon finalising the parser.

Invalid states are encoded with INVALID_STATE. In this state the Parser attempts to find a new block, and thus looks for a new description or name.

Unexpected whitespace before comments, names, and dimensions, as well as empty lines will generate a warning, and will be ignored by the parser.

Source code in hydrolib/core/dflowfm/polyfile/parser.py 
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
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
class Parser:
    """Parser provides the functionality to parse a polyfile line by line.

    The Parser parses blocks describing PolyObject instances by relying on
    a rudimentary state machine. The states are encoded with the StateType.
    New lines are fed through the feed_line method. After each line the
    internal state will be updated. When a complete block is read, it will
    be converted into a PolyObject and stored internally.
    When finalise is called, the constructed objects, as well as any warnings
    and errors describing invalid blocks, will be returned.

    Each state defines a feed_line method, stored in the _feed_line dict,
    which consumes a line and potentially transitions the state into the next.
    Each state further defines a finalise method, stored in the _finalise dict,
    which is called upon finalising the parser.

    Invalid states are encoded with INVALID_STATE. In this state the Parser
    attempts to find a new block, and thus looks for a new description or
    name.

    Unexpected whitespace before comments, names, and dimensions, as well as
    empty lines will generate a warning, and will be ignored by the parser.
    """

    def __init__(self, file_path: Path, has_z_value: bool = False) -> None:
        """Create a new Parser

        Args:
            file_path (Path):
                Name of the file being parsed, only used for providing proper warnings.
            has_z_value (bool, optional):
                Whether to interpret the third column as z-coordinates.
                Defaults to False.
        """
        self._has_z_value = has_z_value
        self._file_path = file_path

        self._line = 0
        self._new_block()

        self._error_builder = ErrorBuilder()

        self._poly_objects: List[PolyObject] = []

        self._current_point: int = 0

        self._feed_line: Dict[StateType, Callable[[str], None]] = {
            StateType.NEW_BLOCK: self._parse_name_or_new_description,
            StateType.PARSED_DESCRIPTION: self._parse_name_or_next_description,
            StateType.PARSED_NAME: self._parse_dimensions,
            StateType.PARSING_POINTS: self._parse_next_point,
            StateType.INVALID_STATE: self._parse_name_or_new_description,
        }

        self._finalise: Dict[StateType, Callable[[], None]] = {
            StateType.NEW_BLOCK: self._noop,
            StateType.PARSED_DESCRIPTION: self._add_current_block_as_incomplete_error,
            StateType.PARSED_NAME: self._add_current_block_as_incomplete_error,
            StateType.PARSING_POINTS: self._add_current_block_as_incomplete_error,
            StateType.INVALID_STATE: self._noop,
        }

    def feed_line(self, line: str) -> None:
        """Parse the next line with this Parser.

        Args:
            line (str): The line to parse
        """

        if not Parser._is_empty_line(line):
            self._feed_line[self._state](line)
        else:
            self._handle_empty_line()

        self._increment_line()

    def finalize(self) -> Sequence[PolyObject]:
        """Finalize parsing and return the constructed PolyObject.

        Raises:
            ValueError: When the plifile is invalid.

        Returns:
            PolyObject:
                A PolyObject containing the constructed PolyObject instances.
        """
        self._error_builder.end_invalid_block(self._line)
        last_error_msg = self._error_builder.finalize_previous_error()
        if last_error_msg is not None:
            self._notify_as_error(last_error_msg)

        self._finalise[self._state]()

        return self._poly_objects

    def _new_block(self, offset: int = 0) -> None:
        self._state = StateType.NEW_BLOCK
        self._current_block = Block(start_line=(self._line + offset))

    def _finish_block(self):
        (obj, warnings) = self._current_block.finalize()  # type: ignore
        self._poly_objects.append(obj)

        for msg in warnings:
            self._notify_as_warning(msg)

        last_error = self._error_builder.finalize_previous_error()
        if last_error is not None:
            self._notify_as_error(last_error)

    def _increment_line(self) -> None:
        self._line += 1

    def _noop(self, *_, **__) -> None:
        # no operation
        pass

    def _add_current_block_as_incomplete_error(self) -> None:
        msg = ParseMsg(
            line_start=self._current_block.start_line,
            line_end=self._line,
            reason="EoF encountered before the block is finished.",
        )
        self._notify_as_error(msg)

    def _parse_name_or_new_description(self, line: str) -> None:
        if Parser._is_comment(line):
            self._handle_new_description(line)
        elif Parser._is_name(line):
            self._handle_parse_name(line)
        elif self._state != StateType.INVALID_STATE:
            self._handle_new_error(
                "Settings of block might be incorrect, expected a valid name or description"
            )
            return

        # If we come from an invalid state, and we started a correct new block
        # we will end the previous invalid block, if it exists.
        self._error_builder.end_invalid_block(self._line)

    def _parse_name_or_next_description(self, line: str) -> None:
        if Parser._is_comment(line):
            self._handle_next_description(line)
        elif Parser._is_name(line):
            self._handle_parse_name(line)
        else:
            self._handle_new_error("Expected a valid name or description")

    def _parse_dimensions(self, line: str) -> None:
        dimensions = Parser._convert_to_dimensions(line)

        if dimensions is not None:
            self._current_block.dimensions = dimensions
            self._current_block.points = []
            self._current_point = 0
            self._state = StateType.PARSING_POINTS
        else:
            self._handle_new_error("Expected valid dimensions")

    def _parse_next_point(self, line: str) -> None:
        point = Parser._convert_to_point(
            line, self._current_block.dimensions[1], self._has_z_value  # type: ignore
        )

        if point is not None:
            self._current_block.points.append(point)  # type: ignore
            self._current_point += 1

            if self._current_block.dimensions[0] == self._current_point:  # type: ignore
                self._finish_block()
                self._new_block(offset=1)

        else:
            self._handle_new_error("Expected a valid next point")
            # we parse the line again, as it might be the first line of a new valid
            # block. For example when the invalid block was missing points.
            self._feed_line[self._state](line)

    def _handle_parse_name(self, line: str) -> None:
        self._current_block.name = Parser._convert_to_name(line)
        self._state = StateType.PARSED_NAME

    def _handle_new_description(self, line: str) -> None:
        comment = Parser._convert_to_comment(line)
        self._current_block.description = [
            comment,
        ]
        self._state = StateType.PARSED_DESCRIPTION

    def _handle_next_description(self, line: str) -> None:
        comment = Parser._convert_to_comment(line)
        self._current_block.description.append(comment)  # type: ignore

    def _handle_empty_line(self) -> None:
        if self._state != StateType.INVALID_STATE:
            self._current_block.empty_lines.append(self._line)

    def _handle_new_error(self, reason: str) -> None:
        self._error_builder.start_invalid_block(
            self._current_block.start_line, self._line, reason
        )
        self._state = StateType.INVALID_STATE

    def _notify_as_warning(self, msg: ParseMsg) -> None:
        warning_message = msg.format_parsemsg_to_string(self._file_path)
        warnings.warn(warning_message)

    def _notify_as_error(self, msg: ParseMsg) -> None:
        error_message = msg.format_parsemsg_to_string(self._file_path)
        raise ValueError(f"Invalid formatted plifile, {error_message}")

    @staticmethod
    def _is_empty_line(line: str) -> bool:
        return len(line.strip()) == 0

    @staticmethod
    def _is_name(line: str) -> bool:
        stripped = line.strip()
        return len(stripped) >= 1 and line[0] != "*"

    @staticmethod
    def _convert_to_name(line: str) -> str:
        return line.strip()

    @staticmethod
    def _is_comment(line: str) -> bool:
        return line.strip().startswith("*")

    @staticmethod
    def _convert_to_comment(line: str) -> str:
        return line.strip()[1:]

    @staticmethod
    def _convert_to_dimensions(line: str) -> Optional[Tuple[int, int]]:
        stripped = line.strip()
        elems = stripped.split()

        if len(elems) != 2:
            return None

        try:
            n_rows = int(elems[0])
            n_cols = int(elems[1])

            if n_rows <= 0 or n_cols <= 0:
                return None

            return (n_rows, n_cols)
        except ValueError:
            return None

    @staticmethod
    def _convert_to_point(
        line: str, expected_n_points: int, has_z: bool
    ) -> Optional[Point]:
        stripped = line.strip()
        elems = stripped.split()

        if len(elems) < expected_n_points:
            return None

        try:
            values = list(float(x) for x in elems[:expected_n_points])

            if has_z:
                x, y, z, *data = values
            else:
                x, y, *data = values
                z = None  # type: ignore

            return Point(x=x, y=y, z=z, data=data)

        except ValueError:
            return None

__init__(file_path, has_z_value=False)

Create a new Parser

Parameters:

Name Type Description Default
file_path Path

Name of the file being parsed, only used for providing proper warnings.

 required
has_z_value bool

Whether to interpret the third column as z-coordinates. Defaults to False.

 False
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
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
def __init__(self, file_path: Path, has_z_value: bool = False) -> None:
    """Create a new Parser

    Args:
        file_path (Path):
            Name of the file being parsed, only used for providing proper warnings.
        has_z_value (bool, optional):
            Whether to interpret the third column as z-coordinates.
            Defaults to False.
    """
    self._has_z_value = has_z_value
    self._file_path = file_path

    self._line = 0
    self._new_block()

    self._error_builder = ErrorBuilder()

    self._poly_objects: List[PolyObject] = []

    self._current_point: int = 0

    self._feed_line: Dict[StateType, Callable[[str], None]] = {
        StateType.NEW_BLOCK: self._parse_name_or_new_description,
        StateType.PARSED_DESCRIPTION: self._parse_name_or_next_description,
        StateType.PARSED_NAME: self._parse_dimensions,
        StateType.PARSING_POINTS: self._parse_next_point,
        StateType.INVALID_STATE: self._parse_name_or_new_description,
    }

    self._finalise: Dict[StateType, Callable[[], None]] = {
        StateType.NEW_BLOCK: self._noop,
        StateType.PARSED_DESCRIPTION: self._add_current_block_as_incomplete_error,
        StateType.PARSED_NAME: self._add_current_block_as_incomplete_error,
        StateType.PARSING_POINTS: self._add_current_block_as_incomplete_error,
        StateType.INVALID_STATE: self._noop,
    }

feed_line(line)

Parse the next line with this Parser.

Parameters:

Name Type Description Default
line str

The line to parse

 required
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
def feed_line(self, line: str) -> None:
    """Parse the next line with this Parser.

    Args:
        line (str): The line to parse
    """

    if not Parser._is_empty_line(line):
        self._feed_line[self._state](line)
    else:
        self._handle_empty_line()

    self._increment_line()

finalize()

Finalize parsing and return the constructed PolyObject.

Raises:

Type Description
ValueError

When the plifile is invalid.

Returns:

Name Type Description
PolyObject Sequence[PolyObject]

A PolyObject containing the constructed PolyObject instances.
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def finalize(self) -> Sequence[PolyObject]:
    """Finalize parsing and return the constructed PolyObject.

    Raises:
        ValueError: When the plifile is invalid.

    Returns:
        PolyObject:
            A PolyObject containing the constructed PolyObject instances.
    """
    self._error_builder.end_invalid_block(self._line)
    last_error_msg = self._error_builder.finalize_previous_error()
    if last_error_msg is not None:
        self._notify_as_error(last_error_msg)

    self._finalise[self._state]()

    return self._poly_objects

StateType

Bases: IntEnum

The types of state of a Parser.

Source code in hydrolib/core/dflowfm/polyfile/parser.py 
233
234
235
236
237
238
239
240
class StateType(IntEnum):
    """The types of state of a Parser."""

    NEW_BLOCK = 0
    PARSED_DESCRIPTION = 1
    PARSED_NAME = 2
    PARSING_POINTS = 3
    INVALID_STATE = 4

read_polyfile(filepath, has_z_values=None)

Read the specified file and return the corresponding data.

The file is expected to follow the .pli(z) / .pol convention. A .pli(z) or .pol file is defined as consisting of a number of blocks of lines adhering to the following format:

  • Optional description record consisting of one or more lines starting with '*'. These will be ignored.
  • Name consisting of a character string
  • Two integers, Nr and Nc, representing the numbers of rows and columns respectively
  • Nr number of data points, consisting of Nc floats separated by whitespace

For example: 

...
*
* Polyline L008
*
L008
4 2
    131595.0 549685.0
    131750.0 549865.0
    131595.0 550025.0
    131415.0 550175.0
...

Note that the points can be arbitrarily indented, and the comments are optional.

if no has_z_value has been defined, it will be based on the file path extensions of the filepath: - .pliz will default to True - .pli and .pol will default to False

Empty lines and unexpected whitespace will be flagged as warnings, and ignored.

If invalid syntax is detected within a block, an error will be created. This block will be ignored for the purpose of creating PolyObject instances. Once an error is encountered, any following lines will be marked as part of the invalid block, until a new valid block is found. Note that this means that sequential invalid blocks will be reported as a single invalid block. Such invalid blocks will be reported as warnings.

Parameters:

Name Type Description Default
filepath Path

Path to the pli(z)/pol convention structured file.

 required
has_z_values Optional[bool]

Whether to create points containing a z-value. Defaults to None.

 None

Raises:

Type Description
ValueError

When the plifile is invalid.

Returns:

Name Type Description
Dict Dict

The dictionary describing the data of a PolyObject.
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
def read_polyfile(filepath: Path, has_z_values: Optional[bool] = None) -> Dict:
    """Read the specified file and return the corresponding data.

    The file is expected to follow the .pli(z) / .pol convention. A .pli(z) or .pol
    file is defined as consisting of a number of blocks of lines adhering to the
    following format:

    - Optional description record consisting of one or more lines starting with '*'.
        These will be ignored.
    - Name consisting of a character string
    - Two integers, Nr and Nc, representing the numbers of rows and columns respectively
    - Nr number of data points, consisting of Nc floats separated by whitespace

    For example:
    ```
    ...
    *
    * Polyline L008
    *
    L008
    4 2
        131595.0 549685.0
        131750.0 549865.0
        131595.0 550025.0
        131415.0 550175.0
    ...
    ```

    Note that the points can be arbitrarily indented, and the comments are optional.

    if no has_z_value has been defined, it will be based on the file path
    extensions of the filepath:
    - .pliz will default to True
    - .pli and .pol will default to False

    Empty lines and unexpected whitespace will be flagged as warnings, and ignored.

    If invalid syntax is detected within a block, an error will be created. This block
    will be ignored for the purpose of creating PolyObject instances.
    Once an error is encountered, any following lines will be marked as part of the
    invalid block, until a new valid block is found. Note that this means that sequential
    invalid blocks will be reported as a single invalid block. Such invalid blocks will
    be reported as warnings.

    Args:
        filepath:
            Path to the pli(z)/pol convention structured file.
        has_z_values:
            Whether to create points containing a z-value. Defaults to None.

    Raises:
        ValueError: When the plifile is invalid.

    Returns:
        Dict: The dictionary describing the data of a PolyObject.
    """
    if has_z_values is None:
        has_z_values = _determine_has_z_value(filepath)

    parser = Parser(filepath, has_z_value=has_z_values)

    with filepath.open("r", encoding="utf8") as f:
        for line in f:
            parser.feed_line(line)

    objs = parser.finalize()

    return {"has_z_values": has_z_values, "objects": objs}

Serializer

Serializer

Serializer provides several static serialize methods for the models.

Source code in hydrolib/core/dflowfm/polyfile/serializer.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
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
class Serializer:
    """Serializer provides several static serialize methods for the models."""

    @staticmethod
    def serialize_description(description: Optional[Description]) -> Iterable[str]:
        """Serialize the Description to a string which can be used within a polyfile.

        Returns:
            str: The serialised equivalent of this Description
        """
        if description is None:
            return []
        if description.content == "":
            return [
                "*",
            ]
        return (f"*{v.rstrip()}" for v in (description.content + "\n").splitlines())

    @staticmethod
    def serialize_metadata(metadata: Metadata) -> Iterable[str]:
        """Serialize this Metadata to a string which can be used within a polyfile.

        The number of rows and number of columns are separated by four spaces.

        Returns:
            str: The serialised equivalent of this Metadata
        """
        return [metadata.name, f"{metadata.n_rows}    {metadata.n_columns}"]

    @staticmethod
    def serialize_point(point: Point, config: SerializerConfig) -> str:
        """Serialize this Point to a string which can be used within a polyfile.

        the point data is indented with 4 spaces, and the individual values are
        separated by 4 spaces as well.

        Args:
            point (Point): The point to serialize.
            config (SerializerConfig): The serialization configuration.

        Returns:
            str: The serialised equivalent of this Point
        """
        space = 4 * " "
        float_format = lambda v: f"{v:{config.float_format}}"
        return space + space.join(
            float_format(v) for v in Serializer._get_point_values(point)
        )

    @staticmethod
    def _get_point_values(point: Point) -> Generator[float, None, None]:
        yield point.x
        yield point.y
        if point.z:
            yield point.z
        for value in point.data:
            yield value

    @staticmethod
    def serialize_poly_object(
        obj: PolyObject, config: SerializerConfig
    ) -> Iterable[str]:
        """Serialize this PolyObject to a string which can be used within a polyfile.

        Args:
            obj (PolyObject): The poly object to serializer.
            config (SerializerConfig): The serialization configuration.

        Returns:
            str: The serialised equivalent of this PolyObject
        """

        description = Serializer.serialize_description(obj.description)
        metadata = Serializer.serialize_metadata(obj.metadata)
        points = [Serializer.serialize_point(obj, config) for obj in obj.points]
        return chain(description, metadata, points)

serialize_description(description) staticmethod

Serialize the Description to a string which can be used within a polyfile.

Returns:

Name Type Description
str Iterable[str]

The serialised equivalent of this Description
Source code in hydrolib/core/dflowfm/polyfile/serializer.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@staticmethod
def serialize_description(description: Optional[Description]) -> Iterable[str]:
    """Serialize the Description to a string which can be used within a polyfile.

    Returns:
        str: The serialised equivalent of this Description
    """
    if description is None:
        return []
    if description.content == "":
        return [
            "*",
        ]
    return (f"*{v.rstrip()}" for v in (description.content + "\n").splitlines())

serialize_metadata(metadata) staticmethod

Serialize this Metadata to a string which can be used within a polyfile.

The number of rows and number of columns are separated by four spaces.

Returns:

Name Type Description
str Iterable[str]

The serialised equivalent of this Metadata
Source code in hydrolib/core/dflowfm/polyfile/serializer.py 
32
33
34
35
36
37
38
39
40
41
@staticmethod
def serialize_metadata(metadata: Metadata) -> Iterable[str]:
    """Serialize this Metadata to a string which can be used within a polyfile.

    The number of rows and number of columns are separated by four spaces.

    Returns:
        str: The serialised equivalent of this Metadata
    """
    return [metadata.name, f"{metadata.n_rows}    {metadata.n_columns}"]

serialize_point(point, config) staticmethod

Serialize this Point to a string which can be used within a polyfile.

the point data is indented with 4 spaces, and the individual values are separated by 4 spaces as well.

Parameters:

Name Type Description Default
point Point

The point to serialize.

 required
config SerializerConfig

The serialization configuration.

 required

Returns:

Name Type Description
str str

The serialised equivalent of this Point
Source code in hydrolib/core/dflowfm/polyfile/serializer.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@staticmethod
def serialize_point(point: Point, config: SerializerConfig) -> str:
    """Serialize this Point to a string which can be used within a polyfile.

    the point data is indented with 4 spaces, and the individual values are
    separated by 4 spaces as well.

    Args:
        point (Point): The point to serialize.
        config (SerializerConfig): The serialization configuration.

    Returns:
        str: The serialised equivalent of this Point
    """
    space = 4 * " "
    float_format = lambda v: f"{v:{config.float_format}}"
    return space + space.join(
        float_format(v) for v in Serializer._get_point_values(point)
    )

serialize_poly_object(obj, config) staticmethod

Serialize this PolyObject to a string which can be used within a polyfile.

Parameters:

Name Type Description Default
obj PolyObject

The poly object to serializer.

 required
config SerializerConfig

The serialization configuration.

 required

Returns:

Name Type Description
str Iterable[str]

The serialised equivalent of this PolyObject
Source code in hydrolib/core/dflowfm/polyfile/serializer.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@staticmethod
def serialize_poly_object(
    obj: PolyObject, config: SerializerConfig
) -> Iterable[str]:
    """Serialize this PolyObject to a string which can be used within a polyfile.

    Args:
        obj (PolyObject): The poly object to serializer.
        config (SerializerConfig): The serialization configuration.

    Returns:
        str: The serialised equivalent of this PolyObject
    """

    description = Serializer.serialize_description(obj.description)
    metadata = Serializer.serialize_metadata(obj.metadata)
    points = [Serializer.serialize_point(obj, config) for obj in obj.points]
    return chain(description, metadata, points)

write_polyfile(path, data, config)

Write the data to a new file at path

Parameters:

Name Type Description Default
path Path

The path to write the data to

 required
data Sequence[PolyObject]

The poly objects to write

 required
config SerializerConfig

The serialization configuration.

 required
Source code in hydrolib/core/dflowfm/polyfile/serializer.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def write_polyfile(
    path: Path, data: Sequence[PolyObject], config: SerializerConfig
) -> None:
    """Write the data to a new file at path

    Args:
        path (Path): The path to write the data to
        data (Sequence[PolyObject]): The poly objects to write
        config (SerializerConfig): The serialization configuration.
    """
    serialized_poly_objects = [
        Serializer.serialize_poly_object(poly_object, config) for poly_object in data
    ]
    serialized_data = chain.from_iterable(serialized_poly_objects)

    path.parent.mkdir(parents=True, exist_ok=True)

    with path.open("w", encoding="utf8") as f:

        for line in serialized_data:
            f.write(line)
            f.write("\n")