Utilities#

CRS#

Coordinate Reference System conversions.

Coordinate Reference Systems (CRSs) differ across datasets, and standardizing and managing these across the workflow is a necessary technical detail. That said, rarely does the user care what coordinate system is being used, as long as it is appropriate for the watershed in question. Watershed Workflow aims to make using datasets in different CRSs as streamlined as possible. Typically, a workflow will pick a CRS based upon either a default for the region or by simply using the CRS of the shapefile that specifies the watershed boundary. This CRS is the passed into each function that acquires more data, and that data’s coordinates are changed to the CRS requested.

This process is made more difficult by the fact that most python GIS packages provide their own class object to store the CRS. This said, nearly all of them are based, to some degree, on the proj4 library and its python wrapper, pyproj for coordinate transformations. Watershed Workflow uses the pyproj.Proj class as its own internal representation of coordinate system, and provides methods for mapping fiona (shapefiles), rasterio (rasters), and cartopy (plotting) CRS objects to and from this type. While this is typically done by calling functions from those libraries, standardizing the API makes dealing with these packages in an integrated form much simpler.

Note

We intend to use the pyproj.Proj object as our standard. But for now we are trying to avoid hard-coding that, so internal code should avoid using that knowledge, and instead map to and from pyproj.Proj objects using the provided interface.

watershed_workflow.crs.from_cartopy(crs: cartopy.crs.CRS) → CRS[source]#

Convert cartopy CRS to workflow CRS.

Parameters:

crs (cartopy.crs.CRS) – Cartopy CRS object to convert.

Returns:

Workflow CRS object.

Return type:

CRS

watershed_workflow.crs.from_epsg(epsg: int) → CRS[source]#

Convert EPSG code to workflow CRS.

Parameters:

epsg (int) – EPSG code identifying the coordinate reference system.

Returns:

Workflow CRS object.

Return type:

CRS

watershed_workflow.crs.from_proj(crs: CRS) → CRS[source]#

Convert a Proj CRS to the workflow CRS standard.

Parameters:

crs (CRS) – Proj CRS object to convert.

Returns:

Workflow CRS standard (currently identical to input).

Return type:

CRS

watershed_workflow.crs.from_rasterio(crs: rasterio.crs.CRS) → CRS[source]#

Convert rasterio CRS to workflow CRS standard.

Parameters:

crs (rasterio.crs.CRS) – Rasterio CRS object to convert.

Returns:

Workflow CRS standard.

Return type:

CRS

Notes

Attempts to use authority codes first for better bounds handling, falls back to general user input conversion.

watershed_workflow.crs.from_string(string: str) → CRS[source]#

Create CRS from proj string specification.

Parameters:

string (str) – Proj string specification of coordinate reference system.

Returns:

Workflow CRS object.

Return type:

CRS

watershed_workflow.crs.from_wkt(string: str) → CRS[source]#

Create CRS from Well-Known Text (WKT) string.

Parameters:

string (str) – WKT string specification of coordinate reference system.

Returns:

Workflow CRS object.

Return type:

CRS

watershed_workflow.crs.from_xarray(array: xarray.DataArray) → CRS | None[source]#

Extract CRS from xarray DataArray.

Parameters:

array (xarray.DataArray) – xarray DataArray that may contain CRS information.

Returns:

Extracted CRS if found, None otherwise.

Return type:

CRS or None

Notes

Attempts to find CRS from rioxarray extension first, then from spatial_ref attributes.

watershed_workflow.crs.isEqual(crs1: CRS, crs2: CRS) → bool[source]#

Check if two CRS objects are equal.

Parameters:

  • crs1 (CRS) – First CRS to compare.

  • crs2 (CRS) – Second CRS to compare.

Returns:

True if CRS objects are considered equal.

Return type:

bool

watershed_workflow.crs.isNative(crs: CRS) → bool[source]#

Check if CRS is in the native format.

Parameters:

crs (CRS) – Coordinate reference system to check.

Returns:

True if CRS is in native format.

Return type:

bool

watershed_workflow.crs.to_cartopy(crs: CRS) → cartopy.crs.CRS[source]#

Convert workflow CRS to cartopy CRS.

Parameters:

crs (CRS) – Workflow CRS to convert.

Returns:

Cartopy CRS object.

Return type:

cartopy.crs.CRS

Notes

Requires cartopy version 0.20.0 or higher. Adopted from: https://pyproj4.github.io/pyproj/stable/crs_compatibility.html

watershed_workflow.crs.to_epsg(crs: CRS) → int[source]#

Convert CRS to EPSG code.

Parameters:

crs (CRS) – Workflow CRS to convert.

