Python新手求助：从RS232读取的秤数据中提取PCS值存入MySQL

一、数据解析：用正则提取PCS数值 #

数组索引提取依赖固定格式，一旦秤的输出格式微调就会失效，用正则表达式更可靠。假设你的串口输出类似[PCS] 5 [WEIGHT] 10.2这类格式，直接匹配[PCS]后面的数字即可：

import re

def extract_pcs(data):
    # 匹配[PCS]后面的数字，支持多位数
    match = re.search(r'\[PCS\]\s*(\d+)', data)
    if match:
        return int(match.group(1))
    return None

测试示例：

test_data = "ABC [PCS] 123 XYZ [WEIGHT] 45.6"
pcs_num = extract_pcs(test_data)
print(pcs_num)  # 输出123

二、MySQL数据库操作 #

先安装数据库驱动：

pip install mysql-connector-python

编写数据库插入函数：

import mysql.connector
from mysql.connector import Error

def insert_pcs_to_db(pcs_value):
    try:
        # 替换为你的数据库信息
        connection = mysql.connector.connect(
            host='localhost',
            database='your_database',
            user='your_username',
            password='your_password'
        )
        if connection.is_connected():
            cursor = connection.cursor()
            # 假设你有scale_data表，pcs字段存件数，可按需调整
            query = "INSERT INTO scale_data (pcs) VALUES (%s)"
            cursor.execute(query, (pcs_value,))
            connection.commit()
            print(f"成功插入PCS数值：{pcs_value}")
    except Error as e:
        print(f"数据库操作错误：{e}")
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()

三、整合串口读取与处理 #

假设你用pyserial读取串口，整合后的完整代码：

import serial
import re
import mysql.connector
from mysql.connector import Error

def extract_pcs(data):
    match = re.search(r'\[PCS\]\s*(\d+)', data)
    return int(match.group(1)) if match else None

def insert_pcs_to_db(pcs_value):
    try:
        connection = mysql.connector.connect(
            host='localhost',
            database='your_database',
            user='your_username',
            password='your_password'
        )
        if connection.is_connected():
            cursor = connection.cursor()
            query = "INSERT INTO scale_data (pcs) VALUES (%s)"
            cursor.execute(query, (pcs_value,))
            connection.commit()
    except Error as e:
        print(f"数据库错误：{e}")
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()

def main():
    # 替换为你的串口参数：端口、波特率等
    ser = serial.Serial('COM3', 9600, timeout=1)
    try:
        while True:
            # 读取串口数据，编码需和秤的输出匹配，比如gbk/utf-8
            data = ser.readline().decode('utf-8').strip()
            if data:
                print(f"原始数据：{data}")
                pcs_num = extract_pcs(data)
                if pcs_num is not None:
                    insert_pcs_to_db(pcs_num)
                else:
                    print("未找到PCS数值")
    except KeyboardInterrupt:
        print("程序终止")
    finally:
        ser.close()

if __name__ == "__main__":
    main()

注意事项 #

• 串口参数（端口号、波特率、编码）要和秤的设置一致，若读取乱码，可尝试将decode('utf-8')改为decode('gbk')。
• 提前创建数据库表，示例SQL：

CREATE TABLE scale_data (
    id INT AUTO_INCREMENT PRIMARY KEY,
    pcs INT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

• 若秤持续发送重复数据，可添加数值变化判断逻辑，避免重复插入。

内容的提问来源于stack exchange，提问作者JDD