feat: detect parallel edges in directed and undirected graphs

CHANGELOG.md

@@ -62,6 +62,10 @@ Deprecated (still work, with a `DeprecationWarning`)
 
 Added
 
+- `Graph.isparallel()` detects two or more distinct edges with the same
+  endpoints; `Graph.parallel_edges()` returns their groups as tuples in edge
+  insertion order, with groups ordered by first appearance (respecting
+  direction in a `DGraph`).
 - `adjacent()` and `isadjacent()`, the graph-theory names, which have one
   spelling.  `neighbours()`, `neighbors()` and `isneighbour()` from 0.6.5 are
   kept as aliases, with `isneighbor()` added, and none of them warn.

README.md

@@ -105,6 +105,8 @@ Graphs belong to the class `UGraph` or `DGraph` for undirected or directed graph
     ***
 - `g.add_edge()` connect two vertices
 - `g.edges()` all edges in the graph
+- `g.isparallel()` true if at least two distinct edges join the same vertices (respecting direction)
+- `g.parallel_edges()` returns `list[tuple[Edge, ...]]` for groups of two or more parallel edges; edges and groups follow insertion order
 - `g.plot()` plots the vertices and edges
 - `g.nc` the number of graph components, 1 if fully connected
 - `g.component(c)` the vertices in component number `c`, numbered from 0

docs/llms_intro.md

@@ -22,7 +22,7 @@ g.highlight_path(path)
 
 - **Classes.** `UGraph` (undirected) and `DGraph` (directed) share the base class `Graph`, which is concrete: `Graph(directed=True)` is a `DGraph`. There is one `Vertex` class for both and one `Edge` class. `UVertex` and `DVertex` are deprecated aliases of `Vertex`. Constructor arguments are keyword-only: `UGraph(dim=2, metric="L2", heuristic=None, vertex_class=None, edge_class=None, verbose=False)`; `UGraph(2)` is a `TypeError`.
 - **Vertices.** `g.add_vertex(coord=None, name=None)` returns the vertex. Default names are `"#0"`, `"#1"`, ..., names are unique, and `g["a"]` or `g[0]` finds a vertex. `for v in g` iterates in the order added; `g.n`, `g.ne` and `g.nc` are the numbers of vertices, edges and components. `g.remove_vertex(v)` and `g.remove_edge(e)` remove them, with their edges.
-- **Edges.** `g.add_edge(v1, v2, cost=None)`; with no `cost` it is the metric distance between the coordinates, and `None` if the vertices have none (planning then raises `ValueError`). `g.edges()` and `v.edges()` return new lists in the order added. `e.v1` and `e.v2` are read-only. `v.adjacent()` is the neighbours, `v.successors()` and `v.predecessors()` are the directed ones, `v.edgeto(w)` the cheapest edge to `w` and `v.edges_to(w)` all of them. A self-loop of an undirected vertex is in `v.edges()` twice and counts 2 in its degree.
+- **Edges.** `g.add_edge(v1, v2, cost=None)`; with no `cost` it is the metric distance between the coordinates, and `None` if the vertices have none (planning then raises `ValueError`). `g.edges()` and `v.edges()` return new lists in the order added. `e.v1` and `e.v2` are read-only. `v.adjacent()` is the neighbours, `v.successors()` and `v.predecessors()` are the directed ones, `v.edgeto(w)` the cheapest edge to `w` and `v.edges_to(w)` all of them. A self-loop of an undirected vertex is in `v.edges()` twice and counts 2 in its degree. `g.isparallel()` reports whether at least two distinct edges join the same vertices (direction matters in a `DGraph`; a single self-loop does not count). `g.parallel_edges()` returns a list of tuples, one per group of two or more parallel edges; each tuple follows edge insertion order and groups follow their first edge's order. Reversed edges form a group only in an undirected graph.
 - **Directed graphs.** `degree` is in plus out (`in_degree`, `out_degree`); `nc` and `component` are weakly connected, ignoring direction.
 - **Planning.** `path_BFS(S, G)` returns `(path, length)` with the fewest edges; `path_UCS(S, G)` and `path_Astar(S, G)` return `(path, length, parents)` of least cost. `S` and `G` are vertices or names, and all three return `None` when there is no path, so check before unpacking.
 - **Metrics.** `"L2"` (default), `"L1"`, `"SE2"` for `(x, y, theta)` poses, `SE2Metric(weight)` to weight the angle, or a function of one difference vector. A* uses `heuristic` (default: the metric), which must be in the same units as the edge costs and must not exceed the true cost to go, or the path is not the shortest.

src/pgraph/PGraph.py

