简介这是一份基于Python实现的公交换乘查询系统源码面向学习数据结构、算法与GUI开发的Python开发者可用于理解公交路线建模与最优路径求解的完整实现。压缩包共62个文件以py源码为主18个配合pyc编译文件、ui界面文件、png示意图及excel/csv数据文件整体约2.24MB结构清晰便于对照学习。系统覆盖了站点数据读取、地图创建、邻接表构建、Dijkstra最短路径计算、PyQt界面展示等环节代码中附有测试数据与多种算法实现适合作为课程设计或算法实践的参考。目前已有1111人学习下载对于希望掌握图算法与桌面应用整合的开发者有一定参考价值。1. 解压“Python公交换乘系统源码.zip”之前先想清楚这三件事如果你是冲着“毕设/课设能跑就行”来的这个标题大概率能满足你一套纯 Python 写的公交换乘系统源码解压出来通常是一堆.py和.csv文件跑起来后输入起点终点能给你返回几条可换乘的公交路线。但如果只是把它跑起来你就亏了。这个标题背后真正值钱的部分是“公交换乘”这四个字对应的那套算法——站点怎么建模、线路怎么组织、换乘次数怎么算才叫“最少”这些才是在面试和答辩时能讲出东西的地方。我建议你拿到任何一份这类源码先别急着双击运行按下面三步走先看数据文件里有没有站点表和线路表再看算法文件里用的是 BFS 还是 Dijkstra最后才是跑起来看效果。数据决定算法天花板上限算法决定结果能不能看。这个顺序比跑通更重要。2. 数据建模决定算法上限站点、线路与邻接表这样组织才不返工公交换乘系统的核心不是算法多花哨而是数据组织得够不够干净。你想想一个城市动辄几千个站点、几百条线路如果站点和线路的关系理不清后面写任何算法都是在沙地上盖楼。这一章我按自己写这类系统的习惯从数据来源、数据结构到可运行的加载脚本一步步拆给你看。2.1 数据从哪来自制CSV与GTFS数据集的取舍做公交换乘第一个绕不开的问题是数据哪来常见的路子有三条我分别说说它们的适用场景和坑。第一条路是自己造数据。适合课程设计、毕设演示、算法验证这类场景。自己手工定义十几条线路、几十个站点把站点名、线路名、站序写进 CSV 文件里。好处是完全可控你能设计出各种边界情况——环形线路、两条线路共享多个站点、单行线——用来测算法的健壮性。坏处是数据规模小跑出来的结果没有“真实感”答辩时容易被追问“你这能扩展到真实城市吗”。第二条路是去找 GTFSGeneral Transit Feed Specification标准数据。GTFS 是公共交通数据的通用格式很多城市都开放了这个格式的数据集里面包含stops.txt站点表、routes.txt线路表、trips.txt班次表、stop_times.txt到站时间表。好处是数据真实、字段规范带有经纬度和精确到秒的时间可以直接用来做带时间的最优路径搜索。坏处是数据量大、关联关系复杂新手第一次接触容易被trip_id、service_id这些 ID 之间的关系绕晕。第三条路是爬虫抓取地图平台的公交数据这个我不太推荐作为主数据源。一方面有使用条款上的风险另一方面实时数据里有很多字段是给前端渲染用的清洗成本高。我一般建议学习阶段用第一条路把算法跑通展示阶段用第二条路的公开数据做扩展爬虫数据只用来补少量缺失站点。2.2 核心数据结构线路表、站点表与站点-线路倒排表无论数据来自哪里落到程序里都需要抽象成几个基础的数据结构。公交换乘系统里最核心的是三样东西站点表、线路表、站点与线路的倒排关系。我先说结论一个能支撑换乘算法的数据组织方式长这样。站点表记录每个站点的 ID、名称有经纬度的话也一并存了。这里要注意一个坑同一个物理位置可能有两个名字比如“市政府西门”和“市政府西”其实是一个站需要在数据里提前做别名归并否则换乘算法会漏掉本该成立的换乘关系。线路表这里说的不是“1路车”这种抽象概念而是具体的运行方向。公交车往返线路往往不完全对称——去程有五个站回程可能有六个因为路口的单行线、立交桥的原因绕了一下。所以一条“公交线路”对应的至少是两条“运行线路”分别存。这是新手最容易搞错的地方很多人直接把往返方向合并成一条线路导致算法给出的路径在现实中根本坐不回去。站点-线路倒排表就是给定一个站点快速查出有哪些线路经过它。这个结构在换乘算法里是查询热点起点站能坐上哪几辆车终点站能被哪几辆车覆盖都是靠它回答。用 Python 的defaultdict(set)实现最省事。三类数据的关系可以用这张表概括数据结构典型字段回答的问题stops.csvstop_id, stop_name, lng, lat这个站点的唯一标识和物理位置routes.csvroute_id, route_name, direction, seq, stop_id某条线路的某个方向依次经过哪些站stop_to_routesstop_id - {route_id, ...}从一个站点可以坐上哪些线路2.3 加载与建图一个直接能跑的CSV加载脚本讲完抽象模型落实到代码。我平时会先把 CSV 加载写成一个独立模块因为后面所有算法都要依赖这份数据。下面这个脚本只用了 Python 标准库不需要装第三方依赖解压源码后新建一个loader.py就能跑。# loader.py import csv from collections import defaultdict, namedtuple # 用 namedtuple 定义站点和线路的数据骨架后续访问字段更直观 Stop namedtuple(Stop, [stop_id, stop_name, lng, lat]) RouteStop namedtuple(RouteStop, [route_id, route_name, direction, seq, stop_id]) def read_stops(stops_pathdata/stops.csv): stops {} with open(stops_path, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: stop Stop( stop_idrow[stop_id], stop_namerow[stop_name], lngfloat(row.get(lng) or 0), latfloat(row.get(lat) or 0), ) stops[stop.stop_id] stop return stops def read_routes(routes_pathdata/routes.csv): 按线路方向组织成有序站点列表同时构建站点-线路倒排表 # 原始行按 (线路, 方向, 站序) 排序 rows [] with open(routes_path, encodingutf-8) as f: reader csv.DictReader(f) for row in reader: rows.append({ route_id: row[route_id], route_name: row[route_name], direction: row[direction], seq: int(row[seq]), stop_id: row[stop_id], }) # route_stops: (route_id, direction) - [stop_id, ...] route_stops defaultdict(list) # stop_to_routes: stop_id - set(route_id)注意把方向也带进去区分上下行 stop_to_routes defaultdict(set) rows.sort(keylambda x: (x[route_id], x[direction], x[seq])) for r in rows: key (r[route_id], r[direction]) route_stops[key].append(r[stop_id]) stop_to_routes[r[stop_id]].add(r[route_id]) return route_stops, stop_to_routes def build_route_graph(route_stops): 构建线路-线路换乘图节点是线路边表示两条线路存在共站 route_graph defaultdict(set) # 先建 站点 - 经过该站的所有 (线路,方向) stop_to_route_keys defaultdict(set) for key in route_stops: for sid in route_stops[key]: stop_to_route_keys[sid].add(key) for routes in stop_to_route_keys.values(): routes list(routes) for i in range(len(routes)): for j in range(i 1, len(routes)): route_graph[routes[i]].add(routes[j]) route_graph[routes[j]].add(routes[i]) return route_graph if __name__ __main__: stops read_stops() route_stops, stop_to_routes read_routes() route_graph build_route_graph(route_stops) print(f站点数: {len(stops)}) print(f线路方向数: {len(route_stops)}) print(f换乘边数: {sum(len(v) for v in route_graph.values()) // 2})这段代码里有几个参数和设计值得展开说。read_routes里我把route_id和direction拼成一个复合 key 存线路站点序列这是刻意的——往返方向必须分开建否则“绕一圈回到起点”这种环形走向会被当成一条直线处理算法算出的最短路径在现实中可能不存在。build_route_graph里有一行关键的判断逻辑stop_to_route_keys这个倒排索引先统计每个站点覆盖了哪些线路然后对同一个站点的线路集合做两两连线这就是“换乘关系”的图模型。两条线路只要有任意一个共享站点就认为可以换乘这是最小换乘算法的基础。换乘边数除以 2 是因为上面是双向加的打印时去重。有个细节容易被忽略csv.DictReader默认把第一行当列名所以 CSV 文件里必须有表头且表头要严格叫stop_id、stop_name、route_id、direction、seq这些名字。如果你拿到的源码里字段名不一样一定要先改这里不然加载出来全是空数据后面跑算法会报一堆KeyError。3. 换乘算法是灵魂BFS 与 Dijkstra 的选型和实现数据建好了接下来是重头戏——换乘算法。这里我先泼一盆冷水很多号称“公交换乘系统”的源码算法其实就是在站点图上跑了一遍最短路径根本不区分“坐同一辆车经过多个站”和“换乘到另一辆车”这两种完全不同的代价。这样算出来的路径换乘次数往往会吓你一跳。真正要解决问题得先搞清楚到底在什么图上搜索。3.1 用什么图搜站点图与线路图两种建模的区别先说站点图。把每个站点当节点相邻两个站点之间有边边权是运行时间或者站数然后跑 Dijkstra。这种建模方式最直观但对公交场景有一个致命问题它计算的是“经过多少站”不是“换乘多少次”。举个例子从 A 站到 B 站坐 1 路车 5 站直达和坐 2 路车 2 站后换乘 3 路 3 站到达站点图上 235 站和 5 站相同算法可能随机给你推荐换乘的方案但现实中没人愿意 5 站路换两次车。解决思路是引入换乘惩罚系数但惩罚系数设多少又成了新的玄学问题。所以做公交换乘我一般推荐用线路图节点是“线路方向”这个组合边连接两条“存在共站”的线路。在这个图上从一个节点跳到另一个节点代表一次换乘路径长度直接对应换乘次数语义非常干净。配合换乘时间惩罚就能做最小换乘或最优时间路径。3.2 最小换乘的BFS在“线路集合”上做广度优先最小换乘次数本质上是一个无权图上的最短路径问题用 BFS 就能解决不需要上 Dijkstra。搜索的起点不是某个站点而是一个“线路集合”——起点站能坐上的所有线路终点同理是终点站能到达的所有线路。BFS 每次向外扩展一层意味着“多换乘一次”。第一层是起点直达线路第二层是换乘一次能到的线路依此类推。代码实现我放在下面逻辑很直接。# min_transfer.py from collections import deque, defaultdict def min_transfer_bfs(start_stop_id, end_stop_id, stop_to_routes, route_graph, route_stops): 最小换乘搜索。 start_stop_id / end_stop_id: 起终点站点ID stop_to_routes: 站点 - 线路集合 route_graph: 线路 - 可换乘线路集合 route_stops: (线路,方向) - 站点列表 返回 (换乘次数, 线路路径, 换乘站点链) 或 None # 起终点能坐的线路集合 start_routes set() end_routes set() for rid in stop_to_routes[start_stop_id]: start_routes.add(rid) for rid in stop_to_routes[end_stop_id]: end_routes.add(rid) # 起点线路与终点线路有交集 直达 direct start_routes end_routes if direct: rid direct.pop() return 0, [rid], [start_stop_id, end_stop_id] # BFS准备 queue deque() prev {} # 记录当前线路是从哪条线路、在哪个站换乘过来的 visited set() for rid in start_routes: visited.add(rid) prev[rid] None queue.append(rid) found None while queue and found is None: current queue.popleft() for nxt in route_graph[current]: if nxt in visited: continue visited.add(nxt) # 找到换乘站两条线路的共同站点 transfer_stop find_transfer_stop(current, nxt, route_stops) prev[nxt] (current, transfer_stop) if nxt in end_routes: found nxt break queue.append(nxt) if found is None: return None # 从终点线路反向回溯还原线路路径和换乘站点 route_path [] transfer_stops [] cur found while cur is not None: route_path.append(cur) if prev[cur] is not None: transfer_stops.append(prev[cur][1]) cur prev[cur][0] if prev[cur] is not None else None route_path.reverse() transfer_stops.reverse() # 换乘次数 经过的线路数 - 1 return len(route_path) - 1, route_path, [start_stop_id] transfer_stops [end_stop_id] def find_transfer_stop(route_a, route_b, route_stops): 返回两条线路的第一个共同站点ID用于换乘站点展示 stops_a set(route_stops[route_a]) stops_b set(route_stops[route_b]) common stops_a stops_b return next(iter(common), None)这段代码里最容易写错的是prev的更新时机。我踩过的坑是当nxt已经等于终点线路时不能先入队再统一处理否则要额外加一层判断“从队列里取出来的这条线路是不是终点线路”逻辑绕且容易漏。上面代码的处理方式是入队前先判断是否命中终点命中就跳出循环不命中才入队这样prev里的信息是完整的。另一个值得注意的参数是find_transfer_stop返回“第一个”共同站点。这里有个潜在问题两条线路可能有多个共同站点默认取第一个但现实中第一个共同站点可能离你的目标方向更远。更精确的做法是结合换乘站点在两条线路上的站序位置选择让总站数最少的那一个但作为最小换乘的基础版取第一个完全够用。BFS 的时间复杂度是 O(VE)V 是线路数E 是换乘关系数。一个城市几百条线路这个规模下 BFS 是微秒级的完全不需要优化。如果你发现跑得很慢先检查是不是visited漏标了导致线路图里的环在反复入队。3.3 最优路径的Dijkstra边权设计与换乘惩罚参数最小换乘只管次数少但现实中“换乘1次但绕路3倍”的方案可能还不如“换乘2次但总时长更短”。这时候需要用 Dijkstra 做加权最短路径。关键在于边权怎么设计这是这类系统最核心的调参点。线路图节点之间的边代价由两部分组成车上运行时间 换乘惩罚。车上运行时间不能直接从 CSV 里读一般用“站数 × 平均单站时间”估算比如市区平均每站 2.5 分钟。换乘惩罚包含步行到站台、等下一班车的时间还有你拖着购物袋走路的体力成本。这个值设多少直接决定了算法给出的路径风格。# optimal_route.py import heapq from collections import defaultdict AVG_STOP_TIME 2.5 # 每站平均行驶分钟数 TRANSFER_PENALTY 15 # 换乘惩罚分钟数关键参数 def dijkstra_route(start_stop_id, end_stop_id, stop_to_routes, route_graph, route_stops): 综合考虑运行时间换乘惩罚的最优路径。 状态定义为 (累计时间, 当前线路ID, 当前所在站点ID) start_routes list(stop_to_routes[start_stop_id]) end_routes set(stop_to_routes[end_stop_id]) # dist[(route_id, stop_id)] 到达该线路且身处该站的最短时间 dist {} heap [] # 初始化从起点站直接坐上某条线路耗时为0等车时间已含在换乘惩罚中 for rid in start_routes: if start_stop_id not in route_stops[rid]: continue key (rid, start_stop_id) dist[key] 0 heapq.heappush(heap, (0, rid, start_stop_id)) prev {} # 记录状态转移信息用于回溯路径 best_total float(inf) best_key None while heap: cost, route, stop heapq.heappop(heap) if cost ! dist.get((route, stop), float(inf)): continue # 惰性删除过期堆项 if stop in end_routes and route in end_routes: if cost best_total: best_total cost best_key (route, stop) break # Dijkstra 首次弹出终点即为最优可提前结束 stops_list route_stops[route] try: idx stops_list.index(stop) except ValueError: continue # 1. 顺着当前线路继续坐 if idx 1 len(stops_list): next_stop stops_list[idx 1] new_cost cost AVG_STOP_TIME key (route, next_stop) if new_cost dist.get(key, float(inf)): dist[key] new_cost prev[key] (route, stop, stay) heapq.heappush(heap, (new_cost, route, next_stop)) # 2. 在当前站换乘到其他线路 for next_route in route_graph[route]: if next_route route: continue if stop in route_stops[next_route]: new_cost cost TRANSFER_PENALTY key (next_route, stop) if new_cost dist.get(key, float(inf)): dist[key] new_cost prev[key] (route, stop, transfer) heapq.heappush(heap, (new_cost, next_route, stop)) if best_key is None: return None return reconstruct_path(best_key, prev, start_stop_id, end_stop_id)这个实现里我把状态定义成了“(线路ID, 站点ID)”二元组这是公交场景和普通图最短路径最本质的区别同一条线路上跨越多站不需要经过中转节点直接顺着线路继续坐每次只加一个平均站时。换乘动作通过“枚举当前站能换到的线路”来实现每次加一个换乘惩罚。TRANSFER_PENALTY这个参数值得单独聊。我一般取 10 到 20 分钟之间太低比如 3 分钟和没有惩罚没区别算法会为省一个站的时间让你换车太高比如 30 分钟又会让算法宁可绕路半小时也不愿意换个车。一个可行的校准方式把惩罚设为 0跑几个起终点组合记录结果里最多的换乘次数是多少然后逐步加大惩罚直到结果里不再出现“换乘次数明显不合理”的路径。这个值本质上是个经验参数不存在绝对正确的答案但 15 分钟对大多数城市公交场景是个不错的起点。4. 避坑公交换乘系统里最常见的五个翻车现场这类系统看着简单真跑起来能翻车的地方不少。我整理了自己在这类项目里遇到的最常见的五个坑按“现象 → 原因 → 解决”的结构写你在跑通源码后如果结果不对优先对照这几条排查。4.1 建图阶段的三类数据坑坑一程序跑很久不结束像死循环一样。现象BFS 或 Dijkstra 在部分起终点组合下运行时间异常甚至卡住。原因线路图里有环而visited没有在线路维度做标记。最常见的情况是两条线路有多个共同站点算法通过不同换乘点反复进入同一条线路每次都当作“新状态”处理导致搜索空间爆炸。解决BFS 里把visited定义成“线路ID集合”入队时立刻标记Dijkstra 里则为每个 (线路, 站点) 组合保存最小代价堆里弹出过期状态直接跳过。上面给出的代码已经处理了这两点如果你拿到的源码没有优先补上。坑二明明两条线路在同一个站算法却说不能换乘。现象两个站点的名字不同但它们实际上是同一个物理位置倒排表里两条线路没有公共站点 ID换乘边没建起来。原因数据源本身存在同站异名问题。比如“市政府”和“市政府西门”是同一个站点的两个名字但 CSV 里给了两个不同的stop_id。解决在数据预处理阶段维护一张站点别名表把同站异名的 ID 映射到一起或者在build_route_graph时以“站点名称”而不是“站点 ID”作为连通关系的主键共享同一名称的站点视为同一节点。课程设计级别的数据量手写一个别名映射表就够。坑三算法给出的路线坐车根本到不了。现象从起始站坐上 5 路车但算法要求 5 路车走完 10 站后在某一站换乘而实际上 5 路车根本没经过那站。原因路线数据里的往返方向混在一起了。5 路车去程经过“图书馆”站回程因为单行道绕行不经过“图书馆”但数据里只存一条“5路车”记录把回程的站点序列也塞进去了导致算法认为 5 路车既经过去程的站也经过回程的站。解决严格遵循“一条 CSV 记录 一个线路的一个方向 站序”的原则往返方向分别赋不同的route_id比如R5_out和R5_back。加载脚本里用 (route_id, direction) 作为完整线路标识这样换乘判断就不会跨方向串站点了。4.2 算法阶段的两类逻辑坑坑四换乘惩罚参数设太小路径结果反直觉。现象从 A 到 B 明明有直达车算法却推荐换乘 3 次理由是这样能省 4 分钟。原因换乘惩罚设成了 0 或者接近 0。算法视角里换乘是零成本的它只看运行时间那当然会把路径拆成若干段“快车”拼接。解决把TRANSFER_PENALTY调大我给的 15 分钟就是经验起点。还有一个更细的改进换乘惩罚不应该是一个固定常数而应该加上“起点线路的等车时间”。如果换乘点正好有一辆车马上到站可以少罚一点如果刚错过一班可能要干等 15 分钟应该多罚一点。基础版用定值就可以进阶版再考虑动态惩罚。坑五BFS 返回“无法到达”但实际上换乘 2 次能到。现象起点和终点之间明明存在可行路径最小换乘算法却返回 None。原因BFS 只搜索了“从起点线路集合出发逐层扩展”这一个方向如果终点线路集合在搜索过程中始终没有与任何已访问线路产生交集就会认为无解。问题通常出在数据上——换乘边构建时漏了某些共站关系比如两个站点名字不同但物理位置相同上面坑二导致的连锁反应。解决先跑一遍build_route_graph后的换乘边统计用 2.3 节里的打印逻辑检查目标线路的route_graph[rid]是否包含预期邻居。更稳妥的做法是给 BFS 加一个“迭代加深”的壳先搜最大换乘 1 次没解再放宽到 2 次逐步增加直到搜到解或达到上限。这样即使图构建有遗漏也能通过放宽约束找到“次优可达路径”而不是直接给一个无解。5. 从能跑到好用结果验证、性能优化与路径可视化最后一章我想把“跑通”变成“放心用”。这里说的放心指的是你拿这个系统去答辩、去演示、去给别人介绍时不会突然翻车而且你心里清楚它好在哪里、差在哪里。先讲验证。公交换乘系统最大的隐藏风险是“路径错误”而且往往是起终点离得越远越容易错。我的习惯是写一个批量回测脚本随机采样 100 组起终点把算法输出的换乘次数和总时长打印出来肉眼扫一遍异常值。具体判据我一般用两条换乘次数超过 4 次的路径要重点看因为大多数城市公交 90% 的情况换乘 2 次内能到总时长超过“直线距离 × 5”分钟的也要重点看说明路径可能绕了大远路。这里直线距离可以从站点经纬度算如果没有经纬度可以按“站数 × 3 分钟”做估算基准。然后是性能。很多源码在数据量小的时候跑得飞快一换成真实城市数据就慢得没法看。这个系统数据规模的瓶颈通常不在 BFS——那部分几百条线路也就毫秒级——而在建图阶段。build_route_graph里的两层循环是 O(站点数 × 线路数²) 的复杂度。数据量一大这层初始化可能吃掉几秒。优化思路有两条一是用空间换时间把“站点→线路集合”的倒排表提前算好并序列化成 pickle 缓存每次启动直接加载二是换乘边只对“同一个站点”的线路组合去重用itertools.combinations替代手工双循环代码更简洁。最后是可视化。如果你的源码里只有命令行输出我建议加一个最轻量级的可视化用matplotlib把换乘路径画出来。不需要底图把站点坐标画成散点把线路依次用折线连起来换乘站用不同颜色标出来就行。这个做法有两层价值第一调试时能直观看到路径是否绕路第二答辩演示的时候放一张路径图比空口说“跑了几个站点”有说服力得多。代码量不大核心就是把route_stops里的站点 ID 映射成坐标然后plt.plot连续画几条折线。如果没有经纬度坐标可以按线路的站序手写一个简单的网格坐标映射效果一样。写到这里我想起一个习惯每次改完数据哪怕只是改了 CSV 里一个站点的名字我都会重新跑一遍第 5 章的批量回测脚本。这个习惯救过我很多次——有一次只是把“人民广场”改成了“人民广场站”没注意别名表里还有旧的映射结果半个城区的换乘路径全部断掉那个问题靠肉眼看根本发现不了。系统不是跑通就没事了数据和算法任何一边动了都要花五分钟做一次回归验证。希望这一章的验证思路和参数起点能让你的公交换乘系统少走点弯路帮你把“能跑的源码”真正变成“能讲清楚的作品”。本文还有配套的精品资源点击获取