Bank Lines Module #

The Bank Lines module is responsible for detecting bank lines from hydrodynamic simulation results. It is one of the core components of the D-FAST Bank Erosion software.

Overview #

The Bank Lines module processes hydrodynamic simulation results to detect bank lines, which are the boundaries between wet and dry areas in the river. These bank lines are then used as input for bank erosion calculations. The module can detect bank lines for multiple simulations and combine them into a single set of bank lines.

classDiagram
    %% Main Classes

    class BankLines {
        -ConfigFile config_file
        -bool gui
        -Path output_dir
        -bool debug
        -dict plot_flags
        -float max_river_width
        +__init__(ConfigFile, bool)
        +detect()
        +mask(GeoSeries, Polygon)
        +plot(array, int, List, Tuple, List, ConfigFile)
        +save(List, GeoSeries, List, List, ConfigFile)
        +detect_bank_lines(BaseSimulationData, float, ConfigFile)
        -_calculate_water_depth(BaseSimulationData)
        -_generate_bank_lines(BaseSimulationData, array, array, array, float)
        -_progress_bar(int, int)
    }

    class ConfigFile {
        -ConfigParser config
        -str path
        +__init__(ConfigParser, Path)
        +read(Path)
        +write(str)
        +make_paths_absolute()
        +get_str(str, str, str)
        +get_bool(str, str, bool)
        +get_float(str, str, float, bool)
        +get_int(str, str, int, bool)
        +get_sim_file(str, str)
        +get_start_end_stations()
        +get_search_lines()
        +read_bank_lines(str)
        +get_parameter(str, str, List, Any, str, bool, List, bool)
        +get_bank_search_distances(int)
        +get_range(str, str)
        +get_river_center_line()
        +resolve(str)
        +relative_to(str)
        +get_plotting_flags(Path)
        +get_output_dir(str)
    }

    %% Data Models - Bank Lines
    class BankLinesRiverData {
        -ConfigFile config_file
        -LineString river_center_line
        -Tuple station_bounds
        +search_lines()
        -_get_bank_lines_simulation_data()
        +simulation_data()
    }

    class SearchLines {
        +List lines
        +LineGeometry mask
        +__init__(List, LineGeometry)
        +mask(List, LineString, float)
        -_select_closest_part(MultiLineString, LineString, float)
        +to_polygons()
    }

    %% Data Models - IO
    class LineGeometry {
        +LineString line
        +dict data
        +__init__(LineString, Tuple, str)
        +as_array()
        +add_data(Dict)
        +to_file(str, Dict)
        +mask(LineString, Tuple)
        -_find_mask_index(float, array)
        -_handle_bound(int, float, bool, array)
        -_interpolate_point(int, float, array)
        +intersect_with_line(array)
    }

    class BaseSimulationData {
        +array x_node
        +array y_node
        +array n_nodes
        +array face_node
        +array bed_elevation_location
        +array bed_elevation_values
        +array water_level_face
        +array water_depth_face
        +array velocity_x_face
        +array velocity_y_face
        +array chezy_face
        +float dry_wet_threshold
        +__init__(array, array, array, array, array, array, array, array, array, array, array, float)
        +read(str, str)
        +clip(LineString, float)
    }

    class BaseRiverData {
        -ConfigFile config_file
        -LineString river_center_line
        -Tuple station_bounds
        +__init__(ConfigFile)
        +get_bbox(array, float)
        +get_erosion_sim_data(int)
    }


    BankLines --> ConfigFile : uses
    BankLines --> BankLinesRiverData : uses
    BankLines --> SearchLines : uses
    BankLines --> LineGeometry : uses
    BankLines --> BaseSimulationData : uses

    SearchLines --> LineGeometry : uses

    BankLinesRiverData --|> BaseRiverData : inherits
    BankLinesRiverData --> ConfigFile : uses
    BankLinesRiverData --> SearchLines : uses
    BankLinesRiverData --> BaseSimulationData : uses

    BaseRiverData --> ConfigFile : uses
    BaseRiverData --> LineGeometry : uses

Components #

The Bank Lines module consists of the following components:

Main Classes #

`dfastbe.bank_lines.bank_lines` #

Bank line detection module.

`BankLines` #

Bases: BaseCalculator

Bank line detection class.

Source code in src/dfastbe/bank_lines/bank_lines.py