@@ -2313,6 +2313,53 @@ def edges(self) -> list[Edge]:
         """
         return list(self._edges)
 
+    def _parallel_edge_key(self, edge: Edge) -> tuple[int, int]:
+        """Identify an edge's endpoints, preserving direction when directed."""
+        assert edge.v1 is not None and edge.v2 is not None
+        a, b = id(edge.v1), id(edge.v2)
+        if not self._directed and a > b:
+            a, b = b, a
+        return a, b
+
+    def isparallel(self) -> bool:
+        """Test whether at least two distinct edges join the same vertices.
+
+        Direction matters in directed graphs: ``a -> b`` and ``b -> a``
+        are not parallel. Undirected edges are orientation-independent.
+        A single self-loop is not parallel, but two self-loops are.
+
+        Checks each edge once and returns as soon as a repeated endpoint
+        pair is found. Worst-case time and space are O(E).
+
+        :return: True if at least two edges share the same endpoints
+        :seealso: :meth:`parallel_edges` :meth:`edges`
+        """
+        seen: set[tuple[int, int]] = set()
+        for edge in self._edges:
+            key = self._parallel_edge_key(edge)
+            if key in seen:
+                return True
+            seen.add(key)
+        return False
+
+    def parallel_edges(self) -> list[tuple[Edge, ...]]:
+        """Return groups of at least two distinct parallel edges.
+
+        Each tuple contains all edges joining the same vertices, in edge
+        insertion order. Groups are ordered by the first edge in each group.
+        Direction matters for directed graphs but not undirected graphs.
+        A single self-loop is excluded; two or more are grouped together.
+        Grouping takes O(E) time and O(E) space.
+
+        :return: List of tuples of parallel edges, or an empty list
+        :seealso: :meth:`isparallel` :meth:`edges`
+        """
+        groups: dict[tuple[int, int], list[Edge]] = {}
+        for edge in self._edges:
+            key = self._parallel_edge_key(edge)
+            groups.setdefault(key, []).append(edge)
+        return [tuple(edges) for edges in groups.values() if len(edges) >= 2]
+
     def _edge_layout(self) -> tuple[dict[Edge, float], dict[Edge, int]]:
         """
         How each edge is to be drawn, apart from its end points (private method)

tests/test_isparallel.py

@@ -0,0 +1,198 @@
+"""Tests for the graph-level parallel-edge predicate."""
+
+import random
+import unittest
+
+from pgraph import DGraph, Graph, UGraph
+
+
+class TestIsParallel(unittest.TestCase):
+    @staticmethod
+    def vertices(graph):
+        a = graph.add_vertex([0, 0], name="a")
+        b = graph.add_vertex([1, 0], name="b")
+        c = graph.add_vertex([2, 0], name="c")
+        return a, b, c
+
+    def test_empty_and_simple_graphs(self):
+        for graph_class in (UGraph, DGraph):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                self.assertFalse(graph.isparallel())
+                self.assertEqual(graph.parallel_edges(), [])
+                a, b, c = self.vertices(graph)
+                self.assertFalse(graph.isparallel())
+                graph.add_edge(a, b, cost=1)
+                graph.add_edge(b, c, cost=2)
+                self.assertFalse(graph.isparallel())
+                self.assertEqual(graph.parallel_edges(), [])
+                self.assertEqual(graph.validate(), [])
+
+    def test_repeated_edge_and_removal(self):
+        for graph_class in (UGraph, DGraph):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                a, b, _ = self.vertices(graph)
+                first = graph.add_edge(a, b, cost=1)
+                second = graph.add_edge(a, b, cost=2)
+                self.assertTrue(graph.isparallel())
+                self.assertEqual(graph.parallel_edges(), [(first, second)])
+                self.assertTrue(graph.isparallel())  # read-only predicate
+                graph.remove_edge(second)
+                self.assertFalse(graph.isparallel())
+                self.assertEqual(graph.parallel_edges(), [])
+                self.assertIn(first, graph.edges())
+                self.assertEqual(graph.validate(), [])
+
+    def test_opposite_directions(self):
+        for graph_class, expected in ((UGraph, True), (DGraph, False)):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                a, b, _ = self.vertices(graph)
+                graph.add_edge(a, b, cost=1)
+                graph.add_edge(b, a, cost=2)
+                self.assertEqual(graph.isparallel(), expected)
+                self.assertEqual(graph.validate(), [])
+
+    def test_self_loops(self):
+        for graph_class in (UGraph, DGraph):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                a, b, _ = self.vertices(graph)
+                first = graph.add_edge(a, a, cost=1)
+                graph.add_edge(b, b, cost=2)
+                self.assertFalse(graph.isparallel())
+                second = graph.add_edge(a, a, cost=3)
+                self.assertTrue(graph.isparallel())
+                graph.remove_edge(first)
+                self.assertFalse(graph.isparallel())
+                self.assertIn(second, graph.edges())
+                self.assertEqual(graph.validate(), [])
+
+    def test_factory_graph_direction(self):
+        for directed, expected in ((False, True), (True, False)):
+            with self.subTest(directed=directed):
+                graph = Graph(directed=directed)
+                a, b, _ = self.vertices(graph)
+                forward = graph.add_edge(a, b)
+                reverse = graph.add_edge(b, a)
+                self.assertEqual(graph.isparallel(), expected)
+                self.assertEqual(
+                    graph.parallel_edges(),
+                    [(forward, reverse)] if expected else [],
+                )
+
+
+    def test_parallel_groups_are_complete_and_ordered_by_first_edge(self):
+        for graph_class in (UGraph, DGraph):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                a, b, c = self.vertices(graph)
+                first_ab = graph.add_edge(a, b, cost=1)
+                first_bc = graph.add_edge(b, c, cost=2)
+                second_bc = graph.add_edge(b, c, cost=3)
+                second_ab = graph.add_edge(a, b, cost=4)
+                third_ab = graph.add_edge(a, b, cost=5)
+
+                # The b-c group completes first, but the a-b group starts first.
+                self.assertEqual(
+                    graph.parallel_edges(),
+                    [(first_ab, second_ab, third_ab), (first_bc, second_bc)],
+                )
+                self.assertTrue(graph.isparallel())
+                self.assertEqual(
+                    graph.edges(),
+                    [first_ab, first_bc, second_bc, second_ab, third_ab],
+                )
+
+                graph.remove_edge(second_ab)
+                self.assertEqual(
+                    graph.parallel_edges(),
+                    [(first_ab, third_ab), (first_bc, second_bc)],
+                )
+
+                # Removing the first edge changes the order of surviving groups.
+                graph.remove_edge(first_ab)
+                self.assertEqual(
+                    graph.parallel_edges(), [(first_bc, second_bc)]
+                )
+                replacement = graph.add_edge(a, b, cost=6)
+                self.assertEqual(
+                    graph.parallel_edges(),
+                    [(first_bc, second_bc), (third_ab, replacement)],
+                )
+                self.assertEqual(graph.validate(), [])
+
+    def test_reversed_undirected_edge_and_directed_edge(self):
+        for graph_class in (UGraph, DGraph):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                a, b, _ = self.vertices(graph)
+                forward = graph.add_edge(a, b, cost=1)
+                reverse = graph.add_edge(b, a, cost=2)
+                if graph_class is UGraph:
+                    self.assertEqual(graph.parallel_edges(), [(forward, reverse)])
+                else:
+                    self.assertEqual(graph.parallel_edges(), [])
+                    repeated = graph.add_edge(a, b, cost=3)
+                    self.assertEqual(graph.parallel_edges(), [(forward, repeated)])
+                self.assertEqual(graph.isparallel(), bool(graph.parallel_edges()))
+
+    def test_single_and_double_self_loop_groups(self):
+        for graph_class in (UGraph, DGraph):
+            with self.subTest(graph_class=graph_class):
+                graph = graph_class()
+                a, b, _ = self.vertices(graph)
+                first = graph.add_edge(a, a, cost=1)
+                graph.add_edge(b, b, cost=2)
+                self.assertEqual(graph.parallel_edges(), [])
+                second = graph.add_edge(a, a, cost=3)
+                self.assertEqual(graph.parallel_edges(), [(first, second)])
+                graph.remove_edge(first)
+                self.assertEqual(graph.parallel_edges(), [])
+                self.assertFalse(graph.isparallel())
+
+    def test_random_graphs_match_independent_pairwise_reference(self):
+        # Reproducible test data only; no security-sensitive randomness.
+        rng = random.Random(20261030)  # nosec B311
+
+        def same_endpoints(left, right, directed):
+            same = left.v1 is right.v1 and left.v2 is right.v2
+            reverse = left.v1 is right.v2 and left.v2 is right.v1
+            return same or (not directed and reverse)
+
+        for graph_class in (UGraph, DGraph):
+            for trial in range(250):
+                with self.subTest(graph_class=graph_class, trial=trial):
+                    graph = graph_class()
+                    vertices = [
+                        graph.add_vertex([i, 0], name=f"v{i}")
+                        for i in range(5)
+                    ]
+                    for _ in range(rng.randrange(31)):
+                        a = rng.choice(vertices)
+                        b = rng.choice(vertices)
+                        graph.add_edge(a, b, cost=1)
+
+                    edges = graph.edges()
+                    used = set()
+                    expected = []
+                    for i, edge in enumerate(edges):
+                        if i in used:
+                            continue
+                        indices = [
+                            j for j in range(i, len(edges))
+                            if same_endpoints(
+                                edge, edges[j], graph_class is DGraph
+                            )
+                        ]
+                        used.update(indices)
+                        if len(indices) >= 2:
+                            expected.append(tuple(edges[j] for j in indices))
+
+                    self.assertEqual(graph.parallel_edges(), expected)
+                    self.assertEqual(graph.isparallel(), bool(expected))
+
+
+if __name__ == "__main__":
+    unittest.main()