5.4.3 计算最短路径
文件connecting_flights.py定义了类ConnectingFlights，用于计算连接航班的最短路径。该类通过调用 Floyd-Warshall 算法计算所有标准的最短路径，并将结果存储在数据库中。它提供了添加单个或多个航班信息的方法，并能够打印每对起始和目标机场之间的最短路径。通过与数据库的交互，该类能够实现航空运输领域中连接航班最短路径的计算与管理。
from shortest.Database import Database # import time class ConnectingFlights: """用于计算连接航班的类""" def __init__(self, database): """构造函数。接受数据库对象""" self.db = database def add_one_flight(self, orig, dest, **kwargs): """向数据库添加一条航班信息。接受多个权重作为关键字参数""" print(kwargs) self.db.add_flight(orig, dest, **kwargs) self.calc_all() # 计算最短路径 def add_many_flights(self, arr): """接受列表的列表，格式为：[orig, dest, **weight]""" for new_flight in arr: orig = new_flight[0] dest = new_flight[1] self.db.add_flight(orig, dest, **new_flight[2]) self.calc_all() # 计算最短路径 def calc_all(self): """计算所有标准的邻接表""" self.db.clear_adj() print("正在计算最短路径...") # 计算每个标准的最短路径 for cri in Database.Criterion: print("正在计算 {}...".format(cri.name)) # t1 = time.time() self.floyd_warshal(cri) # t2 = time.time() # print("用时 {}s.\n".format(t2-t1)) # 打印计算时间 def floyd_warshal(self, criterion): """使用 Floyd Warshal 算法在数据库中生成最短路径数据""" all_airports = self.db.all_airports # 初始化邻接表 self.make_adj(criterion) for thru in all_airports: # thru = 中间顶点 for orig in all_airports: if orig == thru: continue # 如果中间顶点与起点相同，则跳过中间顶点与起点相同的情况 for dest in all_airports: if dest == thru or dest == orig: # 如果目标或中间顶点与起点相同，则跳过目标点等于中间顶点或起点的情况 continue # 起点到中间顶点，中间顶点到目标 orig_to_thru = self.db.get_adj(criterion, orig, thru) thru_to_dest = self.db.get_adj(criterion, thru, dest) if orig_to_thru is None or thru_to_dest is None: # 如果无法通过中间顶点访问，不做任何操作 continue # 计算通过中间顶点 "thru" 的新权重 new_weight = float(orig_to_thru["weight"]) + float(thru_to_dest["weight"]) last_path = thru_to_dest["thru"] last_flight = thru_to_dest["last_flight"] # 获取现有的邻接表条目 orig_adj = self.db.get_adj(criterion, orig, dest) if orig_adj is None: # 如果之前没有发现路径，则添加。 self.db.add_to_adj(criterion, orig, dest, last_path, new_weight, last_flight) else: # 如果新权重总和较小，则使用中间顶点作为最短路径的一部分 old_weight = float(orig_adj["weight"]) if new_weight < old_weight: self.db.update_adj(criterion, orig, dest, last_path, new_weight, last_flight) def print_floyd_warshal(self, criterion): """打印每对起点和目标机场之间的最短路径""" # 获取所有机场的字符串列表 all_airports = self.db.all_airports for orig in all_airports: for dest in all_airports: print("\n{} 到 {}:".format(orig, dest)) # 获取最短路径信息 result = self.get_shortest_floyd_warshal(criterion, orig, dest) if len(result["path"]) != 0: for o, d, w, f in result["path"]: if f is None: f = "" weight = self.get_str_from_cri(criterion, w) print("{} {}->{}: {}".format(f, o, d, weight)) total = self.get_str_from_cri(criterion, result["total_weight"]) print("总 {}: {}".format(criterion.name, total)) else: print("没有数据") def get_shortest_floyd_warshal(self, criterion, orig, dest): """接受标准、起点和目标，并返回最短路径信息""" output = {} shortest = [] start = orig end = dest total = 0.0 while start != end: # 回溯以找到最短路径 # 获取起点到终点的最短路径 adj = self.db.get_adj(criterion, start, end) if adj is None: # 如果没有找到数据，则终止 break mid = adj["thru"] # 获取最短路径中终点之前的最后一个顶点 weight = float(self.db.get_weight(criterion, mid, end)) airline_id = adj["last_flight"] shortest.append((mid, end, weight, airline_id)) total += weight end = mid # 回溯，使用中间点作为终点 output["path"] = shortest[::-1] # 反转列表 output["total_weight"] = total return output @staticmethod def get_str_from_cri(criterion, target): """返回关联单位的格式化字符串""" if criterion == Database.Criterion.price: return "${}".format('%.2f' % target) elif criterion == Database.Criterion.duration: return "{} 小时 {} 分钟".format(int(target // 60), int(target % 60)) elif criterion == Database.Criterion.distance: return "{} 英里".format(target) def make_adj(self, criterion): """使用航班初始化邻接表""" weight_name = criterion.name for flight in self.db.all_flights: orig = flight["orig"] dest = flight["dest"] try: last_flight_info = (flight["airline"], flight["no"]) except KeyError: # 如果没有信息 last_flight_info = None try: weight = float(flight[weight_name]) find_existing = self.db.get_adj(criterion, orig, dest) if find_existing is None: # 如果不存在该起点目标对的邻接表，则添加 self.db.add_to_adj(criterion, orig, dest, orig, weight, last_flight_info) continue if weight < float(find_existing["weight"]): # 如果存在邻接表但新权重较小，则更新邻接表 self.db.update_adj(criterion, orig, dest, orig, weight, last_flight_info) except KeyError: pass # 如果指定的权重属性不存在，则忽略对上述代码的具体说明如下所示。
- __init__(self, database)：初始化方法，接受一个数据库对象作为参数，并将其存储在类的属性中。
- add_one_flight(self, orig, dest, **kwargs)：添加单个航班信息到数据库中的方法。接受起始机场、目的地机场和多个权重作为关键字参数，并将航班信息添加到数据库中，并计算最短路径。
- add_many_flights(self, arr)：添加多个航班信息到数据库中的方法。接受一个包含多个航班信息的列表，并将其添加到数据库中，并计算最短路径。
- calc_all(self)：计算所有标准的最短路径的方法。清除现有的邻接表数据，然后使用 Floyd-Warshall 算法计算所有标准的最短路径，并将结果存储在数据库中。
- floyd_warshal(self, criterion)：使用 Floyd-Warshall 算法计算指定标准的最短路径的方法。根据指定的标准，计算所有顶点对之间的最短路径，并更新数据库中的邻接表。
- print_floyd_warshal(self, criterion)：打印指定标准的所有起始和目标机场之间的最短路径的方法。对于每对起始和目标机场，打印其最短路径以及总权重。
- get_shortest_floyd_warshal(self, criterion, orig, dest)：获取指定标准下起始和目标机场之间的最短路径的方法。根据指定的标准、起始机场和目标机场，返回最短路径的相关信息。
- get_str_from_cri(criterion, target)：根据指定的标准和目标值返回格式化的字符串的静态方法。根据不同的标准，返回对应的格式化字符串。
- make_adj(self, criterion)：使用航班信息初始化指定标准的邻接表的方法。根据指定的标准，遍历数据库中的航班信息，初始化邻接表，并更新数据库中的邻接表。
5.4.4 Flask Web前端
（1）文件server.py是一个基于Flask的Web应用程序的入口文件，实现了一个基于Flask的Web应用程序。用户可以通过网页界面查询航班的最短路径信息，并将结果展示在网页上。
from flask import Flask, render_template from shortest.webapp.forms import find_form from shortest.Database import Database from shortest.connecting_flights import ConnectingFlights import os def setup_app(): # 初始化Flask应用 app = Flask(__name__) # 生成随机密钥 SECRET_KEY = os.urandom(32) # 设置应用的密钥 app.config['SECRET_KEY'] = SECRET_KEY # 初始化数据库对象 db = Database('localhost', 27017, "connecting_flight") # 初始化连接航班对象 cf = ConnectingFlights(db) @app.route("/", methods=["GET", "POST"]) def index(): # 创建表单实例 form = find_form() dbresult = [] result = [] total = "" if form.orig.data != "None" and form.dest.data != "None" and form.cri.data != "None": print(form.orig.data, form.dest.data, form.cri.data) # 获取选择的路径优先标准 cri = Database.Criterion[form.cri.data] # 获取起点和终点 search_orig = form.orig.data.split()[0].upper() search_dest = form.dest.data.split()[0].upper() # 查询最短路径的航班信息 dbresult = cf.get_shortest_floyd_warshal(cri, search_orig, search_dest) # 获取最短路径上的航班 for flight in dbresult["path"]: orig = flight[0] dest = flight[1] weight = cf.get_str_from_cri(cri, flight[2]) airline, no = flight[3] result.append({"orig": orig, "dest": dest, "weight": weight, "airline": airline, "no": no}) # 获取总权重 total = cf.get_str_from_cri(cri, dbresult["total_weight"]) print(result) # 渲染模板 return render_template("index.html", form=form, result=result, total=total) return app if __name__ == "__main__": # 运行Flask应用 setup_app.run()上述代码的实现流程如下所示。
- 首先，导入了必要的模块和类，包括Flask、render_template函数、表单类find_form、Database类和ConnectingFlights类。
- 然后，定义了一个名为setup_app的函数，该函数用于配置和初始化Flask应用，并设置了应用的密钥。接下来，它创建了Database和ConnectingFlights的实例，并将它们传递给Flask应用。
- 然后，定义了路由"/"，它接受GET和POST请求。在这个路由中，首先创建一个表单实例，然后根据用户的输入查询数据库，获取最短路径的航班信息，并将结果渲染到模板index.html中。最后，它使用run()方法运行Flask应用。
（2）文件forms.py定义了一个名为find_form的Flask表单类，用于在Web应用中接收用户输入的起点、目的地和路径优先标准。在这个表单中包含了三个字段：起点（orig）、目的地（dest）和路径优先标准（cri），以及一个提交按钮（submit）。其中，起点和目的地字段是字符串字段，要求用户输入并且不能为空且长度为3个字符以上；路径优先标准字段是下拉选择字段，提供了从数据库中获取的路径优先标准选项。
from flask_wtf import FlaskForm from wtforms import StringField, SubmitField, SelectField from wtforms.validators import DataRequired, Length from shortest.Database import Database class find_form(FlaskForm): orig = StringField(u'Origin', validators=[DataRequired(), Length(3)]) dest = StringField(u'Destination', validators=[DataRequired(), Length(3)]) options = [] for option in Database.Criterion: name = option.name options.append((name, name.capitalize() )) cri = SelectField(u'By',validators=[DataRequired()], choices=options) submit = SubmitField('Search')（3）文件index.html是一个HTML模板文件，用于渲染Web应用的主页。
<!doctype html> <html lang="en"> <head> <!-- Required meta tags --> <meta charset="utf-8"> <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no"> <!-- Bootstrap CSS --> <link rel="stylesheet" href={{ url_for('static', filename='css/bootstrap.min.css') }}> <link rel="stylesheet" href={{ url_for('static', filename='css/index.css') }}> <body> <div class="jumbotron jumbotron-fluid" style="background-image: url({{ url_for('static', filename='conn.jpg') }}); background-size: cover;"> <div class="container"> <div class="back" style="background-image: url({{ url_for('static', filename='back.png') }})"> <h1 class="display-4">Connecting Flights</h1> <p class="lead">Shortest path through Floyd-Warshall algorithm</p> </div> </div> </div> <div class="row"> <div class="container"> <form method="POST" action=""> <div class="form-row"> {{ form.hidden_tag() }} <div class="col"> {{ form.orig.label(class="form-control-lable") }} {{ form.orig(class="form-control") }} </div> <div class="col"> {{ form.dest.label(class="form-control-lable")}} {{ form.dest(class="form-control") }} </div> <div class="col"> {{ form.cri.label(class="form-control-lable")}} {{ form.cri(class="form-control") }} </select> </div> </div> <br> <div class="form-row" style="top-margin: 1rm"> <div class="col"> {{ form.submit(class="btn btn-outline-info") }} </div> </div> </form> </div> </div> <br> <div class="row"> <div class="container"> <ul class="list-group"> {%if total != ""%} <li class="list-group-item"> {%if result|length != 0%} <h3>Lowest {{total}}</h3> {%else%} <p>No data for this query</p> {%endif%} </li> {%endif%} {%for path in result%} <li class="list-group-item"> <div class="container"> <h2 style="color:darkblue">{{path["orig"]}}-{{path["dest"]}}</h2> <h4 style="color:grey">{{path["airline"]}} {{path["no"]}} </h4> <p>{{path["weight"]}}</p> </div> </li> {%endfor%} </ul> </div> </div> </body> </html>在页面中包含了一个带有背景图的jumbotron，展示了"Connecting Flights"的标题和一个简短的描述。页面下方是一个表单，用于用户输入起点、目的地和路径优先标准，并有一个提交按钮。在表单下方，会展示用户查询结果，包括最短路径的航班信息。如果有结果，会显示最短路径的航班信息，包括起点、目的地、航空公司、航班号和权重信息（例如总飞行距离或耗时）。如图5-5所示。
图5-5 主页查询效果
