{ "cells": [ { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import networkx as nx" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [], "source": [ "def bfs_search(G, max_depth, start_node, target_node):\n", " # 待访问的路径\n", " to_search = [(start_node, 0)]\n", " # 存储所有的历史路径，及此路径的距离\n", " bfs_path = []\n", " # 正确的路径列表，及此路径的距离\n", " bfs_correct_path = []\n", " # 当还有待访问的路径时\n", " while to_search:\n", " # 从待访问的路径中取第一个待访问路径及其路径长度，例如 AC\n", " this_path, this_path_dis = to_search.pop(0)\n", " # 如果待访问的路径达到最大搜索深度，跳出循环\n", " if len(this_path) > max_depth :\n", " break\n", " # 把刚取出的路径存入历史路径中\n", " bfs_path.append((this_path, this_path_dis))\n", " # 如果路径的最后一个节点是目标节点，路径 AC 的最后一个节点是 C\n", " if this_path[-1] == target_node:\n", " # 其为一条正确的路径，将存入正确的路径列表中，\n", " # 并不再继续往其子节点进行探索\n", " bfs_correct_path.append((this_path, this_path_dis))\n", " continue\n", " # 找到路径最后一个节点的相邻节点\n", " for ne in sorted(G[this_path[-1]]):\n", " # 如果相邻节点不在路径中，即不存在回路\n", " if ne not in this_path:\n", " # 则加入到待访问的路径中\n", " to_search.append((this_path + ne,\n", " this_path_dis + G[this_path[-1]][ne][\n", " 'weight']))\n", " return bfs_path, bfs_correct_path" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "# 定义节点列表\n", "node_list = ['A', 'B', 'C', 'D', 'E', 'F', 'G']\n", "\n", "# 定义边及权重列表\n", "weighted_edges_list = [('A', 'B', 8), ('A', 'C', 20),\n", " ('B', 'F', 40), ('B', 'E', 30),\n", " ('B', 'D', 20), ('C', 'D', 10), \n", " ('D', 'G', 10), ('D', 'E', 10),\n", " ('E', 'F', 30), ('F', 'G', 30)]\n", "\n", "# 定义绘图中各个节点的坐标\n", "nodes_pos = {\"A\": (1, 1), \"B\": (3, 3), \"C\": (5, 0), \"D\": (9, 2),\n", " \"E\": (7, 4), \"F\": (6,6),\"G\": (11,5)}\n", "\n", "G = nx.Graph()\n", "G.add_nodes_from(node_list)\n", "G.add_weighted_edges_from(weighted_edges_list)\n", "\n" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [], "source": [ "bfs_path, bfs_correct_path = bfs_search(G, 3, 'A', 'G')" ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [], "source": [ "paths = [e[0] for e in bfs_path]" ] }, { "cell_type": "code", "execution_count": 23, "metadata": {}, "outputs": [], "source": [ "import collections\n", "def get_search_tree_node_position(paths):\n", " \"\"\"得到绘图时各个节点的坐标\n", " \"\"\"\n", " max_depth = 3 \n", " # 得到每条路径的子路径\n", " path_childern = {}\n", " for path in paths:\n", " father = path[:-1]\n", " if father in paths:\n", " if father in path_childern:\n", " path_childern[father].append(path)\n", " else:\n", " path_childern[father] = [path]\n", " # 对每条子路径排序\n", " o_path_childern = collections.OrderedDict(\n", " sorted(path_childern.items()))\n", " # 计算每个树图中每个节点的位置\n", " tree_node_position = {paths[0][0]: (1, 0, 2)}\n", " for path, sub_paths in o_path_childern.items():\n", " y_pos = -1.0 / max_depth * len(path)\n", " dx = tree_node_position[path][2] / len(sub_paths)\n", " sub_paths.sort()\n", " for index, e_s in enumerate(sub_paths):\n", " x_pos = tree_node_position[path][0] - tree_node_position[path][\n", " 2] / 2 + dx / 2 + dx * index\n", " tree_node_position[e_s] = (x_pos, y_pos, dx)\n", " print(tree_node_position)" ] }, { "cell_type": "code", "execution_count": 24, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "{'A': (1, 0, 2), 'ACD': (1.5, -0.6666666666666666, 1.0), 'AC': (1.5, -0.3333333333333333, 1.0), 'ABD': (0.16666666666666666, -0.6666666666666666, 0.3333333333333333), 'AB': (0.5, -0.3333333333333333, 1.0), 'ABF': (0.8333333333333333, -0.6666666666666666, 0.3333333333333333), 'ABE': (0.5, -0.6666666666666666, 0.3333333333333333)}\n" ] } ], "source": [ "get_search_tree_node_position(paths)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import matplotlib.pyplot as plt\n", "import collections\n", "from IPython import display\n", "import networkx as nx\n", "import numpy as np\n", "import time\n", "\n", "\n", "class SearchGraph():\n", " def __init__(self,\n", " node_list, \n", " weighted_edges_list, \n", " start_node,\n", " target_node,\n", " max_depth=1000,\n", " nodes_pos=None,\n", " help_info=None,):\n", " # 图中的节点\n", " self.node_list = node_list\n", " self.weighted_edges_list = weighted_edges_list\n", " self.start_node = start_node\n", " self.target_node = target_node\n", " self.nodes_pos = nodes_pos\n", " self.max_depth = min(max_depth, len(node_list))\n", " self.temp_best_path = None\n", " \n", " self.weighted_edges_dic = {frozenset([e[0],e[1]]):e[2] for e in weighted_edges_list}\n", " self.help_info = help_info\n", " self.path_score={self.start_node:0}\n", " \n", " self.animation_type = 'dfs'\n", " \n", " self.basic_node_color = '#6CB6FF'\n", " self.start_node_color = 'y'\n", " self.target_node_color = 'r'\n", " self.visited_node_color = 'g'\n", " \n", " self.basic_edge_color = 'b'\n", " self.visited_edge_color = 'g'\n", " \n", " self.success_color = 'r'\n", " \n", " self.correct_paths={}\n", " self.show_correct_path = []\n", " self.build_graph()\n", " self.get_search_tree_node_position()\n", " self.bfs_search()\n", " \n", " \n", "\n", " def build_graph(self):\n", " self.G = nx.Graph()\n", " self.G.add_nodes_from(self.node_list)\n", " self.G.add_weighted_edges_from(self.weighted_edges_list)\n", " \n", " def get_search_tree_node_position(self):\n", " \"\"\"得到绘图的点的坐标\n", " \"\"\"\n", " self.dfs_search()\n", " # 得到 dfs 的搜索路径图\n", " paths = self.dfs_path\n", " # 得到每条路径的子路径\n", " path_childern = {}\n", " for path in paths:\n", " father = path[:-1]\n", " if father in paths:\n", " if father in path_childern:\n", " path_childern[father].append(path)\n", " else:\n", " path_childern[father] = [path]\n", " # 对每条子路径排序\n", " o_path_childern = collections.OrderedDict(sorted(path_childern.items()))\n", " # 计算每个树图中每个节点的位置\n", " tree_node_position = {self.start_node:(1, 0, 2)}\n", " for path, sub_paths in o_path_childern.items():\n", " y_pos = -1.0/self.max_depth * len(path)\n", " dx = tree_node_position[path][2]/len(sub_paths) \n", " sub_paths.sort()\n", " for index, e_s in enumerate(sub_paths):\n", " x_pos = tree_node_position[path][0] - tree_node_position[path][2]/2 + dx/2 + dx*index\n", " tree_node_position[e_s]=(x_pos,y_pos, dx)\n", " self.tree_node_position = tree_node_position\n", " \n", " def show_edge_labels(self, ax, pos1, pos2, label):\n", " (x1, y1) = pos1\n", " (x2, y2) = pos2\n", " (x, y) = (x1*0.5 + x2*0.5, y1*0.5 + y2*0.5)\n", "\n", " angle = np.arctan2(y2 - y1, x2 - x1) / (2.0 * np.pi) * 360\n", " if angle > 90:\n", " angle -= 180\n", " if angle < - 90:\n", " angle += 180\n", " xy = np.array((x, y))\n", " trans_angle = ax.transData.transform_angles(np.array((angle,)),\n", " xy.reshape((1, 2)))[0]\n", " bbox = dict(boxstyle='round',\n", " ec=(1.0, 1.0, 1.0),\n", " fc=(1.0, 1.0, 1.0),\n", " )\n", " label = str(label) \n", " ax.text(x, y,\n", " label,\n", " size=16,\n", " color='k',\n", " alpha=1,\n", " horizontalalignment='center',\n", " verticalalignment='center',\n", " rotation=trans_angle,\n", " transform=ax.transData,\n", " bbox=bbox,\n", " zorder=1,\n", " clip_on=True,\n", " )\n", " \n", " def show_search_tree(self, \n", " this_path=None, \n", " show_success_color=False,\n", " best_path=None\n", " ):\n", " \"\"\"展示搜索树\n", " \"\"\"\n", " # 画出树图 \n", " fig, ax = plt.subplots()\n", " fig.set_figwidth(15)\n", " fig.set_figheight(self.max_depth*1.5)\n", " plt.axis('off')\n", " \n", " for path, pos in self.tree_node_position.items():\n", " if path[-1] == self.start_node:\n", " node_color = self.start_node_color\n", " edge_color = self.basic_edge_color\n", " elif this_path and path in this_path:\n", " if show_success_color:\n", " node_color = self.success_color\n", " edge_color = self.success_color\n", " else:\n", " node_color = self.visited_node_color\n", " edge_color = self.visited_edge_color\n", " elif path[-1] == self.target_node:\n", " node_color = self.target_node_color\n", " edge_color = self.basic_edge_color\n", " else:\n", " node_color = self.basic_node_color\n", " edge_color = self.basic_edge_color\n", " ax.scatter(pos[0], pos[1], c=node_color, s=1000,zorder=1)\n", " plt.annotate(\n", " path[-1],\n", " xy=(pos[0], pos[1]),\n", " xytext=(0, 0),\n", " textcoords='offset points',\n", " ha='center',\n", " va='center',\n", " size=15,)\n", " if len(path)>1:\n", " plt.plot([self.tree_node_position[path[:-1]][0],pos[0]], \n", " [self.tree_node_position[path[:-1]][1],pos[1]], \n", " color=edge_color,\n", " zorder=0)\n", " if len(path)>1:\n", " label = self.weighted_edges_dic[frozenset([path[-2],path[-1]])]\n", " if self.animation_type in ['greedy','a_star']:\n", " label = self.help_info_weight*self.help_info[path[-1]] + self.origin_info_weight*label\n", " self.show_edge_labels(ax, self.tree_node_position[path[:-1]][0:2], pos[0:2], label)\n", " display.clear_output(wait=True)\n", " \n", " show_res_text = \"\"\n", " for e_c in self.show_correct_path:\n", " show_res_text += '找到一条路径: %-7s' % e_c + '。距离为:' +str(self.correct_paths[e_c]) + '\\n'\n", " plt.text(0, -1.1, show_res_text, fontsize=18,horizontalalignment='left', verticalalignment='top',)\n", " \n", " if best_path:\n", " top_text = '最终最短路径为: %-7s' % this_path + '。距离为:' +str(self.correct_paths[this_path]) + '\\n'\n", " elif this_path and self.animation_type in ['dfs','bfs']:\n", " top_text = '当前路径: %-7s' % this_path + '。距离为:' +str(self.path_score[this_path]) + '\\n' \n", " if self.temp_best_path:\n", " top_text += '当前最短路径为: %-7s' % self.temp_best_path + '。距离为:' +str(self.correct_paths[self.temp_best_path]) + '\\n'\n", " else:\n", " top_text = ''\n", "\n", " plt.text(0, 0, \n", " top_text, \n", " fontsize=18,\n", " horizontalalignment='left', \n", " verticalalignment='top',)\n", " \n", " if self.animation_type in ['greedy','a_star']:\n", " show_greedy_text = self.generate_greedy_help_text(this_path)\n", " plt.text(0, 0, show_greedy_text, fontsize=18, horizontalalignment='left', verticalalignment='top',)\n", " plt.show()\n", " \n", " def animation_search_tree(self,search_method='dfs', help_info_weight=1, origin_info_weight=1):\n", " \"\"\"动画展示搜索过程\n", " \"\"\"\n", " self.animation_type = search_method\n", " self.show_correct_path = []\n", " self.temp_best_path = None\n", " if search_method == 'bfs':\n", " paths = self.bfs_path\n", " elif search_method == 'dfs':\n", " paths = self.dfs_path\n", " elif search_method == 'greedy':\n", " self.greedy_search()\n", " paths = self.greedy_search_path\n", " elif search_method == 'a_star':\n", " self.a_star_search(help_info_weight=help_info_weight, origin_info_weight=origin_info_weight)\n", " paths = self.greedy_search_path\n", " else:\n", " paths = []\n", " for e_path in paths:\n", " self.show_search_tree(e_path)\n", " if e_path in self.correct_paths:\n", " if not self.temp_best_path:\n", " self.temp_best_path = e_path\n", " elif self.path_score[e_path] < self.path_score[self.temp_best_path]:\n", " self.temp_best_path = e_path\n", " self.show_correct_path.append(e_path)\n", " self.show_search_tree(e_path, True)\n", " if search_method in ['greedy', 'a_star']:\n", " time.sleep(5)\n", " if search_method in ['bfs', 'dfs']:\n", " if self.correct_paths:\n", " best_path = min(self.correct_paths, key=self.correct_paths.get)\n", " self.show_search_tree(best_path, True, True)\n", " \n", " def animation_graph(self, search_method='bfs', help_info_weight=1, origin_info_weight=1):\n", " \n", " \"\"\"\n", " \"\"\"\n", " self.animation_type = search_method\n", " self.show_correct_path = []\n", " if search_method == 'bfs':\n", " paths = self.bfs_path\n", " elif search_method == 'dfs':\n", " paths = self.dfs_path\n", " elif search_method == 'greedy':\n", " self.greedy_search()\n", " paths = self.greedy_search_path\n", " elif search_method == 'a_star':\n", " self.a_star_search(help_info_weight=help_info_weight, origin_info_weight=origin_info_weight)\n", " paths = self.greedy_search_path\n", " else:\n", " paths = []\n", " for e_path in paths:\n", " self.show_graph(e_path)\n", " if e_path in self.correct_paths:\n", " self.show_correct_path.append(e_path)\n", " self.show_graph(e_path, True)\n", " time.sleep(5)\n", " if search_method in ['bfs', 'dfs']:\n", " best_path = min(self.correct_paths, key=self.correct_paths.get)\n", " self.show_graph(best_path, True, True)\n", " \n", " def show_graph(self, this_path='', \n", " show_success_color=False,\n", " best_path=None):\n", " \"\"\"\n", " 绘制图\n", " :return:\n", " \"\"\"\n", " fig, ax = plt.subplots()\n", " fig.set_figwidth(6)\n", " fig.set_figheight(8)\n", " plt.axis('off')\n", "\n", " # 绘制节点与边颜色\n", " visited_edges = []\n", " if not this_path:\n", " this_path = self.start_node\n", " path_node_list = list(this_path)\n", " for i in range(1,len(path_node_list)):\n", " visited_edges.append(frozenset([path_node_list[i],path_node_list[i-1]]))\n", " \n", " # 节点与标识\n", " nlabels = dict(zip(self.node_list, self.node_list))\n", " edge_labels = dict([((u, v,), d['weight']) for u, v, d in self.G.edges(data=True)])\n", " \n", " # 节点颜色变化\n", " val_map = {self.target_node: self.target_node_color}\n", " if path_node_list:\n", " for i in path_node_list:\n", " if show_success_color:\n", " val_map[i] = self.success_color\n", " else:\n", " val_map[i] = self.visited_node_color\n", " val_map[self.start_node] = self.start_node_color \n", " values = [val_map.get(node, self.basic_node_color) for node in self.G.nodes()]\n", "\n", " # 处理边的颜色\n", " edge_colors = []\n", " for edge in self.G.edges():\n", " # 如果边在result_red_edges,分2种情况:\n", " # 如果this_path[0]/this_path[-1] 对应起始点和终点，颜色为绿色，否则颜色为红色\n", " # 如果边不在result_red_edges,则初始化边的颜色为黑色\n", " if frozenset(edge) in visited_edges:\n", " if show_success_color:\n", " edge_colors.append(self.success_color)\n", " else:\n", " edge_colors.append(self.visited_edge_color)\n", " else:\n", " edge_colors.append(self.basic_edge_color)\n", "\n", " # 绘制节点及其标签\n", " nx.draw_networkx_nodes(self.G, self.nodes_pos, node_size=800, node_color=values, width=6.0)\n", " nx.draw_networkx_labels(self.G, self.nodes_pos, nlabels, font_size=20)\n", " # 绘制边及其标签\n", " nx.draw_networkx_edges(self.G, self.nodes_pos, edge_color=edge_colors, width=2.0, alpha=1.0)\n", " nx.draw_networkx_edge_labels(self.G, self.nodes_pos, edge_labels=edge_labels, font_size=18)\n", "\n", " display.clear_output(wait=True)\n", " # show_text = \"\"\n", " # for e_c in self.show_correct_path:\n", " # show_text += '找到一条路径: %-7s' % e_c + '。距离为:' +str(self.correct_paths[e_c]) + '\\n'\n", " # plt.text(0, -2.6, show_text, fontsize=18, horizontalalignment='left', verticalalignment='top', )\n", " \n", "# if best_path:\n", "# top_text = '最佳路径为: %-7s' % this_path + '。 距离为:' +str(self.correct_paths[this_path]) + '\\n'\n", "# elif this_path and self.animation_type in ['dfs','bfs']:\n", "# top_text = '当前路径: %-7s' % this_path + '。 距离为:' +str(self.cal_dis(this_path)) + '\\n'\n", "# else:\n", "# top_text = ''\n", "# plt.text(0, 0, \n", "# top_text, \n", "# fontsize=18,\n", "# horizontalalignment='left', \n", "# verticalalignment='top',)\n", " plt.show()\n", " \n", " def _dfs_helper(self, G, node, father, target_node,level, res, path):\n", " path+=str(node)\n", " if len(path)>1:\n", " self.path_score[path] = self.path_score[path[:-1]] + self.weighted_edges_dic[frozenset([path[-2],path[-1]])]\n", " res.append(path)\n", " # 找到目标，停止搜索\n", " if node==target_node:\n", " return\n", " if level< self.max_depth:\n", " for neighbor in sorted(G[node]):\n", " if str(neighbor) not in path:\n", " self._dfs_helper(G, neighbor, node, target_node, level+1, res, path)\n", " \n", " def dfs_search(self):\n", " dfs_path=[]\n", " this_path=''\n", " if self.start_node:\n", " self._dfs_helper(self.G, self.start_node, None, self.target_node, 0, dfs_path, this_path)\n", " self.dfs_path = dfs_path\n", " for p in dfs_path:\n", " if p[-1]==self.target_node and p not in self.correct_paths:\n", " self.correct_paths[p] = self.cal_dis(p) \n", " \n", " def bfs_search(self):\n", " to_search=[self.start_node]\n", " bfs_path = []\n", " bfs_correct_path = []\n", " depth = 0\n", " while to_search:\n", " this_search = to_search.pop(0)\n", " if len(this_search)>self.max_depth+1 :\n", " break\n", " bfs_path.append(this_search)\n", " if this_search[-1]==self.target_node:\n", " bfs_correct_path.append(this_search)\n", " continue\n", " for ne in sorted(self.G[this_search[-1]]):\n", " if ne not in this_search:\n", " to_search.append(this_search+ne)\n", " self.bfs_path = bfs_path\n", " for p in bfs_path:\n", " if p[-1]==self.target_node and p not in self.correct_paths:\n", " self.correct_paths[p] = self.cal_dis(p)\n", " \n", " def greedy_search(self, help_info_weight=1, origin_info_weight=0):\n", " self.help_info_weight = help_info_weight\n", " self.origin_info_weight = origin_info_weight\n", " search_path = self.start_node\n", " # 存储每一步的可选项及其分数，用来在动态演示时显示出来\n", " search_scores = {}\n", " while len(search_path) <= self.max_depth:\n", " this_node = search_path[-1]\n", " neighbour_nodes = [e_n for e_n in sorted(self.G[this_node]) if e_n not in search_path]\n", " if len(neighbour_nodes) == 0:\n", " search_scores[search_path]={}\n", " break\n", " if self.help_info:\n", " scores = {e_n:help_info_weight*self.help_info[e_n]+origin_info_weight*self.weighted_edges_dic[frozenset([this_node,e_n])] for e_n in neighbour_nodes }\n", " else:\n", " scores = {e_n:self.weighted_edges_dic[frozenset([this_node,e_n])]\n", " for e_n in neighbour_nodes }\n", " search_scores[search_path]=scores\n", " nearest_node = min(scores, key=scores.get)\n", " search_path += nearest_node\n", " if nearest_node == self.target_node:\n", " break\n", " self.greedy_search_path = [search_path[0:index+1] for index in range(len(search_path))]\n", " self.search_scores = search_scores\n", " \n", " def a_star_search(self, help_info_weight=1, origin_info_weight=1):\n", " self.greedy_search(help_info_weight, origin_info_weight)\n", " \n", "\n", " def generate_greedy_help_text(self,path):\n", " if path[-1] == self.target_node:\n", " return '抵达目标节点' + str(self.target_node)\n", " elif path not in self.search_scores:\n", " return '抵达最大搜索深度，未找到目标节点'\n", " \n", " base_text = '当前可选的子节点及其信息值为 \\n'+ \\\n", " str(self.search_scores[path]) + '\\n'\n", " if self.target_node in self.search_scores[path]:\n", " return base_text + '当前可选的子节点包含了目标节点，\\n所以选择目标节点'\n", " elif len(self.search_scores[path]) == 1:\n", " return base_text + '因为只有一个子节点，所以选择此节点'\n", " else:\n", " return base_text + '因为'+ \\\n", " str(min(self.search_scores[path], key=self.search_scores[path].get)) + \\\n", " '的值最小，所以选择此节点'\n", " \n", " def cal_dis(self,path):\n", " dis = 0\n", " if len(path) > 1:\n", " for i in range(len(path)-1):\n", " dis += self.weighted_edges_dic[frozenset([path[i],path[i+1]])]\n", " return dis\n", " " ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.5.2" } }, "nbformat": 4, "nbformat_minor": 2 }