简介:水质检测.rar 面向环境科学、数据分析与机器学习初学者及科研人员,围绕水质安全影响因素分析与预测,提供从数据预处理到多模型对比的一套轻量实践材料。压缩包共10个文件,以8个Python脚本为主,辅以1份CSV水质参数数据和1份开题报告文档,整体约221KB,便于下载后在本地复现实验。CSV涵盖pH值、溶解氧、氨氮、浊度、电导率等常用指标;脚本涉及数据清洗、缺失值与异常值处理、特征工程及时间序列处理,并给出XGBoost、随机森林、LightGBM、KNN、AdaBoost等算法的对比实验代码,可用于观察不同模型在水质预测任务中的表现差异。开题报告则帮助梳理研究背景、目标、方法选择与实验设计。已有195人学习,适合用作课程设计、毕业设计前期的方案参考,也可帮助环境监测人员理解机器学习在水质预测中的落地流程。
1. 从一沓手写台账说起:这套水质检测包能替掉哪些活
去年帮人看一个小型水站的记录本,A4 纸订成册,每页写着日期、pH、浊度、谁测的。翻到中间有三天是空白,问起来才知道那几天电极线松了,读数一直跳,索性就不记了。纸质台账最要命的地方不是写起来麻烦,是它没有后悔药:数据丢了就是丢了,你想回头查换滤芯前后浊度有没有变化,只能靠回忆。这套水质检测.rar 解决的就是这一段——把 pH、浊度、TDS、温度这些探头的读数自动采上来、存进本地库、画成能回看的曲线,超限了给个提示,而不是靠人隔两个小时跑过去抄一次。
它适合三类人:做课程设计或毕业设计、需要一套能跑通完整采集链路的同学;手上已经有传感器、缺上位机软件的自搭水站玩家;以及想把手工记录换成电子台账、又不想上云的小型站点。整套东西是单机形态,不依赖服务器,插上串口就能用,这对没有运维的环境来说反而是优点。
2. 资源构成与技术选型:四层结构如何把传感器读数送到屏幕上
先把这份包的整体骨架说清楚,不然后面改代码会找不到落点。它按「采集 → 解析 → 存储 → 呈现」切成四层,层与层之间用队列和函数调用隔开,没有互相 import 成一团。这种切法不是什么架构美学,是踩出来的:传感器读数一旦和界面刷新写在一个循环里,串口一卡,窗口就假死,任务管理器里只能强杀。
2.1 目录里都有什么
包解压之后大致是下面这个形状,不同版本的文件名可能略有出入,但职责划分基本一致:
水质检测/ ├── main.py # 入口,读配置、装配线程、拉起窗口 ├── config.yaml # 串口参数、采样周期、告警阈值,改这里不用动代码 ├── core/ │ ├── serial_reader.py # 串口线程:只负责把字节流切成完整帧 │ ├── protocol.py # 帧解析、校验、单位换算 │ ├── repository.py # SQLite 建表与读写 │ └── alarm.py # 阈值判定、去抖、告警分发 ├── ui/ │ ├── main_window.py # 主窗口与菜单 │ ├── chart_view.py # 实时曲线控件 │ └── history_view.py # 历史查询与导出 ├── tools/ │ └── fake_source.py # 没有硬件时的模拟数据源 ├── data/ # 运行时生成 water.db ├── requirements.txt └── README.md| 文件/目录 | 职责 | 二次开发时改它的频率 |
|---|---|---|
config.yaml | 串口、阈值、采样周期 | 高,几乎每次部署都改 |
core/protocol.py | 帧格式与字段换算 | 高,换探头必改 |
core/serial_reader.py | 字节流切帧、断线重连 | 低,逻辑稳定 |
core/repository.py | 建表、批量写入、查询 | 中,加参数要加列 |
ui/chart_view.py | 曲线刷新与坐标范围 | 中,加参数要加曲线 |
tools/fake_source.py | 造数据,联调必备 | 低,但很值得读 |
提示:先看
config.yaml和protocol.py这两个文件,基本就能判断这套包能不能直接对接你手上的硬件,其他文件可以晚点再看。
2.2 技术栈为什么是这几样
上位机用 Python,串口靠pyserial,本地库用标准库自带的sqlite3,界面走PyQt5加pyqtgraph画曲线,导出报表用pandas,最后PyInstaller打成一个 exe 丢给现场。这套组合的共同点是「装得少、部署轻」:SQLite 不需要单独起服务,一个文件就是一个库,拷走就能备份;pyqtgraph画几千个点的动态曲线比 matplotlib 顺得多,后者更适合出静态图。
有人会问为什么不上 MySQL 或者时序库。小型站点的现实是,现场那台机器常年没人维护,多一个服务就多一个半夜挂掉的可能。日均几千条记录的量级,SQLite 开着 WAL 模式完全扛得住,写入是毫秒级的事。真要上多站点集中管理,那才轮到换存储,这时候repository.py单独抽出来的价值就体现出来了——只改这一个文件。
如果这份包是 Web 形态的变体(Flask 起服务、前端用 ECharts 画图),结构上只有ui/这一层被替换,core/里的采集、解析、入库三块原样不动。判断方法很简单:看requirements.txt里有没有 flask,有就是 Web 版。
2.3 一次采样从串口到曲线的完整链路
传感器输出的是模拟电压或数字信号,下位机做完 ADC 之后按约定打包成帧,从串口吐给上位机。包里常见的一种帧定义是这样,总共 19 字节,固定长度:
| 偏移 | 长度 | 含义 | 说明 |
|---|---|---|---|
| 0 | 1 | 帧头 | 固定0xAA |
| 1 | 1 | 帧头 | 固定0x55 |
| 2 | 1 | 设备地址 | 多探头时用来区分 |
| 3 | 1 | 数据类型 | 0x01表示采样数据 |
| 4–7 | 4 | 设备时间戳 | 秒级,小端无符号整数 |
| 8–9 | 2 | pH | 实际值 ×100,有符号 |
| 10–11 | 2 | 浊度 | 实际值 ×10,单位 NTU |
| 12–13 | 2 | TDS | 单位 mg/L |
| 14–15 | 2 | 温度 | 实际值 ×100 |
| 16 | 1 | 状态位 | bit0 电极异常,bit1 校准过期 |
| 17 | 1 | 校验和 | 第 2 到 16 字节累加取低 8 位 |
| 18 | 1 | 帧尾 | 固定0x0D |
固定长度帧的好处是切分简单:找到帧头、数够 19 字节、验校验和,不合格就丢。代价是扩展性差,加一路参数就得改帧长,这一点在第 6 章会专门处理。整条链路的时序是:采集线程阻塞读串口 → 切出完整帧塞进queue.Queue→ 解析线程取出解析成 dict → 一个分支进写库线程,一个分支进界面刷新。用队列而不是回调,是为了让慢的那一端(写库、画图)不拖累快的那一端(读串口)。
3. 核心链路拆解:串口取帧、协议校验、入库与告警判定
这一章是整套东西最值得细看的部分,也是最容易翻车的地方。串口读取看着就几行代码,实际上半帧、粘包、拔线重连这些情况一个都跑不掉,写得不严实,跑一晚上早上起来就会发现曲线断了一截。
3.1 串口读取与断线重连
采集线程只干一件事:把字节流切成完整帧。它不做解析,也不碰数据库,这样即使解析逻辑改坏了,采集这一层还是干净的。
import queue import threading import time import serial class SerialReader(threading.Thread): """把串口字节流切成完整帧塞进队列,不做任何业务解析""" def __init__(self, port: str, baudrate: int, out_q: "queue.Queue[bytes]", frame_head: bytes = b"\xaa\x55", frame_size: int = 19): super().__init__(daemon=True, name=f"reader-{port}") self.port = port self.baudrate = baudrate self.out_q = out_q self.frame_head = frame_head self.frame_size = frame_size self._buf = bytearray() self._stop = threading.Event() def stop(self): self._stop.set() def run(self): # 外层 while 负责重连,内层 while 负责读数据,拔线后回到外层 while not self._stop.is_set(): ser = self._open() if ser is None: time.sleep(2) # 打不开就退避,别死循环刷日志把磁盘写满 continue try: while not self._stop.is_set(): chunk = ser.read(ser.in_waiting or 1) if chunk: self._feed(chunk) except (serial.SerialException, OSError): pass # USB 被拔掉走这里,直接进 finally 重连 finally: ser.close() def _open(self): try: return serial.Serial(self.port, self.baudrate, timeout=0.5) except serial.SerialException: return None def _feed(self, chunk: bytes): self._buf.extend(chunk) while True: idx = self._buf.find(self.frame_head) if idx < 0: self._buf.clear() # 没有帧头说明全是噪声,直接丢 return if idx > 0: del self._buf[:idx] # 帧头之前的脏数据不要 if len(self._buf) < self.frame_size: return # 半帧,留着等下一批数据 frame = bytes(self._buf[:self.frame_size]) del self._buf[:self.frame_size] self.out_q.put(frame)几个参数和设计点值得说明。timeout=0.5让read不会永久阻塞,线程才能响应stop(),否则退出程序时窗口关了线程还挂着。ser.in_waiting or 1是常见的省事写法:缓冲区里有数据就一次读完,没有就阻塞读 1 字节,避免空转占满 CPU。daemon=True保证主程序退出时不留孤儿线程,但对写库线程不能这么干,那个必须优雅停止并 commit,不然最后一批数据会丢。
真正容易出问题的是_buf的清理时机。噪声大的现场(比如探头线跟变频器走一个线槽),帧头之前可能堆着几百字节的垃圾,必须在find之后立即删掉,否则内存只涨不降。另外,如果下位机发送的帧长不固定,这里就要改成先读长度字节再判断,固定帧长是最省事的形态。
3.2 帧解析与校验和
拿到完整帧之后才做校验,顺序不能反。先验帧头帧尾,再验校验和,最后才解析字段——校验不过的帧直接返回None,绝不进库。
import struct def calc_checksum(frame: bytes) -> int: """第 2 到 16 字节累加,取低 8 位""" return sum(frame[2:17]) & 0xFF def parse_frame(frame: bytes) -> dict | None: if len(frame) != 19 or frame[0] != 0xAA or frame[1] != 0x55: return None if frame[18] != 0x0D or calc_checksum(frame) != frame[17]: return None # <IhhhH 共 12 字节:时间戳4 + pH2 + 浊度2 + TDS2 + 温度2 ts, ph, ntu, tds, temp = struct.unpack("<IhhhH", frame[4:16]) return { "device_id": f"{frame[2]:02X}", "ts": ts, "ph": round(ph / 100.0, 2), "turbidity": round(ntu / 10.0, 1), "tds": float(tds), "temp": round(temp / 100.0, 2), "status": frame[16], }struct的格式串是小端解析的核心。<表示小端,I是无符号 4 字节,h是有符号 2 字节,H是无符号 2 字节。pH 用有符号是因为有些探头在校准时会输出负值,用无符号就直接变成 65000 多了。缩放系数(÷100、÷10)写在解析层而不是界面层,这样库里存的永远是物理量,导出报表时不用再换算一遍。
校验和用累加而不是 CRC16,是这类小数据包常见的取舍:出错概率低、计算量小,下位机几个时钟周期就算完。但如果现场干扰特别强,累加校验会漏掉某些双比特翻转,那就得换 CRC16,客户端和服务端一起改。
字段返回dict而不是自定义类,是为了后面写库和画图都能直接按键取值,也方便 JSON 序列化。状态位单独留着,电极异常的时候可以在界面上直接标红,不用等到读数超限才发现探头已经没浸在水里了。
3.3 建表与批量入库
每条采样单独 commit 是最常见的性能坑,一天下来几万次磁盘同步,树莓派上能把 SD 卡写废。正确做法是攒一批再写。
CREATE TABLE IF NOT EXISTS sample ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, -- 设备或采集时间,秒级 device_id TEXT NOT NULL, ph REAL, turbidity REAL, tds REAL, temp REAL, status INTEGER NOT NULL DEFAULT 0, level INTEGER NOT NULL DEFAULT 0 -- 0 正常 1 预警 2 报警 ); CREATE INDEX IF NOT EXISTS idx_sample_ts ON sample (ts); CREATE INDEX IF NOT EXISTS idx_sample_dev_ts ON sample (device_id, ts);import queue import sqlite3 import threading DDL = open("core/schema.sql", encoding="utf-8").read() class SampleWriter(threading.Thread): """单写线程独占一个连接,批量提交""" def __init__(self, db_path: str, in_q: "queue.Queue[dict]", batch: int = 20, flush_sec: float = 1.0): super().__init__(daemon=False, name="db-writer") self.db_path = db_path self.in_q = in_q self.batch = batch self.flush_sec = flush_sec self._stop = threading.Event() def stop(self): self._stop.set() def run(self): conn = sqlite3.connect(self.db_path) conn.executescript(DDL) conn.execute("PRAGMA journal_mode=WAL") # 读写并发不互相阻塞 conn.execute("PRAGMA synchronous=NORMAL") # 掉电最多丢最后一批,速度翻倍 buf = [] while not self._stop.is_set() or buf: try: buf.append(self.in_q.get(timeout=self.flush_sec)) except queue.Empty: pass if len(buf) >= self.batch or (buf and self.in_q.empty()): conn.executemany( "INSERT INTO sample (ts, device_id, ph, turbidity, tds, temp, status, level)" " VALUES (?,?,?,?,?,?,?,?)", [(r["ts"], r["device_id"], r["ph"], r["turbidity"], r["tds"], r["temp"], r["status"], r["level"]) for r in buf], ) conn.commit() buf.clear() conn.close()WAL模式是 SQLite 在「一边写一边查」场景下的关键开关,它让读操作不被写事务阻塞,历史查询界面刷新时不会卡住采集。synchronous=NORMAL是拿一点掉电安全性换吞吐,对水质监测这种数据来说可以接受——丢最后一两秒的记录,比卡顿一整晚强。
daemon=False配合stop()是故意的:程序退出时要先停采集线程、再停写库线程,让缓冲区里的数据落盘。顺序反了,最后一批就永远留在内存里了。批量大小 20 条、1 秒兜底,是按常见 1 秒一条的采样率定的,采样更快就把batch调大。
3.4 阈值判定与去抖
告警不能一超限就弹窗,模拟量探头本身就有几百分之几的抖动,pH 在 6.5 边界上下飘的时候,一晚上能给你弹一百次。加个连续计数去抖,问题基本消失。
# config.yaml 里对应的规则 # rules: # ph: [6.5, 8.5] # turbidity: [0, 1.0] # tds: [0, 1000] # temp: [0, 40] def judge(sample: dict, rules: dict, state: dict, need: int = 3) -> int: """连续 need 次越界才升级,任一次回正常立即复位""" hit = 0 for name, (lo, hi) in rules.items(): v = sample.get(name) if v is None: continue if v < lo or v > hi: state[name] = state.get(name, 0) + 1 else: state[name] = 0 if state[name] >= need: hit += 1 if hit == 0: return 0 return 1 if hit == 1 else 2state是一个跨调用保留的字典,每个参数一个计数器,这是去抖的全部秘密。need=3意味着连续 3 个采样点都越界才报,按 1 秒一条算就是 3 秒延迟,对水质这种变化缓慢的对象完全够用。如果某些参数要求更灵敏(比如余氯),单独给它配need=1就行,把规则从二元组改成三元组即可。
判定结果写回level字段,同时推一条消息给界面。这里不要再加一层「告警处理器类」,两三个函数就够,过度设计在这个体量的项目里只会让后来接手的人更懵。
4. 界面与可视化:实时曲线、历史回溯与指标限值表
界面这一层的目标很明确:让值班的人扫一眼就知道现在正不正常,点两下就能查到上周三下午的数据。花哨的仪表盘不如一条清晰的曲线加一个会变色的状态灯。
4.1 实时曲线的刷新方式
曲线最容易踩的坑是把刷新逻辑写进采集线程里,跨线程直接操作控件,跑着跑着就崩。稳妥的做法是界面用QTimer定时去队列里取数据,采集那边完全不知道界面的存在。
from collections import deque import pyqtgraph as pg from PyQt5 import QtCore, QtWidgets class PhChart(pg.PlotWidget): def __init__(self, window: int = 600, parent=None): super().__init__(parent) self.setLabel("left", "pH") self.setLabel("bottom", "采样点") self.setYRange(0, 14) self.showGrid(x=True, y=True, alpha=0.3) self.curve = self.plot(pen=pg.mkPen("#2e86de", width=2)) self.buf = deque(maxlen=window) # 定长,内存不会一直涨 def push(self, value: float): self.buf.append(value) self.curve.setData(list(self.buf)) class MainWindow(QtWidgets.QMainWindow): def __init__(self, data_q, interval_ms: int = 500): super().__init__() self.data_q = data_q self.chart = PhChart() self.setCentralWidget(self.chart) self.timer = QtCore.QTimer(self) self.timer.timeout.connect(self.drain) self.timer.start(interval_ms) # 界面 2Hz 刷新,采集再快也不影响 def drain(self): latest = None while True: try: latest = self.data_q.get_nowait() except Exception: break if latest is not None: self.chart.push(latest["ph"])deque(maxlen=600)是定长队列,超出自动丢最老的,跑一个月内存也不会涨。界面刷新频率(500ms)和采样频率(1s)解耦,是这套结构最实在的好处:采样调快了界面不卡,采样断了界面也不崩,只是曲线不动。
drain里那个while True是故意把队列抽干,只显示最后一条。如果只取一条,队列会越积越多,最后内存爆掉。判断队列积压还有个土办法:在状态栏显示队列长度,超过 100 就说明写库或解析跟不上了,得查。
Web 版本的对应位置是前端setInterval定时拉/api/latest,后端从同一个队列里取,逻辑完全一样,只是把QTimer换成了浏览器定时器。
4.2 历史查询与报表导出
历史页面的核心是一条带时间范围条件的查询。索引建在ts上,几百万行也是毫秒级。
import pandas as pd def query_range(conn, start_ts: int, end_ts: int, device_id: str | None = None): sql = ("SELECT ts, device_id, ph, turbidity, tds, temp, level " "FROM sample WHERE ts BETWEEN ? AND ?") args = [start_ts, end_ts] if device_id: sql += " AND device_id = ?" args.append(device_id) sql += " ORDER BY ts ASC" return pd.read_sql_query(sql, conn, params=args) def export_csv(df: pd.DataFrame, path: str): out = df.copy() out["时间"] = pd.to_datetime(out["ts"], unit="s", utc=True).dt.tz_convert("Asia/Shanghai") out = out.drop(columns=["ts"]) out.to_csv(path, index=False, encoding="utf-8-sig") # 带 BOM,Excel 打开不乱码utf-8-sig这个细节坑过不少人:不加 BOM,Excel 双击打开中文列名就是乱码,现场的人会以为数据坏了。utc=True再tz_convert是为了处理跨时区的情况,单站点可以简化成pd.to_datetime(df["ts"], unit="s")。
查询结果直接给pandas是为了导出和统计(日均值、最大值)都方便,但如果你的环境里不想装 pandas,用csv模块手写循环也完全可行,就是代码长一点。
4.3 参数与限值对照表
界面上的颜色和报警都依赖这张表,改阈值只改配置,不要散落在代码里:
| 参数 | 常见参考限值 | 单位 | 超限最常见的原因 |
|---|---|---|---|
| pH | 6.5 – 8.5 | 无量纲 | 电极未校准、参比液干涸 |
| 浊度 | ≤ 1.0 | NTU | 探头窗口结垢、气泡附着 |
| TDS | ≤ 1000 | mg/L | 探头常数设置错、水温补偿缺失 |
| 温度 | 0 – 40 | ℃ | 探头脱落、线缆断芯 |
| 电极状态位 | 应为 0 | — | 校准过期、信号超量程 |
注意:表里的限值只是常见参考区间,不同用途的水标准不一样,正式使用前请按你所在场景的实际要求核对,别直接照抄。
5. 常见问题排查:串口、毛刺、时间戳与打包的五个坑
这套东西跑通不难,难的是跑稳。下面五条基本覆盖了我在类似项目里遇到的大半问题,按「现象 → 原因 → 解决」写,照着对号入座就行。
5.1 从零启动的四步
先把环境跑起来再谈别的。没硬件的时候用包里的fake_source.py造数据,能省掉一大半联调时间。
# 1. 建虚拟环境,别用系统 Python 直接装 python -m venv .venv source .venv/bin/activate # Windows 下是 .venv\Scripts\activate # 2. 装依赖 pip install -r requirements.txt # 3. 没有硬件:起一个模拟数据源,往虚拟串口里灌帧 # Windows 用 com0com 建一对虚拟串口,Linux 用 socat socat -d -d pty,raw,echo=0 pty,raw,echo=0 python tools/fake_source.py --port /dev/pts/3 --interval 1.0 # 4. 把主程序指向另一端 python main.py --config config.yaml模拟数据源的核心就是按 3.2 节的帧格式反向打包,造出带毛刺、带偶发超限的数据,用来验证去抖和告警是否生效:
import random import struct import time import serial def build_frame(ph: float, ntu: float, tds: float, temp: float, addr: int = 0x01) -> bytes: body = struct.pack("<IhhhH", int(time.time()), int(ph * 100), int(ntu * 10), int(tds), int(temp * 100)) frame = b"\xAA\x55" + bytes([addr, 0x01]) + body + b"\x00" assert len(frame) == 17, len(frame) return frame + bytes([sum(frame[2:17]) & 0xFF, 0x0D]) if __name__ == "__main__": ser = serial.Serial("/dev/pts/4", 9600) while True: # 在 7.0 附近加噪声,偶尔串到 9.0 触发一次告警 ph = 7.0 + random.gauss(0, 0.15) + (2.0 if random.random() < 0.02 else 0) ser.write(build_frame(ph, 0.3 + random.random() * 0.5, 220, 24.5)) time.sleep(1)assert len(frame) == 17这行别删,改协议的时候它是第一道防线,帧长算错了立刻报错,比上位机那边收到一堆解析失败的None好查得多。
5.2 五个高频坑的定位与处理
坑一:串口打不开,提示 Permission denied 或拒绝访问。现象是程序一启动就退出,日志里只有一句SerialException。原因是 Linux 下当前用户不在dialout组,或者 Windows 下串口被另一个软件(串口助手、厂商的上位机)占着。解决:Linux 执行sudo usermod -aG dialout $USER后重新登录;Windows 先关掉其他串口工具,设备管理器里看看是不是被识别成了两个 COM 口(有些 CH340 驱动会这样)。这个坑属于玄学范畴,插不同的 USB 口表现还不一样,建议在_open失败时把可用串口列表打出来:
from serial.tools import list_ports print([p.device for p in list_ports.comports()])坑二:曲线毛刺,pH 时不时跳到 1 或者 13。现象是数据里混着明显不合理的尖峰,一天能有几十个。原因是模拟量探头共地没做好、信号线跟电源线捆在一起,或者电极表面结垢导致接触电阻变化。解决:硬件上把探头地和主板地单点连接、信号线用屏蔽线且屏蔽层单端接地;软件上在解析层之后加一个中值滤波,窗口取 5,能滤掉绝大部分孤立尖峰,比滑动平均更保边。别用平均,平均会把真实的突变抹平,反而漏掉污染事件。
坑三:时间戳错乱,历史曲线的时间轴对不上。现象是查昨天下午的数据,结果显示在凌晨,或者两个探头的记录差了 8 小时。原因分两种:一是ts用的是设备时间,而设备没有 RTC、断电重启后从 0 开始;二是存的是 UTC 秒,展示时忘了做时区转换。解决:入库前统一用采集端时间覆盖设备时间(前提是采集机时间准,建议开 NTP),并在字段名上标明ts_utc;展示层统一在一处做转换,别每个界面各转一次,那种写法必出不一致。
坑四:打包成 exe 之后一启动就闪退。现象是源码跑得好好的,PyInstaller打完双击没反应。原因十有八九是相对路径:源码里的open("config.yaml")在 PyInstaller 解压的临时目录里找不到文件。解决:所有资源路径走统一入口——
import sys from pathlib import Path def res_path(name: str) -> Path: base = Path(getattr(sys, "_MEIPASS", Path(__file__).resolve().parent)) return base / name def data_path(name: str) -> Path: """可写数据放 exe 同级的 data 目录,不要放 _MEIPASS,那是临时的""" base = Path(sys.executable).parent if getattr(sys, "frozen", False) \ else Path(__file__).resolve().parent p = base / "data" / name p.parent.mkdir(parents=True, exist_ok=True) return p配置文件和图标用res_path,数据库和导出文件用data_path。这两个混用,就会出现「配置读到了、数据库写完重启就没了」的诡异现象。
坑五:跑一段时间报 database is locked。现象是采集正常,但历史查询偶尔报错,或者写库线程卡住。原因是多个线程共用一个sqlite3连接,或者事务没提交就开了新事务。解决:写库线程独占一个连接(3.3 节就是这么做的),读操作各自开连接或者走连接池;必须共用时加check_same_thread=False并配一把锁,但更推荐直接按线程分开。另外记得开 WAL,很多人以为开了没用,其实是开在了错误的连接上——PRAGMA是连接级的,每个新连接都要执行一次。
6. 二次开发:给系统加一路新参数,从协议表改到界面
最常见的改造需求就是加探头,比如再上一路余氯或者溶解氧。如果按前面的写法硬改,你会同时动到解析函数、建表语句、插入语句、曲线控件、导出列五六个地方,改漏一个就出问题。我一般会先把参数抽成一张注册表:
# core/params.py PARAMS = { "ph": {"offset": 8, "fmt": "h", "scale": 0.01, "unit": "", "range": (0, 14)}, "turbidity": {"offset": 10, "fmt": "h", "scale": 0.1, "unit": "NTU", "range": (0, 1000)}, "tds": {"offset": 12, "fmt": "h", "scale": 1.0, "unit": "mg/L", "range": (0, 2000)}, "temp": {"offset": 14, "fmt": "H", "scale": 0.01, "unit": "℃", "range": (0, 60)}, # 新增一路余氯:改下位机帧长,这里加一行,其他地方基本不用动 # "chlorine": {"offset": 16, "fmt": "H", "scale": 0.001, "unit": "mg/L", "range": (0, 5)}, }解析函数改成遍历注册表,struct.unpack_from按偏移取值,加参数时帧长和校验范围跟着调一下就行。建表语句用" , ".join(f"{k} REAL" for k in PARAMS)动态拼,入库的INSERT同理用参数名列表拼占位符。前端按PARAMS的 key 循环生成曲线控件和 Y 轴标签,加一路就自动多一条线。
改完之后的验证顺序我固定成三步:先跑fake_source.py造新参数的数据,确认解析出来的值和造的值对得上(差一个量级就是scale写错了);再开历史页面确认新列进了表和导出文件;最后拔一次串口,看重连之后新参数是否还能继续写入。第三步最容易忘,也最容易出问题——很多人的新参数只在启动时注册了一次,重连后初始化没走,字段就一直是空的。
从那以后我每接一个新探头,都先把它的量程、单位、零点偏移记进params.py再动别的地方,绝不先在界面里手写一遍曲线。这个顺序反过来,血泪经验就是:界面能画出来但库里没存,等要出报表的时候才发现历史数据缺一路,那时候只能重采。希望帮到你。
本文还有配套的精品资源,点击获取