Skip to content

Polyline/Polygon Files Documentation

Overview

Polyline (.pli/.pliz) 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. The hydrolib.core.dflowfm.polyfile module provides functionality for reading, writing, and manipulating these files.

Polyline files consist of one or more blocks, each containing: 1. An optional description (lines starting with *) 2. A name line 3. A dimensions line (number of rows and columns) 4. A series of points with coordinates and optional data

File Format

A typical polyline file has the following structure:

* Optional description line 1
* Optional description line 2
name_of_polyline
2    3    # number_of_points number_of_columns
    10.0    20.0    0.0    # x y z
    30.0    40.0    0.0    # x y z

The file can contain multiple such blocks, each defining a separate polyline object.

Class Diagram

Mermaid Class Diagram

ASCII Class Diagram

┌───────────────┐     ┌───────────────┐
│  Description  │     │    Metadata   │
├───────────────┤     ├───────────────┤
│ content: str  │     │ name: str     │
└───────────────┘     │ n_rows: int   │
                      │ n_columns: int│
                      └───────────────┘
                             ▲
                             │
┌───────────────┐     ┌─────┴─────────┐     ┌───────────────┐
│     Point     │◄────│   PolyObject  │     │   PolyFile    │
├───────────────┤     ├───────────────┤     ├───────────────┤
│ x: float      │     │ description   │◄────│ has_z_values  │
│ y: float      │     │ metadata      │     │ objects       │
│ z: Optional   │     │ points        │     └───────────────┘
│ data: List    │     └───────────────┘             │
└───────────────┘                                   │
                                                    ▼
                      ┌───────────────┐     ┌───────────────┐
                      │    Parser     │     │  Serializer   │
                      ├───────────────┤     ├───────────────┤
                      │ feed_line()   │     │ serialize_*() │
                      │ finalize()    │     └───────────────┘
                      └───────────────┘

Core Classes

PolyFile

PolyFile is the main class representing a polyline file. It inherits from ParsableFileModel and provides methods for reading, writing, and accessing polyline data.

Key methods: - x property: Returns all x-coordinates as a list - y property: Returns all y-coordinates as a list - get_z_sources_sinks(): Gets z-values for source and sink points - number_of_points property: Returns the total number of points

PolyObject

PolyObject represents a single block in a polyline file, containing a description, metadata, and a list of points.

Point

Point represents a single point in a polyline, with x and y coordinates, an optional z coordinate, and additional data values.

Metadata

Metadata contains information about a polyline block, including the name, number of rows (points), and number of columns (values per point).

Description

Description represents comments for a polyline block, with content starting with * in the file.

Sequence Diagram: Reading a Polyfile

┌─────────┐          ┌──────────┐          ┌────────┐          ┌────────────┐
│  Client │          │ PolyFile │          │ Parser │          │ PolyObject │
└────┬────┘          └────┬─────┘          └───┬────┘          └─────┬──────┘
     │                     │                    │                     │
     │ PolyFile(filepath)  │                    │                     │
     │────────────────────>│                    │                     │
     │                     │                    │                     │
     │                     │ read_polyfile()    │                     │
     │                     │───────────────────>│                     │
     │                     │                    │                     │
     │                     │                    │ feed_line()         │
     │                     │                    │─────────────────────│
     │                     │                    │                     │
     │                     │                    │ finalize()          │
     │                     │                    │─────────────────────│
     │                     │                    │                     │
     │                     │<───────────────────│                     │
     │                     │                    │                     │
     │<────────────────────│                    │                     │
     │                     │                    │                     │

Sequence Diagram: Writing a Polyfile

┌─────────┐          ┌──────────┐          ┌────────────┐          ┌────────────┐
│  Client │          │ PolyFile │          │ Serializer │          │    File    │
└────┬────┘          └────┬─────┘          └─────┬──────┘          └─────┬──────┘
     │                     │                      │                       │
     │ polyfile.save()     │                      │                       │
     │────────────────────>│                      │                       │
     │                     │                      │                       │
     │                     │ write_polyfile()     │                       │
     │                     │─────────────────────>│                       │
     │                     │                      │                       │
     │                     │                      │ serialize_poly_object()│
     │                     │                      │───────────────────────│
     │                     │                      │                       │
     │                     │                      │ write lines           │
     │                     │                      │───────────────────────│
     │                     │                      │                       │
     │                     │<─────────────────────│                       │
     │                     │                      │                       │
     │<────────────────────│                      │                       │
     │                     │                      │                       │

