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 (BaseModel)` `pydantic-model` ¶

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.

`Metadata (BaseModel)` `pydantic-model` ¶

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.

`Point (BaseModel)` `pydantic-model` ¶

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.

`PolyFile (FileModel)` `pydantic-model` ¶

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

`PolyObject (BaseModel)` `pydantic-model` ¶

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

Parser¶

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

`Block (BaseModel)` `pydantic-model` ¶

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.

`finalize(self) -> Optional[Tuple[hydrolib.core.io.polyfile.models.PolyObject, List[hydrolib.core.io.polyfile.parser.ParseMsg]]]` ¶

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]]]`	The constructed PolyObject and warnings encountered while parsing it.

Source code in hydrolib/core/io/polyfile/parser.py

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.

`init(self) -> None` `special` ¶

Create a new ErorrorBuilder

Source code in hydrolib/core/io/polyfile/parser.py

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

`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.

Parameters:

Name	Type	Description	Default
`line`	`int`	the final line of this invalid block	required

Source code in hydrolib/core/io/polyfile/parser.py

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(self) -> Optional[hydrolib.core.io.polyfile.parser.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:

Type	Description
`Optional[ParseMsg]`	The corresponding ParseMsg if an InvalidBlock exists.

Source code in hydrolib/core/io/polyfile/parser.py

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(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.

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/io/polyfile/parser.py

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 (BaseModel)` `pydantic-model` ¶

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.

`to_msg(self) -> ParseMsg` ¶

Convert this InvalidBlock to the corresponding ParseMsg

Returns:

Type	Description
`ParseMsg`	The ParseMsg corresponding with this InvalidBlock

Source code in hydrolib/core/io/polyfile/parser.py

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 (BaseModel)` `pydantic-model` ¶

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.

`notify_as_warning(self, file_path: Optional[pathlib.Path] = None)` ¶

Call warnings.warn with a formatted string describing this ParseMsg

Parameters:

Name	Type	Description	Default
`file_path`	`Optional[Path]`	The file path mentioned in the warning if specified. Defaults to None.	`None`

Source code in hydrolib/core/io/polyfile/parser.py

def notify_as_warning(self, file_path: Optional[Path] = None):
    """Call warnings.warn with a formatted string describing this ParseMsg

    Args:
        file_path (Optional[Path], optional):
            The file path mentioned in the warning 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 ""

    warnings.warn(f"{self.reason}{block_suffix}{col_suffix}{file_suffix}")

`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.

`init(self, file_path: Path, has_z_value: bool = False) -> None` `special` ¶

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/io/polyfile/parser.py

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,
    }

    self._handle_ws: Dict[StateType, Callable[[str], None]] = {
        StateType.NEW_BLOCK: self._log_ws_warning,
        StateType.PARSED_DESCRIPTION: self._log_ws_warning,
        StateType.PARSED_NAME: self._log_ws_warning,
        StateType.PARSING_POINTS: self._noop,
        StateType.INVALID_STATE: self._noop,
    }

`feed_line(self, line: str) -> None` ¶

Parse the next line with this Parser.

Parameters:

Name	Type	Description	Default
`line`	`str`	The line to parse	required

Source code in hydrolib/core/io/polyfile/parser.py

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._handle_ws[self._state](line)
        self._feed_line[self._state](line)
    else:
        self._handle_empty_line()

    self._increment_line()

`finalize(self) -> Sequence[hydrolib.core.io.polyfile.models.PolyObject]` ¶

Finalize parsing and return the constructed PolyObject.

Returns:

Type	Description
`PolyObject`	A PolyObject containing the constructed PolyObject instances.

Source code in hydrolib/core/io/polyfile/parser.py

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

    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._handle_parse_msg(last_error_msg)

    self._finalise[self._state]()

    return self._poly_objects

`StateType (IntEnum)` ¶

The types of state of a Parser.

`read_polyfile(filepath: Path, has_z_values: bool) -> 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 non-blank 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`	`bool`	Whether to create points containing a z-value. Defaults to None.	required

Returns:

Type	Description
`Dict`	The dictionary describing the data of a PolyObject.

Source code in hydrolib/core/io/polyfile/parser.py

def read_polyfile(filepath: Path, has_z_values: bool) -> 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 non-blank 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.

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

`serialize_description(description: Optional[hydrolib.core.io.polyfile.models.Description]) -> Iterable[str]` `staticmethod` ¶

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

Returns:

Type	Description
`str`	The serialised equivalent of this Description

Source code in hydrolib/core/io/polyfile/serializer.py

@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.splitlines())

