各位读者朋友大家好。今天这篇博文想从一个有点“中二”的项目标题说起“列车即将到站下一站——至冬”我第一次看到这句话时第一反应是某个游戏剧情预告又或者是某个漫展的主题活动。但冷静下来一想这句话本身其实是一个非常典型的轨道交通语音播报场景列车即将进站系统需要动态计算出“当前位置”和“下一站”然后生成类似“列车即将到站下一站——XX”的播报信息。如果把这个场景落地成一套可运行的代码就是一个很有意思的实战项目。这篇文章我打算就用“至冬”这个虚构站名作为演示数据带大家从零实现一套列车到站播报系统。系统会包含站台数据建模与线路管理到站信息计算与播报文案生成中文语音合成播报Web API 查询接口定时任务自动播报常见问题排查与工程化建议。无论你是 Python 初学者还是工作中需要做语音提醒、定时播报、信息提示类小系统的开发者都可以从这篇文章里找到可复用的思路。代码我会尽量给全并且保证在本地可以跑起来。接下来我们正式开工。1. 背景与核心概念1.1 什么是列车到站播报系统列车到站播报系统简单来说就是一套“根据列车实时位置计算下一站信息并生成语音/文字提示”的程序。它在现实生活中非常常见地铁车厢内的“列车即将到站下一站人民广场”公交车的“下一站XX站请下车的乘客做好准备”高铁、机场摆渡车、景区观光车上的到站提醒企业内部班车、校园摆渡车的到站通知。这类系统虽然业务规模不大但背后涉及的逻辑很完整线路数据管理、位置状态维护、文案生成、语音合成、信息发布。作为一个练手项目它能很好地锻炼数据建模、模块拆分和工程化能力。1.2 系统核心组成从软件架构角度看一个最小可用的到站播报系统包含以下部分模块职责线路数据模块定义线路名称、站点列表、站点顺序、换乘信息位置状态模块记录列车当前在哪个站或刚从哪个站出发文案生成模块根据当前位置生成“下一站”播报文本语音合成模块将文本转为语音并播放定时任务模块按照时间间隔自动触发播报对外接口模块提供查询下一站/触发播报的 API1.3 为什么自己做一套而不是直接用现成的有人可能会问地铁里已经有现成的播报系统了自己写一套有什么意义其实意义在于这种“信息播报类”系统在很多非公共交通场景下是没有现成产品的。比如公司班车到站提醒学校图书馆闭馆提醒仓库叉车调度提示展会摆渡车播报个人开发的小型语音助手。在这些场景里你需要的往往是“一套轻量、可配置、能快速接入业务数据”的播报服务而不是地铁级的信号控制系统。我们用 Python 写一个简化版正好覆盖了这类需求的核心逻辑。2. 环境准备与版本说明2.1 运行环境本文示例代码使用 Python 3 开发主要依赖以下库库名用途pyttsx3离线文字转语音支持 Windows/macOS/Linuxedge-tts微软 Edge 在线语音合成音质更好Flask提供 Web API 查询接口APScheduler定时触发播报任务PyYAML读取 YAML 配置文件需要说明的是语音合成库的版本和音色在不同操作系统上差异较大本文以“能跑通”为目标版本不需要刻意追求最新。如果你本地环境没有这些库可以直接用 pip 安装pip install pyttsx3 edge-tts flask apscheduler pyyaml如果你的网络环境无法安装edge-tts系统会自动降级为pyttsx3控制台输出模式不影响核心流程。2.2 项目目录结构我们先规划一个清晰的项目结构方便后续扩展train-announcer/ ├── app/ │ ├── __init__.py │ ├── models.py # 数据模型站点、线路 │ ├── announcer.py # 播报文案生成与语音播报 │ ├── scheduler.py # 定时任务 │ ├── web.py # Flask Web API │ └── config.py # 配置加载 ├── config/ │ └── line_config.yaml # 线路与站点配置 ├── main.py # 程序入口 ├── requirements.txt └── README.md这个结构不算复杂但已经把“数据”“业务”“服务”三层分开了。3. 核心模块设计与实现原理3.1 站点与线路的数据建模到站播报系统里最重要的数据是“线路”和“站点”。一个合理的建模方式是每一条线路包含若干站点站点按顺序排列每个站点可能有英文名、拼音名、是否换乘站等属性列车当前状态记录的是“已经离开哪个站”或者“当前正驶向哪个站”。下面我们先用 Python 的dataclass定义站点和线路模型。代码路径app/models.pyfrom dataclasses import dataclass, field from typing import List, Optional dataclass class Station: 站点数据模型 id: str # 站点唯一标识 name: str # 站点中文名 pinyin: str # 站点拼音 english_name: str # 站点英文名 is_transfer: bool False # 是否换乘站 arrival_time: str # 预计到达时间例如 10:30 dataclass class Line: 线路数据模型 name: str # 线路名 stations: List[Station] field(default_factorylist) def get_station_by_id(self, station_id: str) - Optional[Station]: 根据站点 ID 查找站点对象 for station in self.stations: if station.id station_id: return station return None def get_next_station(self, current_station_id: str) - Optional[Station]: 返回当前站点的下一个站点。 如果当前站点是终点站返回 None。 for index, station in enumerate(self.stations): if station.id current_station_id: if index 1 len(self.stations): return self.stations[index 1] return None return None这里需要注意get_next_station是关键方法。它决定了播报系统能不能准确找到“下一站”。如果站点 ID 不存在我们返回None上层调用时要做空值判断避免程序崩溃。3.2 播报文案生成逻辑有了线路模型之后我们可以这样设计播报逻辑如果当前站是普通站播放“列车即将到站下一站——XX”如果当前站是终点站播放“列车即将到达终点站XX”如果下一站是换乘站可以追加一句“可换乘 XX 号线”。为了避免把文案写死在代码里我建议用一个独立函数来生成播报文本后续上线不同场景时可以直接替换。代码路径app/announcer.pyfrom app.models import Line, Station def build_announcement(line: Line, current_station_id: str) - str: 根据当前线路和当前站点 ID生成播报文案。 返回字符串如果找不到当前站点则返回空字符串。 current_station line.get_station_by_id(current_station_id) if current_station is None: return next_station line.get_next_station(current_station_id) if next_station is None: # 当前站是终点站 return f列车即将到达终点站{current_station.name}感谢您的乘坐 # 常规到站播报 text f列车即将到站下一站——{next_station.name} if next_station.is_transfer: text 可换乘 、.join(next_station.transfer_lines) text return text这里为了演示换乘信息我在 Station 模型里加了一个transfer_lines字段。你可以在实际项目中补充这个字段这里不额外展开。3.3 语音合成方案对比语音播报是整个系统的灵魂。在 Python 里常用的方案有三种方案离线/在线音质跨平台适用场景pyttsx3离线一般好本地快速测试、离线环境edge-tts在线较好好对音质有要求可以联网商用 TTS SDK在线最好一般生产环境、携带品牌音色我的推荐是本地开发用pyttsx3跑通流程音质敏感场景用edge-tts。edge-tts使用微软 Edge 的在线语音服务音色自然支持中文代码也不复杂。下面封装一个支持“自动降级”的播报类。代码路径app/announcer.pyimport asyncio import subprocess import sys class VoiceAnnouncer: 语音播报器支持 pyttsx3 和 edge-tts 自动切换 def __init__(self, engineauto): self.engine engine self._tts_engine None def _init_pyttsx3(self): try: import pyttsx3 engine pyttsx3.init() # 尝试设置中文语音不同平台 voice id 不同 voices engine.getProperty(voices) for voice in voices: if chinese in voice.name.lower() or zh in voice.id.lower(): engine.setProperty(voice, voice.id) break return engine except Exception as e: print(f[语音引擎] pyttsx3 初始化失败: {e}) return None def _play_pyttsx3(self, text: str): if self._tts_engine is None: self._tts_engine self._init_pyttsx3() if self._tts_engine is None: # 最终降级方案控制台输出 self._console_print(text) return self._tts_engine.say(text) self._tts_engine.runAndWait() async def _play_edge(self, text: str): try: import edge_tts communicate edge_tts.Communicate(text, zh-CN-XiaoxiaoNeural) await communicate.save(announcement.mp3) # 播放生成的 mp3 if sys.platform.startswith(win): subprocess.run([start, announcement.mp3], shellTrue) elif sys.platform darwin: subprocess.run([afplay, announcement.mp3]) else: subprocess.run([mpg123, announcement.mp3]) except Exception as e: print(f[语音引擎] edge-tts 播放失败: {e}) self._console_print(text) def _console_print(self, text: str): print(f[播报] {text}) def announce(self, text: str): 对外统一播报入口 print(f[播报内容] {text}) if self.engine edge: asyncio.run(self._play_edge(text)) elif self.engine pyttsx3: self._play_pyttsx3(text) else: # auto 模式优先 edge失败后降级 try: asyncio.run(self._play_edge(text)) except Exception: self._play_pyttsx3(text)这个类的设计思路是“入口统一内部降级”。不管底层用哪个引擎上层调用announce(text)都能完成播报避免因为某个语音库不可用导致整个程序崩溃。3.4 定时任务的实现在真实场景中播报不是手动的而是根据列车运行计划自动触发。我们可以用APScheduler实现定时任务。代码路径app/scheduler.pyfrom apscheduler.schedulers.blocking import BlockingScheduler from app.announcer import build_announcement, VoiceAnnouncer from app.models import Line class AnnounceScheduler: def __init__(self, line: Line, interval_seconds: int 30): self.line line self.interval interval_seconds self.announcer VoiceAnnouncer() self.scheduler BlockingScheduler() # 记录当前模拟位置从第 0 个站开始 self.current_index 0 def job(self): current_station self.line.stations[self.current_index] text build_announcement(self.line, current_station.id) if text: self.announcer.announce(text) # 移动到下一个站如果到达终点则回到起点模拟循环运行 self.current_index (self.current_index 1) % len(self.line.stations) def start(self): self.scheduler.add_job( self.job, interval, secondsself.interval, idannounce_job, max_instances1, coalesceTrue, ) print(f定时播报已启动每 {self.interval} 秒触发一次。) self.scheduler.start()这里的current_index模拟列车当前位置。实际项目中位置状态应该来自信号系统或数据库而不是内存变量。本文用循环递增来演示“下一站”的动态变化。4. 完整实战案例为了让代码可以完整运行我准备了一个示例线路配置一条名为“冬旅线”的虚构线路包含 5 个站点最后一站命名为“至冬站”呼应项目标题。4.1 配置文件config/line_config.yamlline_name: 冬旅线 stations: - id: D1 name: 始发广场 pinyin: shifa guangchang english_name: Start Square - id: D2 name: 雪原路口 pinyin: xueyuan lukou english_name: Snowfield Road - id: D3 name: 冰湖码头 pinyin: binghu matou english_name: Ice Lake Pier is_transfer: true transfer_lines: [环湖线] - id: D4 name: 北风站 pinyin: beifeng zhan english_name: Northwind Station - id: D5 name: 至冬站 pinyin: zhidong zhan english_name: Snezhnaya Station is_transfer: true transfer_lines: [愚人众快线]4.2 配置加载模块app/config.pyimport yaml from app.models import Line, Station def load_line_from_yaml(path: str) - Line: with open(path, r, encodingutf-8) as f: data yaml.safe_load(f) line_name data.get(line_name, 未知线路) stations [] for item in data.get(stations, []): station Station( iditem[id], nameitem[name], pinyinitem.get(pinyin, ), english_nameitem.get(english_name, ), is_transferitem.get(is_transfer, False), transfer_linesitem.get(transfer_lines, []), ) stations.append(station) return Line(nameline_name, stationsstations)注意上面代码里用到了transfer_lines字段所以我们需要在Station数据类中追加这个字段dataclass class Station: 站点数据模型 id: str name: str pinyin: str english_name: str is_transfer: bool False transfer_lines: list field(default_factorylist) arrival_time: str 4.3 命令行入口main.pyimport sys from app.config import load_line_from_yaml from app.announcer import build_announcement, VoiceAnnouncer def main(): config_path config/line_config.yaml line load_line_from_yaml(config_path) print(f线路加载成功{line.name}) print(站点列表) for i, station in enumerate(line.stations): print(f {i 1}. {station.name}) if len(sys.argv) 1: # 支持命令行指定当前站点 ID current_id sys.argv[1] else: # 默认使用第一个站作为当前站 current_id line.stations[0].id text build_announcement(line, current_id) if not text: print(未找到当前站点请检查站点 ID。) return print(生成播报文案, text) announcer VoiceAnnouncer(engineauto) announcer.announce(text) if __name__ __main__: main()4.4 Web API 服务app/web.py除了命令行播报我们再提供一个 Flask Web 接口方便其他系统调用查询“下一站”信息。from flask import Flask, jsonify, request from app.config import load_line_from_yaml from app.announcer import build_announcement, VoiceAnnouncer app Flask(__name__) line load_line_from_yaml(config/line_config.yaml) announcer VoiceAnnouncer(engineauto) app.route(/api/next-station, methods[GET]) def get_next_station(): 查询下一站信息 current_id request.args.get(current_id, ) current_station line.get_station_by_id(current_id) if current_station is None: return jsonify({error: station not found}), 404 next_station line.get_next_station(current_id) if next_station is None: return jsonify({ current_station: current_station.name, message: 当前站点是终点站, }) return jsonify({ current_station: current_station.name, next_station: next_station.name, announcement: build_announcement(line, current_id), }) app.route(/api/announce, methods[POST]) def trigger_announce(): 触发语音播报 data request.get_json(forceTrue, silentTrue) or {} current_id data.get(current_id, ) text build_announcement(line, current_id) if not text: return jsonify({error: station not found}), 404 # 异步播报避免阻塞 HTTP 响应 import threading thread threading.Thread(targetannouncer.announce, args(text,), daemonTrue) thread.start() return jsonify({status: ok, announcement: text}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)启动 Web 服务python -m app.web然后访问接口测试curl http://127.0.0.1:5000/api/next-station?current_idD4预期返回{ current_station: 北风站, next_station: 至冬站, announcement: 列车即将到站下一站——至冬站可换乘愚人众快线 }4.5 运行与验证场景一命令行模式python main.py D3预期输出线路加载成功冬旅线 站点列表 1. 始发广场 2. 雪原路口 3. 冰湖码头 4. 北风站 5. 至冬站 生成播报文案 列车即将到站下一站——北风站 [播报内容] 列车即将到站下一站——北风站场景二定时任务模式from app.scheduler import AnnounceScheduler from app.config import load_line_from_yaml line load_line_from_yaml(config/line_config.yaml) scheduler AnnounceScheduler(line, interval_seconds10) scheduler.start()每 10 秒会模拟一次列车到站并依次播报下一站当到达终点站“至冬站”后会循环回始发站。5. 常见问题与排查思路在实际开发或者运行这个项目时大家可能会遇到下面这些问题。我把常见现象、原因和解决思路整理成一张表方便快速排查。问题现象常见原因解决思路提示ModuleNotFoundError: No module named pyttsx3本地没有安装依赖执行pip install pyttsx3pyttsx3 初始化失败Linux 缺少 espeak 或 libespeak1Ubuntu 执行sudo apt-get install espeak中文语音不生效系统语音库中没有中文音色遍历engine.getProperty(voices)找到中文语音 ID 并设置Windows 控制台输出中文乱码控制台编码不是 UTF-8在代码开头设置sys.stdout.reconfigure(encodingutf-8)edge-tts 无法生成语音网络无法访问微软服务切换为 pyttsx3 或检查网络代理设置Flask 接口返回 404请求的站点 ID 不存在检查配置文件和请求参数是否一致定时任务不触发BlockingScheduler被其他代码阻塞确保scheduler.start()是主线程最后调用的方法音频播放没有声音系统没有安装播放器Linux 安装mpg123或改用自己的播放逻辑这里重点说两个高频问题。5.1 pyttsx3 在 Linux 上不发声pyttsx3在 Linux 上依赖espeak。如果系统没有安装执行时可能不报错但也没有声音。解决方法是安装 espeaksudo apt-get update sudo apt-get install espeak如果你在容器环境里运行还需要额外安装音频驱动相关的包。如果只是测试逻辑不要求声音输出可以临时把引擎改成console模式也就是只打印文案。5.2 Windows 下中文语音不生效pyttsx3默认会使用系统第一个语音可能是英文。我们需要手动选择中文语音。可以使用下面这段调试代码来查看当前系统有哪些语音import pyttsx3 engine pyttsx3.init() for voice in engine.getProperty(voices): print(voice.id, voice.name)如果输出结果里有类似HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Speech\Voices\Tokens\TTS_MS_ZH-CN_HUIHUI_11.0这样的 ID就可以用代码设置engine.setProperty(voice, HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Speech\\Voices\\Tokens\\TTS_MS_ZH-CN_HUIHUI_11.0)如果你的系统里完全没有中文语音包需要先安装 Windows 的中文语音包或者直接用edge-tts在线方案。6. 最佳实践与工程建议6.1 不要硬编码线路数据我在文中把站点数据放在了 YAML 配置文件里这是一个很好的习惯。硬编码的坏处很明显新增站点需要改代码、重新发布不同线路需要复制代码运维人员无法独立维护数据。推荐的做法是线路数据放数据库或配置文件程序启动时加载支持热更新。6.2 语音引擎要做降级处理语音合成服务可能因为网络、系统依赖、授权等问题随时不可用。我们要保证“语音挂了播报系统不能挂”。建议实现降级链路edge-tts 在线合成 → 失败降级 pyttsx3 本地合成 → 再失败降级控制台日志输出这样即使没有任何发声设备系统依然可以记录日志、输出文案业务不受影响。6.3 定时任务要防止重复播报如果上一轮播报还没结束下一轮任务又触发了就会导致语音重叠。在使用APScheduler时可以给任务添加max_instances1和coalesceTruescheduler.add_job( job, interval, seconds30, max_instances1, coalesceTrue, )这两个参数的含义是同一时间只允许一个实例运行如果任务积压了多次触发只合并执行一次。6.4 对外接口注意权限控制如果你的 Web API 暴露在公网任何人都可以调用/api/announce接口触发播报轻则骚扰重则被刷流量。建议部署到内网不直接暴露公网在 Nginx 层做 IP 白名单API 增加简单的 Token 校验。6.5 日志记录要完整每次播报都应该记录以下信息播报时间当前站点下一站信息使用的语音引擎播报是否成功。用 Python 的logging模块实现即可不要用print代替。代码中可以用类似这样的写法import logging logger logging.getLogger(train_announcer) def log_announcement(current_station, next_station_name, engine_type): logger.info( 播报成功 | 当前站: %s | 下一站: %s | 引擎: %s, current_station, next_station_name, engine_type, )6.6 生产环境进一步扩展方向以上代码是一个 MVP最小可行产品版本。如果要在真实轨道交通或班车系统中落地还需要考虑实时位置源接入GPS、信号系统、刷卡数据多线路同时播报并发控制异常场景处理列车晚点、跳过站点、终点站变更多语言播报切换前端大屏显示语音音量、语速动态调整。这些方向都可以在现有代码基础上逐步叠加。7. 总结与下一步学习建议这篇文章围绕“列车即将到站下一站——至冬”这个场景完整实现了一个轻量级的列车到站播报系统。我们完成了线路与站点的数据建模“下一站”计算逻辑播报文案生成基于pyttsx3和edge-tts的双引擎语音播报基于 Flask 的 Web 查询接口基于 APScheduler 的定时任务常见问题排查与工程化建议。如果你是从零跟着文章写到这里建议你先在命令行模式跑通流程然后把线路配置替换成自己熟悉的地铁线路或班车线路再逐步加入 Web API 和定时任务。只有亲自动手改一遍才能理解“模型、配置、播报逻辑、服务化”是怎么串起来的。接下来你可以继续学习的方向包括用 FastAPI 替代 Flask体验自动生成 OpenAPI 文档把线路数据存到 SQLite/MySQL实现动态增删站点接入消息队列让其他系统通过 MQTT/Redis 触发播报做一个简单的 Web 管理界面可视化编辑线路和站点部署到树莓派 音箱做成一个真正的“到站提醒机器人”。如果这篇教程对你有帮助可以收藏备用。后续我会继续分享语音合成、Flask 接口设计、定时任务调度的更多实战细节欢迎在评论区留言交流你的想法和遇到的问题。
Python实战：从零构建列车到站播报系统（语音合成+定时任务+Web API）
各位读者朋友大家好。今天这篇博文想从一个有点“中二”的项目标题说起“列车即将到站下一站——至冬”我第一次看到这句话时第一反应是某个游戏剧情预告又或者是某个漫展的主题活动。但冷静下来一想这句话本身其实是一个非常典型的轨道交通语音播报场景列车即将进站系统需要动态计算出“当前位置”和“下一站”然后生成类似“列车即将到站下一站——XX”的播报信息。如果把这个场景落地成一套可运行的代码就是一个很有意思的实战项目。这篇文章我打算就用“至冬”这个虚构站名作为演示数据带大家从零实现一套列车到站播报系统。系统会包含站台数据建模与线路管理到站信息计算与播报文案生成中文语音合成播报Web API 查询接口定时任务自动播报常见问题排查与工程化建议。无论你是 Python 初学者还是工作中需要做语音提醒、定时播报、信息提示类小系统的开发者都可以从这篇文章里找到可复用的思路。代码我会尽量给全并且保证在本地可以跑起来。接下来我们正式开工。1. 背景与核心概念1.1 什么是列车到站播报系统列车到站播报系统简单来说就是一套“根据列车实时位置计算下一站信息并生成语音/文字提示”的程序。它在现实生活中非常常见地铁车厢内的“列车即将到站下一站人民广场”公交车的“下一站XX站请下车的乘客做好准备”高铁、机场摆渡车、景区观光车上的到站提醒企业内部班车、校园摆渡车的到站通知。这类系统虽然业务规模不大但背后涉及的逻辑很完整线路数据管理、位置状态维护、文案生成、语音合成、信息发布。作为一个练手项目它能很好地锻炼数据建模、模块拆分和工程化能力。1.2 系统核心组成从软件架构角度看一个最小可用的到站播报系统包含以下部分模块职责线路数据模块定义线路名称、站点列表、站点顺序、换乘信息位置状态模块记录列车当前在哪个站或刚从哪个站出发文案生成模块根据当前位置生成“下一站”播报文本语音合成模块将文本转为语音并播放定时任务模块按照时间间隔自动触发播报对外接口模块提供查询下一站/触发播报的 API1.3 为什么自己做一套而不是直接用现成的有人可能会问地铁里已经有现成的播报系统了自己写一套有什么意义其实意义在于这种“信息播报类”系统在很多非公共交通场景下是没有现成产品的。比如公司班车到站提醒学校图书馆闭馆提醒仓库叉车调度提示展会摆渡车播报个人开发的小型语音助手。在这些场景里你需要的往往是“一套轻量、可配置、能快速接入业务数据”的播报服务而不是地铁级的信号控制系统。我们用 Python 写一个简化版正好覆盖了这类需求的核心逻辑。2. 环境准备与版本说明2.1 运行环境本文示例代码使用 Python 3 开发主要依赖以下库库名用途pyttsx3离线文字转语音支持 Windows/macOS/Linuxedge-tts微软 Edge 在线语音合成音质更好Flask提供 Web API 查询接口APScheduler定时触发播报任务PyYAML读取 YAML 配置文件需要说明的是语音合成库的版本和音色在不同操作系统上差异较大本文以“能跑通”为目标版本不需要刻意追求最新。如果你本地环境没有这些库可以直接用 pip 安装pip install pyttsx3 edge-tts flask apscheduler pyyaml如果你的网络环境无法安装edge-tts系统会自动降级为pyttsx3控制台输出模式不影响核心流程。2.2 项目目录结构我们先规划一个清晰的项目结构方便后续扩展train-announcer/ ├── app/ │ ├── __init__.py │ ├── models.py # 数据模型站点、线路 │ ├── announcer.py # 播报文案生成与语音播报 │ ├── scheduler.py # 定时任务 │ ├── web.py # Flask Web API │ └── config.py # 配置加载 ├── config/ │ └── line_config.yaml # 线路与站点配置 ├── main.py # 程序入口 ├── requirements.txt └── README.md这个结构不算复杂但已经把“数据”“业务”“服务”三层分开了。3. 核心模块设计与实现原理3.1 站点与线路的数据建模到站播报系统里最重要的数据是“线路”和“站点”。一个合理的建模方式是每一条线路包含若干站点站点按顺序排列每个站点可能有英文名、拼音名、是否换乘站等属性列车当前状态记录的是“已经离开哪个站”或者“当前正驶向哪个站”。下面我们先用 Python 的dataclass定义站点和线路模型。代码路径app/models.pyfrom dataclasses import dataclass, field from typing import List, Optional dataclass class Station: 站点数据模型 id: str # 站点唯一标识 name: str # 站点中文名 pinyin: str # 站点拼音 english_name: str # 站点英文名 is_transfer: bool False # 是否换乘站 arrival_time: str # 预计到达时间例如 10:30 dataclass class Line: 线路数据模型 name: str # 线路名 stations: List[Station] field(default_factorylist) def get_station_by_id(self, station_id: str) - Optional[Station]: 根据站点 ID 查找站点对象 for station in self.stations: if station.id station_id: return station return None def get_next_station(self, current_station_id: str) - Optional[Station]: 返回当前站点的下一个站点。 如果当前站点是终点站返回 None。 for index, station in enumerate(self.stations): if station.id current_station_id: if index 1 len(self.stations): return self.stations[index 1] return None return None这里需要注意get_next_station是关键方法。它决定了播报系统能不能准确找到“下一站”。如果站点 ID 不存在我们返回None上层调用时要做空值判断避免程序崩溃。3.2 播报文案生成逻辑有了线路模型之后我们可以这样设计播报逻辑如果当前站是普通站播放“列车即将到站下一站——XX”如果当前站是终点站播放“列车即将到达终点站XX”如果下一站是换乘站可以追加一句“可换乘 XX 号线”。为了避免把文案写死在代码里我建议用一个独立函数来生成播报文本后续上线不同场景时可以直接替换。代码路径app/announcer.pyfrom app.models import Line, Station def build_announcement(line: Line, current_station_id: str) - str: 根据当前线路和当前站点 ID生成播报文案。 返回字符串如果找不到当前站点则返回空字符串。 current_station line.get_station_by_id(current_station_id) if current_station is None: return next_station line.get_next_station(current_station_id) if next_station is None: # 当前站是终点站 return f列车即将到达终点站{current_station.name}感谢您的乘坐 # 常规到站播报 text f列车即将到站下一站——{next_station.name} if next_station.is_transfer: text 可换乘 、.join(next_station.transfer_lines) text return text这里为了演示换乘信息我在 Station 模型里加了一个transfer_lines字段。你可以在实际项目中补充这个字段这里不额外展开。3.3 语音合成方案对比语音播报是整个系统的灵魂。在 Python 里常用的方案有三种方案离线/在线音质跨平台适用场景pyttsx3离线一般好本地快速测试、离线环境edge-tts在线较好好对音质有要求可以联网商用 TTS SDK在线最好一般生产环境、携带品牌音色我的推荐是本地开发用pyttsx3跑通流程音质敏感场景用edge-tts。edge-tts使用微软 Edge 的在线语音服务音色自然支持中文代码也不复杂。下面封装一个支持“自动降级”的播报类。代码路径app/announcer.pyimport asyncio import subprocess import sys class VoiceAnnouncer: 语音播报器支持 pyttsx3 和 edge-tts 自动切换 def __init__(self, engineauto): self.engine engine self._tts_engine None def _init_pyttsx3(self): try: import pyttsx3 engine pyttsx3.init() # 尝试设置中文语音不同平台 voice id 不同 voices engine.getProperty(voices) for voice in voices: if chinese in voice.name.lower() or zh in voice.id.lower(): engine.setProperty(voice, voice.id) break return engine except Exception as e: print(f[语音引擎] pyttsx3 初始化失败: {e}) return None def _play_pyttsx3(self, text: str): if self._tts_engine is None: self._tts_engine self._init_pyttsx3() if self._tts_engine is None: # 最终降级方案控制台输出 self._console_print(text) return self._tts_engine.say(text) self._tts_engine.runAndWait() async def _play_edge(self, text: str): try: import edge_tts communicate edge_tts.Communicate(text, zh-CN-XiaoxiaoNeural) await communicate.save(announcement.mp3) # 播放生成的 mp3 if sys.platform.startswith(win): subprocess.run([start, announcement.mp3], shellTrue) elif sys.platform darwin: subprocess.run([afplay, announcement.mp3]) else: subprocess.run([mpg123, announcement.mp3]) except Exception as e: print(f[语音引擎] edge-tts 播放失败: {e}) self._console_print(text) def _console_print(self, text: str): print(f[播报] {text}) def announce(self, text: str): 对外统一播报入口 print(f[播报内容] {text}) if self.engine edge: asyncio.run(self._play_edge(text)) elif self.engine pyttsx3: self._play_pyttsx3(text) else: # auto 模式优先 edge失败后降级 try: asyncio.run(self._play_edge(text)) except Exception: self._play_pyttsx3(text)这个类的设计思路是“入口统一内部降级”。不管底层用哪个引擎上层调用announce(text)都能完成播报避免因为某个语音库不可用导致整个程序崩溃。3.4 定时任务的实现在真实场景中播报不是手动的而是根据列车运行计划自动触发。我们可以用APScheduler实现定时任务。代码路径app/scheduler.pyfrom apscheduler.schedulers.blocking import BlockingScheduler from app.announcer import build_announcement, VoiceAnnouncer from app.models import Line class AnnounceScheduler: def __init__(self, line: Line, interval_seconds: int 30): self.line line self.interval interval_seconds self.announcer VoiceAnnouncer() self.scheduler BlockingScheduler() # 记录当前模拟位置从第 0 个站开始 self.current_index 0 def job(self): current_station self.line.stations[self.current_index] text build_announcement(self.line, current_station.id) if text: self.announcer.announce(text) # 移动到下一个站如果到达终点则回到起点模拟循环运行 self.current_index (self.current_index 1) % len(self.line.stations) def start(self): self.scheduler.add_job( self.job, interval, secondsself.interval, idannounce_job, max_instances1, coalesceTrue, ) print(f定时播报已启动每 {self.interval} 秒触发一次。) self.scheduler.start()这里的current_index模拟列车当前位置。实际项目中位置状态应该来自信号系统或数据库而不是内存变量。本文用循环递增来演示“下一站”的动态变化。4. 完整实战案例为了让代码可以完整运行我准备了一个示例线路配置一条名为“冬旅线”的虚构线路包含 5 个站点最后一站命名为“至冬站”呼应项目标题。4.1 配置文件config/line_config.yamlline_name: 冬旅线 stations: - id: D1 name: 始发广场 pinyin: shifa guangchang english_name: Start Square - id: D2 name: 雪原路口 pinyin: xueyuan lukou english_name: Snowfield Road - id: D3 name: 冰湖码头 pinyin: binghu matou english_name: Ice Lake Pier is_transfer: true transfer_lines: [环湖线] - id: D4 name: 北风站 pinyin: beifeng zhan english_name: Northwind Station - id: D5 name: 至冬站 pinyin: zhidong zhan english_name: Snezhnaya Station is_transfer: true transfer_lines: [愚人众快线]4.2 配置加载模块app/config.pyimport yaml from app.models import Line, Station def load_line_from_yaml(path: str) - Line: with open(path, r, encodingutf-8) as f: data yaml.safe_load(f) line_name data.get(line_name, 未知线路) stations [] for item in data.get(stations, []): station Station( iditem[id], nameitem[name], pinyinitem.get(pinyin, ), english_nameitem.get(english_name, ), is_transferitem.get(is_transfer, False), transfer_linesitem.get(transfer_lines, []), ) stations.append(station) return Line(nameline_name, stationsstations)注意上面代码里用到了transfer_lines字段所以我们需要在Station数据类中追加这个字段dataclass class Station: 站点数据模型 id: str name: str pinyin: str english_name: str is_transfer: bool False transfer_lines: list field(default_factorylist) arrival_time: str 4.3 命令行入口main.pyimport sys from app.config import load_line_from_yaml from app.announcer import build_announcement, VoiceAnnouncer def main(): config_path config/line_config.yaml line load_line_from_yaml(config_path) print(f线路加载成功{line.name}) print(站点列表) for i, station in enumerate(line.stations): print(f {i 1}. {station.name}) if len(sys.argv) 1: # 支持命令行指定当前站点 ID current_id sys.argv[1] else: # 默认使用第一个站作为当前站 current_id line.stations[0].id text build_announcement(line, current_id) if not text: print(未找到当前站点请检查站点 ID。) return print(生成播报文案, text) announcer VoiceAnnouncer(engineauto) announcer.announce(text) if __name__ __main__: main()4.4 Web API 服务app/web.py除了命令行播报我们再提供一个 Flask Web 接口方便其他系统调用查询“下一站”信息。from flask import Flask, jsonify, request from app.config import load_line_from_yaml from app.announcer import build_announcement, VoiceAnnouncer app Flask(__name__) line load_line_from_yaml(config/line_config.yaml) announcer VoiceAnnouncer(engineauto) app.route(/api/next-station, methods[GET]) def get_next_station(): 查询下一站信息 current_id request.args.get(current_id, ) current_station line.get_station_by_id(current_id) if current_station is None: return jsonify({error: station not found}), 404 next_station line.get_next_station(current_id) if next_station is None: return jsonify({ current_station: current_station.name, message: 当前站点是终点站, }) return jsonify({ current_station: current_station.name, next_station: next_station.name, announcement: build_announcement(line, current_id), }) app.route(/api/announce, methods[POST]) def trigger_announce(): 触发语音播报 data request.get_json(forceTrue, silentTrue) or {} current_id data.get(current_id, ) text build_announcement(line, current_id) if not text: return jsonify({error: station not found}), 404 # 异步播报避免阻塞 HTTP 响应 import threading thread threading.Thread(targetannouncer.announce, args(text,), daemonTrue) thread.start() return jsonify({status: ok, announcement: text}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)启动 Web 服务python -m app.web然后访问接口测试curl http://127.0.0.1:5000/api/next-station?current_idD4预期返回{ current_station: 北风站, next_station: 至冬站, announcement: 列车即将到站下一站——至冬站可换乘愚人众快线 }4.5 运行与验证场景一命令行模式python main.py D3预期输出线路加载成功冬旅线 站点列表 1. 始发广场 2. 雪原路口 3. 冰湖码头 4. 北风站 5. 至冬站 生成播报文案 列车即将到站下一站——北风站 [播报内容] 列车即将到站下一站——北风站场景二定时任务模式from app.scheduler import AnnounceScheduler from app.config import load_line_from_yaml line load_line_from_yaml(config/line_config.yaml) scheduler AnnounceScheduler(line, interval_seconds10) scheduler.start()每 10 秒会模拟一次列车到站并依次播报下一站当到达终点站“至冬站”后会循环回始发站。5. 常见问题与排查思路在实际开发或者运行这个项目时大家可能会遇到下面这些问题。我把常见现象、原因和解决思路整理成一张表方便快速排查。问题现象常见原因解决思路提示ModuleNotFoundError: No module named pyttsx3本地没有安装依赖执行pip install pyttsx3pyttsx3 初始化失败Linux 缺少 espeak 或 libespeak1Ubuntu 执行sudo apt-get install espeak中文语音不生效系统语音库中没有中文音色遍历engine.getProperty(voices)找到中文语音 ID 并设置Windows 控制台输出中文乱码控制台编码不是 UTF-8在代码开头设置sys.stdout.reconfigure(encodingutf-8)edge-tts 无法生成语音网络无法访问微软服务切换为 pyttsx3 或检查网络代理设置Flask 接口返回 404请求的站点 ID 不存在检查配置文件和请求参数是否一致定时任务不触发BlockingScheduler被其他代码阻塞确保scheduler.start()是主线程最后调用的方法音频播放没有声音系统没有安装播放器Linux 安装mpg123或改用自己的播放逻辑这里重点说两个高频问题。5.1 pyttsx3 在 Linux 上不发声pyttsx3在 Linux 上依赖espeak。如果系统没有安装执行时可能不报错但也没有声音。解决方法是安装 espeaksudo apt-get update sudo apt-get install espeak如果你在容器环境里运行还需要额外安装音频驱动相关的包。如果只是测试逻辑不要求声音输出可以临时把引擎改成console模式也就是只打印文案。5.2 Windows 下中文语音不生效pyttsx3默认会使用系统第一个语音可能是英文。我们需要手动选择中文语音。可以使用下面这段调试代码来查看当前系统有哪些语音import pyttsx3 engine pyttsx3.init() for voice in engine.getProperty(voices): print(voice.id, voice.name)如果输出结果里有类似HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Speech\Voices\Tokens\TTS_MS_ZH-CN_HUIHUI_11.0这样的 ID就可以用代码设置engine.setProperty(voice, HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Speech\\Voices\\Tokens\\TTS_MS_ZH-CN_HUIHUI_11.0)如果你的系统里完全没有中文语音包需要先安装 Windows 的中文语音包或者直接用edge-tts在线方案。6. 最佳实践与工程建议6.1 不要硬编码线路数据我在文中把站点数据放在了 YAML 配置文件里这是一个很好的习惯。硬编码的坏处很明显新增站点需要改代码、重新发布不同线路需要复制代码运维人员无法独立维护数据。推荐的做法是线路数据放数据库或配置文件程序启动时加载支持热更新。6.2 语音引擎要做降级处理语音合成服务可能因为网络、系统依赖、授权等问题随时不可用。我们要保证“语音挂了播报系统不能挂”。建议实现降级链路edge-tts 在线合成 → 失败降级 pyttsx3 本地合成 → 再失败降级控制台日志输出这样即使没有任何发声设备系统依然可以记录日志、输出文案业务不受影响。6.3 定时任务要防止重复播报如果上一轮播报还没结束下一轮任务又触发了就会导致语音重叠。在使用APScheduler时可以给任务添加max_instances1和coalesceTruescheduler.add_job( job, interval, seconds30, max_instances1, coalesceTrue, )这两个参数的含义是同一时间只允许一个实例运行如果任务积压了多次触发只合并执行一次。6.4 对外接口注意权限控制如果你的 Web API 暴露在公网任何人都可以调用/api/announce接口触发播报轻则骚扰重则被刷流量。建议部署到内网不直接暴露公网在 Nginx 层做 IP 白名单API 增加简单的 Token 校验。6.5 日志记录要完整每次播报都应该记录以下信息播报时间当前站点下一站信息使用的语音引擎播报是否成功。用 Python 的logging模块实现即可不要用print代替。代码中可以用类似这样的写法import logging logger logging.getLogger(train_announcer) def log_announcement(current_station, next_station_name, engine_type): logger.info( 播报成功 | 当前站: %s | 下一站: %s | 引擎: %s, current_station, next_station_name, engine_type, )6.6 生产环境进一步扩展方向以上代码是一个 MVP最小可行产品版本。如果要在真实轨道交通或班车系统中落地还需要考虑实时位置源接入GPS、信号系统、刷卡数据多线路同时播报并发控制异常场景处理列车晚点、跳过站点、终点站变更多语言播报切换前端大屏显示语音音量、语速动态调整。这些方向都可以在现有代码基础上逐步叠加。7. 总结与下一步学习建议这篇文章围绕“列车即将到站下一站——至冬”这个场景完整实现了一个轻量级的列车到站播报系统。我们完成了线路与站点的数据建模“下一站”计算逻辑播报文案生成基于pyttsx3和edge-tts的双引擎语音播报基于 Flask 的 Web 查询接口基于 APScheduler 的定时任务常见问题排查与工程化建议。如果你是从零跟着文章写到这里建议你先在命令行模式跑通流程然后把线路配置替换成自己熟悉的地铁线路或班车线路再逐步加入 Web API 和定时任务。只有亲自动手改一遍才能理解“模型、配置、播报逻辑、服务化”是怎么串起来的。接下来你可以继续学习的方向包括用 FastAPI 替代 Flask体验自动生成 OpenAPI 文档把线路数据存到 SQLite/MySQL实现动态增删站点接入消息队列让其他系统通过 MQTT/Redis 触发播报做一个简单的 Web 管理界面可视化编辑线路和站点部署到树莓派 音箱做成一个真正的“到站提醒机器人”。如果这篇教程对你有帮助可以收藏备用。后续我会继续分享语音合成、Flask 接口设计、定时任务调度的更多实战细节欢迎在评论区留言交流你的想法和遇到的问题。
