YZDR's Blog
校园充电桩监控系统实战 - Flask + SQLite 实现
YZDR Lv2
  2026-05-07 17:00:00   2026-05-07 17:36:11    1k 字  4 分钟

校园充电桩监控系统实战 - Flask + SQLite 实现

本文介绍一个校园充电桩监控系统的设计与实现，该系统基于 Flask + SQLite 构建，支持多建筑充电桩状态实时监控。

项目背景

校园充电桩分布广泛，传统方式需要人工巡查确认空闲状态。本系统通过网络接口实时获取充电桩状态，用户可在 Web 界面上直观查看空闲充电桩数量和位置。

技术架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
┌──────────────────────────────────────────────────────────────┐
│                     系统架构图                               │
├──────────────────────────────────────────────────────────────┤
│                                                              │
│   ┌─────────────┐                    ┌─────────────┐        │
│   │  前端页面   │ ◀────── REST API ──▶ │  Flask后端  │        │
│   │  (HTML/CSS)│                    │  (Python)   │        │
│   └─────────────┘                    └──────┬──────┘        │
│                                              │               │
│   ┌─────────────┐                    ┌──────┴──────┐        │
│   │  图片代理   │                    │  多线程调度器 │        │
│   │  (COS缓存) │                    │  (charger_worker)│    │
│   └─────────────┘                    └──────┬──────┘        │
│                                              │               │
│   ┌─────────────┐                    ┌──────┴──────┐        │
│   │  SQLite数据库│                    │  状态采集模块 │        │
│   │  (WAL模式)  │                    │  (Web Scraping)│      │
│   └─────────────┘                    └─────────────┘        │
│                                                              │
└──────────────────────────────────────────────────────────────┘

核心模块设计

1. 数据库层

采用 SQLite 的 WAL 模式，提升并发读写性能：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def init_db():
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    # 启用 WAL 模式，支持并发写入
    cursor.execute('PRAGMA journal_mode=WAL;')
    
    cursor.execute('''
    CREATE TABLE IF NOT EXISTS chargers (
        id TEXT PRIMARY KEY,
        device_id TEXT,
        online BOOLEAN DEFAULT 0,
        sytime REAL DEFAULT 0.0,
        SwitchState BOOLEAN DEFAULT 0,
        devicestat TEXT DEFAULT ''
    )
    ''')

关键字段说明：

字段 类型 说明
id TEXT 充电桩位置编号
device_id TEXT 设备唯一标识
online BOOLEAN 是否在线
sytime REAL 累计使用时长
SwitchState BOOLEAN 开关状态
devicestat TEXT 设备状态描述

2. 多线程调度器

采用生产者-消费者模式，每 30 秒发起新一轮巡检：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def scheduler(data_file, load_func, name):
    task_queue = Queue()
    worker_threads = []
    max_workers = 4
    
    while True:
        if task_queue.empty() and (current_time - last_round_time > 30):
            data = load_func()
            
            # 按设备ID排序后加入队列
            sorted_chargers = sorted(data.items(), key=lambda x: x[1]['id'])
            for pos, charger_data in sorted_chargers:
                task_queue.put((pos, charger_data['id']))
            
            last_round_time = current_time

3. 状态采集模块

通过模拟浏览器请求获取设备状态：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def fetch_status_with_cookie(device_id, cookie, session):
    # 三步走：Index -> AddSession -> getWxInfo
    session.get(url_index, headers=base_headers, verify=False)
    
    session.post(url_add, headers=add_headers, data={
        "AdminName": admin,
        "DeviceNumber": str(device_id)
    }, verify=False)
    
    # 等待后端 Session 挂载
    time.sleep(1.5)
    
    info_res = session.post(url_info, headers=info_headers, data={
        "AdminName": admin,
        "DeviceNumber": str(device_id),
        "openid": current_openid
    }, verify=False)
    
    return res.json()

4. 自动续期机制

定时刷新 Cookie 保持会话有效：