`serialize_metadata(metadata: Metadata) -> Iterable[str]` `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:

Type	Description
`str`	The serialised equivalent of this Metadata

Source code in hydrolib/core/io/polyfile/serializer.py

@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: Point) -> str` `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.

Returns:

Type	Description
`str`	The serialised equivalent of this Point

Source code in hydrolib/core/io/polyfile/serializer.py

@staticmethod
def serialize_point(point: Point) -> 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.

    Returns:
        str: The serialised equivalent of this Point
    """
    z_val = f"{point.z}    " if point.z is not None else ""
    data_vals = "    ".join(str(v) for v in point.data)
    return f"    {point.x}    {point.y}    {z_val}{data_vals}".rstrip()

`serialize_poly_object(obj: PolyObject) -> Iterable[str]` `staticmethod` ¶

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

Returns:

Type	Description
`str`	The serialised equivalent of this Point

Source code in hydrolib/core/io/polyfile/serializer.py

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

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

    description = Serializer.serialize_description(obj.description)
    metadata = Serializer.serialize_metadata(obj.metadata)
    points = map(Serializer.serialize_point, obj.points)
    return chain(description, metadata, points)

`write_polyfile(path: Path, data: Dict) -> None` ¶

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`	`PolyFile`	The data to write	required

Source code in hydrolib/core/io/polyfile/serializer.py

def write_polyfile(path: Path, data: Dict) -> None:
    """Write the data to a new file at path

    Args:
        path (Path): The path to write the data to
        data (PolyFile): The data to write
    """
    serialized_data = chain.from_iterable(
        map(Serializer.serialize_poly_object, data["objects"])
    )

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

    with path.open("w") as f:

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

Polyline/polygon files¶

Model¶

Description (BaseModel) pydantic-model ¶

Metadata (BaseModel) pydantic-model ¶

Point (BaseModel) pydantic-model ¶

PolyFile (FileModel) pydantic-model ¶

PolyObject (BaseModel) pydantic-model ¶

Parser¶

Block (BaseModel) pydantic-model ¶

finalize(self) -> Optional[Tuple[hydrolib.core.io.polyfile.models.PolyObject, List[hydrolib.core.io.polyfile.parser.ParseMsg]]] ¶

ErrorBuilder ¶

__init__(self) -> None special ¶

end_invalid_block(self, line: int) -> None ¶

finalize_previous_error(self) -> Optional[hydrolib.core.io.polyfile.parser.ParseMsg] ¶

start_invalid_block(self, block_start: int, invalid_line: int, reason: str) -> None ¶

InvalidBlock (BaseModel) pydantic-model ¶

to_msg(self) -> ParseMsg ¶

ParseMsg (BaseModel) pydantic-model ¶

notify_as_warning(self, file_path: Optional[pathlib.Path] = None) ¶

Parser ¶

__init__(self, file_path: Path, has_z_value: bool = False) -> None special ¶

feed_line(self, line: str) -> None ¶

finalize(self) -> Sequence[hydrolib.core.io.polyfile.models.PolyObject] ¶

StateType (IntEnum) ¶

read_polyfile(filepath: Path, has_z_values: bool) -> Dict ¶

Serializer¶

Serializer ¶

serialize_description(description: Optional[hydrolib.core.io.polyfile.models.Description]) -> Iterable[str] staticmethod ¶

serialize_metadata(metadata: Metadata) -> Iterable[str] staticmethod ¶

serialize_point(point: Point) -> str staticmethod ¶

serialize_poly_object(obj: PolyObject) -> Iterable[str] staticmethod ¶

write_polyfile(path: Path, data: Dict) -> None ¶

`Description (BaseModel)` `pydantic-model` ¶

`Metadata (BaseModel)` `pydantic-model` ¶

`Point (BaseModel)` `pydantic-model` ¶

`PolyFile (FileModel)` `pydantic-model` ¶

`PolyObject (BaseModel)` `pydantic-model` ¶

`Block (BaseModel)` `pydantic-model` ¶

`finalize(self) -> Optional[Tuple[hydrolib.core.io.polyfile.models.PolyObject, List[hydrolib.core.io.polyfile.parser.ParseMsg]]]` ¶

`ErrorBuilder` ¶

`init(self) -> None` `special` ¶

`end_invalid_block(self, line: int) -> None` ¶

`finalize_previous_error(self) -> Optional[hydrolib.core.io.polyfile.parser.ParseMsg]` ¶

`start_invalid_block(self, block_start: int, invalid_line: int, reason: str) -> None` ¶

`InvalidBlock (BaseModel)` `pydantic-model` ¶

`to_msg(self) -> ParseMsg` ¶

`ParseMsg (BaseModel)` `pydantic-model` ¶

`notify_as_warning(self, file_path: Optional[pathlib.Path] = None)` ¶

`Parser` ¶

`init(self, file_path: Path, has_z_value: bool = False) -> None` `special` ¶

`feed_line(self, line: str) -> None` ¶

`finalize(self) -> Sequence[hydrolib.core.io.polyfile.models.PolyObject]` ¶

`StateType (IntEnum)` ¶

`read_polyfile(filepath: Path, has_z_values: bool) -> Dict` ¶

`Serializer` ¶

`serialize_description(description: Optional[hydrolib.core.io.polyfile.models.Description]) -> Iterable[str]` `staticmethod` ¶

`serialize_metadata(metadata: Metadata) -> Iterable[str]` `staticmethod` ¶

`serialize_point(point: Point) -> str` `staticmethod` ¶

`serialize_poly_object(obj: PolyObject) -> Iterable[str]` `staticmethod` ¶

`write_polyfile(path: Path, data: Dict) -> None` ¶