Returns:

EPSG code.

Return type:

int

Raises:

ValueError – If CRS cannot be represented as an EPSG code.

watershed_workflow.crs.to_proj(crs: CRS) → CRS[source]#

Convert workflow CRS standard to a Proj CRS.

Parameters:

crs (CRS) – Workflow CRS to convert.

Returns:

Proj CRS object (currently identical to input).

Return type:

CRS

watershed_workflow.crs.to_rasterio(crs: CRS) → rasterio.crs.CRS[source]#

Convert workflow CRS to rasterio CRS.

Parameters:

crs (CRS) – Workflow CRS to convert.

Returns:

Rasterio CRS object.

Return type:

rasterio.crs.CRS

watershed_workflow.crs.to_wkt(crs: CRS) → str[source]#

Convert CRS to Well-Known Text (WKT) string.

Parameters:

crs (CRS) – Workflow CRS to convert.

Returns:

WKT string representation of the CRS.

Return type:

str

Package Configuration#

Configuration and global defaults.

watershed_workflow.config.getDefaultConfig() → ConfigParser[source]#

Dictionary of all config option defaults.

Returns:

rcParams – A dict-like object containing parameters.

Return type:

configparser.ConfigParser

watershed_workflow.config.setDataDirectory(path: str) → None[source]#

Sets the directory in which all data is stored.

Plotting#

watershed_workflow.colors.blackzerojet_cmap(data: ndarray) → LinearSegmentedColormap[source]#

Create a jet colormap with zero values set to black.

Parameters:

data (np.ndarray) – Data array used to determine color scaling.

Returns:

Custom colormap with zero values as black.

Return type:

matplotlib.colors.LinearSegmentedColormap

watershed_workflow.colors.cm_discrete(ncolors: int, cmap: str | ~matplotlib.colors.Colormap = <matplotlib.colors.LinearSegmentedColormap object>) → LinearSegmentedColormap[source]#

Calculate a discrete colormap with N entries from the continuous colormap cmap.

Parameters:

  • ncolors (int) – Number of colors.

  • cmap (str or matplotlib.colors.Colormap instance, optional) – The colormap to discretize. Default is ‘jet’.

Return type:

matplotlib.colors.LinearSegmentedColormap instance

Example

# plot 4 lines
x = np.arange(10)
colors = cmap_discretize('jet', 4)
for i in range(4):
    plt.plot(x, x**i, color=colors[i])
watershed_workflow.colors.cm_mapper(vmin: float = 0.0, vmax: float = 1.0, cmap: str | Colormap | None = None, norm: Normalize | None = None) → Callable[[float], Tuple[float, float, float, float]][source]#

Provide a function that maps scalars to colors in a given colormap.

Parameters:

  • vmin (scalar) – Min and max scalars to be mapped.

  • vmax (scalar) – Min and max scalars to be mapped.

  • cmap (str or matplotlib.colors.Colormap instance) – The colormap to discretize.

  • norm (optional, matplotlib Norm object) – A normalization.

Return type:

Function, cmap(scalar) -> (r,g,b,a)

Example

# plot 4 lines
x = np.arange(10)
cm = cm_mapper(0,3,'jet')
for i in range(4):
    plt.plot(x, x**i, color=cm(i))
watershed_workflow.colors.createIndexedColorbar(ncolors: int, cmap: Colormap, labels: List[str] | None = None, **kwargs: Any) → Colorbar[source]#

Add an indexed colorbar based on a given colormap.

This sets ticks in the middle of each color range, adds the colorbar, and sets the labels if provided.

Parameters:

  • ncolors (int) – Number of colors to display.

  • cmap (matplotlib.colors.Colormap instance) – The colormap used in the image.

  • labels (list) – Optional list of label strings that equal to the number of colors. If not provided, labels are set to range(ncolors).

  • kwargs (dict) – Other arguments are passed on to the plt.colorbar() call, which can be used for things like fraction, pad, etc to control the location/spacing of the colorbar.

Returns:

colorbar

Return type:

the colorbar object

watershed_workflow.colors.createIndexedColormap(indices: List[int] | ndarray, cmap: str | Colormap | None = None) → Tuple[List[int], ListedColormap, BoundaryNorm, List[int], List[str]][source]#

Generates an indexed colormap and labels for imaging, e.g. soil indices. :param indices: Collection of indices that will be used in this colormap. :type indices: iterable(int) :param cmap: Name of matplotlib colormap to sample. :type cmap: optional, str