class BankLines(BaseCalculator):
    """Bank line detection class."""

    def __init__(self, config_file: ConfigFile, gui: bool = False):
        """Bank line initializer.

        Args:
            config_file : configparser.ConfigParser
                Analysis configuration settings.
            gui : bool
                Flag indicating whether this routine is called from the GUI.

        Examples:
            ```python
            >>> from unittest.mock import patch
            >>> from dfastbe.io.config import ConfigFile
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
            >>> bank_lines = BankLines(config_file)  # doctest: +ELLIPSIS
            N...e
            >>> isinstance(bank_lines, BankLines)
            True

            ```
        """
        super().__init__(config_file, gui)
        # the root_dir is used to get the FigureDir in the `_get_plotting_flags`
        self.bank_output_dir = config_file.get_output_dir("banklines")

        # set plotting flags
        self.plot_flags = config_file.get_plotting_flags(self.root_dir)
        self.river_data = BankLinesRiverData(config_file)
        self.search_lines = self.river_data.search_lines
        self.simulation_data, self.critical_water_depth = (
            self.river_data.simulation_data()
        )
        if self.plot_flags.plot_data:
            self.plotter = self.get_plotter()

    @property
    def max_river_width(self) -> int:
        """int: Maximum river width in meters."""
        return MAX_RIVER_WIDTH

    def get_plotter(self) -> BankLinesPlotter:
        return BankLinesPlotter(
            self.gui, self.plot_flags, self.config_file.crs, self.simulation_data, self.river_data.river_center_line,
            self.river_data.river_center_line.station_bounds,
        )

    def detect(self) -> None:
        """Run the bank line detection analysis for a specified configuration.

        This method performs bank line detection using the provided configuration file.
        It generates shapefiles that can be opened with GeoPandas or QGIS, and also
        creates a plot of the detected bank lines along with the simulation data.

        Examples:
            ```python
            >>> import matplotlib
            >>> matplotlib.use('Agg')
            >>> from dfastbe.io.config import ConfigFile
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
            >>> bank_lines = BankLines(config_file)  # doctest: +ELLIPSIS
            N...e
            >>> bank_lines.detect()
            >>> bank_lines.plot()
            >>> bank_lines.save()
               0...-

            ```
            In the BankDir directory specified in the .cfg, the following files are created:
            - "raw_detected_bankline_fragments.shp"
            - "bank_areas.shp"
            - "bankline_fragments_per_bank_area.shp"
            - "bankfile.shp"
            In the FigureDir directory specified in the .cfg, the following files are created:
            - "1_banklinedetection.png"
        """
        timed_logger("-- start analysis --")

        log_text(
            "header_banklines",
            data={
                "version": __version__,
                "location": "https://github.com/Deltares/D-FAST_Bank_Erosion",
            },
        )
        log_text("-")

        # clip the chainage path to the range of chainages of interest
        river_center_line = self.river_data.river_center_line
        river_center_line_values = river_center_line.values
        center_line_arr = river_center_line.as_array()

        bank_areas: List[Polygon] = self.search_lines.to_polygons()

        to_right = [True] * self.search_lines.size
        for ib in range(self.search_lines.size):
            to_right[ib] = on_right_side(
                np.array(self.search_lines.values[ib].coords), center_line_arr[:, :2]
            )

        log_text("identify_banklines")
        banklines = self.detect_bank_lines(
            self.simulation_data, self.critical_water_depth, self.config_file
        )

        # clip the set of detected bank lines to the bank areas
        log_text("simplify_banklines")
        bank = []
        masked_bank_lines = []
        for ib, bank_area in enumerate(bank_areas):
            log_text("bank_lines", data={"ib": ib + 1})
            masked_bank_line = self.mask(banklines, bank_area)
            if not masked_bank_line.is_empty:
                masked_bank_lines.append(masked_bank_line)
                bank.append(sort_connect_bank_lines(masked_bank_line, river_center_line_values, to_right[ib]))

        self.results = {
            "bank": bank,
            "banklines": banklines,
            "masked_bank_lines": masked_bank_lines,
            "bank_areas": bank_areas,
        }

        log_text("end_banklines")
        timed_logger("-- stop analysis --")

    @staticmethod
    def mask(banklines: GeoSeries, bank_area: Polygon) -> MultiLineString:
        """
        Clip the bank line segments to the area of interest.

        Args:
            banklines (GeoSeries):
                Unordered set of bank line segments.
            bank_area (Polygon):
                A search area corresponding to one of the bank search lines.

        Returns:
            MultiLineString: Un-ordered set of bank line segments, clipped to bank area.

        Examples:
            ```python
            >>> from dfastbe.io.config import ConfigFile
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
            >>> river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
            N...e
            >>> bank_lines = BankLines(config_file)
            N...e
            >>> simulation_data, critical_water_depth = river_data.simulation_data()
            N...e
            >>> banklines = bank_lines.detect_bank_lines(simulation_data, critical_water_depth, config_file)
            P...)
            >>> bank_area = bank_lines.search_lines.to_polygons()[0]
            >>> bank_lines.mask(banklines, bank_area)
            <MULTILINESTRING ((207830.389 392063.658, 2078...>

            ```
        """
        # intersection returns one MultiLineString object
        masked_bank_lines = banklines.intersection(bank_area)[0]

        return masked_bank_lines

    def plot(self):
        if self.plot_flags.plot_data:
            self.plotter.plot(
                self.search_lines.size,
                self.results["bank"],
                self.results["bank_areas"],
            )

    def save(self):
        """Save results to files."""
        if self.results is None:
            raise ValueError("No results to save. Run the detect method first.")

        bank_name = self.config_file.get_str("General", "BankFile", "bankfile")
        bank_file = self.bank_output_dir / f"{bank_name}{EXTENSION}"
        log_text("save_banklines", data={"file": bank_file})
        gpd.GeoSeries(self.results["bank"], crs=self.config_file.crs).to_file(bank_file)

        gpd.GeoSeries(self.results["masked_bank_lines"], crs=self.config_file.crs).to_file(
            self.bank_output_dir / f"{BANKLINE_FRAGMENTS_PER_BANK_AREA_FILE}{EXTENSION}"
        )
        self.results["banklines"].to_file(
            self.bank_output_dir / f"{RAW_DETECTED_BANKLINE_FRAGMENTS_FILE}{EXTENSION}"
        )
        gpd.GeoSeries(self.results["bank_areas"], crs=self.config_file.crs).to_file(
            self.bank_output_dir / f"{BANK_AREAS_FILE}{EXTENSION}"
        )

    @staticmethod
    def detect_bank_lines(
        simulation_data: BaseSimulationData,
        critical_water_depth: float,
        config_file: ConfigFile,
    ) -> gpd.GeoSeries:
        """Detect all possible bank line segments based on simulation data.

        Use a critical water depth critical_water_depth as a water depth threshold for dry/wet boundary.

        Args:
            simulation_data (BaseSimulationData):
                Simulation data: mesh, bed levels, water levels, velocities, etc.
            critical_water_depth (float):
                Critical water depth for determining the banks.

        Returns:
            geopandas.GeoSeries:
                The collection of all detected bank segments in the remaining model area.

        Examples:
            ```python
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
            >>> river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
            N...e
            >>> simulation_data, critical_water_depth = river_data.simulation_data()
            N...e
            >>> BankLines.detect_bank_lines(simulation_data, critical_water_depth, config_file)
            P...
            0    MULTILINESTRING ((207927.151 391960.747, 20792...
            dtype: geometry

            ```
        """
        h_node = BankLines._calculate_water_depth(simulation_data)
        lines = BankLines._generate_bank_lines(
            simulation_data, h_node, critical_water_depth
        )
        multi_line = union_all(lines)
        merged_line = line_merge(multi_line)

        return gpd.GeoSeries(merged_line, crs=config_file.crs)

    @staticmethod
    def _calculate_water_depth(
        simulation_data: BaseSimulationData,
    ) -> np.ndarray:
        """Calculate the water depth at each node in the simulation data.

        This method computes the water depth for each node by considering the
        water levels at the faces and the bed elevation values at the nodes.
        There are two cases to consider:
           (1) the water levels at the "dry" faces are equal to the bed level
               at the faces (Delft3D-FLOW, WAQUA, D-Flow FM before June 2023)
           (2) the water levels at the "dry" faces are undefined (D-Flow FM
               after June 2023)

        Args:
            simulation_data (BaseSimulationData):
                Simulation data containing face-node relationships, water levels,
                and bed elevation values.

        Returns:
            np.ndarray:
                An array representing the water depth at each node.
        """
        face_node = simulation_data.face_node
        max_num_nodes = face_node.shape[1]
        num_nodes_total = len(simulation_data.x_node)

        # mask all nodes that shouldn't be used to compute the water level at the nodes.
        # start with all unused entries in the face node connectivity
        # mask all node indices if the water level at the face is undefined (dry)
        mask = face_node.mask.copy()
        mask[np.isnan(simulation_data.water_level_face)] = True

        # construct two arrays of equal size:
        # * an array of the unmasked node indices in face_node, and
        # * an array of water levels at the corresponding face
        total_num_node_indices = face_node.size
        num_masked_node_indices = sum(mask.reshape(total_num_node_indices))
        num_unmasked_node_indices = total_num_node_indices - num_masked_node_indices
        unmasked = ~mask
        node_indices = face_node[unmasked]
        water_levels = np.repeat(simulation_data.water_level_face, max_num_nodes)[
            unmasked.flatten()
        ]

        # compute the water level at nodes as the average of the water levels at all corresponding faces and use the bed level if none of the corresponding faces has a water level
        zw_node = np.bincount(node_indices, weights=water_levels, minlength=num_nodes_total)
        num_val = np.bincount(
            node_indices, weights=np.ones(num_unmasked_node_indices), minlength=num_nodes_total
        )
        zw_node = zw_node / np.maximum(num_val, 1)
        zw_node[num_val == 0] = simulation_data.bed_elevation_values[num_val == 0]
        h_node = zw_node - simulation_data.bed_elevation_values

        return h_node

    @staticmethod
    def _generate_bank_lines(
        simulation_data: BaseSimulationData,
        h_node: np.ndarray,
        critical_water_depth: float,
    ) -> List[LineString]:
        """Detect bank lines based on wet/dry nodes.

        Args:
            simulation_data (BaseSimulationData):
                Simulation data: mesh, bed levels, water levels, velocities, etc.
            h_node (np.ndarray):
                Water depth at each node.
            critical_water_depth (float):
                Critical water depth for determining the banks.

        Returns:
            List[LineString or MultiLineString]:
                List of detected bank lines.
        """
        num_faces = len(simulation_data.face_node)
        lines = []

        for i in range(num_faces):
            BankLines._progress_bar(i, num_faces)

            n_node = simulation_data.n_nodes[i]
            face_nodes = simulation_data.face_node[i,:n_node]
            x_face_nodes = simulation_data.x_node[face_nodes]
            y_face_nodes = simulation_data.y_node[face_nodes]
            h_face_nodes = h_node[face_nodes]

            wet_face_nodes = h_face_nodes > critical_water_depth
            n_wet = wet_face_nodes.sum()
            if n_wet == 0 or n_wet == n_node:
                continue

            if n_node == 3:
                line = tri_to_line(
                    x_face_nodes, y_face_nodes, wet_face_nodes, h_face_nodes, critical_water_depth
                )
            else:
                line = poly_to_line(
                    n_node,
                    x_face_nodes,
                    y_face_nodes,
                    wet_face_nodes,
                    h_face_nodes,
                    critical_water_depth,
                )

            if line is not None:
                lines.append(line)

        return lines

    @staticmethod
    def _progress_bar(current: int, total: int) -> None:
        """Print progress bar.

        Args:
            current (int): Current iteration.
            total (int): Total iterations.
        """
        if current % 100 == 0:
            percent = (current / total) * 100
            print(f"Progress: {percent:.2f}% ({current}/{total})", end="\r")
        if current == total - 1:
            print("Progress: 100.00% (100%)")