Usage Examples

Reading a Polyline File

from pathlib import Path
from hydrolib.core.dflowfm.polyfile.models import PolyFile

# Read a polyline file
polyfile = PolyFile(filepath=Path("path/to/file.pli"))

# Access the polyline objects
for poly_obj in polyfile.objects:
    print(f"Name: {poly_obj.metadata.name}")
    print(f"Number of points: {len(poly_obj.points)}")

    # Access points
    for point in poly_obj.points:
        print(f"  Point: ({point.x}, {point.y})")
        if point.z is not None:
            print(f"  Z-value: {point.z}")
        if point.data:
            print(f"  Data: {point.data}")

Creating and Writing a Polyline File

from pathlib import Path
from hydrolib.core.dflowfm.polyfile.models import PolyFile, PolyObject, Metadata, Point, Description

# Create a new polyline file
polyfile = PolyFile(has_z_values=True)

# Create a polyline object
poly_obj = PolyObject(
    description=Description(content="This is a test polyline"),
    metadata=Metadata(name="test_polyline", n_rows=2, n_columns=3),
    points=[
        Point(x=10.0, y=20.0, z=0.0, data=[]),
        Point(x=30.0, y=40.0, z=1.0, data=[])
    ]
)

# Add the polyline object to the file
polyfile.objects = [poly_obj]

# Save the file
polyfile.save(filepath=Path("path/to/output.pli"))

Accessing Coordinates

from pathlib import Path
from hydrolib.core.dflowfm.polyfile.models import PolyFile

# Read a polyline file
polyfile = PolyFile(filepath=Path("path/to/file.pli"))

# Get all x and y coordinates
x_coords = polyfile.x
y_coords = polyfile.y

# Get the total number of points
total_points = polyfile.number_of_points

# Get z-values for source and sink points (for .pliz files)
z_source, z_sink = polyfile.get_z_sources_sinks()

File Extensions

The module supports different file extensions: - .pli: Standard polyline file - .pol: Polygon file (same format as .pli) - .pliz: Polyline file with z-values

Error Handling

The parser includes robust error handling for various issues: - Invalid dimensions - Missing points - Invalid point data - Empty lines (generates warnings) - Incomplete blocks

API Reference

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 
11
12
13
14
15
16
17
18
19
20
21
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 
24
25
26
27
28
29
30
31
32
33
34
35
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 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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.

Notes
  • The has_z_values attribute is used to determine if the PolyFile contains z-values.
  • The has_z_values is false by default and should be set to true if the PolyFile path ends with .pliz.
  • The ***.pliz file should have a 2*3 structure, where the third column contains the z-values, otherwise (the parser will give an error).
  • If there is a label in the file, the parser will ignore the label and read the file as a normal polyline file. 
    tfl_01
    2 2
    0.00 1.00 #zee
    0.00 2.00 #zee
  • if the file is .pliz, and the dimensions are 2*5 the first three columns will be considered as x, y, z values and the last two columns will be considered as data values. 
    L1
    2 5
    63.35 12.95 -4.20 -5.35 0
    45.20 6.35 -3.00 -2.90 0
Source code in hydrolib/core/dflowfm/polyfile/models.py 
 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