Returns:

  • indices_out (list(int)) – The unique, sorted list of indices found.

  • cmap (cmap-type) – A linestringed map for use with plots.

  • norm (BoundaryNorm) – A norm for use in plot_trisurf() or other plotting methods to ensure correct NLCD colors.

  • ticks (list(int)) – A list of tick locations for the requested indices. For use with set_ticks().

  • labels (list(str)) – A list of labels associated with the ticks. For use with set_{x,y}ticklabels().

watershed_workflow.colors.createMODISColormap(indices: List[int] | None = None, formatted: bool = False) → Tuple[List[int], ListedColormap, BoundaryNorm, List[float], List[str]]#

Generates a colormap and labels for imaging with the MODIS colors.

Parameters:

indices (iterable(int), optional) – Collection of MODIS indices that will be used in this colormap. If None (default), uses all MODIS indices.

Returns:

  • indices_out (list(int)) – The unique, sorted list of indices found.

  • cmap (cmap-type) – A linestringed map for use with plots.

  • norm (BoundaryNorm) – A norm for use in plot_trisurf() or other plotting methods to ensure correct MODIS colors.

  • ticks (list(int)) – A list of tick locations for the requested indices. For use with set_ticks().

  • labels (list(str)) – A list of labels associated with the ticks. For use with set_{x,y}ticklabels().

  • formatted (bool,) – To make the labels formatted nicely (i.e. add newline in long label names)

Example

Plot a triangulation given a set of MODIS colors on those triangles.

Given a triangluation mesh_points, mesh_tris and MODIS color indices for each triangle (tri_modis):

indices, cmap, norm, ticks, labels = generate_modis_colormap(set(tri_modis))

mp = ax.plot_trisurf(mesh_points[:,0], mesh_points[:,1], mesh_points[:,2],
        triangles=mesh_tris, color=tri_modis,
        cmap=cmap, norm=norm)
cb = fig.colorbar(mp, orientation='horizontal')
cb.set_ticks(ticks)
cb.ax.set_xticklabels(labels, rotation=45)
watershed_workflow.colors.createNLCDColormap(indices: List[int] | None = None, formatted: bool = False) → Tuple[List[int], ListedColormap, BoundaryNorm, List[float], List[str]]#

Generates a colormap and labels for imaging with the NLCD colors.

Parameters:

indices (iterable(int), optional) – Collection of NLCD indices that will be used in this colormap. If None (default), uses all NLCD indices.

Returns:

  • indices_out (list(int)) – The unique, sorted list of indices found.

  • cmap (cmap-type) – A linestringed map for use with plots.

  • norm (BoundaryNorm) – A norm for use in plot_trisurf() or other plotting methods to ensure correct NLCD colors.

  • ticks (list(int)) – A list of tick locations for the requested indices. For use with set_ticks().

  • labels (list(str)) – A list of labels associated with the ticks. For use with set_{x,y}ticklabels().

  • formatted (bool,) – To make the labels formatted nicely (i.e. add newline in long label names)

Example

Plot a triangulation given a set of NLCD colors on those triangles.

Given a triangluation mesh_points, mesh_tris and NLCD color indices for each triangle (tri_nlcd):

indices, cmap, norm, ticks, labels = generate_nlcd_colormap(set(tri_nlcd))

mp = ax.plot_trisurf(mesh_points[:,0], mesh_points[:,1], mesh_points[:,2],
        triangles=mesh_tris, color=tri_nlcd,
        cmap=cmap, norm=norm)
cb = fig.colorbar(mp, orientation='horizontal')
cb.set_ticks(ticks)
cb.ax.set_xticklabels(labels, rotation=45)
watershed_workflow.colors.darken(color: str | Tuple[float, float, float] | Tuple[float, float, float, float], fraction: float = 0.6) → Tuple[float, float, float][source]#

Darken a color by reducing its brightness.

Parameters:

  • color (Color) – Color to darken.

  • fraction (float, optional) – Fraction of brightness to remove (0-1). Default is 0.6.

Returns:

Darkened color in RGB format.

Return type:

Tuple[float, float, float]

watershed_workflow.colors.desaturate(color: str | Tuple[float, float, float] | Tuple[float, float, float, float], amount: float = 0.4, is_hsv: bool = False) → ndarray[source]#

Desaturate a color by reducing its saturation.

Parameters:

  • color (Color or np.ndarray) – Color to desaturate. Can be in RGB or HSV format.

  • amount (float, optional) – Amount of desaturation to apply (0-1). Default is 0.4.

  • is_hsv (bool, optional) – If True, input color is in HSV format. Default is False (RGB).

Returns:

Desaturated color in RGB format.

Return type:

np.ndarray

