Utilities#

pysewer.helper.get_upstream_nodes(G: DiGraph, start_node, field: str, value: str) → list[source]#

Returns a list of all upstream nodes in a directed graph G that have a node attribute field with value value, starting from start_node and traversing the graph in reverse order using a breadth-first search algorithm.

Parameters:

  • G (networkx.DiGraph) – The directed graph to traverse.

  • start_node (hashable) – The node to start the traversal from.

  • field (str) – The name of the node attribute to filter by.

  • value (Any) – The value of the node attribute to filter by.

Returns:

A list of all upstream nodes in G that have a node attribute field with value value.

Return type:

List

pysewer.helper.get_path_distance(detailed_path: list[tuple]) → float[source]#

Calculates the total distance of a path given a list of detailed path coordinates.

Parameters:

detailed_path (List[tuple]) – A list of tuples representing the detailed path coordinates.

Returns:

The total distance of the path.

Return type:

float

Examples

>>> get_path_distance([(0, 0), (3, 4), (7, 1)])
9.848857801796104
pysewer.helper.points_close(a, b, tol: float = 1e-06) → bool[source]#

True if two (x, y) coordinates coincide within tol (CRS units).

pysewer.helper.dedupe_coords(coords, tol: float = 1e-06) → list[source]#

Drop consecutive vertices closer than tol (CRS units).

