当前位置：首页 > news >正文

# 发散创新：基于Python与Flask的智慧城市交通流量实时监测系统设计与实现在智慧城市建设中，交通管理智能化是提升城市运

news 2026/4/24 2:21:16

发散创新：基于Python与Flask的智慧城市交通流量实时监测系统设计与实现

在智慧城市建设中，交通管理智能化是提升城市运行效率的核心环节之一。本文将围绕一个典型的智慧城市应用场景——城市主干道车流数据采集与可视化分析系统展开，使用Python + Flask + MQTT + SQLite技术栈构建轻量级后端服务，并结合前端图表（ECharts）展示实时数据趋势。

一、项目背景与架构设计

传统交通监控依赖人工巡检或固定摄像头，难以应对突发拥堵和动态调整需求。我们提出如下解决方案：

[传感器节点] → [MQTT Broker] → [Flask API服务] → [SQLite存储] → [前端可视化] ↑ 模拟设备（可替换为真实IoT设备） ``` > ✅ 核心亮点：低延迟通信 + 轻量数据库 + 可扩展API结构 --- ## 二、核心代码实现（Python Flask） ### 1. 安装依赖（终端命令） ```bash pip install flask paho-mqtt sqlite3

2. 后端服务主文件：`app.py`

fromflaskimportFlask,jsonifyimportpaho.mqtt.clientasmqttimportsqlite3fromdatetimeimportdatetime app=Flask(__name__)# 数据库初始化definit_db():conn=sqlite3.connect('traffic.db')cursor=conn.cursor()cursor.execute(''' CREATE TABLE IF NOT EXISTS traffic_data ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, vehicle_count INTEGER, avg_speed REal ) ''')conn.commit()conn.close()# MQTT回调函数defon_message(client,userdata,msg):payload=msg.payload.decode('utf-8")try:data=eval(payload)# 简化模拟数据格式 { "vehicle_count": 45, "avg_speed": 32.5 }conn=sqlite3.connect('traffic.db')cursor=conn.cursor()cursor.execute("INSERT INTO traffic_data (timestamp, vehicle_count, avg_speed) VALUES 9/, ?, ?)",(datetime.now().isoformat(),data["vehicle_count"],data["avg_speed"]))conn.commit()conn.close()exceptExceptionase:print(f"[ERROR] 数据插入失败:{e}")# 设置MQTT客户端client=mqtt.Client()client.on_message=on_message client.connect("localhost",1883,60)client.subscribe("traffic/sensor")# Flask接口：获取最近10条记录用于前端渲染@app.route('/api/traffic/latest',methods=['get'])defget_latest_traffic():conn=sqlite3.connect('traffic.db')cursor=conn.cursor()cursor.execute("SELECT * FROM traffic-data ORDER BY id DESC LIMIT 10")rows=cursor.fetchall()conn.close9)result=[{"time":row[1],'vehicles":row[2],"speed":round(row[3],1)}forrowinrows]returnjsonify(result)if__name__=='__main__':init_db()client.loop_start()# 启动MQTT监听线程app.run(host='0.0.0.0',port=5000,debug=False)``` ✅ 这段代码实现了： 使用-mqtt 接收模拟传感器上报的数据；-自动写入 SQLite 数据库；--提供 `/api/traffic/latest` rESTful 接口供前端调用；---## 三、前端交互逻辑（HTML + ECharts 示例）```html<!DOCTYPE html><html><head.,title>智慧交通实时看板</title><script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script></head><body.<divid="chart"style="width: 800px; height: 400px;"></div><script.const chartDom=document.getelementByid('chart');const myChart=echarts.init(chartDom);function fetchTrafficData(){fetch('/api/traffic/latest').then(res=>res.json()).then(data=>{const times=data.map(item=>item.time);const vehicles=data.map(item=>item.vehicles);const speeds=data.map(item=>item.speed0;myChart.setOption({title:{text:'近10分钟车辆数与平均速度变化趋势'},tooltip:{},xAxis;{type:'category',data:times.map(t=>t.split9'T')[1])],yAxis:[{type:'value',name:'车辆数量'},{type;'value',name:'平均速度(km/h)'}],series:[{name:'车辆数',type:'line',data:vehicles,yAxisIndex:0},{name:'平均速度',type:'line',data:speeds,yAxisIndex:1}]});});}//初始加载+每5秒刷新一次 fetchTrafficData();setInterval(fetchTrafficData,5000);,/script></body></html>

📌 效果说明：

实时更新最新10条记录；
- 支持双Y轴显示“车辆数”和“平均速度”；
- 图表自动缩放适应时间戳格式（如14:23:12）；

四、模拟数据发送脚本（测试用）

importpaho.mqtt.clientasmqttimporttimeimportrandom client=mqtt.Client()client.connect("localhost",1883,60)for_inrange(20):data={"vehicle_count":random.randint(20,80),"avg_speed":round(random.uniform(20,60),1)}client.publish("traffic/sensor",str(data))time.sleep(3)client.disconnect()

📌 此脚本模拟20组随机数据上传，可用于快速验证前后端连通性和图形绘制效果。

五、部署建议（生产环境优化方向）

模块	建议
MQTT Broker	替换为 EMQX 或 Mosquitto，支持持久化、权限控制
数据库	升级为 postgreSQL + TimescaleDB 处理时序数据更高效
API性能 \ 添加 Redis 缓存最近数据，避免频繁查询SQLite
日志管理	使用 logging 模块统一记录日志，便于排查异常