一个基于 PyQT 框架的地铁线路规划与查询程序

1 题目要求

1.1 背景

随着城市轨道交通的快速发展，地铁已成为现代都市出行的核心方式之一。乘客在出行时，往往需要根据多种因素（如时间、距离、费用）选择最优路径。然而，地铁网络结构复杂，站点间存在多条路径，单一指标无法满足用户多样化需求。因此，开发一个智能的地铁线路规划查询程序，能够为任意两站提供多目标路径推荐，具有重要的实际应用价值。

1.2 目标

开发一个地铁线路规划查询程序，实现以下核心功能：

路径查询：允许用户输入任意起始站和终点站，查询所有可行的路径。
多目标推荐：根据价格、距离和时间三个指标，计算并推荐多种路径选项。
数据集成：构建结构化数据库，支持路径计算。
网状规划：将地铁网络建模为图结构，利用算法解决多目标优化问题。
用户界面：提供友好的图形界面，方便用户输入查询条件和查看结果。

设计与实现

系统架构

为了使项目具备更好的可维护性和可拓展性，我们采取了模块化设计思想，将系统划分为多个独立的模块，包括数据获取模块、路径计算模块和用户界面模块。各个模块之间通过清晰的接口进行交互，便于后续的功能扩展和维护。

有关于项目的模块划分，请参见下图：

modules

数据获取模块

地铁数据来源于高德地图开放的地铁线路 API，每个城市对应唯一的数据链接。例如，西安的数据接口为：

https://map.amap.com/service/subway?_1759306864569&srhdata=6101_drw_xian.json

然而，原始数据为 JSON 格式，结构较为复杂。项目通过parse_metro_info()函数，将原始数据解析为统一的数据结构：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
[
    {
        'line_name': '1号线',
        'is_loop': '0',
        'stations': {
            '136 338': {
                'station_name': '咸阳西站',
                'line_id': '136 338',
                'latitude': '108.123',
                'longitude': '34.567'
            },
            ...
        }
    },
    ...
]

核心算法模块

路径规划核心模块主要负责根据用户输入的起点和终点，结合当前地铁线路数据，计算出最优的地铁出行方案。该模块实现了多种策略的路线规划算法，并确保结果的准确性与实用性。

模块首先通过解析后的地铁数据，将整个地铁网络抽象为图结构：

站点建模：每个地铁站点（Station类）拥有唯一 ID、名称、所属线路及经纬度坐标。
邻接表构建：每个站点与其前后相邻站点通过邻接表连接，形成完整的地铁路网。对于环线，首尾站点相互连接，支持环线路径规划。

为了适应换乘站的情况，我们在Station类中的所属线路属性中，使用列表来存储一个站点可能属于的多个线路信息。每个线路信息由一个StationInLine类表示，包含线路名称、相邻站点等信息。

用户界面模块

本项目的 UI 模块采用了 PyQt5 和 qfluentwidgets 库实现。

ui_design

其他模块

在assets模块中，我们储存了项目所需的静态资产，如背景图片、图标等。在utils模块中，我们实现了一些通用的工具函数，如票价计算、距离计算，便于其他模块调用。

测试与验证

为了确保系统的稳定性和准确性，我们对各个模块进行了充分的测试。测试主要分为单元测试和集成测试两部分。

单元测试主要针对各个函数和类的功能进行验证，确保其在各种输入下都能返回正确的结果。我们使用了 unittest 框架编写了详细的测试用例，并覆盖了常见的边界情况。

集成测试则关注模块之间的协作，确保数据流和控制流的正确性。我们模拟了用户的实际操作场景，验证了从数据获取到路径规划再到结果展示的整个流程。

封装与打包

我们使用 Nuitka 工具对项目进行封装。Nuitka 将 Python 模块翻译成 C 级程序，然后使用 libpython 和自己的静态 C 文件，以 CPython 的方式执行。这种方法相比于pyinstaller 等工具的简单封装，能够生成更小的可执行文件，并且有着更高的运行效率。

在使用过程中，我们发现 Nuitka 工具在处理 PyQt5 的依赖库 SciPy 时，存在一定的兼容性问题，导致部分功能无法正常使用。为了解决这个问题，我们尝试了多种方案，通过手动指定依赖库的方法，最终成功实现了与 Nuitka 的兼容。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
python
-m
nuitka
xianmetro/main.py
--standalone
--enable-plugin=pyqt5
--output-dir=dist
--include-package=xianmetro
--include-package-data=scipy
--include-data-dir=xianmetro/assets=xianmetro/assets
--nofollow-import-to=test
--windows-console-mode=disable
--windows-icon-from-ico=xianmetro/assets/icon.png