`max_river_width: int` `property` #

int: Maximum river width in meters.

`init(config_file: ConfigFile, gui: bool = False)` #

Bank line initializer.

Parameters:

Name	Type	Description	Default
`config_file`		configparser.ConfigParser Analysis configuration settings.	required
`gui`		bool Flag indicating whether this routine is called from the GUI.	`False`

Examples:

>>> from unittest.mock import patch
>>> from dfastbe.io.config import ConfigFile
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
>>> bank_lines = BankLines(config_file)  # doctest: +ELLIPSIS
N...e
>>> isinstance(bank_lines, BankLines)
True

Source code in src/dfastbe/bank_lines/bank_lines.py

def __init__(self, config_file: ConfigFile, gui: bool = False):
    """Bank line initializer.

    Args:
        config_file : configparser.ConfigParser
            Analysis configuration settings.
        gui : bool
            Flag indicating whether this routine is called from the GUI.

    Examples:
        ```python
        >>> from unittest.mock import patch
        >>> from dfastbe.io.config import ConfigFile
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
        >>> bank_lines = BankLines(config_file)  # doctest: +ELLIPSIS
        N...e
        >>> isinstance(bank_lines, BankLines)
        True

        ```
    """
    super().__init__(config_file, gui)
    # the root_dir is used to get the FigureDir in the `_get_plotting_flags`
    self.bank_output_dir = config_file.get_output_dir("banklines")

    # set plotting flags
    self.plot_flags = config_file.get_plotting_flags(self.root_dir)
    self.river_data = BankLinesRiverData(config_file)
    self.search_lines = self.river_data.search_lines
    self.simulation_data, self.critical_water_depth = (
        self.river_data.simulation_data()
    )
    if self.plot_flags.plot_data:
        self.plotter = self.get_plotter()

