Demo showing how to invoke the `floodplain_extraction` function

example_floodplain_call.py

"""Tracer for floodplain extraction function"""

"""
def floodplain_extraction(
        t_return_parameter,
        min_flow_accum_threshold,
        dem_path,
        stream_gauge_vector_path,
        gauge_table_path,
        gauge_id_field,
        table_field_prefix,
        target_stream_vector_path,
        target_watershed_boundary_vector_path,
        target_floodplain_raster_path,
        target_snap_point_vector_path):
"""
"""Build a floodplain simulation with given parameters.

Args:
    t_return_parameter (float): the return year to estimate the flood
        height. Bankfull heights are estimated at
        ``t_return_parameter``=1.5. A 10 year flood would be 10.0, etc.
    min_flow_accum_threshold (int): this is the upstream flow
        accumulation used to classify streams.
    dem_path (str): path to DEM raster used for determining streams,
        subwatersheds, and upstream drainage area.
    stream_gauge_vector_path (str): path to a point vector used to
        represent stream gauges. If these points don't lie directly on
        a stream they will be snapped to the nearest stream. The model
        will create an output vector called ``snapped_gauge_points.gpkg``
        in the base directory of the target floodplain raster path.
    gauge_table_path (str): path to CSV file containing columns with
        stream guage ids and rows containing either positive values
        indicating flood heights or negative values indicating no
        data. Generally each row can be thought of as a year, though
        no chronological order is necessary.
    gauge_id_field (str): name of the field in the
        ``stream_gauge_vector_path`` to use as the unique identifier for
        each point.
    table_field_prefix (str): this value is prepended to the value
        in the guage vector's ``gauge_id_field`` when being used to
        link it to
    target_stream_vector_path (str): this is the desired path to the
        output stream vector generated by this function.
    target_watershed_boundary_vector_path (str): this is the desired
        path to the subwatershed vector generated by this function.
    target_floodplain_raster_path (str): this is the desired path to the
        floodplain raster
    target_snap_point_vector_path (str): this is the desired target path
        to the snapped stream gauge vector.

Return:
    None.
"""
import logging
import os
import sys

from inspring.floodplain_extraction.floodplain_extraction import floodplain_extraction


logging.basicConfig(
    level=logging.DEBUG,
    format=(
        '%(asctime)s (%(relativeCreated)d) %(processName)s %(levelname)s '
        '%(name)s [%(funcName)s:%(lineno)d] %(message)s'),
    stream=sys.stdout)
LOGGER = logging.getLogger(__name__)
logging.getLogger('taskgraph').setLevel(logging.INFO)
WORKSPACE_DIR = 'workspace'


def main():
    """Entry point."""
    os.makedirs(WORKSPACE_DIR, exist_ok=True)

    t_return_parameter = 10
    min_flow_accum_threshold = 1000
    dem_path = "sample_data/MERIT DEM Pro Agua Purus Acre clip2.tif"
    gauge_vector_path = "sample_data/GaugesBrazil_fromML_all_gauges.gpkg"
    gauge_table_path = "sample_data/Observed_waterlevels.csv"
    gauge_id_field = 'StationID'
    table_field_prefix = 'G'
    target_stream_vector_path = os.path.join(
        WORKSPACE_DIR, f'stream_segments_fa{min_flow_accum_threshold}.gpkg')
    target_watershed_boundary_vector_path = os.path.join(
        WORKSPACE_DIR, f'subwatersheds_fa{min_flow_accum_threshold}.gpkg')
    target_snap_point_vector_path = os.path.join(
        WORKSPACE_DIR, f'snapped_gauge_points_fa{min_flow_accum_threshold}.gpkg')
    target_floodplain_raster_path = os.path.join(
        WORKSPACE_DIR, f'floodplain_t{t_return_parameter}_fa{min_flow_accum_threshold}.tif')
    floodplain_extraction(
        t_return_parameter,
        min_flow_accum_threshold,
        dem_path, gauge_vector_path,
        gauge_table_path,
        gauge_id_field,
        table_field_prefix,
        target_stream_vector_path,
        target_watershed_boundary_vector_path,
        target_floodplain_raster_path,
        target_snap_point_vector_path)


if __name__ == '__main__':
    main()