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.

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

Data Models

dfastbe.bank_lines.data_models

BankLinesRiverData

Bases: BaseRiverData

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

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:

1. Initialize the BankLines object with a configuration file
2. Call the detect method to start the bank line detection process
3. The detect method orchestrates the entire process:
4. Loads hydrodynamic simulation data
5. Calculates water depth
6. Generates bank lines
7. Masks bank lines with bank areas
8. Saves bank lines to output files
9. 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.