watershed_workflow.colors.enumerated_colors(count: int, palette: int | List[str | Tuple[float, float, float] | Tuple[float, float, float, float]] = 1, chain: bool = True) → List[str | Tuple[float, float, float] | Tuple[float, float, float, float]][source]#

Gets an enumerated list of count independent colors.

Parameters:

  • count (int) – Number of colors to return.

  • palette (int or List[Color], optional) – Palette identifier (int) or explicit list of colors. Default is 1.

  • chain (bool, optional) – If True, cycle through palette colors when count exceeds palette size. Default is True.

Returns:

List of colors from the specified palette.

Return type:

List[Color]

Raises:

ValueError – If no enumerated palette of sufficient length exists and chain is False.

watershed_workflow.colors.gas_cmap() → LinearSegmentedColormap[source]#

Create a gas-themed colormap.

Returns:

Gas-themed colormap from white to red.

Return type:

matplotlib.colors.LinearSegmentedColormap

watershed_workflow.colors.ice_cmap() → LinearSegmentedColormap[source]#

Create an ice-themed colormap.

Returns:

Ice-themed colormap from white to blue.

Return type:

matplotlib.colors.LinearSegmentedColormap

watershed_workflow.colors.isNearlyGrey(color: str | Tuple[float, float, float] | Tuple[float, float, float, float], tolerance: float = 0.1) → bool[source]#

Determines whether a color is nearly grey.

Parameters:

  • color (Color) – Color to test for greyness.

  • tolerance (float, optional) – Tolerance for RGB component differences. Default is 0.1.

Returns:

True if color is nearly grey (RGB components within tolerance).

Return type:

bool

watershed_workflow.colors.lighten(color: str | Tuple[float, float, float] | Tuple[float, float, float, float], fraction: float = 0.6) → Tuple[float, float, float][source]#

Lighten a color by increasing its brightness.

Parameters:

  • color (Color) – Color to lighten.

  • fraction (float, optional) – Fraction of brightness to add (0-1). Default is 0.6.

Returns:

Lightened color in RGB format.

Return type:

Tuple[float, float, float]

watershed_workflow.colors.measureBoldness(color: str | Tuple[float, float, float] | Tuple[float, float, float, float]) → float[source]#

Calculate a vibrancy and boldness score for a given color.

Parameters:

color (Color) – Color to measure boldness for.

Returns:

A score representing how vibrant and bold the color is (0 to 100).

Return type:

float

watershed_workflow.colors.water_cmap() → LinearSegmentedColormap[source]#

Create a water-themed colormap.

Returns:

Water-themed colormap from white to dark blue.

Return type:

matplotlib.colors.LinearSegmentedColormap

Generic utilities#

Shape and geometry utilities for working with fiona and shapely objects.

Note this module contains a lot of other simple functions that are commonly used by other functions, but are not included in documentation because they are likely not useful to users.

exception watershed_workflow.utils.CutError(message: str, line: LineString, seg: LineString, cutline: LineString)[source]#

Exception raised when cutting geometries fails.

Parameters:

  • message (str) – Error message.

  • line (shapely.geometry.LineString) – The line being cut.

  • seg (shapely.geometry.LineString) – The segment causing issues.

  • cutline (shapely.geometry.LineString) – The cutting line.

watershed_workflow.utils.breakLineStringCollinearity(linestring_coords: ndarray, tol: float = 1e-05) → ndarray[source]#

Remove collinearity from linestring by adding small orthogonal perturbations.

Parameters:

  • linestring_coords (np.ndarray) – Array of linestring coordinates, shape (n, 2).

  • tol (float, optional) – Tolerance for collinearity detection and perturbation size. Default is 1e-5.

Returns:

Modified coordinates with collinearity removed.

Return type:

np.ndarray

watershed_workflow.utils.cluster(points: np.ndarray, tol: float) → Tuple[np.ndarray, np.ndarray][source]#

Cluster points based on distance tolerance.

Parameters:

  • points (np.ndarray) – Array of points to cluster, shape (n, 2).

  • tol (float) – Distance tolerance for clustering.

Returns:

Tuple containing (cluster_indices, cluster_centroids) where cluster_indices is an array of cluster assignments and cluster_centroids are the centroid coordinates of each cluster.

Return type:

tuple of np.ndarray

watershed_workflow.utils.computeAngle(v1: Tuple[float, float] | shapely.geometry.LineString, v2: Tuple[float, float] | shapely.geometry.LineString) → float[source]#

Compute the angle (in degrees) of v2 relative to v1 in clockwise direction.

Parameters:

  • v1 (tuple of float or shapely.geometry.LineString) – First vector as (x, y) coordinates or downstream-oriented LineString.

  • v2 (tuple of float or shapely.geometry.LineString) – Second vector as (x, y) coordinates or downstream-oriented LineString.

