wolfhece.hydrology.flowaccdir
Author: HECE - University of Liege, Pierre Archambeau Date: 2024
Copyright (c) 2024 University of Liege. All rights reserved.
This script and its content are protected by copyright law. Unauthorized copying or distribution of this file, via any medium, is strictly prohibited.
Module Contents
- wolfhece.hydrology.flowaccdir.get_direction_mapping(value)[source]
Get the mapping of flow direction value to neighbor indices (i, j).
The array convention is WOLF array with X-axis along (i) and Y-axis along (j), with (0, 0) at the lower-left corner.
The flow direction values are based on the D8 algorithm, where each value corresponds to a specific direction of flow from a cell to one of its 8 neighbors.
- wolfhece.hydrology.flowaccdir._is_neighbor_flows_towards_me(flow_dir, start_i, start_j, visited, watershed)[source]
Check if any unvisited neighbors flow towards the current cell. This function examines the 8 neighboring cells of a given position and determines if any of them have flow directions pointing towards the current cell. Neighbors that flow towards the current cell are marked in the watershed and added to the output queue.
- Parameters:
flow_dir – 2D array containing flow direction values for each cell.
start_i – Row index of the current cell.
start_j – Column index of the current cell.
visited – 2D boolean array tracking which cells have been processed.
watershed – 2D boolean array marking cells that belong to the watershed.
- Returns:
A list of tuples (neighbor_i, neighbor_j) representing neighbors that flow towards the current cell. Returns an empty list if the starting cell has already been visited.
- wolfhece.hydrology.flowaccdir._inverse_flow_dir_numba(flow_dir, seed_i, seed_j, n_seeds, visited, watershed)[source]
Fully Numba-compiled BFS for inverse flow direction.
The entire BFS runs in compiled code, avoiding repeated Python-Numba boundary crossings that occur in the per-cell approach.
- Parameters:
flow_dir – 2D array containing flow direction values for each cell.
seed_i – 1D int32 array of row indices for seed cells.
seed_j – 1D int32 array of column indices for seed cells.
n_seeds – Number of seed cells.
visited – 2D boolean array tracking which cells have been processed.
watershed – 2D boolean array marking cells that belong to the watershed.
- Returns:
A 2D boolean array where True values indicate cells that belong to the watershed defined by the flow direction.
- wolfhece.hydrology.flowaccdir.inverse_flow_dir(flow_dir: numpy.ndarray, queue: collections.deque, visited: numpy.ndarray, watershed: numpy.ndarray)[source]
Inverse flow direction to create watershed.
- Parameters:
flow_dir – 2D array containing flow direction values for each cell.
queue – A deque containing tuples of (i, j) indices for cells to process. Initially contains the starting cell.
visited – 2D boolean array tracking which cells have been processed.
watershed – 2D boolean array marking cells that belong to the watershed.
- Returns:
A 2D boolean array where True values indicate cells that belong to the watershed defined by the flow direction.
- wolfhece.hydrology.flowaccdir._get_largest_neighbor_flowing_towards_me(flow_dir, accumul, start_i, start_j)[source]
Find the largest neighbor cell that flows towards the current cell. This function examines the 8 neighboring cells of a given position and determines if any of them have flow directions pointing towards the current cell. The neighbor with the highest accumulation value is selected.
- Parameters:
flow_dir – 2D array containing flow direction values for each cell.
accumul – 2D array containing accumulation values for each cell.
start_i – Row index of the current cell.
start_j – Column index of the current cell.
visited – 2D boolean array tracking which cells have been processed.
- Returns:
A tuple (neighbor_i, neighbor_j) representing the neighbor that flows towards the current cell and has the highest accumulation value. Returns an empty tuple if no such neighbor exists or if the starting cell has already been visited.
- class wolfhece.hydrology.flowaccdir.FlowAcc(filename=None, data_dir=None)[source]
Bases:
wolfhece.wolf_tiles.Tiles
Class LIDAXE flow accumulation
- _extract_data(bounds_vec: wolfhece.PyVertexvectors.vector, buffer: float = 0.0)[source]
Extract flow accumulation data for the given bounds vector.
This method checks for the presence of the required tiles on disk, downloads them if necessary, creates a VRT file, and crops it to the bounding box defined by the bounds vector.
The resulting cropped raster is returned as a WolfArray.
- Parameters:
bounds_vec – A vector object defining the bounding area for which flow accumulation data is to be extracted.
buffer – A buffer distance in meters to expand the bounding box. This can be used to ensure that the extracted data includes a margin around the area of interest.
- Returns:
A WolfArray containing the flow accumulation data for the specified bounds.
- get_accumulation_from_catchment(catchment: wolfhece.wolf_array.WolfArray) wolfhece.wolf_array.WolfArray[source]
Get accumulation from a catchment.
- Parameters:
catchment – A WolfArray representing the catchment area for which flow accumulation is to be calculated.
- Returns:
A WolfArray containing the flow accumulation values for the specified catchment area.
- get_accumulation_around(x: float, y: float) wolfhece.wolf_array.WolfArray[source]
Get accumulation from flow accumulation.
- Parameters:
x – X coordinate of the point around which flow accumulation is to be calculated.
y – Y coordinate of the point around which flow accumulation is to be calculated.
- Returns:
A WolfArray containing the flow accumulation values for the area around the specified point.
- class wolfhece.hydrology.flowaccdir.FlowDir(filename=None, data_dir=None)[source]
Bases:
wolfhece.wolf_tiles.Tiles
Class to inverse flow direction to create watershed.
- _search_length = 500.0[source]
The search length is the distance in meters that will be used to search for the flow direction tiles around the point of interest. If the catchment touches the border of the extracted flow direction array, it means that the catchment is not complete and we need to search for a larger area. The search length will be increased by this value until we find a sufficiently large area that contains the entire catchment.
- _extract_data(bounds_vec: wolfhece.PyVertexvectors.vector, buffer: float = 0.0)[source]
Extract flow direction data for the given bounds vector.
- Parameters:
bounds_vec – A vector object defining the bounding area for which flow direction data is to be extracted.
buffer – A buffer distance in meters to expand the bounding box.
- Returns:
A WolfArray containing the flow direction data for the specified bounds.
- get_catchment(x: float, y: float) wolfhece.wolf_array.WolfArray[source]
Get catchment from flow direction.
- Parameters:
x – X coordinate of the point for which catchment is to be calculated.
y – Y coordinate of the point for which catchment is to be calculated.
- Returns:
A WolfArray containing the catchment area for the specified point.
- _inverse_flow_dir(flow_dir: wolfhece.wolf_array.WolfArray, x: float, y: float) tuple[wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray][source]
Inverse flow direction to create watershed.
Begin with a quite small area around the point of interest, extract the flow direction data for this area, and inverse the flow direction to create the watershed. If the catchment touches the border of the extracted flow direction array, it means that the catchment is not complete and we need to search for a larger area. The search length will be increased by the value of self._search_length until we find a sufficiently large area that contains the entire catchment.
- Parameters:
flow_dir – A WolfArray containing the flow direction data for the area around the point of interest.
x – X coordinate of the point for which catchment is to be calculated.
y – Y coordinate of the point for which catchment is to be calculated.
- Returns:
A tuple of WolfArrays containing the catchment area and flow direction for the specified point.
- class wolfhece.hydrology.flowaccdir.Lidaxe(flow_direction: pathlib.Path | None = None, flow_accumulation: pathlib.Path | None = None)[source]
Class to handle LIDAXE flow direction and flow accumulation data.
- snap_to_max(x: float, y: float, snap_length: float = 10.0) tuple[float, float][source]
Snap the point to the nearest maximum flow accumulation value within a certain distance.
- Parameters:
x – X coordinate of the point to be snapped.
y – Y coordinate of the point to be snapped.
snap_length – The distance in meters within which to search for the maximum flow accumulation value to snap to.
- Returns:
A tuple containing the new X and Y coordinates of the snapped point.
- get_catchment(x: float, y: float, snap_to_max: bool = True, snap_length: float = 10.0) tuple[wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray, wolfhece.wolf_array.WolfArray][source]
Get catchment from flow direction and flow accumulation.
- Parameters:
x – X coordinate of the point for which catchment is to be calculated.
y – Y coordinate of the point for which catchment is to be calculated.
snap_to_max – If True, the point will be snapped to the nearest maximum flow accumulation value within a certain distance defined by snap_length.
snap_length – The distance in meters within which to search for the maximum flow accumulation value to snap to if snap_to_max is True.
- Returns:
A tuple of WolfArrays containing the catchment area, flow accumulation, and flow direction for the specified point.
- get_catchments(working_area: wolfhece.PyVertexvectors.vector, buffer: float = 2.0) tuple[wolfhece.wolf_array.WolfArrayMB, wolfhece.wolf_array.WolfArrayMB, wolfhece.wolf_array.WolfArrayMB][source]
Get catchments from flow direction and flow accumulation for a working area.
- Parameters:
working_area – A vector object defining the working area for which catchments are to be calculated.
buffer – A buffer distance in meters to expand the bounding box of the working polyline. This can be used to ensure that the extracted data includes a margin around the area of interest.
- Returns:
a tuple of three WolfArrayMB, the first one containing the catchments, the second one containing the flow accumulations, and the third one containing the flow directions for each catchment.
- get_catchments_on_border(working_polyline: wolfhece.PyVertexvectors.vector, exclude_interior: bool = True, buffer: float = 2.0, ignore_smaller_than: float = 100.0) tuple[wolfhece.wolf_array.WolfArrayMB, wolfhece.wolf_array.WolfArrayMB, wolfhece.wolf_array.WolfArrayMB][source]
Get catchments from flow direction and flow accumulation for a working polyline.
- Parameters:
working_polyline – A vector object defining the working polyline for which catchments are to be calculated.
exclude_interior – If True, exclude interior points from the catchment calculation.
buffer – A buffer distance in meters to expand the bounding box of the working polyline. This can be used to ensure that the extracted data includes a margin around the area of interest.
ignore_smaller_than – Minimum area in square meters for a catchment to be included in the results. Catchments smaller than this threshold will be ignored.
- Returns:
a tuple of three WolfArrayMB, the first one containing the catchments, the second one containing the flow accumulations, and the third one containing the flow directions for each catchment.
- get_longest_flow_path(x: float, y: float, snap_to_max: bool = True, snap_length: float = 10.0, prepared_catchment: list[wolfhece.wolf_array.WolfArray] | tuple[wolfhece.wolf_array.WolfArray | None] = None) wolfhece.PyVertexvectors.vector[source]
Find the longest flow path in the catchment.
- Parameters:
x – X coordinate of the point for which the longest flow path is to be calculated.
y – Y coordinate of the point for which the longest flow path is to be calculated.
snap_to_max – If True, the point will be snapped to the nearest maximum flow accumulation value within a certain distance defined by snap_length before calculating the longest flow path.
snap_length – The distance in meters within which to search for the maximum flow accumulation value to snap to if snap_to_max is True.
prepared_catchment – A list or tuple containing pre-extracted catchment, accumulation, and flow direction arrays. If provided, these will be used instead of extracting them again from the flow direction and flow accumulation data. The expected format is (catchment: WolfArray, accumulation: WolfArray, flow_direction: WolfArray). If any of these is None, it will be extracted as needed.
- Returns:
A vector containing the vertices of the longest flow path in the catchment.