Routing Solvers#

pysewer.routing.rsph_tree_fast(connection_graph: Graph, sink: list[tuple[float, float]], from_type: str = 'building') → DiGraph[source]#

Returns the directed routed steiner tree that connects all terminal nodes to the sink using the repeated shortest path heuristic.

Parameters:#

connection_graphnx.Graph

The graph representing the sewer network.

sinkList[Tuple[float, float]]

The coordinates of the sink node.

from_typestr, optional

The type of the source nodes, by default “building”.

to_typestr, optional

The type of the destination nodes, by default “wwtp”.

Returns:#

nx.DiGraph

A directed graph representing the routed steiner tree that connects all terminal nodes to the sink.

exception pysewer.routing.UnroutedTerminalsError(unrouted)[source]#

Bases: ValueError

Raised by rsph_tree() with strict=True when terminals cannot reach a sink.

pysewer.routing.rsph_tree(connection_graph: Graph, sinks: list[int | str | Hashable], from_type: str = 'building', skip_nodes: list[int | str | Hashable] | None = None, *, strict: bool = False) → DiGraph[source]#

Returns the directed routed steiner tree that connects all terminal nodes to the sink using the repeated shortest path heuristic.

Parameters:

  • connection_graph (nx.Graph) – The graph representing the sewer network.

  • sinks (List[NodeType]) – The nodes representing the sinks in the sewer network.

  • from_type (str, optional) – The type of the source nodes, by default “building”.

  • to_type (str, optional) – The type of the sink nodes, by default “wwtp”.

  • skip_nodes (List[NodeType], optional) – The nodes to skip in the routing, by default [].

  • strict (bool, optional, keyword-only) – Raise UnroutedTerminalsError (a ValueError) if any terminal cannot reach a sink, by default False.

Returns:

The directed routed steiner tree that connects all terminal nodes to the sink using the repeated shortest path heuristic. Terminals without a path to any sink stay in the graph as isolated nodes; they are listed in graph["unrouted_terminals"] (empty list if every terminal is connected) and a warning is issued.

Return type:

nx.DiGraph

pysewer.routing.find_rsph_path(connection_graph: Graph, subgraph_nodes: list[int | str | Hashable], terminals: list[int | str | Hashable], all_paths: dict, all_lengths: dict) → list[int | str | Hashable][source]#