Functions Module

This module contains functions related to networks and routing

wayfarer.functions.add_direction_flag(start_node, end_node, attributes, **kwargs)

kwargs so an edge with any properties can be easily passed to the function

wayfarer.functions.add_edge(net, start_node, end_node, key, attributes)

Add an edge to a network. When adding an edge to a network, nodes are automatically added

Parameters:

• net (MultiGraph | MultiDiGraph) – A network

• start_node (int | str) – The start node of the edge

• end_node (int | str) – The end node of the edge

• key (int | str) – The unique key of the edge

• attributes (dict) – Any attributes associated with the edge

Return type:

Edge

>>> net = networkx.MultiGraph()
>>> add_edge(net, 1, 2, "A", {"LEN_": 10})
Edge(start_node=1, end_node=2, key='A', attributes={'LEN_': 10})
>>> print(net)
MultiGraph with 2 nodes and 1 edges

wayfarer.functions.add_single_edge(net, edge)

Add an edge class to a network

Parameters:

• net (MultiGraph | MultiDiGraph) – A network

• edge (Edge) – The edge to add to the network

• end_node – The end node of the edge

>>> net = networkx.MultiGraph()
>>> edge = Edge(start_node=1, end_node=2, key='A', attributes={'LEN_': 10})
>>> add_single_edge(net, edge)
Edge(start_node=1, end_node=2, key='A', attributes={'LEN_': 10})
>>> print(net)
MultiGraph with 2 nodes and 1 edges

wayfarer.functions.edges_to_graph(edges)

Utility function to create a new MultiGraph based on a list of edges

Parameters:

edges (list[Edge] | list[tuple]) – A list of Edge named tuples, or simply tuples

Return type:

MultiGraph

Returns:

A new network

>>> edges = [Edge(0, 1, "A", {}), Edge(1, 2, "B", {})]
>>> net = edges_to_graph(edges)
>>> print(net)
MultiGraph with 3 nodes and 2 edges

wayfarer.functions.get_all_complex_paths(net, node_list)

For a node path, find any self-looping edges on a node and add these combinations to the possible path list

wayfarer.functions.get_all_paths_from_nodes(net, node_list, with_direction_flag=False)

Get all combinations of edges along a set of nodes, rather than just the shortest edges l = [(“a”,”b”), (“c”), (“d”,”e”)] [(‘a’, ‘c’, ‘d’), (‘a’, ‘c’, ‘e’), (‘b’, ‘c’, ‘d’), (‘b’, ‘c’, ‘e’)]

list(itertools.combinations([[1],[1,2]], r=2)) [([1], [1, 2])]

Check for self loops on every node

wayfarer.functions.get_edge_by_key(net, key, with_data=True)

Go through all edge property dicts until the relevant key is found Using an iterator faster than list comprehension See https://stackoverflow.com/a/36413569/179520 answer from networkx dev

https://networkx.github.io/documentation/stable/reference/classes/generated/networkx.Graph.edges.html

400ns with reverse lookup 1.16 seconds with generators

Return type:

Edge

>>> edges = [Edge(0, 1, "A", {}), Edge(1, 2, "B", {"LEN_": 10})]
>>> net = edges_to_graph(edges)
>>> get_edge_by_key(net, "B", with_data=True)
Edge(start_node=1, end_node=2, key='B', attributes={'attributes': {'LEN_': 10}})

wayfarer.functions.get_edges(net, nodes, in_edges=True)

Get all the edges in the network that touch the nodes in the nodes list. Note only works with DiGraph or MultiDiGraph.

Parameters:

• net (object) – a networkx network

• nodes – (list): a list of network nodes e.g. [(224966, 437657), (225195, 437940)]

• in_edges – (bool): set to True to get edges coming into the node, and False to edges leaving the node

Returns:

a dict containing segment codes and their associated network edge

Return type:

end_edges (dict)

wayfarer.functions.get_edges_by_attribute(net, attribute_field, value)

Go through all edge property dicts until the relevant attribute is found. Edges can have different attribute dicts, so add a check that the key exists If the attribute_field does not exist for a particular edge then it is ignored. Returns an iterator

Return type:

Iterable[Edge]

wayfarer.functions.get_edges_from_node_pair(net, start_node, end_node, hide_log_output=False)

Get all edges between two nodes

This function can return multiple edges when there is more than one edge between two nodes

Return type:

AtlasView | None

>>> net = networkx.MultiGraph()
>>> net.add_edges_from([(0, 1, "A", {'val': 'foo'}), (0, 1, "B", {'val': 'bar'})])
['A', 'B']
>>> get_edges_from_node_pair(net, start_node=0, end_node=1)
AtlasView({'A': {'val': 'foo'}, 'B': {'val': 'bar'}})

