串口温湿度采集上位机 python

### Python串口温湿度数据采集及上位机开发

#### 项目概述
为了实现从单片机串口接收温度和湿度并将这些数据存储到数据库中，同时通过图形界面展示出来，可以采用如下技术栈：

- **操作系统**: Linux系统[^1]
- **编程语言**: Python
- **集成开发环境(IDE)**: PyCharm
- **主要使用的Python库**:
- `pymysql`用于连接MySQL数据库并执行SQL查询操作。
- `matplotlib`负责绘制动态更新的折线图来直观呈现历史记录中的变化趋势。
- `pyserial`处理与硬件设备之间的RS232/USB接口通讯。

#### 示例代码结构

##### 导入库文件
首先安装所需的第三方模块：

pip install pyserial pymysql matplotlib

接着导入必要的包：

import serial
import time
from datetime import datetime
import pymysql
import matplotlib.pyplot as plt
plt.switch_backend('TkAgg')  # 设置Matplotlib后端以便于嵌入GUI应用

##### 初始化配置参数
定义一些全局变量用来保存设置好的串口号、波特率以及数据库连接信息等：

SERIAL_PORT = '/dev/ttyUSB0'    # 根据实际情况修改此路径
BAUD_RATE = 9600               # 波特率需匹配下位机设定值
DATABASE_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',            # 用户名密码请按自己情况调整
    'password': '',
    'database': 'sensor_data'
}
TABLE_NAME = 'temperature_humidity_records'

# 创建表格语句（仅当表不存在时创建）
CREATE_TABLE_SQL = f"""
CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP(),
temperature FLOAT COMMENT '摄氏度单位',
humidity FLOAT COMMENT '百分比形式表示相对湿度');
"""

INSERT_RECORD_SQL = f"INSERT INTO `{TABLE_NAME}` (temperature,humidity) VALUES (%s,%s)"

##### 建立数据库链接函数
封装一个简单的工具类来进行增删改查的操作：

class DatabaseHelper(object):

    def __init__(self, config):
        self.connection = None
        try:
            self.connection = pymysql.connect(**config)
            cursor = self.connection.cursor()
            cursor.execute(CREATE_TABLE_SQL)
            self.connection.commit()
        except Exception as e:
            print(f'Database connection failed due to {e}')

    def insert(self, temp_value, humi_value):
        with self.connection.cursor() as cur:
            affected_rows = cur.execute(
                INSERT_RECORD_SQL, args=(temp_value, humi_value))
            if not affected_rows:
                raise RuntimeError("Failed inserting record.")
        self.connection.commit()

    def close(self):
        if self.connection is not None and self.connection.open:
            self.connection.close()


db_helper = DatabaseHelper(DATABASE_CONFIG)


def cleanup():
    global db_helper
    if isinstance(db_helper, DatabaseHelper):
        db_helper.close()

##### 处理接收到的数据流
编写逻辑解析来自串行端口的消息，并将其转换成浮点数类型的数值对儿；然后调用上述方法完成持久化工作：

ser = serial.Serial(SERIAL_PORT, BAUD_RATE)

while True:
    line = ser.readline().decode('ascii').strip('\r\n')
    parts = line.split(',')
    
    if len(parts)==2 :
        try:
            temperature=float(parts[0])
            humidity=float(parts[1])

            now=datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print(f'{now}: T={temperature:.1f}, H={humidity:.1f}')
            
            db_helper.insert(temperature=temperature, humi_value=humidity)

            # 更新图表...
        
        except ValueError as ve:
            print(f"Parsing error occurred at {datetime.now()}, detail:{ve}")

##### 绘制实时曲线图
最后一步就是利用`matplotlib.animation.FuncAnimation()`定时刷新图像窗口内的内容，从而达到近似“实况转播”的效果：

fig, ax = plt.subplots(figsize=(8, 6))

xdata, y_temp, y_humi=[], [], []

line_t, =ax.plot([],[],label='Temperature')
line_h, =ax.plot([],[],label='Humidity')

ax.set_xlim(-5, 5)
ax.legend(loc="upper left")

def init_plot():
    """初始化绘图区域."""
    ax.set_ylim(0, 100)
    return line_t,line_h,

def update(frame):
    """获取最新一条记录作为新坐标加入列表末端."""
    last_record=db_helper.query_lastest_one()[0]

    current_time=str(datetime.fromisoformat(last_record['timestamp']))
    xdata.append(current_time[-8:])     # 只保留时间部分HH:mm:ss
    
    new_yt,new_yh=[float(v)for v in[last_record[col]for col in ['temperature','humidity']]]
    y_temp.append(new_yt);y_humi.append(new_yh)

    min_x,max_x=min(xdata),max(xdata)
    ax.set_xlim(min_x,max_x)

    line_t.set_data(xdata,y_temp)
    line_h.set_data(xdata,y_humi)

ani = animation.FuncAnimation(fig,update,interval=1000*60,blit=True,repeat=False,frames=None,init_func=init_plot)

try:
    plt.show(block=True)
finally:
    cleanup()