1
2
3
4
5
6
7
8
def auto_refresh_cookie():
    while True:
        time.sleep(600)  # 每10分钟
        new_cookie = refresh_session_id()
        if new_cookie:
            with COOKIE_LOCK:
                global COOKIE_POOL
                COOKIE_POOL = [new_cookie]

API 设计

路由 方法 说明
/api/chargers GET 获取 23 号楼充电桩状态
/api/chargers/wenke GET 获取文科楼充电桩状态
/api/cookies/status GET 查看 Cookie 状态
/proxy-image/<path> GET 图片代理（避免跨域）

数据流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
┌─────────────────────────────────────────────────────────────┐
│                      数据流程                                │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│  1. 调度器创建任务队列 ──▶ 分配给 Worker 线程                 │
│                                                               │
│  2. Worker 调用 fetch_status_with_cookie()                    │
│     ├── 访问 Index 页面                                       │
│     ├── 挂号 AddSession                                       │
│     ├── 等待 1.5s                                             │
│     └── 调用 getWxInfo 获取数据                               │
│                                                               │
│  3. 数据写入 SQLite 数据库                                     │
│                                                               │
│  4. Flask API 从数据库读取返回给前端                          │
│                                                               │
│  5. 前端轮询 /api/chargers 更新显示                          │
│                                                               │
└─────────────────────────────────────────────────────────────┘

前端展示

前端使用原生 HTML/CSS/JavaScript 实现，支持：

  • 实时状态统计（在线/离线/空闲数量）
  • 充电桩网格展示
  • 状态颜色编码：绿色空闲、橙色使用中、灰色离线
  • 定时自动刷新
1
2
3
4
5
6
7
function updateDisplay(data) {
    const onlineCount = Object.values(data).filter(c => c.online).length;
    const freeCount = Object.values(data).filter(c =>
        c.online && !c.SwitchState
    ).length;
    // 更新统计面板...
}

部署方式

1
2
3
4
5
6
# 直接运行
python app.py

# 或使用 gunicorn（生产环境）
pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:23422 app:app

在线体验

在线访问地址：https://charger.yzdrserver0106.top/

支持多建筑充电桩状态查看，实时显示空闲桩位数量和位置分布。

技术总结

技术点 实现方式
Web 框架 Flask
数据库 SQLite (WAL 模式)
并发处理 threading + Queue
状态采集 requests Session 复用
会话保持 Cookie 自动刷新
图片代理 腾讯云 COS 缓存

适用场景

本系统架构可应用于：

  • 校园充电桩监控：实时查看空闲桩位
  • 共享设备管理：工位、储物柜等状态监控
  • 物联网数据采集：传感器网络状态聚合

相关阅读：

 评论
评论插件加载失败
正在加载评论插件
  1. 1. 校园充电桩监控系统实战 - Flask + SQLite 实现
    1. 1.1. 项目背景
    2. 1.2. 技术架构
    3. 1.3. 核心模块设计
      1. 1.3.1. 1. 数据库层
      2. 1.3.2. 2. 多线程调度器
      3. 1.3.3. 3. 状态采集模块
      4. 1.3.4. 4. 自动续期机制
    4. 1.4. API 设计
    5. 1.5. 数据流程
    6. 1.6. 前端展示
    7. 1.7. 部署方式
    8. 1.8. 在线体验
    9. 1.9. 技术总结
    10. 1.10. 适用场景
  1. 1. 校园充电桩监控系统实战 - Flask + SQLite 实现
    1. 1.1. 项目背景
    2. 1.2. 技术架构
    3. 1.3. 核心模块设计
      1. 1.3.1. 1. 数据库层
      2. 1.3.2. 2. 多线程调度器
      3. 1.3.3. 3. 状态采集模块
      4. 1.3.4. 4. 自动续期机制
    4. 1.4. API 设计
    5. 1.5. 数据流程
    6. 1.6. 前端展示
    7. 1.7. 部署方式
    8. 1.8. 在线体验
    9. 1.9. 技术总结
    10. 1.10. 适用场景