Returns:

Angle in degrees of v2 relative to v1 in clockwise direction (0-360).

Return type:

float

watershed_workflow.utils.computeArclengths(ls: LineString) → ndarray[source]#

Compute cumulative arc length at each coordinate in a LineString.

Parameters:

ls (shapely.geometry.LineString) – LineString to analyze.

Returns:

Array of cumulative arc lengths, shape (n,) for n coordinates.

Return type:

np.ndarray

watershed_workflow.utils.computeArea(vertices: Iterable[Tuple[float]]) → float[source]#

Compute the area of a polygon given its vertices.

Parameters:

vertices (iterable of tuple) – Vertices of the polygon as coordinate tuples.

Returns:

Area of the polygon.

Return type:

float

watershed_workflow.utils.computeDistance(p1: Tuple[float, float], p2: Tuple[float, float]) → float[source]#

Compute Euclidean distance between two points.

Parameters:

  • p1 (tuple of float) – First point as (x, y) coordinates.

  • p2 (tuple of float) – Second point as (x, y) coordinates.

Returns:

Euclidean distance between the two points.

Return type:

float

watershed_workflow.utils.computeMidpoint(p1: Tuple[float, float] | np.ndarray, p2: Tuple[float, float] | np.ndarray) → Tuple[float, float][source]#

Compute the midpoint between two points.

Parameters:

  • p1 (tuple of float) – First point as (x, y) coordinates.

  • p2 (tuple of float) – Second point as (x, y) coordinates.

Returns:

Midpoint as (x, y) coordinates.

Return type:

tuple of float

watershed_workflow.utils.computeSegmentLengths(ls: LineString) → ndarray[source]#

Compute the length of each segment in a LineString.

Parameters:

ls (shapely.geometry.LineString) – LineString to analyze.

Returns:

Array of segment lengths, shape (n-1,) for n coordinates.

Return type:

np.ndarray

watershed_workflow.utils.computeTriangleArea(p1: Tuple[float, float], p2: Tuple[float, float], p3: Tuple[float, float]) → float[source]#

Compute the area of a triangle given three vertices.

Parameters:

  • p1 (tuple of float) – First vertex as (x, y) coordinates.

  • p2 (tuple of float) – Second vertex as (x, y) coordinates.

  • p3 (tuple of float) – Third vertex as (x, y) coordinates.

Returns:

Area of the triangle.

Return type:

float

watershed_workflow.utils.computeTriangleCentroid(p1: Tuple[float, float], p2: Tuple[float, float], p3: Tuple[float, float]) → Tuple[float, float][source]#

Compute the centroid of a triangle given three vertices.

Parameters:

  • p1 (tuple of float) – First vertex as (x, y) coordinates.

  • p2 (tuple of float) – Second vertex as (x, y) coordinates.

  • p3 (tuple of float) – Third vertex as (x, y) coordinates.

Returns:

Centroid of the triangle as (x, y) coordinates.

Return type:

tuple of float

watershed_workflow.utils.contains(s1: BaseGeometry, s2: BaseGeometry, tol: float = 1e-07) → bool[source]#

Check if one geometry contains another with tolerance for roundoff issues.

Parameters:

  • s1 (BaseGeometry) – Container geometry.

  • s2 (BaseGeometry) – Geometry to test for containment.

  • tol (float, optional) – Tolerance buffer for containment test. Default uses module tolerance.

Returns:

True if s1 contains s2 within tolerance.

Return type:

bool

watershed_workflow.utils.cut(line1: shapely.geometry.LineString, line2: shapely.geometry.LineString) → Tuple[List[shapely.geometry.LineString], List[shapely.geometry.LineString]][source]#

Cut two LineStrings at their intersection point.

Parameters:

  • line1 (shapely.geometry.LineString) – First LineString to cut.

  • line2 (shapely.geometry.LineString) – Second LineString to cut.

Returns:

Tuple containing (line1_segments, line2_segments) where each is a list of LineString segments created by the cut operation.

Return type:

tuple of list

watershed_workflow.utils.filterToShape(df: gpd.GeoDataFrame, shape: BaseGeometry, shape_crs: CRS, method: str = 'contains', tol: float | None = None) → gpd.GeoDataFrame[source]#

Filter GeoDataFrame features based on spatial relationship with shape.

Parameters:

  • df (gpd.GeoDataFrame) – GeoDataFrame to filter.

  • shape (BaseGeometry) – Shapely geometry to use as filter.

  • shape_crs (CRS) – Coordinate reference system of the shape.

  • method (str, optional) – Spatial relationship method: ‘contains’, ‘intersects’, or ‘non_point_intersection’. Default is ‘contains’.

  • tol (float, optional) – Tolerance for spatial operations. Uses default if None.