class PolyFile(ParsableFileModel):
    """
    Poly-file (.pol/.pli/.pliz) representation.

    Notes:
        - The `has_z_values` attribute is used to determine if the PolyFile contains z-values.
        - The `has_z_values` is false by default and should be set to true if the PolyFile path ends with `.pliz`.
        - The `***.pliz` file should have a 2*3 structure, where the third column contains the z-values, otherwise
        (the parser will give an error).
        - If there is a label in the file, the parser will ignore the label and read the file as a normal polyline file.
        ```
        tfl_01
            2 2
            0.00 1.00 #zee
            0.00 2.00 #zee
        ```
        - if the file is .pliz, and the dimensions are 2*5 the first three columns will be considered as x, y, z values
        and the last two columns will be considered as data values.
        ```
        L1
            2 5
            63.35 12.95 -4.20 -5.35 0
            45.20 6.35 -3.00 -2.90 0
        ```
    """

    has_z_values: bool = False
    objects: Sequence[PolyObject] = Field(default_factory=list)

    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:
        # Prevent circular dependency in Parser
        from hydrolib.core.dflowfm.polyfile.parser import read_polyfile

        return read_polyfile

    @property
    def x(self) -> List[float]:
        """X-coordinates of all points in the PolyFile."""
        return [point.x for obj in self.objects for point in obj.points]

    @property
    def y(self) -> List[float]:
        """Y-coordinates of all points in the PolyFile."""
        return [point.y for obj in self.objects for point in obj.points]

    def get_z_sources_sinks(self) -> Tuple[List[float], List[float]]:
        """
        Get the z values of the source and sink points from the polyline file.

        Returns:
            z_source, z_sinkA: Tuple[List[float]]:
            If the polyline has data (more than 3 columns), then both the z_source and z_sink will be a list of two values.
            Otherwise, the z_source and the z_sink will be a single value each.

        Note:
             - calling this method on a polyline file that does not have z-values will return a list of None.

        Examples:
        in case the polyline has 3 columns:
            >>> polyline = PolyFile("tests/data/input/source-sink/leftsor.pliz")
            >>> z_source, z_sink = polyline.get_z_sources_sinks()
            >>> print(z_source, z_sink)
            [-3.0] [-4.2]

        in case the polyline has more than 3 columns:
            >>> polyline = PolyFile("tests/data/input/source-sink/leftsor-5-columns.pliz") #Doctest: +SKIP
            >>> z_source, z_sink = polyline.get_z_sources_sinks()
            >>> print(z_source, z_sink)
            [-3.0, -2.9] [-4.2, -5.35]

        in case the polyline does not have z-values:
            >>> root_dir = "tests/data/input/dflowfm_individual_files/polylines"
            >>> polyline = PolyFile(f"{root_dir}/boundary-polyline-no-z-no-label.pli")
            >>> z_source, z_sink = polyline.get_z_sources_sinks()
            >>> print(z_source, z_sink)
            [None] [None]
        """
        has_data = True if self.objects[0].points[0].data else False

        z_source_sink = []
        for elem in [0, -1]:
            point = self.objects[0].points[elem]
            if has_data:
                z_source_sink.append([point.z, point.data[0]])
            else:
                z_source_sink.append([point.z])

        z_sink: list[float | None] = z_source_sink[0]
        z_source: list[float | None] = z_source_sink[1]
        return z_source, z_sink

    @property
    def number_of_points(self) -> int:
        """Total number of points in the PolyFile."""
        return len(self.x)

number_of_points property

Total number of points in the PolyFile.

x property

X-coordinates of all points in the PolyFile.

y property

Y-coordinates of all points in the PolyFile.

get_z_sources_sinks()

Get the z values of the source and sink points from the polyline file.

Returns:

Type Description
List[float]

z_source, z_sinkA: Tuple[List[float]]:
List[float]

If the polyline has data (more than 3 columns), then both the z_source and z_sink will be a list of two values.
Tuple[List[float], List[float]]

Otherwise, the z_source and the z_sink will be a single value each.
Note
  • calling this method on a polyline file that does not have z-values will return a list of None.

Examples:

in case the polyline has 3 columns: >>> polyline = PolyFile("tests/data/input/source-sink/leftsor.pliz") >>> z_source, z_sink = polyline.get_z_sources_sinks() >>> print(z_source, z_sink) [-3.0][-4.2]

in case the polyline has more than 3 columns

polyline = PolyFile("tests/data/input/source-sink/leftsor-5-columns.pliz") #Doctest: +SKIP z_source, z_sink = polyline.get_z_sources_sinks() print(z_source, z_sink) [-3.0, -2.9][-4.2, -5.35]

in case the polyline does not have z-values

