From 81501062a978f0d2eb63f45197d89047426e8087 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Bertrand=20N=C3=A9ron?= <freeh4cker@gmail.com>
Date: Fri, 6 Sep 2019 17:11:41 +0200
Subject: [PATCH] :sparkles: add support to direted graph and weighted graph
new data structure to support weigthed graph
add 2 concrete implementaion of graph directed and undirected
---
Graph/graph/graph_3.py | 350 +++++++++++++++++++-------------------
Graph/graph/traversing.py | 67 ++++++++
2 files changed, 241 insertions(+), 176 deletions(-)
create mode 100644 Graph/graph/traversing.py
diff --git a/Graph/graph/graph_3.py b/Graph/graph/graph_3.py
index 86f4494..ea05577 100644
--- a/Graph/graph/graph_3.py
+++ b/Graph/graph/graph_3.py
@@ -1,37 +1,67 @@
+from typing import Iterator, Dict
import itertools
-from abc import ABCMeta
-from typing import Iterator, Dict, Set, Sequence, Union, Tuple
+from abc import ABCMeta, abstractmethod
-class Graph(metaclass=ABCMeta):
+class Node:
+ """
+ Graph node representation
+ """
- def __init__(self):
- self.nodes: set[Node] = set()
- self.vertices: Dict[Node: set[Node]]
+ _id = itertools.count(0)
- def add_node(self):
- pass
+ def __init__(self, **kwargs) -> None:
+ self.id: int = next(self._id)
+ for k, v in kwargs:
+ setattr(self, k, v)
- def del_node(self):
- pass
+
+ def __hash__(self) -> int:
+ # to be usable in set an object must be hashable
+ # so we need to implement __hash__
+ # which must return an int unique per object
+ # here we have the Node identifier
+ return self.id
+
+ def __str__(self) -> str:
+ return f"node_{self.id}"
+
+ def __repr__(self) -> str:
+ return str(self)
-class NDGraph:
+class Edge:
"""
- To handle Non Directed Graph
- A graph is a collection of Nodes linked by edges
- there are basically to way to implement a graph
- - The graph handles Nodes, each nodes know it's neighbors
- - The graph handles nodes and the edges between nodes
- below I implemented the 2nd version.
+ modelize edge between two nodes, each edge can support properties
"""
- def __init__(self, nodes: Sequence['Node']) -> None:
- self.nodes: Set['Node'] = {n for n in nodes}
- self.vertices: Dict['Node': Set['Node']] = {n: set() for n in self.nodes}
+ def __init__(self, src: Node, target: Node, directed=True, **kwargs) -> None:
+ self.src = src
+ self.target = target
+ self.directed = directed
+ for k, v in kwargs.items():
+ setattr(self, k, v)
+ def __str__(self):
+ """string representation of this edge"""
+ return f"{self.src} -{'>' if self.directed else ''} {self.target}"
- def add_node(self, node: 'Node') -> None:
+
+ def properties(self) -> Dict:
+ """the edge properties"""
+ return {k: v for k, v in self.__dict__.items()}
+
+
+class Graph(metaclass=ABCMeta):
+ """
+ Base class for graphs.
+ A graph is a collection of Nodes linked by edges
+ """
+
+ def __init__(self) -> None:
+ self._nodes: {}
+
+ def add_node(self, node) -> None:
"""
Add a node to the graph
@@ -39,12 +69,11 @@ class NDGraph:
:type node: :class:`Node`
:return: None
"""
- self.nodes.add(node)
- if node not in self.vertices:
- self.vertices[node] = set()
+ if node not in self._nodes:
+ self._nodes[node] = {}
- def del_node(self, node: 'Node') -> None:
+ def del_node(self, node: Node) -> None:
"""
Remove Node from Graph
@@ -52,33 +81,12 @@ class NDGraph:
:type node: :class:`Node`
:return: None
"""
- self.nodes.remove(node)
- neighbors = self.vertices[node]
- for nbh in neighbors:
- self.vertices[nbh].remove(node)
- del self.vertices[node]
+ for nbh in self.neighbors(node):
+ del self._nodes[nbh][node]
+ del self._nodes[node]
- def add_edge(self, node_1: 'Node', nodes: Sequence['Node']):
- """
- Add vertice between node1 and all nodes in nodes
-
- :param node_1: the reference node
- :type node_1: :class:`Node`
- :param nodes: the nodes connected to node_1
- :type nodes: Sequence of :class:`Node`
- :return: Node
- """
- if node_1 not in self.nodes:
- raise ValueError("the node_1 must be in Graph")
- for n in nodes:
- if n not in self.nodes:
- raise ValueError("node must be add to Graph before creating edge")
- self.vertices[node_1].add(n)
- self.vertices[n].add(node_1)
-
-
- def neighbors(self, node: 'Node') -> Set['Node']:
+ def neighbors(self, node: Node) -> Iterator[Node]:
"""
return the nodes connected to node
@@ -86,174 +94,164 @@ class NDGraph:
:type node: :class:`Node`
:return: a set of :class:`Node`
"""
- return {n for n in self.vertices[node]}
+ for n in self._nodes[node]:
+ yield n
- def get_weight(self, node_1, node_2):
+ def nodes(self) -> Iterator[Node]:
"""
+ Iterates over all nodes belonging to the graph
- :param node_1:
- :param node_2:
- :return:
+ :return: iterator
"""
- return 1
+ for n in self._nodes:
+ yield n
+ def edges(self, node: Node) -> Iterator[Edge]:
+ """
+ Iterates over edge connecting to/from this node
+ :param node:
+ :return:
+ """
+ for e in self._nodes[node].values():
+ yield e
-class Node:
+ def has_edge(self, edge) -> bool:
+ """
+ :param edge:
+ :return: True if it exists an edge connecting edge nodes src and target
+ """
+ return edge.src in self._nodes and edge.target in self._nodes[edge.src]
- _id = itertools.count(0)
+ @abstractmethod
+ def add_edge(self, src: Node, target: Node, **edge_prop) -> None:
+ """
- def __init__(self) -> None:
- self.id: int = next(self._id)
+ :param edge:
+ :return:
+ """
+ pass
- def __hash__(self) -> int:
- # to be usable in set an object must be hashable
- # so we need to implement __hash__
- # which must return an int unique per object
- # here we ad the identifier of the Node
- return self.id
+ @abstractmethod
+ def del_edge(self, src: Node, target: Node) -> None:
+ """
+ remove the edge between nodes src and target
+ :param src:
+ :param target:
+ :return:
+ """
+ pass
- def __str__(self):
- return f"node_{self.id}"
- def __repr__(self):
- return str(self)
+ @staticmethod
+ @abstractmethod
+ def is_directed() -> bool:
+ pass
-class Edge:
+ def order(self):
+ """
+ The Order of a graph is the number of Nodes in the graph
+ :return:
+ """
+ return len(self._nodes)
- def __init__(self, src, target, weight):
- self.src = src
- self.target = target
- self.weight = weight
+ @abstractmethod
+ def size(self):
+ """
+ The size of a graph is the number of edges in the graph.
+ :return:
+ """
+ size = 0
+ for n in self.nodes():
+ size += len(self._nodes[n])
+ return size
-class NDWGraph:
+class UnDirectedGraph(Graph):
"""
- To handle Non Directed Graph
+ To handle UnDirected Graph
A graph is a collection of Nodes linked by edges
- there are basically to way to implement a graph
- - The graph handles Nodes, each nodes know it's neighbors
- - The graph handles nodes and the edges between nodes
- below I implemented the 2nd version.
"""
- def __init__(self, node) -> None:
- self.nodes: Set[Node] = {}
+ def __init__(self) -> None:
+ self._nodes = {}
+ @staticmethod
+ def is_directed() -> bool:
+ return False
- def add_node(self, node: Node) -> None:
+ def size(self) -> int:
"""
- Add a node to the graph
-
- :param node: the node to add
- :type node: :class:`Node`
- :return: None
+ :return: The number of edges of this graph
"""
- if node not in self.nodes:
- self.nodes[node] = {}
+ return super().size() // 2
- def del_node(self, node: Node) -> None:
+ def add_edge(self, node_1, node_2, **prop):
"""
- Remove Node from Graph
+ Add vertex between node1 and all nodes in nodes
- :param node: the node to add
- :type node: :class:`Node`
- :return: None
+ :return: Node
"""
- neighbors = self.nodes[node]
- for nbh in neighbors:
- del self.vertices[nbh][node]
- del self.nodes[node]
+ self.add_node(node_1)
+ self.add_node(node_2)
+ self._nodes[node_1][node_2] = Edge(node_1, node_2, directed=False, **prop)
+ self._nodes[node_2][node_1] = Edge(node_2, node_1, directed=False, **prop)
- def add_edge(self, node_1: Node, node_2: Node, weight: Union[int, float]):
+ def del_edge(self, src: Node, target: Node) -> None:
"""
- Add vertex between node1 and all nodes in nodes
-
- :param node_1: the reference node
- :param node_2: the nodes connected to node_1
- :param weight: the weight of the edge between node_1 and node_2
- :return: Node
+ remove edge between nodes src and target
+ :param src:
+ :param target:
+ :return:
"""
- for n in node_1, node_2:
- if n not in self.nodes:
- raise ValueError(f"node {n.id} not found in Graph. The node must be in Graph.")
- self.nodes[node_1][node_2] = weight
- self.nodes[node_2][node_1] = weight
+ del self._nodes[src][target]
+ del self._nodes[target][src]
- def neighbors(self, node):
- """
- return the nodes connected to node
+class DirectedGraph(Graph):
+ """
+ To handle Directed Graph
+ A graph is a collection of Nodes linked by edges
+ """
- :param node: the reference node
- :type node: :class:`Node`
- :return: a set of :class:`Node`
- """
- return {n for n, w in self.nodes[node].items()}
+ def __init__(self) -> None:
+ self._nodes = {}
- def get_weight(self, node_1, node_2):
- """
+ @staticmethod
+ def is_directed() -> bool:
+ return True
- :param node_1:
- :param node_2:
- :return:
+
+ def size(self):
"""
- return self.nodes[node_1][node_2]
+ :return: The number of edges of this graph
+ """
+ return super().size()
-def DFS(graph: NDGraph, node: Node) -> Iterator[Node]:
- """
- **D**epth **F**irst **S**earch.
- We start the path from the node given as argument,
- This node is labeled as 'visited'
- The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit'
- We pick the last element to visit and visit it
- (The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit').
- on so on until there no nodes to visit anymore.
-
- :param graph:
- :param node:
- :return:
- """
- to_visit = [node]
- visited = set()
- while to_visit:
- n = to_visit.pop(-1)
- visited.add(n)
- new_to_visit = graph.neighbors(n) - visited - set(to_visit)
- to_visit.extend(new_to_visit)
- yield n
+ def add_edge(self, src: Node, target: Node, **prop) -> None:
+ """
+ Add edge from node *src* to node *target*
+
+ :param src: the src node
+ :param target: the target node
+ :param prop: the edge properties
+ :return: None
+ """
+ self.add_node(src)
+ self.add_node(target)
+ self._nodes[src][target] = Edge(src, target, directed=True, **prop)
-def BFS(graph: NDGraph, node: Node) -> Iterator[Node]:
- """
- **B**readth **F**irst **s**earch
- We start the path from the node given as argument,
- This node is labeled as 'visited'
- The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit'
- we pick the **first** element of the nodes to visit and visit it.
- (The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit')
- on so on until there no nodes to visit anymore.
-
- :param graph:
- :param node:
- :return:
- """
- to_visit = [node]
- visited = set()
- parent = None
- while to_visit:
- n = to_visit.pop(0)
- visited.add(n)
- new_to_visit = graph.neighbors(n) - visited - set(to_visit)
- to_visit.extend(new_to_visit)
- if not parent:
- weight = 0
- else:
- weight = graph.get_weight(parent, n)
- parent = n
- yield n, weight
+ def del_edge(self, src: Node, target: Node) -> Node:
+ """
+ Remove edge between nodes src and target
+ :param src:
+ :param target:
+ :return:
+ """
+ del self._nodes[src][target]
diff --git a/Graph/graph/traversing.py b/Graph/graph/traversing.py
new file mode 100644
index 0000000..07f7da4
--- /dev/null
+++ b/Graph/graph/traversing.py
@@ -0,0 +1,67 @@
+from typing import Iterator, List, Union, Tuple
+
+from .helpers import LIFO, FIFO
+from .graph_3 import Node, Edge, Graph
+
+
+def _traversing(to_visit: Union[FIFO, LIFO], graph: Graph, node: Node) -> Iterator[Tuple[Node, List[Edge]]]:
+ """
+ function that traverse the graph starting from node
+
+ :param to_visit:
+ :param graph:
+ :param node:
+ :return: an iterator at each step return the visiting node and the path
+ """
+ to_visit.add(node)
+ visited = set()
+ parent = {}
+ path = []
+ while to_visit:
+ node = to_visit.pop()
+ # it is important to add node in visited right now
+ # and not a the end of the block
+ # otherwise node is not anymore in to_visit and not yet in visited
+ # so when we explore neighbors we add it again in to_visit
+ visited.add(node)
+ for edge in graph.edges(node):
+ if edge.target not in visited and edge.target not in to_visit:
+ parent[edge.target] = edge
+ to_visit.add(edge.target)
+ if node in parent:
+ path.append(parent[node])
+ yield node, path
+
+
+def DFS(graph: Graph, node: Node) -> Iterator[Tuple[Node, List[Edge]]]:
+ """
+ **D**\ epth **F**\ irst **S**\ earch.
+ We start the path from the node given as argument,
+ This node is labeled as 'visited'
+ The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit'
+ We pick the last element to visit and visit it
+ (The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit').
+ on so on until there no nodes to visit anymore.
+
+ :param graph: The graph to traverse
+ :param node: The starting node
+ :return: iterator on nodes
+ """
+ return _traversing(FIFO(), graph, node)
+
+
+def BFS(graph: Graph, node: Node) -> Iterator[Tuple[Node, List[Edge]]]:
+ """
+ **B**\ readth **F**\ irst **s**\ earch
+ We start the path from the node given as argument,
+ This node is labeled as 'visited'
+ The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit'
+ we pick the **first** element of the nodes to visit and visit it.
+ (The neighbors of this node which have not been already 'visited' nor 'to visit' are labelled as 'to visit')
+ on so on until there no nodes to visit anymore.
+
+ :param graph:
+ :param node:
+ :return:
+ """
+ return _traversing(LIFO(), graph, node)
--
GitLab