More refactoring to graphutils

src/macrogen/graph.py

@@ -10,7 +10,7 @@
 
 import networkx as nx
 
-from macrogen.graphutils import mark_edges_to_delete
+from .graphutils import mark_edges_to_delete, remove_edges, in_path
 from .bibliography import BiblSource
 from .config import config
 from .datings import build_datings_graph
@@ -282,24 +282,6 @@ def scc_subgraphs(graph: nx.MultiDiGraph) -> List[nx.MultiDiGraph]:
 #     return [('List of conflicts', conflicts_file_name)]
 
 
-def remove_edges(source: nx.MultiDiGraph, predicate: Callable[[Any, Any, Dict[str, Any]], bool]):
-    """
-    Returns a subgraph of source that does not contain the edges for which the predicate returns true.
-    Args:
-        source: source graph
-
-        predicate: a function(u, v, attr) that returns true if the edge from node u to node v with the attributes attr should be removed.
-
-    Returns:
-        the subgraph of source induced by the edges that are not selected by the predicate.
-        This is a read-only view, you may want to use copy() on  the result.
-    """
-    to_keep = [(u, v, k) for u, v, k, attr in source.edges(data=True, keys=True)
-               if not predicate(u, v, attr)]
-    return source.edge_subgraph(to_keep)
-    # return nx.restricted_view(source, source.nodes, [(u,v,k) for u,v,k,attr in source.edges if predicate(u,v,attr)])
-
-
 def prepare_timeline_for_keeping(graph: nx.MultiDiGraph, weight=0.1) -> List[Tuple[V, V]]:
     result = []
     for u, v, k, attr in graph.edges(keys=True, data=True):

src/macrogen/graphutils.py

@@ -1,7 +1,7 @@
 from collections import defaultdict
 from datetime import date
 from pathlib import Path
-from typing import List, Iterable, Tuple, Any, Generator, Union, TypeVar
+from typing import List, Iterable, Tuple, Any, Generator, Union, TypeVar, Callable, Dict, Sequence
 
 import networkx as nx
 
@@ -15,7 +15,6 @@
 logger = config.getLogger(__name__)
 
 
-
 def pathlink(*nodes) -> Path:
     """
     Creates a file name for the given path.
@@ -163,4 +162,43 @@ def mark_edges_to_delete(graph: nx.MultiDiGraph, edges: List[Tuple[Any, Any, int
 def mark_edges(graph: nx.MultiDiGraph, edges: List[Tuple[Any, Any, int, Any]], **new_attrs):
     """Mark all edges in the given graph by updating their attributes with the keyword arguments. """
     for u, v, k, *_ in edges:
-        graph.edges[u, v, k].update(new_attrs)
+        graph.edges[u, v, k].update(new_attrs)
+
+
+def remove_edges(source: nx.MultiDiGraph, predicate: Callable[[Any, Any, Dict[str, Any]], bool]):
+    """
+    Returns a subgraph of source that does not contain the edges for which the predicate returns true.
+    Args:
+        source: source graph
+
+        predicate: a function(u, v, attr) that returns true if the edge from node u to node v with the attributes attr should be removed.
+
+    Returns:
+        the subgraph of source induced by the edges that are not selected by the predicate.
+        This is a read-only view, you may want to use copy() on  the result.
+    """
+    to_keep = [(u, v, k) for u, v, k, attr in source.edges(data=True, keys=True)
+               if not predicate(u, v, attr)]
+    return source.edge_subgraph(to_keep)
+    # return nx.restricted_view(source, source.nodes, [(u,v,k) for u,v,k,attr in source.edges if predicate(u,v,attr)])
+
+
+def in_path(edge: Tuple[T, T], path: Sequence[T], cycle=False) -> bool:
+    """
+    Whether edge is part of the given path.
+
+    Args:
+        edge: the edge we search, as a pair of nodes
+        path: the path we search in, as a sequence of nodes
+        cycle: if True, assume path is a cycle, i.e. there is an edge from path[-1] to path[0]
+    """
+    try:
+        first_index = path.index(edge[0])
+        if first_index < len(path) - 1:
+            return path[first_index + 1] == edge[1]
+        elif cycle:
+            return path[0] == edge[1]
+        else:
+            return False
+    except ValueError:
+        return False