Returns:

Filtered GeoDataFrame containing features that satisfy the spatial relationship.

Return type:

gpd.GeoDataFrame

Raises:

ValueError – If method is not one of the supported options.

watershed_workflow.utils.findClosestPointIndex(point: Tuple[float, float], points: npt.ArrayLike) → int[source]#

Find the index of the closest point in an array of points.

Parameters:

  • point (tuple of float) – Reference point as (x, y) coordinates.

  • points (array_like) – Array of points to search, shape (n, 2).

Returns:

Index of the closest point in the points array.

Return type:

int

watershed_workflow.utils.findNearestPoint(point: shapely.geometry.Point | Tuple[float, float], line: shapely.geometry.LineString, tol: float | None = None) → Tuple[float, float] | None[source]#

Find the nearest point on a LineString to a given point.

Parameters:

  • point (shapely.geometry.Point or tuple of float) – Point to find nearest location for.

  • line (shapely.geometry.LineString) – LineString to search along.

  • tol (float, optional) – Distance tolerance. If provided, returns None if nearest point is farther than tolerance.

Returns:

Nearest point coordinates as (x, y) tuple, or None if outside tolerance.

Return type:

tuple of float or None

watershed_workflow.utils.flatten(list_of_shps: Any) → List[BaseGeometry][source]#

Flatten a list of shapes by expanding Multi-objects into individual geometries.

Parameters:

list_of_shps (iterable) – List of shapely geometry objects, may contain Multi-objects.

Returns:

Flattened list containing only single geometry objects.

Return type:

list of BaseGeometry

watershed_workflow.utils.generateCoords(obj: Any) → Iterable[Tuple[float, float]][source]#

Generate all coordinates from a fiona geometry object.

Parameters:

obj (dict) – Fiona shape dictionary with ‘coordinates’ or ‘geometry’ key.

Yields:

tuple of float – Individual coordinate tuples.

watershed_workflow.utils.generateRings(obj: Any) → Iterable[List[Tuple[float, float]]][source]#

Generate coordinate rings from a fiona shape object.

Parameters:

obj (dict) – Fiona shape dictionary with ‘coordinates’ or ‘geometry’ key.

Yields:

list of tuple – Each ring as a list of coordinate tuples.

watershed_workflow.utils.inNeighborhood(shp1: BaseGeometry, shp2: BaseGeometry, tol: float = 0.1) → bool[source]#

Check if two geometries can possibly intersect using bounding box test.

Parameters:

  • shp1 (BaseGeometry) – First shapely geometry object.

  • shp2 (BaseGeometry) – Second shapely geometry object.

  • tol (float, optional) – Tolerance for bounding box expansion. Default is 0.1.

Returns:

True if bounding boxes indicate possible intersection.

Return type:

bool

watershed_workflow.utils.intersectPointToSegment(point: Point, seg_start: Point, seg_end: Point) → Point[source]#

Find the nearest point on a line segment to a given point.

Parameters:

  • point (shapely.geometry.Point) – Point to find nearest location for.

  • seg_start (shapely.geometry.Point) – Start point of the line segment.

  • seg_end (shapely.geometry.Point) – End point of the line segment.

Returns:

Nearest point on the line segment to the input point.

Return type:

shapely.geometry.Point

watershed_workflow.utils.intersects(shp1: BaseGeometry, shp2: BaseGeometry) → bool[source]#

Check whether two geometries intersect.

Parameters:

  • shp1 (BaseGeometry) – First shapely geometry object.

  • shp2 (BaseGeometry) – Second shapely geometry object.

Returns:

True if geometries intersect, False otherwise.

Return type:

bool

Notes

This function computes the actual intersection rather than using shapely.intersects() for more reliable results.

watershed_workflow.utils.isClose(s1: BaseGeometry, s2: BaseGeometry, tol: float = 1e-07) → bool[source]#

Check if two shapely geometries are topologically equivalent and geometrically close.

Parameters:

  • s1 (BaseGeometry) – First shapely geometry object.

  • s2 (BaseGeometry) – Second shapely geometry object.

  • tol (float, optional) – Tolerance for geometric comparison. Default uses module tolerance.

Returns:

True if geometries are topologically equivalent and geometrically close.

Return type:

bool

Notes

This function handles complex cases like polygon rotations and coordinate ordering that simple coordinate comparison would miss.

Raises:

NotImplementedError – If geometry type is not supported.

