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

Bank line detection class.

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

class BankLines:
    """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

            ```
        """
        # the root_dir is used to get the FigureDir in the `_get_plotting_flags`
        self.root_dir = config_file.root_dir

        self._config_file = config_file
        self.gui = gui
        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()
        )

    @property
    def config_file(self) -> ConfigFile:
        """ConfigFile: object containing the configuration file."""
        return self._config_file

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

    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()
               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"
        """
        config_file = self.config_file
        river_data = self.river_data
        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 = river_data.river_center_line
        station_bounds = river_center_line.station_bounds
        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, 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_lines.append(self.mask(banklines, bank_area))
            bank.append(sort_connect_bank_lines(masked_bank_lines[ib], river_center_line_values, to_right[ib]))

        self.save(bank, banklines, masked_bank_lines, bank_areas, config_file)

        if self.plot_flags["plot_data"]:
            self.plot(
                center_line_arr,
                self.search_lines.size,
                bank,
                station_bounds,
                bank_areas,
                config_file,
            )

        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,
        station_coords: np.ndarray,
        num_search_lines: int,
        bank: List[LineString],
        stations_bounds: Tuple[float, float],
        bank_areas: List[Polygon],
        config_file: ConfigFile,
    ):
        """Plot the bank lines and the simulation data.

        Args:
            station_coords (np.ndarray):
                Array of x and y coordinates in km.
            num_search_lines (int):
                Number of search lines.
            bank (List):
                List of bank lines.
            stations_bounds (Tuple[float, float]):
                Minimum and maximum km bounds.
            bank_areas (List[Polygon]):
                A search area corresponding to one of the bank search lines.
            config_file (ConfigFile):
                Configuration file object.

        Examples:
            ```python
            >>> import matplotlib
            >>> matplotlib.use('Agg')
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")  # doctest: +ELLIPSIS
            >>> bank_lines = BankLines(config_file)
            N...e
            >>> bank_lines.plot_flags["save_plot"] = False
            >>> station_coords = np.array([[0, 0, 0], [1, 1, 0]])
            >>> num_search_lines = 1
            >>> bank = [LineString([(0, 0), (1, 1)])]
            >>> stations_bounds = (0, 1)
            >>> bank_areas = [Polygon([(0, 0), (1, 1), (1, 0)])]
            >>> bank_lines.plot(station_coords, num_search_lines, bank, stations_bounds, bank_areas, config_file)
            N...s

            ```
        """
        log_text("=")
        log_text("create_figures")
        i_fig = 0
        bbox = get_bbox(station_coords)

        if self.plot_flags["save_plot_zoomed"]:
            bank_crds: List[np.ndarray] = []
            bank_km: List[np.ndarray] = []
            for ib in range(num_search_lines):
                bcrds_numpy = np.array(bank[ib].coords)
                line_geom = LineGeometry(bcrds_numpy, crs=config_file.crs)
                km_numpy = line_geom.intersect_with_line(station_coords)
                bank_crds.append(bcrds_numpy)
                bank_km.append(km_numpy)
            km_zoom, xy_zoom = get_zoom_extends(
                stations_bounds[0],
                stations_bounds[1],
                self.plot_flags["zoom_km_step"],
                bank_crds,
                bank_km,
            )

        fig, ax = df_plt.plot_detect1(
            bbox,
            station_coords,
            bank_areas,
            bank,
            self.simulation_data.face_node,
            self.simulation_data.n_nodes,
            self.simulation_data.x_node,
            self.simulation_data.y_node,
            self.simulation_data.water_depth_face,
            1.1 * self.simulation_data.water_depth_face.max(),
            "x-coordinate [m]",
            "y-coordinate [m]",
            "water depth and detected bank lines",
            "water depth [m]",
            "bank search area",
            "detected bank line",
            config_file,
        )
        if self.plot_flags["save_plot"]:
            i_fig = i_fig + 1
            fig_base = (
                f"{self.plot_flags.get('fig_dir')}{os.sep}{i_fig}_banklinedetection"
            )
            if self.plot_flags["save_plot_zoomed"]:
                df_plt.zoom_xy_and_save(
                    fig, ax, fig_base, self.plot_flags.get("plot_ext"), xy_zoom, scale=1
                )
            fig_file = fig_base + self.plot_flags["plot_ext"]
            df_plt.savefig(fig, fig_file)

        if self.plot_flags["close_plot"]:
            plt.close("all")
        else:
            plt.show(block=not self.gui)

    def save(
        self,
        bank: List[LineString],
        banklines: GeoSeries,
        masked_bank_lines: List[MultiLineString],
        bank_areas: List[Polygon],
        config_file: ConfigFile,
    ):
        """Save results to files.

        Args:
            bank (List[LineString]):
                List of bank lines.
            banklines (GeoSeries):
                Un-ordered set of bank line segments.
            masked_bank_lines (List[MultiLineString]):
                Un-ordered set of bank line segments, clipped to bank area.
            bank_areas (List[Polygon]):
                A search area corresponding to one of the bank search lines.
            config_file (ConfigFile):
                Configuration file object.

        Examples:
            ```python
            >>> from dfastbe.io.config import ConfigFile
            >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")  # doctest: +ELLIPSIS
            >>> bank_lines = BankLines(config_file)
            N...e
            >>> bank = [LineString([(0, 0), (1, 1)])]
            >>> banklines = gpd.GeoSeries([LineString([(0, 0), (1, 1)])])
            >>> masked_bank_lines = [MultiLineString([LineString([(0, 0), (1, 1)])])]
            >>> bank_areas = [Polygon([(0, 0), (1, 1), (1, 0)])]
            >>> bank_lines.save(bank, banklines, masked_bank_lines, bank_areas, config_file)
            No message found for save_banklines

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

        gpd.GeoSeries(masked_bank_lines, crs=config_file.crs).to_file(
            self.bank_output_dir / f"{BANKLINE_FRAGMENTS_PER_BANK_AREA_FILE}{EXTENSION}"
        )
        banklines.to_file(
            self.bank_output_dir / f"{RAW_DETECTED_BANKLINE_FRAGMENTS_FILE}{EXTENSION}"
        )
        gpd.GeoSeries(bank_areas, crs=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)

        wet_node = h_node > critical_water_depth
        num_wet_arr = wet_node.sum(axis=1)

        lines = BankLines._generate_bank_lines(
            simulation_data, wet_node, num_wet_arr, 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,
    ) -> Tuple[np.ndarray, 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.

        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 = simulation_data.face_node.shape[1]
        num_nodes_total = len(simulation_data.x_node)

        if hasattr(face_node, "mask"):
            mask = ~face_node.mask
            non_masked = sum(mask.reshape(face_node.size))
            f_nc_m = face_node[mask]
            zwm = np.repeat(simulation_data.water_level_face, max_num_nodes)[
                mask.flatten()
            ]
        else:
            non_masked = face_node.size
            f_nc_m = face_node.reshape(non_masked)
            zwm = np.repeat(simulation_data.water_level_face, max_num_nodes).reshape(
                non_masked
            )

        zw_node = np.bincount(f_nc_m, weights=zwm, minlength=num_nodes_total)
        num_val = np.bincount(
            f_nc_m, weights=np.ones(non_masked), 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[face_node] - simulation_data.bed_elevation_values[face_node]
        return h_node

    @staticmethod
    def _generate_bank_lines(
        simulation_data: BaseSimulationData,
        wet_node: np.ndarray,
        num_wet_arr: np.ndarray,
        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.
            wet_node (np.ndarray):
                Wet/dry boolean array for each face node.
            num_wet_arr (np.ndarray):
                Number of wet nodes for each face.
            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)
        x_node = simulation_data.x_node[simulation_data.face_node]
        y_node = simulation_data.y_node[simulation_data.face_node]
        mask = num_wet_arr.mask.size > 1
        lines = []

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

            n_wet = num_wet_arr[i]
            n_node = simulation_data.n_nodes[i]
            if (mask and n_wet.mask) or n_wet == 0 or n_wet == n_node:
                continue

            if n_node == 3:
                line = tri_to_line(
                    x_node[i], y_node[i], wet_node[i], h_node[i], critical_water_depth
                )
            else:
                line = poly_to_line(
                    n_node,
                    x_node[i],
                    y_node[i],
                    wet_node[i],
                    h_node[i],
                    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%)")

config_file: ConfigFile property

ConfigFile: object containing the configuration file.

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

        ```
    """
    # the root_dir is used to get the FigureDir in the `_get_plotting_flags`
    self.root_dir = config_file.root_dir

    self._config_file = config_file
    self.gui = gui
    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()
    )

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()
   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()
           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"
    """
    config_file = self.config_file
    river_data = self.river_data
    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 = river_data.river_center_line
    station_bounds = river_center_line.station_bounds
    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, 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_lines.append(self.mask(banklines, bank_area))
        bank.append(sort_connect_bank_lines(masked_bank_lines[ib], river_center_line_values, to_right[ib]))

    self.save(bank, banklines, masked_bank_lines, bank_areas, config_file)

    if self.plot_flags["plot_data"]:
        self.plot(
            center_line_arr,
            self.search_lines.size,
            bank,
            station_bounds,
            bank_areas,
            config_file,
        )

    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)

    wet_node = h_node > critical_water_depth
    num_wet_arr = wet_node.sum(axis=1)

    lines = BankLines._generate_bank_lines(
        simulation_data, wet_node, num_wet_arr, 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

plot(station_coords: np.ndarray, num_search_lines: int, bank: List[LineString], stations_bounds: Tuple[float, float], bank_areas: List[Polygon], config_file: ConfigFile)

Plot the bank lines and the simulation data.

Parameters:

Name	Type	Description	Default
station_coords	ndarray	Array of x and y coordinates in km.	required
num_search_lines	int	Number of search lines.	required
bank	List	List of bank lines.	required
stations_bounds	Tuple[float, float]	Minimum and maximum km bounds.	required
bank_areas	List[Polygon]	A search area corresponding to one of the bank search lines.	required
config_file	ConfigFile	Configuration file object.	required

Examples:

>>> import matplotlib
>>> matplotlib.use('Agg')
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")  # doctest: +ELLIPSIS
>>> bank_lines = BankLines(config_file)
N...e
>>> bank_lines.plot_flags["save_plot"] = False
>>> station_coords = np.array([[0, 0, 0], [1, 1, 0]])
>>> num_search_lines = 1
>>> bank = [LineString([(0, 0), (1, 1)])]
>>> stations_bounds = (0, 1)
>>> bank_areas = [Polygon([(0, 0), (1, 1), (1, 0)])]
>>> bank_lines.plot(station_coords, num_search_lines, bank, stations_bounds, bank_areas, config_file)
N...s

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

def plot(
    self,
    station_coords: np.ndarray,
    num_search_lines: int,
    bank: List[LineString],
    stations_bounds: Tuple[float, float],
    bank_areas: List[Polygon],
    config_file: ConfigFile,
):
    """Plot the bank lines and the simulation data.

    Args:
        station_coords (np.ndarray):
            Array of x and y coordinates in km.
        num_search_lines (int):
            Number of search lines.
        bank (List):
            List of bank lines.
        stations_bounds (Tuple[float, float]):
            Minimum and maximum km bounds.
        bank_areas (List[Polygon]):
            A search area corresponding to one of the bank search lines.
        config_file (ConfigFile):
            Configuration file object.

    Examples:
        ```python
        >>> import matplotlib
        >>> matplotlib.use('Agg')
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")  # doctest: +ELLIPSIS
        >>> bank_lines = BankLines(config_file)
        N...e
        >>> bank_lines.plot_flags["save_plot"] = False
        >>> station_coords = np.array([[0, 0, 0], [1, 1, 0]])
        >>> num_search_lines = 1
        >>> bank = [LineString([(0, 0), (1, 1)])]
        >>> stations_bounds = (0, 1)
        >>> bank_areas = [Polygon([(0, 0), (1, 1), (1, 0)])]
        >>> bank_lines.plot(station_coords, num_search_lines, bank, stations_bounds, bank_areas, config_file)
        N...s

        ```
    """
    log_text("=")
    log_text("create_figures")
    i_fig = 0
    bbox = get_bbox(station_coords)

    if self.plot_flags["save_plot_zoomed"]:
        bank_crds: List[np.ndarray] = []
        bank_km: List[np.ndarray] = []
        for ib in range(num_search_lines):
            bcrds_numpy = np.array(bank[ib].coords)
            line_geom = LineGeometry(bcrds_numpy, crs=config_file.crs)
            km_numpy = line_geom.intersect_with_line(station_coords)
            bank_crds.append(bcrds_numpy)
            bank_km.append(km_numpy)
        km_zoom, xy_zoom = get_zoom_extends(
            stations_bounds[0],
            stations_bounds[1],
            self.plot_flags["zoom_km_step"],
            bank_crds,
            bank_km,
        )

    fig, ax = df_plt.plot_detect1(
        bbox,
        station_coords,
        bank_areas,
        bank,
        self.simulation_data.face_node,
        self.simulation_data.n_nodes,
        self.simulation_data.x_node,
        self.simulation_data.y_node,
        self.simulation_data.water_depth_face,
        1.1 * self.simulation_data.water_depth_face.max(),
        "x-coordinate [m]",
        "y-coordinate [m]",
        "water depth and detected bank lines",
        "water depth [m]",
        "bank search area",
        "detected bank line",
        config_file,
    )
    if self.plot_flags["save_plot"]:
        i_fig = i_fig + 1
        fig_base = (
            f"{self.plot_flags.get('fig_dir')}{os.sep}{i_fig}_banklinedetection"
        )
        if self.plot_flags["save_plot_zoomed"]:
            df_plt.zoom_xy_and_save(
                fig, ax, fig_base, self.plot_flags.get("plot_ext"), xy_zoom, scale=1
            )
        fig_file = fig_base + self.plot_flags["plot_ext"]
        df_plt.savefig(fig, fig_file)

    if self.plot_flags["close_plot"]:
        plt.close("all")
    else:
        plt.show(block=not self.gui)

save(bank: List[LineString], banklines: GeoSeries, masked_bank_lines: List[MultiLineString], bank_areas: List[Polygon], config_file: ConfigFile)

Save results to files.

Parameters:

Name	Type	Description	Default
bank	List[LineString]	List of bank lines.	required
banklines	GeoSeries	Un-ordered set of bank line segments.	required
masked_bank_lines	List[MultiLineString]	Un-ordered set of bank line segments, clipped to bank area.	required
bank_areas	List[Polygon]	A search area corresponding to one of the bank search lines.	required
config_file	ConfigFile	Configuration file object.	required

Examples:

>>> from dfastbe.io.config import ConfigFile
>>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")  # doctest: +ELLIPSIS
>>> bank_lines = BankLines(config_file)
N...e
>>> bank = [LineString([(0, 0), (1, 1)])]
>>> banklines = gpd.GeoSeries([LineString([(0, 0), (1, 1)])])
>>> masked_bank_lines = [MultiLineString([LineString([(0, 0), (1, 1)])])]
>>> bank_areas = [Polygon([(0, 0), (1, 1), (1, 0)])]
>>> bank_lines.save(bank, banklines, masked_bank_lines, bank_areas, config_file)
No message found for save_banklines

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

def save(
    self,
    bank: List[LineString],
    banklines: GeoSeries,
    masked_bank_lines: List[MultiLineString],
    bank_areas: List[Polygon],
    config_file: ConfigFile,
):
    """Save results to files.

    Args:
        bank (List[LineString]):
            List of bank lines.
        banklines (GeoSeries):
            Un-ordered set of bank line segments.
        masked_bank_lines (List[MultiLineString]):
            Un-ordered set of bank line segments, clipped to bank area.
        bank_areas (List[Polygon]):
            A search area corresponding to one of the bank search lines.
        config_file (ConfigFile):
            Configuration file object.

    Examples:
        ```python
        >>> from dfastbe.io.config import ConfigFile
        >>> config_file = ConfigFile.read("tests/data/bank_lines/meuse_manual.cfg")  # doctest: +ELLIPSIS
        >>> bank_lines = BankLines(config_file)
        N...e
        >>> bank = [LineString([(0, 0), (1, 1)])]
        >>> banklines = gpd.GeoSeries([LineString([(0, 0), (1, 1)])])
        >>> masked_bank_lines = [MultiLineString([LineString([(0, 0), (1, 1)])])]
        >>> bank_areas = [Polygon([(0, 0), (1, 1), (1, 0)])]
        >>> bank_lines.save(bank, banklines, masked_bank_lines, bank_areas, config_file)
        No message found for save_banklines

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

    gpd.GeoSeries(masked_bank_lines, crs=config_file.crs).to_file(
        self.bank_output_dir / f"{BANKLINE_FRAGMENTS_PER_BANK_AREA_FILE}{EXTENSION}"
    )
    banklines.to_file(
        self.bank_output_dir / f"{RAW_DETECTED_BANKLINE_FRAGMENTS_FILE}{EXTENSION}"
    )
    gpd.GeoSeries(bank_areas, crs=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()

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