拓展与反思

项目的拓展

由于在项目设计时我们使用了模块化的设计思路，使得各个模块之间的耦合度较低，便于后续的功能扩展和维护。而由于我们通过远程数据接口获取地铁线路和站点信息，未来可以方便地接入更多城市的地铁数据，扩展系统的适用范围。

随着城市地铁数据的不断增加，我们需要对数据获取模块进行重构，以支持更多城市的地铁线路和站点信息。我们计划将数据获取模块设计为一个可扩展的插件系统，允许用户根据需要动态加载不同城市的地铁数据。

我们在数据获取模块中引入了城市参数，以便在请求地铁数据时指定城市，通过城市请求相对应的地铁数据，从而实现对不同城市地铁数据的灵活支持。

同时，我们在用户界面模块中引入了城市选择功能，允许用户在不同城市之间切换，从而获取相应城市的地铁线路和站点信息。我们还优化了界面的布局和交互逻辑，以提升用户体验。

项目的反思

在项目的后期开发中，我们发现了一些潜在的问题和改进的空间。如：

在路径规划算法中，我们使用广度优先搜索（BFS）算法进行路径搜索，但在某些复杂情况下，可能会导致搜索效率低下。未来可以考虑引入其他算法，如 A* 搜索算法，以提高路径规划的效率。
由于缺少资料，我们没有在其他城市中支持票价计算功能。
由于对 PyQt5 的不了解，我们在界面设计上遇到了一些困难，也尚未支持线路图示的功能。未来可以考虑进行更多的学习和研究，以提升界面的美观性和易用性。

参见

部分源代码

路径规划算法

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
# xianmetro/core/planner.py
def plan_route(start_station, end_station, strategy):
    """
    规划路线
    :param start_station: 起始站ID
    :param end_station: 目标站ID
    :param strategy: 选择策略，1-最少换乘，2-最少站点，3-最短距离
    :return: 一个包含路线信息的字典
    {
        "route": [{"line": "1号线", "stations": ["ID1", "ID2", ...]}, {"line": "2号线", "stations": ["ID3", "ID4", ...]}],
        "total_stops": 总站点数,
        "total_distance": 总距离（公里）,
        "transfers": 换乘次数
    }
    """
    stations = parse_stations()

    # 构建邻接表，站点ID为节点，边为相邻站
    adj = defaultdict(list)
    line_map = defaultdict(list)  # 站点ID到其所有线路
    for station_id, station_obj in stations.items():
        for st_line in station_obj.line:
            line_map[station_id].append(st_line.line_name)
            if st_line.prev_station_id:
                adj[station_id].append((st_line.prev_station_id, st_line.line_name))
            if st_line.next_station_id:
                adj[station_id].append((st_line.next_station_id, st_line.line_name))

    # 状态：(权重, 当前站ID, 当前线路, 路径列表, 已走距离, 换乘次数, 经过站点数)
    queue = []
    visited = dict()  # (station_id, line_name): 权重，防止回头/环线死循环

    # 初始化起点入队
    for line_name in line_map[start_station]:
        heappush(queue, (0, start_station, line_name, [(start_station, line_name)], 0.0, 0, 1))

    while queue:
        # 根据策略排序
        if strategy == 1:
            # 换乘优先权重：换乘次数、站点数
            queue.sort(key=lambda x: (x[5], x[6]))
        elif strategy == 2:
            # 站点数优先权重：站点数、换乘
            queue.sort(key=lambda x: (x[6], x[5]))
        elif strategy == 3:
            # 距离优先权重：距离、换乘
            queue.sort(key=lambda x: (x[4], x[5]))
        item = queue.pop(0)
        _, curr_id, curr_line, path, curr_dist, curr_transfer, curr_stops = item

        # 到达终点
        if curr_id == end_station:
            # 整理路线分段
            route = []
            temp = []
            last_line = path[0][1]
            for sid, lname in path:
                if lname != last_line:
                    route.append({"line": last_line, "stations": temp})
                    temp = [sid]
                    last_line = lname
                else:
                    temp.append(sid)
            if temp:
                route.append({"line": last_line, "stations": temp})

            transfers = max(0, len(route) - 1)  # 换乘次数为分段数减一

            return {
                "route": route,
                "total_stops": curr_stops,
                "total_distance": round(curr_dist, 5),
                "transfers": transfers
            }

        # 防止回头/死循环，记录最优权重
        state_key = (curr_id, curr_line)
        weight = (curr_transfer, curr_stops, curr_dist)
        if state_key in visited:
            # 如果已访问且当前权重不优则跳过
            if visited[state_key] <= weight:
                continue
        visited[state_key] = weight

        # 扩展邻居
        for neighbor_id, neighbor_line in adj[curr_id]:
            # 换乘判断
            next_transfer = curr_transfer
            if neighbor_line != curr_line:
                next_transfer += 1
            # 距离累加
            lat1, lon1 = stations[curr_id].coords
            lat2, lon2 = stations[neighbor_id].coords
            d = haversine(lat1, lon1, lat2, lon2)
            # 入队
            heappush(queue, (
                0, neighbor_id, neighbor_line,
                path + [(neighbor_id, neighbor_line)],
                curr_dist + d,
                next_transfer,
                curr_stops + 1
            ))
    return None  # 未找到路径