watershed_workflow.utils.isCollinear(p1: Tuple[float, float], p2: Tuple[float, float], p3: Tuple[float, float], tol: float = 1e-06) → bool[source]#

Check if three points are collinear within a given tolerance.

Parameters:

  • p1 (tuple of float) – First point as (x, y) coordinates.

  • p2 (tuple of float) – Second point as (x, y) coordinates.

  • p3 (tuple of float) – Third point as (x, y) coordinates.

  • tol (float, optional) – Tolerance for collinearity test. Default is 1e-6.

Returns:

True if the three points are collinear within tolerance.

Return type:

bool

watershed_workflow.utils.isConvex(points: Iterable[Tuple[float, float]]) → bool[source]#

Check if a set of points forms a convex polygon.

Parameters:

points (iterable of tuple) – Points as coordinate tuples to check for convexity.

Returns:

True if points form a convex polygon.

Return type:

bool

watershed_workflow.utils.isEmpty(shply: BaseGeometry | None) → bool[source]#

Check if a shapely geometry is None or empty.

Parameters:

shply (BaseGeometry or None) – Shapely geometry object to check.

Returns:

True if geometry is None or empty.

Return type:

bool

watershed_workflow.utils.isNonPointIntersection(shp1: BaseGeometry, shp2: BaseGeometry) → bool[source]#

Check whether two geometries intersect with more than just a point.

Parameters:

  • shp1 (BaseGeometry) – First shapely geometry object.

  • shp2 (BaseGeometry) – Second shapely geometry object.

Returns:

True if geometries intersect with line or area, False if no intersection or point intersection only.

Return type:

bool

Notes

This function computes the actual intersection to check its dimensionality.

watershed_workflow.utils.isVolumetricIntersection(shp1: BaseGeometry, shp2: BaseGeometry) → bool[source]#

Check whether two geometries have a volumetric (area) intersection.

Parameters:

  • shp1 (BaseGeometry) – First shapely geometry object.

  • shp2 (BaseGeometry) – Second shapely geometry object.

Returns:

True if intersection has positive area, False otherwise.

Return type:

bool

watershed_workflow.utils.logMinMaxMedianSegment(iterable: Iterable[shapely.geometry.LineString], name: str, assert_on_zero: bool = False, ax: matplotlib.axes.Axes | None = None, color: str | Tuple | None = None) → None[source]#

Log statistics and optionally plot histogram of segment lengths.

Parameters:

  • iterable (iterable of shapely.geometry.LineString) – Collection of LineStrings to analyze.

  • name (str) – Name for logging and plot labels.

  • assert_on_zero (bool, optional) – If True, raise assertion error if any segment has zero length.

  • ax (matplotlib.axes.Axes, optional) – Axes to plot histogram on. No plot if None.

  • color (str or tuple, optional) – Color for histogram plot.

Raises:

AssertionError – If assert_on_zero is True and zero-length segments are found.

watershed_workflow.utils.projectVectorAtAngle(v1: Tuple[float, float] | shapely.geometry.LineString, angle: float, distance: float) → Tuple[float, float][source]#

Project a vector from v1 at a specified angle and distance.

Parameters:

  • v1 (tuple of float or shapely.geometry.LineString) – Reference vector as (x, y) coordinates or downstream-oriented LineString.

  • angle (float) – Angle in degrees for the projection.

  • distance (float) – Distance (magnitude) of the projected vector.

Returns:

Projected vector as (x, y) coordinates with specified angle and distance.

Return type:

tuple of float

watershed_workflow.utils.recenter(objects: Iterable[BaseGeometry], centering: bool | str | shapely.geometry.Point = True) → Tuple[List[BaseGeometry], shapely.geometry.Point][source]#

Center a collection of objects by translating to remove their centroid.

Parameters:

  • objects (iterable of BaseGeometry) – Collection of shapely geometry objects to center.

  • centering (bool, str, or shapely.geometry.Point, optional) – Centering method: True or ‘geometric’ for geometric center, ‘mass’ for center of mass, or Point for custom center.

Returns:

Tuple containing (centered_objects, centroid_point) where centered_objects is the list of translated geometries and centroid_point is the original centroid.

Return type:

tuple

Raises:

ValueError – If centering method is not recognized.

watershed_workflow.utils.removeThirdDimension(geom: BaseGeometry) → BaseGeometry[source]#

Remove the third dimension (Z-coordinate) from a shapely geometry.

Parameters:

geom (shapely.geometry.base.BaseGeometry) – Input geometry that may have Z-coordinates.

Returns:

Geometry with only X and Y coordinates.

Return type:

shapely.geometry.base.BaseGeometry

watershed_workflow.utils.reverseLineString(ls: LineString) → LineString[source]#