`detect() -> None` #

Run the bank line detection analysis for a specified configuration.

This method performs bank line detection using the provided configuration file. It generates shapefiles that can be opened with GeoPandas or QGIS, and also creates a plot of the detected bank lines along with the simulation data.

Examples:

>>> import matplotlib
>>> matplotlib.use('Agg')
>>> from dfastbe.io.config import ConfigFile
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
>>> bank_lines = BankLines(config_file)  # doctest: +ELLIPSIS
N...e
>>> bank_lines.detect()
>>> bank_lines.plot()
>>> bank_lines.save()
   0...-

In the BankDir directory specified in the .cfg, the following files are created: - "raw_detected_bankline_fragments.shp" - "bank_areas.shp" - "bankline_fragments_per_bank_area.shp" - "bankfile.shp" In the FigureDir directory specified in the .cfg, the following files are created: - "1_banklinedetection.png"

Source code in src/dfastbe/bank_lines/bank_lines.py

def detect(self) -> None:
    """Run the bank line detection analysis for a specified configuration.

    This method performs bank line detection using the provided configuration file.
    It generates shapefiles that can be opened with GeoPandas or QGIS, and also
    creates a plot of the detected bank lines along with the simulation data.

    Examples:
        ```python
        >>> import matplotlib
        >>> matplotlib.use('Agg')
        >>> from dfastbe.io.config import ConfigFile
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
        >>> bank_lines = BankLines(config_file)  # doctest: +ELLIPSIS
        N...e
        >>> bank_lines.detect()
        >>> bank_lines.plot()
        >>> bank_lines.save()
           0...-

        ```
        In the BankDir directory specified in the .cfg, the following files are created:
        - "raw_detected_bankline_fragments.shp"
        - "bank_areas.shp"
        - "bankline_fragments_per_bank_area.shp"
        - "bankfile.shp"
        In the FigureDir directory specified in the .cfg, the following files are created:
        - "1_banklinedetection.png"
    """
    timed_logger("-- start analysis --")

    log_text(
        "header_banklines",
        data={
            "version": __version__,
            "location": "https://github.com/Deltares/D-FAST_Bank_Erosion",
        },
    )
    log_text("-")

    # clip the chainage path to the range of chainages of interest
    river_center_line = self.river_data.river_center_line
    river_center_line_values = river_center_line.values
    center_line_arr = river_center_line.as_array()

    bank_areas: List[Polygon] = self.search_lines.to_polygons()

    to_right = [True] * self.search_lines.size
    for ib in range(self.search_lines.size):
        to_right[ib] = on_right_side(
            np.array(self.search_lines.values[ib].coords), center_line_arr[:, :2]
        )

    log_text("identify_banklines")
    banklines = self.detect_bank_lines(
        self.simulation_data, self.critical_water_depth, self.config_file
    )

    # clip the set of detected bank lines to the bank areas
    log_text("simplify_banklines")
    bank = []
    masked_bank_lines = []
    for ib, bank_area in enumerate(bank_areas):
        log_text("bank_lines", data={"ib": ib + 1})
        masked_bank_line = self.mask(banklines, bank_area)
        if not masked_bank_line.is_empty:
            masked_bank_lines.append(masked_bank_line)
            bank.append(sort_connect_bank_lines(masked_bank_line, river_center_line_values, to_right[ib]))

    self.results = {
        "bank": bank,
        "banklines": banklines,
        "masked_bank_lines": masked_bank_lines,
        "bank_areas": bank_areas,
    }

    log_text("end_banklines")
    timed_logger("-- stop analysis --")

`detect_bank_lines(simulation_data: BaseSimulationData, critical_water_depth: float, config_file: ConfigFile) -> gpd.GeoSeries` `staticmethod` #

Detect all possible bank line segments based on simulation data.