Pipe LineString geometries are built directly from connection-graph node coordinates, so a projected connection point landing on an existing road vertex — e.g. buildings pre-snapped onto road vertices — leaves a duplicate vertex offset only by floating-point noise (~1e-10 at metric coordinates), and a zero-length segment when it coincides exactly. Both stay OGC-valid but break consumers that assume strictly increasing curvilinear distance along the pipe (see issue #4). This removes such near-duplicates while always preserving at least the two endpoints.

pysewer.helper.closest_edges(points: list, edge_gdf: GeoDataFrame, column: str = 'closest_edge', tol: float = 1e-06) → list[source]#

For each point, the column value of the edge nearest to it.

Distance is the true point-to-segment distance (GEOS, via an STRtree), not the distance to the nearest edge vertex. Edges whose distance is within tol (map units) of the minimum count as tied and are resolved by their endpoint coordinates, so the choice does not depend on the platform or on edge order. (A KD-tree over edge vertices stores every shared junction once per edge; which copy it returns on an exact tie depends on the C++ standard library, which made networks differ between macOS and Linux.)

Parameters:

  • points (list of shapely.geometry.Point)

  • edge_gdf (geopandas.GeoDataFrame) – Edges with LineString geometries.

  • column (str) – Column of edge_gdf to return for the chosen edge.

  • tol (float) – Distance tolerance for treating edges as equally close.

Returns:

One column value per point.

Return type:

list

pysewer.helper.get_closest_edge(G: Graph, point: Point) → tuple[source]#

Returns the closest edge to a given point in a networkx graph.

Parameters:#

Gnetworkx.Graph

The graph to search for the closest edge.

pointshapely.geometry.Point

The point to search for the closest edge.

Returns:#

closest_edgetuple

A tuple representing the closest edge to the given point.

pysewer.helper.get_closest_edge_multiple(G: Graph, list_of_points: list) → list[source]#

Returns a list of the closest edges in a networkx graph to a list of points.

Parameters:

  • G (networkx.Graph) – A networkx graph object.

  • list_of_points (list) – A list of shapely Point objects.

Returns:

A list of shapely LineString objects representing the closest edges to the input points.

Return type:

list

pysewer.helper.get_edge_gdf(G: Graph, field: str | None = None, value: Any | None = None, detailed: bool = False) → GeoDataFrame[source]#

Returns a GeoDataFrame of edges in a networkx graph that match a given field and value.

Parameters:

  • G (networkx.Graph) – The graph to extract edges from.

  • field (str, optional) – The edge attribute to filter on.

  • value (any, optional) – The value to filter the edge attribute on.

  • detailed (bool, optional) – If True, returns a detailed GeoDataFrame with all edge attributes. If False, returns a simplified GeoDataFrame with only the edge geometry.

Returns:

A GeoDataFrame of edges that match the given field and value.

Return type:

gpd.GeoDataFrame

pysewer.helper.get_node_gdf(G: Graph, field=None, value=None) → GeoDataFrame[source]#

Returns a GeoDataFrame of nodes in the graph that match the specified field and value.

Parameters:

  • G (nx.Graph) – The input graph.

  • field (str, optional) – The node attribute to filter on.

  • value (str, optional) – The value to filter on.

Returns:

gdf – A GeoDataFrame of nodes that match the specified field and value.

Return type:

gpd.GeoDataFrame

Raises:

ValueError – If no geometry column is found in the GeoDataFrame.

Notes

This function filters the nodes in the input graph based on the specified field and value, and returns a GeoDataFrame containing the filtered nodes and their attributes. The GeoDataFrame is created using the coordinates of the filtered nodes and their attributes.

pysewer.helper.get_node_keys(G: Graph, field: str | None = None, value: str | None = None)[source]#

Returns a list of keys for nodes in the graph G that have a specified field with a specified value.

Parameters:#

Gnetworkx.Graph

The graph to search for nodes.

fieldstr, optional

The field to search for in the node data dictionary. If None, all nodes are returned.

valuestr

The value to search for in the specified field. If None, all nodes with the specified field are returned.

Returns:#

list

A list of keys for nodes in the graph G that have a specified field with a specified value.

pysewer.helper.get_edge_keys(G, field=None, value=None)[source]#

Returns a list of edge keys (tuples of nodes) for the given graph G that have an edge attribute field with value value.

Parameters:#

Gnetworkx.Graph

The graph to search for edges.

fieldstr, optional

The name of the edge attribute to filter on.

valueany, optional

The value of the edge attribute to filter on.

Returns:#

list of tuples

A list of edge keys (tuples of nodes) that have an edge attribute field with value value.

pysewer.helper.get_path_gdf(G, upstream, downstream)[source]#

Returns a GeoDataFrame containing the geometry of the shortest path between two nodes in a graph.

Parameters:#

Gnetworkx.Graph

The graph to find the shortest path in.

upstreamint

The starting node of the path.

downstreamint

The ending node of the path.

Returns:#

gpd.GeoDataFrame

A GeoDataFrame containing the geometry of the shortest path between the upstream and downstream nodes.

pysewer.helper.get_mean_slope(G: MultiDiGraph, upstream: int, downstream: int, us_td: float, ds_td: float) → float[source]#

Calculates the mean slope of a path between two nodes in a graph.

Parameters:#

Gnetworkx.MultiDiGraph

A directed graph object.

upstreamint

The upstream node ID.

downstreamint

The downstream node ID.

us_tdfloat

The upstream node topographic elevation.

ds_tdfloat

The downstream node topographic elevation.

Returns:#

float

The mean slope of the path between the upstream and downstream nodes.

pysewer.helper.ckdnearest(gdfA: GeoDataFrame, gdfB: GeoDataFrame, gdfB_cols=None) → GeoDataFrame[source]#

Returns a GeoDataFrame containing the closest geometry and attributes from gdfB to each geometry in gdfA.

pysewer.helper.remove_third_dimension(geom)[source]#

remove the third dimension of a shapely geometry

pysewer.helper.is_valid_geometry(geometry)[source]#

Check if a geometry is valid.

pysewer.helper.get_sewer_info(G)[source]#

Returns a dictionary with information about the sewer network.

Parameters:

G (networkx.Graph) – The sewer network graph.

Returns:

A dictionary containing the following information: - Total Buildings: Total number of buildings in the network. - Connected Buildings: Buildings connected to a sink. - Unconnected Buildings: Buildings without a connection to a sink

(graph["unrouted_terminals"] set by rsph_tree(), otherwise buildings without any pipe).

  • Pressurized Sewer Length [m]: Total length of pressurized sewers in meters.

  • Gravity Sewer Length [m]: Total length of gravity sewers in meters.

  • Lifting Stations: Total number of lifting stations in the network.

  • Pumping Stations: Total number of pumping stations in the network (excluding those located in buildings).

  • Private Pumps: Total number of pumps located in buildings.

Return type:

dict