Reverse the direction of a LineString.

Parameters:

ls (shapely.geometry.LineString) – LineString to reverse.

Returns:

LineString with reversed coordinate order.

Return type:

shapely.geometry.LineString

Warping#

Used to warp shapefiles and rasters into new coordinate systems.

watershed_workflow.warp.bounds(bounds: Tuple[float, float, float, float], old_crs: CRS, new_crs: CRS) → Tuple[float, float, float, float][source]#

Warp a bounding box from old_crs to new_crs.

Parameters:

  • bounds (tuple of float) – Bounding box as (minx, miny, maxx, maxy).

  • old_crs (CRS) – Source coordinate reference system.

  • new_crs (CRS) – Target coordinate reference system.

Returns:

Transformed bounding box as (minx, miny, maxx, maxy).

Return type:

tuple of float

Notes

Creates a box geometry from the bounds, transforms it, then returns the new bounds of the transformed box.

watershed_workflow.warp.dataset(ds: Dataset, target_crs: CRS, resampling_method: str = 'nearest') → Dataset[source]#

Reproject an xarray Dataset from its current CRS to a target CRS using rioxarray. Maintains the same width and height as the original dataset.

Parameters:

  • ds (xr.Dataset) – Input dataset with CRS information (ds.rio.crs must be set)

  • target_crs (pyproj.CRS) – Target coordinate reference system as a pyproj.CRS object

  • resampling_method (str, default "nearest") – Resampling method for reprojection (nearest, bilinear, cubic, etc.)

Returns:

Reprojected dataset with x/y coordinates instead of lat/lon

Return type:

xr.Dataset

Examples

>>> from pyproj import CRS
>>> # Load a dataset with lat/lon coordinates
>>> ds = xr.open_dataset('data.nc')
>>> ds = ds.rio.write_crs("EPSG:4326")  # Set CRS if not already set
>>>
>>> # Reproject to Albers Equal Area
>>> target_crs = CRS.from_epsg(5070)
>>> ds_projected = dataset(ds, target_crs=target_crs)
watershed_workflow.warp.points(array: ndarray, old_crs: CRS, new_crs: CRS) → ndarray[source]#

Warp an array of points from old_crs to new_crs.

Parameters:

  • array (numpy.ndarray) – Array of shape (N, 2) containing x,y coordinates.

  • old_crs (CRS) – Source coordinate reference system.

  • new_crs (CRS) – Target coordinate reference system.

Returns:

Transformed array of shape (N, 2) with warped coordinates.

Return type:

numpy.ndarray

watershed_workflow.warp.shply(shp: BaseGeometry, old_crs: CRS, new_crs: CRS) → BaseGeometry[source]#

Warp a shapely geometry object from old_crs to new_crs.

Parameters:

  • shp (shapely.geometry.base.BaseGeometry) – Shapely geometry object to transform.

  • old_crs (CRS) – Source coordinate reference system.

  • new_crs (CRS) – Target coordinate reference system.

Returns:

Transformed shapely geometry object.

Return type:

shapely.geometry.base.BaseGeometry

Notes

If the coordinate systems are equal, returns the input geometry unchanged. Preserves any ‘properties’ attribute on the input geometry.

watershed_workflow.warp.shplys(shps: list, old_crs: CRS, new_crs: CRS) → list[source]#

Warp a collection of shapely geometry objects from old_crs to new_crs.

Parameters:

  • shps (list of shapely.geometry.base.BaseGeometry) – Collection of shapely geometry objects to transform.

  • old_crs (CRS) – Source coordinate reference system.

  • new_crs (CRS) – Target coordinate reference system.

Returns:

List of transformed shapely geometry objects.

Return type:

list of shapely.geometry.base.BaseGeometry

Notes

If the coordinate systems are equal, returns the input geometries unchanged. Preserves any ‘properties’ attribute on each input geometry.

watershed_workflow.warp.xy(x: ndarray, y: ndarray, old_crs: CRS, new_crs: CRS) → Tuple[ndarray, ndarray][source]#

Warp a set of points from old_crs to new_crs.

Parameters:

  • x (numpy.ndarray) – X coordinates in the old coordinate system.

  • y (numpy.ndarray) – Y coordinates in the old coordinate system.

  • old_crs (CRS) – Source coordinate reference system.

  • new_crs (CRS) – Target coordinate reference system.

Returns:

  • x_transformed (numpy.ndarray) – X coordinates in the new coordinate system.

  • y_transformed (numpy.ndarray) – Y coordinates in the new coordinate system.

Notes

If the coordinate systems are equal, returns the input coordinates unchanged.