构建站点对象

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# xianmetro/core/load_graph.py
def parse_stations():
    metro_data = load_from_file()
    station_dict = {}  # key: id, value: Station object
    transfer_map = {}

    for line_info in metro_data:
        line_name = line_info['line_name']
        is_loop = line_info.get('is_loop', "0") == "1"
        stations_data = line_info['stations']
        station_ids = list(stations_data.keys())
        n = len(station_ids)

        for idx, station_id in enumerate(station_ids):
            # Get prev/next for current station (loop if is_loop)
            prev_idx = (idx - 1) % n if is_loop else (idx - 1 if idx > 0 else None)
            next_idx = (idx + 1) % n if is_loop else (idx + 1 if idx < n - 1 else None)
            prev_station_id = station_ids[prev_idx] if prev_idx is not None else None
            next_station_id = station_ids[next_idx] if next_idx is not None else None

            info = stations_data[station_id]
            station_name = info['station_name']
            latitude = float(info['latitude'])
            longitude = float(info['longitude'])
            line_id = info['line_id']

            station_in_line = StationInLine(
                station_id=station_id,
                line_id=line_id,
                line_name=line_name,
                prev_station_id=prev_station_id,
                next_station_id=next_station_id
            )

            if station_id not in station_dict:
                # First occurrence, create the Station object
                station_dict[station_id] = Station(
                    name=station_name,
                    id=station_id,
                    line=[station_in_line],
                    coords=(latitude, longitude)
                )
            else:
                # Transfer station: add new line info if not already present
                # Prevent duplicate line_name in line list
                if not any(l.line_name == line_name for l in station_dict[station_id].line):
                    station_dict[station_id].line.append(station_in_line)

    return station_dict

获取信息

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
# xianmetro/fetch/fetch_data.py
import requests
from assets import UPDATE_LINK
import json

def get_metro_info(city = "西安"):
    """
    Fetch metro station information from AMAP API, and return it as a JSON object.
    :return: json object containing metro station information
    """
    api_url = UPDATE_LINK.get(city, UPDATE_LINK["西安"])
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"}
    response = requests.get(api_url,headers=headers)
    return json.loads(response.text)

def parse_metro_info(metro_json):
    """
    Parse metro station information from JSON object.
    :param metro_json: json object containing metro station information
    :return: list of dictionaries containing parsed metro station information
    """
    _metro_info = []
    lines = metro_json['l']
    for line in lines:
        line_name = line['ln']
        is_loop = line['lo']
        stations = {}
        for station in line['st']:
            station_sl = station['sl'].split(',')
            station_id = station['rs']
            station_info = {
                'line': line_name,
                'station_name': station['n'],
                'line_id': station_id,
                'latitude': station_sl[1],
                'longitude': station_sl[0]
            }
            stations[station_id] = station_info
        _metro_info.append({
            'line_name': line_name,
            'is_loop': is_loop,
            'stations': stations
        })
    return _metro_info

