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

hyhyyyds1234

193人浏览 · 2026-04-23 10:09:05

hyhyyyds1234 · 2026-04-23 10:09:05 发布

发散创新：基于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`

from flask import Flask, jsonify
import paho.mqtt.client as mqtt
import sqlite3
from datetime import datetime

app = Flask(__name__)

# 数据库初始化
def init_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回调函数
def on_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()
                                                                                            except Exception as e:
                                                                                                    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'])
def get_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)
                                                                                } for row in rows
                                                                                    ]
                                                                                        return jsonify(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.
<div id="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）；

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

import paho.mqtt.client as mqtt
import time
import random

client = mqtt.Client()
client.connect("localhost", 1883, 60)

for _ in range(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 模块统一记录日志，便于排查异常