root_dir = "tests/data/input/dflowfm_individual_files/polylines" polyline = PolyFile(f"{root_dir}/boundary-polyline-no-z-no-label.pli") z_source, z_sink = polyline.get_z_sources_sinks() print(z_source, z_sink) [None][]

Source code in hydrolib/core/dflowfm/polyfile/models.py 
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
def get_z_sources_sinks(self) -> Tuple[List[float], List[float]]:
    """
    Get the z values of the source and sink points from the polyline file.

    Returns:
        z_source, z_sinkA: Tuple[List[float]]:
        If the polyline has data (more than 3 columns), then both the z_source and z_sink will be a list of two values.
        Otherwise, the z_source and the z_sink will be a single value each.

    Note:
         - calling this method on a polyline file that does not have z-values will return a list of None.

    Examples:
    in case the polyline has 3 columns:
        >>> polyline = PolyFile("tests/data/input/source-sink/leftsor.pliz")
        >>> z_source, z_sink = polyline.get_z_sources_sinks()
        >>> print(z_source, z_sink)
        [-3.0] [-4.2]

    in case the polyline has more than 3 columns:
        >>> polyline = PolyFile("tests/data/input/source-sink/leftsor-5-columns.pliz") #Doctest: +SKIP
        >>> z_source, z_sink = polyline.get_z_sources_sinks()
        >>> print(z_source, z_sink)
        [-3.0, -2.9] [-4.2, -5.35]

    in case the polyline does not have z-values:
        >>> root_dir = "tests/data/input/dflowfm_individual_files/polylines"
        >>> polyline = PolyFile(f"{root_dir}/boundary-polyline-no-z-no-label.pli")
        >>> z_source, z_sink = polyline.get_z_sources_sinks()
        >>> print(z_source, z_sink)
        [None] [None]
    """
    has_data = True if self.objects[0].points[0].data else False

    z_source_sink = []
    for elem in [0, -1]:
        point = self.objects[0].points[elem]
        if has_data:
            z_source_sink.append([point.z, point.data[0]])
        else:
            z_source_sink.append([point.z])

    z_sink: list[float | None] = z_source_sink[0]
    z_source: list[float | None] = z_source_sink[1]
    return z_source, z_sink

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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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 
 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
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]]]:
        """Finalize 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):
        warnings = []
        if len(self.empty_lines) > 0:
            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()

Finalize 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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def finalize(self) -> Optional[Tuple[PolyObject, List[ParseMsg]]]:
    """Finalize 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 
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
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 
181
182
183
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 
204
205
206
207
208
209
210
211
212
213
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 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
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 
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
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 
165
166
167
168
169
170
171
172
173
174
175
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 
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
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 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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 
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
517
518
519
520
521
522
523
524
525
526
527
528
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, verbose: 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.
            verbose (bool, optional):
                Whether to show warnings when parsing the file. Defaults to False.
        """
        self._has_z_value = has_z_value
        self._file_path = file_path
        self._verbose = verbose

        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:
        """Add the empty line to the current block if it is not in an invalid state.

        Notes:
            - Empty lines are ignored, but we want to keep track of them for the purpose of warnings.
            - Empty lines are not added to the current block if it is in an invalid state.
        """
        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)
        if self._verbose:
            warnings.warn(warning_message, stacklevel=2)

    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, verbose=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
verbose bool

Whether to show warnings when parsing the file. Defaults to False.

 False
Source code in hydrolib/core/dflowfm/polyfile/parser.py 
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
def __init__(
    self, file_path: Path, has_z_value: bool = False, verbose: 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.
        verbose (bool, optional):
            Whether to show warnings when parsing the file. Defaults to False.
    """
    self._has_z_value = has_z_value
    self._file_path = file_path
    self._verbose = verbose

    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 
311
312
313
314
315
316
317
318
319
320
321
322
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 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
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 
234
235
236
237
238
239
240
241
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, verbose=False)

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
verbose bool

Whether to show warnings when parsing the file. Defaults to False.

 False

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 
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
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
def read_polyfile(
    filepath: Path, has_z_values: Optional[bool] = None, verbose: bool = False
) -> 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.
        verbose:
            Whether to show warnings when parsing the file. Defaults to False.

    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, verbose=verbose)

    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")