wayfarer.functions.get_edges_from_nodes(net, node_list, with_direction_flag=False, length_field='LEN_', return_unique=True, shortest_path_only=True)

From a list of nodes, create pairs and then get all the edges between the two nodes. By default only the shortest edges between the edges will be returned, but setting shortest_path_only to False will return all edges ignoring the length_field. Set with_direction_flag to add a new attribute to the edge to show if it is matching the direction of the path. return_unique ensures only one copy of each edge is returned in the case of loops

Return type:

list[Edge]

wayfarer.functions.get_end_nodes(edges, ordered=True)

Find the end nodes of a graph

Return type:

list

>>> edges = [Edge(0 , 1, "A", {}), Edge(1, 2, "B", {})]
>>> get_end_nodes(edges)
[0, 2]

wayfarer.functions.get_multiconnected_nodes(edges, connections=2)

Get all nodes which have at least n connections

>>> edges = [Edge(0 , 1, "A", {}), Edge(1, 2, "B", {})]
>>> get_multiconnected_nodes(edges, connections=2)
[1]

wayfarer.functions.get_nodes_from_edges(edges)

From a list of edges return a unique, ordered list of nodes along the path

wayfarer.functions.get_ordered_end_nodes(edges, end_nodes)

Ensures the end nodes are returned in the same order as the input edges

Return type:

list[int]

>>> edges = [Edge(1 , 2, "A", {}), Edge(1, 0, "B", {})]
>>> nodes = [0, 1]
>>> get_ordered_end_nodes(edges, nodes)
[1, 0]

wayfarer.functions.get_shortest_edge(edges, length_field='LEN_')

From a dictionary of edges, get the shortest edge by its length If two edges have identical lengths then the edge with the higher Id will be returned

wayfarer.functions.get_sink_edges(net)

Get all the sinks edges in the network. These are edges which have no edges attached at their downstream end

Parameters:

net (object) – a networkx network

Returns:

a dict containing sink segment codes and their associated network edge

Return type:

end_edges (dict)

wayfarer.functions.get_sink_nodes(net)

Get all the sinks in the network.

out_degree_iter is the number of edges pointing out of the node http://networkx.lanl.gov/reference/generated/networkx.DiGraph.out_degree_iter.html

Parameters:

net (object) – a networkx network

Returns:

a list of sink nodes [(0,1), (1,0)]

Return type:

sinks (list)

wayfarer.functions.get_source_edges(net)

Get all the source edges in the network. These are edges which have no edges attached at their upstream end

Parameters:

net (object) – a networkx network

Returns:

a dict containing source segment codes and their associated network edge

Return type:

end_edges (dict)

wayfarer.functions.get_source_nodes(net)

Get all the source in the network.

in_degree_iter is the number of edges pointing into of the node http://networkx.lanl.gov/reference/generated/networkx.DiGraph.in_degree_iter.html

Parameters:

net (object) – a networkx network

Returns:

a list of sink nodes [(0,1), (1,0)]

Return type:

sinks (list)

wayfarer.functions.get_unattached_node(net, edge)

Given an edge find which end is unattached on the network If the edge is part of a loop then return the start node If the edge is attached at both ends then None is returned

wayfarer.functions.get_unique_ordered_list(items)

An order-preserving function to get a unique list of edges or nodes

>>> get_unique_ordered_list([1, 1, 2, 3, 4, 5, 6, 5])
[1, 2, 3, 4, 5, 6]

Parameters:

items (Iterable) – Any iterable

Return type:

list

Returns:

A list of unique values

wayfarer.functions.has_overlaps(edges)

Checks if the provided list of edges has overlaps If an edge is split but the measures don’t overlap then this will return False The same edge id can be found at the start and end of the sequence if there is no overlap of measures

Return type:

bool

>>> edges = [Edge(0, 1, "A", {"EDGE_ID": 1, "OFFSET": 0, "LEN_": 10}), Edge(1, 2, "B", {"EDGE_ID": 2, "OFFSET": 0, "LEN_": 10})]
>>> has_overlaps(edges)
False

wayfarer.functions.pairwise(iterable)

Loops through the iterable, returning adjacent pairs each time. Useful for returning point coordinates.

Return type:

Iterable[tuple]

>>> list(pairwise((0, 10, 20, 30)))
[(0, 10), (10, 20), (20, 30)]

wayfarer.functions.remove_edge_by_key(net, key)

Remove an edge from a network using its unique key.

Parameters:

• net (MultiGraph | MultiDiGraph) – A network

• key (int | str) – The unique key of the edge

>>> net = networkx.MultiGraph()
>>> add_edge(net, 1, 2, "A", {"LEN_": 10})
Edge(start_node=1, end_node=2, key='A', attributes={'LEN_': 10})
>>> print(len(net.edges()))
1
>>> remove_edge_by_key(net, "A")
>>> print(len(net.edges()))
0