主程序

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
# xianmetro/main.py
import sys
from PyQt5.QtWidgets import QApplication
from xianmetro.ui.main_window import MetroPlannerUI
from xianmetro.core import plan_route, parse_stations, id_to_name, name_to_id
from xianmetro.fetch import get_metro_info, parse_metro_info, save_to_file
from xianmetro.utils import calc_price
from qfluentwidgets import MessageBox, InfoBarIcon

def format_route_output_verbose(route, stations):
    """
    格式化路线输出：每行为一个站点，包含上车、换乘和下车提示
    """
    ...

def show_message(window, msg):
    ...

def get_price_text(distance, city):
    ...

def main():
    app = QApplication(sys.argv)
    window = MetroPlannerUI()
    stations = parse_stations()
    # 设置默认城市
    current_city = window.get_city() or "西安"

    def load_city_data(city):
        metro_json = get_metro_info(city)
        metro_info = parse_metro_info(metro_json)
        save_to_file(metro_info)
        return parse_stations()

    stations = load_city_data(current_city)

    def refresh_station_inputs(city):
        # 更新下拉选项
        from xianmetro.fetch import get_id_list, get_station_list
        station_names = get_station_list()
        station_ids = get_id_list()
        start_options = list(dict.fromkeys(station_names + station_ids))
        window.start_input.clear()
        window.end_input.clear()
        window.start_input.addItems(start_options)
        window.end_input.addItems(start_options)

    def update_routes():
        """
        更新并显示三种策略的路线规划结果
        1. 最少换乘
        2. 最少站点
        3. 最短距离
        允许输入站名或ID，优先ID
        结果显示每个站点一行，包含上车、换乘和下车提示
        结果显示总站点数、总距离、换乘次数和票价信息
        处理无效输入和相同起终点的情况
        """
        start_input = window.get_start_station().strip()
        end_input = window.get_end_station().strip()
        ...
        # 允许输入站名或ID，优先ID
        start_id = stations.get(start_input) and start_input
        end_id = stations.get(end_input) and end_input

        if not start_id:
            start_id = name_to_id(stations, start_input)
        if not end_id:
            end_id = name_to_id(stations, end_input)

        if not start_id or not end_id:
            show_message(window, "请输入有效的起点和终点（支持站名或ID）！")
            for idx in range(3):
                window.clear_result_area(idx)
                window.result_info_labels[idx].setText("")
            return
        if start_id == end_id:
            show_message(window, "你搁这原地TP呢？起点和终点不能相同！")
            for idx in range(3):
                window.clear_result_area(idx)
                window.result_info_labels[idx].setText("")
            return
        # 路径规划
        results = []
        for strategy in [1, 2, 3]:
            result = plan_route(start_id, end_id, strategy=strategy)
            results.append(result)

        # 输出各方案
        for idx, result in enumerate(results):
            if result:
                route_lines, icon_list = format_route_output_verbose(result["route"], stations)
                info_text = (
                    f"总站点数: {result['total_stops']}\n"
                    f"总距离: {result['total_distance']} km\n"
                    f"换乘次数: {result['transfers']}\n"
                    f"{get_price_text(result['total_distance'], window.get_city())}"
                )
                window.clear_result_area(idx)
                for item, icon in zip(route_lines, icon_list):
                    window.add_result_item(idx, item, icon)
                window.result_info_labels[idx].setText(info_text)
            else:
                window.clear_result_area(idx)
                window.add_result_item(idx, "未找到方案")
                window.result_info_labels[idx].setText("")

    def on_plan_clicked():
        """
        规划路线按钮点击事件
        """
        update_routes()

    def on_refresh_clicked():
        nonlocal stations
        city = window.get_city() or "西安"
        try:
            stations = load_city_data(city)
            refresh_station_inputs(city)
            show_message(window, f"{city}地铁数据已刷新成功！")
        except Exception as e:
            show_message(window, f"{city}地铁数据刷新失败: {e}")

    def on_city_changed():
        nonlocal stations
        city = window.get_city() or "西安"
        stations = load_city_data(city)
        refresh_station_inputs(city)
        show_message(window, f"已切换至{city}，地铁数据已更新！")

    window.plan_btn.clicked.connect(on_plan_clicked)
    window.refresh_btn.clicked.connect(on_refresh_clicked)
    window.city_input.currentTextChanged.connect(on_city_changed)

    window.show()
    sys.exit(app.exec_())


if __name__ == "__main__":
    main()