Use a critical water depth critical_water_depth as a water depth threshold for dry/wet boundary.

Parameters:

Name	Type	Description	Default
`simulation_data`	`BaseSimulationData`	Simulation data: mesh, bed levels, water levels, velocities, etc.	required
`critical_water_depth`	`float`	Critical water depth for determining the banks.	required

Returns:

Type	Description
`GeoSeries`	geopandas.GeoSeries: The collection of all detected bank segments in the remaining model area.

Examples:

>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
>>> river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
N...e
>>> simulation_data, critical_water_depth = river_data.simulation_data()
N...e
>>> BankLines.detect_bank_lines(simulation_data, critical_water_depth, config_file)
P...
0    MULTILINESTRING ((207927.151 391960.747, 20792...
dtype: geometry

Source code in src/dfastbe/bank_lines/bank_lines.py

@staticmethod
def detect_bank_lines(
    simulation_data: BaseSimulationData,
    critical_water_depth: float,
    config_file: ConfigFile,
) -> gpd.GeoSeries:
    """Detect all possible bank line segments based on simulation data.

    Use a critical water depth critical_water_depth as a water depth threshold for dry/wet boundary.

    Args:
        simulation_data (BaseSimulationData):
            Simulation data: mesh, bed levels, water levels, velocities, etc.
        critical_water_depth (float):
            Critical water depth for determining the banks.

    Returns:
        geopandas.GeoSeries:
            The collection of all detected bank segments in the remaining model area.

    Examples:
        ```python
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
        >>> river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
        N...e
        >>> simulation_data, critical_water_depth = river_data.simulation_data()
        N...e
        >>> BankLines.detect_bank_lines(simulation_data, critical_water_depth, config_file)
        P...
        0    MULTILINESTRING ((207927.151 391960.747, 20792...
        dtype: geometry

        ```
    """
    h_node = BankLines._calculate_water_depth(simulation_data)
    lines = BankLines._generate_bank_lines(
        simulation_data, h_node, critical_water_depth
    )
    multi_line = union_all(lines)
    merged_line = line_merge(multi_line)

    return gpd.GeoSeries(merged_line, crs=config_file.crs)

`mask(banklines: GeoSeries, bank_area: Polygon) -> MultiLineString` `staticmethod` #

Clip the bank line segments to the area of interest.

Parameters:

Name	Type	Description	Default
`banklines`	`GeoSeries`	Unordered set of bank line segments.	required
`bank_area`	`Polygon`	A search area corresponding to one of the bank search lines.	required

Returns:

Name	Type	Description
`MultiLineString`	`MultiLineString`	Un-ordered set of bank line segments, clipped to bank area.

Examples:

>>> from dfastbe.io.config import ConfigFile
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
>>> river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
N...e
>>> bank_lines = BankLines(config_file)
N...e
>>> simulation_data, critical_water_depth = river_data.simulation_data()
N...e
>>> banklines = bank_lines.detect_bank_lines(simulation_data, critical_water_depth, config_file)
P...)
>>> bank_area = bank_lines.search_lines.to_polygons()[0]
>>> bank_lines.mask(banklines, bank_area)
<MULTILINESTRING ((207830.389 392063.658, 2078...>

Source code in src/dfastbe/bank_lines/bank_lines.py

@staticmethod
def mask(banklines: GeoSeries, bank_area: Polygon) -> MultiLineString:
    """
    Clip the bank line segments to the area of interest.

    Args:
        banklines (GeoSeries):
            Unordered set of bank line segments.
        bank_area (Polygon):
            A search area corresponding to one of the bank search lines.

    Returns:
        MultiLineString: Un-ordered set of bank line segments, clipped to bank area.

    Examples:
        ```python
        >>> from dfastbe.io.config import ConfigFile
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
        >>> river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
        N...e
        >>> bank_lines = BankLines(config_file)
        N...e
        >>> simulation_data, critical_water_depth = river_data.simulation_data()
        N...e
        >>> banklines = bank_lines.detect_bank_lines(simulation_data, critical_water_depth, config_file)
        P...)
        >>> bank_area = bank_lines.search_lines.to_polygons()[0]
        >>> bank_lines.mask(banklines, bank_area)
        <MULTILINESTRING ((207830.389 392063.658, 2078...>

        ```
    """
    # intersection returns one MultiLineString object
    masked_bank_lines = banklines.intersection(bank_area)[0]

    return masked_bank_lines

`save()` #

Save results to files.

Source code in src/dfastbe/bank_lines/bank_lines.py

def save(self):
    """Save results to files."""
    if self.results is None:
        raise ValueError("No results to save. Run the detect method first.")

    bank_name = self.config_file.get_str("General", "BankFile", "bankfile")
    bank_file = self.bank_output_dir / f"{bank_name}{EXTENSION}"
    log_text("save_banklines", data={"file": bank_file})
    gpd.GeoSeries(self.results["bank"], crs=self.config_file.crs).to_file(bank_file)

    gpd.GeoSeries(self.results["masked_bank_lines"], crs=self.config_file.crs).to_file(
        self.bank_output_dir / f"{BANKLINE_FRAGMENTS_PER_BANK_AREA_FILE}{EXTENSION}"
    )
    self.results["banklines"].to_file(
        self.bank_output_dir / f"{RAW_DETECTED_BANKLINE_FRAGMENTS_FILE}{EXTENSION}"
    )
    gpd.GeoSeries(self.results["bank_areas"], crs=self.config_file.crs).to_file(
        self.bank_output_dir / f"{BANK_AREAS_FILE}{EXTENSION}"
    )

class BankLinesRiverData(BaseRiverData):

    @property
    def search_lines(self) -> SearchLines:
        """Get search lines for bank lines.

        Returns:
            SearchLines:
                Search lines for bank lines.

        Examples:
            ```python
            >>> from dfastbe.io.config import ConfigFile
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
            >>> bank_lines_river_data = BankLinesRiverData(config_file)
            No message found for read_chainage
            No message found for clip_chainage
            >>> search_lines = bank_lines_river_data.search_lines
            No message found for read_search_line
            No message found for read_search_line
            >>> len(search_lines.values)
            2

            ```
        """
        search_lines = SearchLines(self.config_file.get_search_lines(), self.river_center_line)
        search_lines.d_lines = self.config_file.get_bank_search_distances(search_lines.size)
        return search_lines

    def _get_bank_lines_simulation_data(self) -> Tuple[BaseSimulationData, float]:
        """read simulation data and drying flooding threshold dh0

        Returns:
            Tuple[BaseSimulationData, float]:
                simulation data and critical water depth (h0).
        """
        sim_file = self.config_file.get_sim_file("Detect", "")
        log_text("read_simdata", data={"file": sim_file})
        simulation_data = BaseSimulationData.read(sim_file)
        # increase critical water depth h0 by flooding threshold dh0
        # get critical water depth used for defining bank line (default = 0.0 m)
        critical_water_depth = self.config_file.get_float(
            "Detect", "WaterDepth", default=0
        )
        h0 = critical_water_depth + simulation_data.dry_wet_threshold
        return simulation_data, h0

    def simulation_data(self) -> Tuple[BaseSimulationData, float]:
        """Get simulation data and critical water depth and clip to river center line.

        Returns:
            Tuple[BaseSimulationData, float]:
                simulation data and critical water depth (h0).

        Examples:
            ```python
            >>> from dfastbe.io.config import ConfigFile
            >>> from unittest.mock import patch
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
            >>> bank_lines_river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
            N...e
            >>> simulation_data, h0 = bank_lines_river_data.simulation_data()
            N...e
            >>> h0
            0.1

            ```
        """
        simulation_data, h0 = self._get_bank_lines_simulation_data()
        # clip simulation data to boundaries ...
        log_text("clip_data")
        simulation_data.clip(self.river_center_line.values, self.search_lines.max_distance)

        return simulation_data, h0

`search_lines: SearchLines` `property` #

Get search lines for bank lines.

Returns:

Name	Type	Description
`SearchLines`	`SearchLines`	Search lines for bank lines.

Examples:

>>> from dfastbe.io.config import ConfigFile
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
>>> bank_lines_river_data = BankLinesRiverData(config_file)
No message found for read_chainage
No message found for clip_chainage
>>> search_lines = bank_lines_river_data.search_lines
No message found for read_search_line
No message found for read_search_line
>>> len(search_lines.values)
2

`simulation_data() -> Tuple[BaseSimulationData, float]` #

Get simulation data and critical water depth and clip to river center line.

Returns:

Type	Description
`Tuple[BaseSimulationData, float]`	Tuple[BaseSimulationData, float]: simulation data and critical water depth (h0).

Examples:

>>> from dfastbe.io.config import ConfigFile
>>> from unittest.mock import patch
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
>>> bank_lines_river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
N...e
>>> simulation_data, h0 = bank_lines_river_data.simulation_data()
N...e
>>> h0
0.1

Source code in src/dfastbe/bank_lines/data_models.py

def simulation_data(self) -> Tuple[BaseSimulationData, float]:
    """Get simulation data and critical water depth and clip to river center line.

    Returns:
        Tuple[BaseSimulationData, float]:
            simulation data and critical water depth (h0).

    Examples:
        ```python
        >>> from dfastbe.io.config import ConfigFile
        >>> from unittest.mock import patch
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")
        >>> bank_lines_river_data = BankLinesRiverData(config_file)  # doctest: +ELLIPSIS
        N...e
        >>> simulation_data, h0 = bank_lines_river_data.simulation_data()
        N...e
        >>> h0
        0.1

        ```
    """
    simulation_data, h0 = self._get_bank_lines_simulation_data()
    # clip simulation data to boundaries ...
    log_text("clip_data")
    simulation_data.clip(self.river_center_line.values, self.search_lines.max_distance)

    return simulation_data, h0

`SearchLines` #

Source code in src/dfastbe/bank_lines/data_models.py

class SearchLines:

    def __init__(self, lines: List[LineString], mask: LineGeometry = None):
        """Search lines initialization.

        Args:
            lines (List[LineString]):
                List of search lines.
            mask (LineGeometry, optional):
                Center line for masking the search lines. Defaults to None.
        """
        if mask is None:
            self.values = lines
            self.max_distance = None
        else:
            self.values, self.max_distance = self.mask(lines, mask.values)

        self.size = len(lines)

    @property
    def d_lines(self) -> List[float]:
        if hasattr(self, "_d_lines"):
            return self._d_lines
        else:
            raise ValueError("The d_lines property has not been set yet.")

    @d_lines.setter
    def d_lines(self, value: List[float]):
        self._d_lines = value

    @staticmethod
    def mask(
        search_lines: List[LineString],
        river_center_line: LineString,
        max_river_width: float = MAX_RIVER_WIDTH,
    ) -> Tuple[List[LineString], float]:
        """
        Clip the list of lines to the envelope of a certain size surrounding a reference line.

        Args:
            search_lines (List[LineString]):
                List of lines to be clipped.
            river_center_line (LineString):
                Reference line to which the search lines are clipped.
            max_river_width: float
                Maximum distance away from river_profile.

        Returns:
            Tuple[List[LineString], float]:
                - List of clipped search lines.
                - Maximum distance from any point within line to reference line.

        Examples:
            ```python
            >>> from shapely.geometry import LineString
            >>> search_lines = [LineString([(0, 0), (1, 1)]), LineString([(2, 2), (3, 3)])]
            >>> river_center_line = LineString([(0, 0), (2, 2)])
            >>> search_lines_clipped, max_distance = SearchLines.mask(search_lines, river_center_line)
            >>> max_distance
            2.0

            ```
        """
        num = len(search_lines)
        profile_buffer = river_center_line.buffer(max_river_width, cap_style=2)

        # The algorithm uses simplified geometries for determining the distance between lines for speed.
        # Stay accurate to within about 1 m
        profile_simplified = river_center_line.simplify(1)

        max_distance = 0
        for ind in range(num):
            # Clip the bank search lines to the reach of interest (indicated by the reference line).
            search_lines[ind] = search_lines[ind].intersection(profile_buffer)

            # If the bank search line breaks into multiple parts, select the part closest to the reference line.
            if search_lines[ind].geom_type == "MultiLineString":
                search_lines[ind] = SearchLines._select_closest_part(
                    search_lines[ind], profile_simplified, max_river_width
                )

            # Determine the maximum distance from a point on this line to the reference line.
            line_simplified = search_lines[ind].simplify(1)
            max_distance = max(
                [Point(c).distance(profile_simplified) for c in line_simplified.coords]
            )

            # Increase the value of max_distance by 2 to account for error introduced by using simplified lines.
            max_distance = max(max_distance, max_distance + 2)

        return search_lines, max_distance

    @staticmethod
    def _select_closest_part(
        search_lines_segments: MultiLineString,
        reference_line: LineString,
        max_river_width: float,
    ) -> LineString:
        """Select the closest part of a MultiLineString to the reference line.

        Args:
            search_lines_segments (MultiLineString):
                The MultiLineString containing multiple line segments to evaluate.
            reference_line (LineString):
                The reference line to calculate distances.
            max_river_width (float):
                Maximum allowable distance.

        Returns:
            LineString: The closest part of the MultiLineString.
        """
        closest_part = search_lines_segments.geoms[0]
        min_distance = max_river_width

        for part in search_lines_segments.geoms:
            simplified_part = part.simplify(1)
            distance = simplified_part.distance(reference_line)
            if distance < min_distance:
                min_distance = distance
                closest_part = part

        return closest_part

    def to_polygons(self) -> List[Polygon]:
        """
        Construct a series of polygons surrounding the bank search lines.

        Returns:
            bank_areas:
                Array containing the areas of interest surrounding the bank search lines.

        Examples:
            ```python
            >>> search_lines = [LineString([(0, 0), (1, 1)]), LineString([(2, 2), (3, 3)])]
            >>> search_lines_clipped = SearchLines(search_lines)
            >>> search_lines_clipped.d_lines = [10, 20]
            >>> bank_areas = search_lines_clipped.to_polygons()
            >>> len(bank_areas)
            2

            ```
        """
        bank_areas = [
            self.values[b].buffer(distance, cap_style=2)
            for b, distance in enumerate(self.d_lines)
        ]
        return bank_areas

`init(lines: List[LineString], mask: LineGeometry = None)` #

Search lines initialization.

Parameters:

Name	Type	Description	Default
`lines`	`List[LineString]`	List of search lines.	required
`mask`	`LineGeometry`	Center line for masking the search lines. Defaults to None.	`None`

Source code in src/dfastbe/bank_lines/data_models.py

def __init__(self, lines: List[LineString], mask: LineGeometry = None):
    """Search lines initialization.

    Args:
        lines (List[LineString]):
            List of search lines.
        mask (LineGeometry, optional):
            Center line for masking the search lines. Defaults to None.
    """
    if mask is None:
        self.values = lines
        self.max_distance = None
    else:
        self.values, self.max_distance = self.mask(lines, mask.values)

    self.size = len(lines)

`mask(search_lines: List[LineString], river_center_line: LineString, max_river_width: float = MAX_RIVER_WIDTH) -> Tuple[List[LineString], float]` `staticmethod` #

Clip the list of lines to the envelope of a certain size surrounding a reference line.

Parameters:

Name	Type	Description	Default
`search_lines`	`List[LineString]`	List of lines to be clipped.	required
`river_center_line`	`LineString`	Reference line to which the search lines are clipped.	required
`max_river_width`	`float`	float Maximum distance away from river_profile.	`MAX_RIVER_WIDTH`

Returns:

Type	Description
`Tuple[List[LineString], float]`	Tuple[List[LineString], float]: - List of clipped search lines. - Maximum distance from any point within line to reference line.

Examples:

>>> from shapely.geometry import LineString
>>> search_lines = [LineString([(0, 0), (1, 1)]), LineString([(2, 2), (3, 3)])]
>>> river_center_line = LineString([(0, 0), (2, 2)])
>>> search_lines_clipped, max_distance = SearchLines.mask(search_lines, river_center_line)
>>> max_distance
2.0

Source code in src/dfastbe/bank_lines/data_models.py

@staticmethod
def mask(
    search_lines: List[LineString],
    river_center_line: LineString,
    max_river_width: float = MAX_RIVER_WIDTH,
) -> Tuple[List[LineString], float]:
    """
    Clip the list of lines to the envelope of a certain size surrounding a reference line.

    Args:
        search_lines (List[LineString]):
            List of lines to be clipped.
        river_center_line (LineString):
            Reference line to which the search lines are clipped.
        max_river_width: float
            Maximum distance away from river_profile.

    Returns:
        Tuple[List[LineString], float]:
            - List of clipped search lines.
            - Maximum distance from any point within line to reference line.

    Examples:
        ```python
        >>> from shapely.geometry import LineString
        >>> search_lines = [LineString([(0, 0), (1, 1)]), LineString([(2, 2), (3, 3)])]
        >>> river_center_line = LineString([(0, 0), (2, 2)])
        >>> search_lines_clipped, max_distance = SearchLines.mask(search_lines, river_center_line)
        >>> max_distance
        2.0

        ```
    """
    num = len(search_lines)
    profile_buffer = river_center_line.buffer(max_river_width, cap_style=2)

    # The algorithm uses simplified geometries for determining the distance between lines for speed.
    # Stay accurate to within about 1 m
    profile_simplified = river_center_line.simplify(1)

    max_distance = 0
    for ind in range(num):
        # Clip the bank search lines to the reach of interest (indicated by the reference line).
        search_lines[ind] = search_lines[ind].intersection(profile_buffer)

        # If the bank search line breaks into multiple parts, select the part closest to the reference line.
        if search_lines[ind].geom_type == "MultiLineString":
            search_lines[ind] = SearchLines._select_closest_part(
                search_lines[ind], profile_simplified, max_river_width
            )

        # Determine the maximum distance from a point on this line to the reference line.
        line_simplified = search_lines[ind].simplify(1)
        max_distance = max(
            [Point(c).distance(profile_simplified) for c in line_simplified.coords]
        )

        # Increase the value of max_distance by 2 to account for error introduced by using simplified lines.
        max_distance = max(max_distance, max_distance + 2)

    return search_lines, max_distance

`to_polygons() -> List[Polygon]` #

Construct a series of polygons surrounding the bank search lines.

Returns:

Name	Type	Description
`bank_areas`	`List[Polygon]`	Array containing the areas of interest surrounding the bank search lines.

Examples:

>>> search_lines = [LineString([(0, 0), (1, 1)]), LineString([(2, 2), (3, 3)])]
>>> search_lines_clipped = SearchLines(search_lines)
>>> search_lines_clipped.d_lines = [10, 20]
>>> bank_areas = search_lines_clipped.to_polygons()
>>> len(bank_areas)
2

Source code in src/dfastbe/bank_lines/data_models.py

def to_polygons(self) -> List[Polygon]:
    """
    Construct a series of polygons surrounding the bank search lines.

    Returns:
        bank_areas:
            Array containing the areas of interest surrounding the bank search lines.

    Examples:
        ```python
        >>> search_lines = [LineString([(0, 0), (1, 1)]), LineString([(2, 2), (3, 3)])]
        >>> search_lines_clipped = SearchLines(search_lines)
        >>> search_lines_clipped.d_lines = [10, 20]
        >>> bank_areas = search_lines_clipped.to_polygons()
        >>> len(bank_areas)
        2

        ```
    """
    bank_areas = [
        self.values[b].buffer(distance, cap_style=2)
        for b, distance in enumerate(self.d_lines)
    ]
    return bank_areas

Utility Functions #

The Bank Lines module includes several utility functions for processing bank lines:

sort_connect_bank_lines: Sorts and connects bank line fragments
poly_to_line: Converts polygons to lines
tri_to_line: Converts triangles to lines

Workflow #

The typical workflow for bank line detection is:

Initialize the BankLines object with a configuration file
Call the detect method to start the bank line detection process
The detect method orchestrates the entire process:
Loads hydrodynamic simulation data
Calculates water depth
Generates bank lines
Masks bank lines with bank areas
Saves bank lines to output files
Generates plots

Usage Example #

from dfastbe.io.config import ConfigFile
from dfastbe.bank_lines.bank_lines import BankLines

# Load configuration file
config_file = ConfigFile.read("config.cfg")

# Initialize BankLines object
bank_lines = BankLines(config_file)

# Run bank line detection
bank_lines.detect()

# plot results
bank_lines.plot()

# Save results
bank_lines.save()

For more details on the specific methods and classes, refer to the API reference below.

Bank Lines Module #

Overview #

Components #

Main Classes #

`dfastbe.bank_lines.bank_lines` #

`BankLines` #

`max_river_width: int` `property` #

`init(config_file: ConfigFile, gui: bool = False)` #

`detect() -> None` #

`detect_bank_lines(simulation_data: BaseSimulationData, critical_water_depth: float, config_file: ConfigFile) -> gpd.GeoSeries` `staticmethod` #

`mask(banklines: GeoSeries, bank_area: Polygon) -> MultiLineString` `staticmethod` #

`save()` #

Data Models #

`dfastbe.bank_lines.data_models` #

`BankLinesRiverData` #

`search_lines: SearchLines` `property` #

`simulation_data() -> Tuple[BaseSimulationData, float]` #

`SearchLines` #

`init(lines: List[LineString], mask: LineGeometry = None)` #

`mask(search_lines: List[LineString], river_center_line: LineString, max_river_width: float = MAX_RIVER_WIDTH) -> Tuple[List[LineString], float]` `staticmethod` #

`to_polygons() -> List[Polygon]` #

Utility Functions #

Workflow #

Usage Example #