pyhton自动化采集数据脚本-EW帮帮网

点关注不迷路哟。你的点赞、收藏，一键三连，是我持续更新的动力哟！！！

持续关注我~~~主页，查看更多内容哟（希望你能在这里有所收获🤭）。点关注，不迷路，哈哈哈！~~~

一位搞嵌入式的 genius-CSDN博客https://blog.csdn.net/m0_73589512?spm=1000.2115.3001.5343你是否曾因物联网设备传来的串口数据杂乱无章而束手无策？是否为如何高效存储传感器数据到 MySQL 数据库而发愁？本文拆解的parse_serial_data与DatabaseManager代码，就像两把精准的瑞士军刀 —— 前者能将串口原始数据瞬间解码为规范格式，后者则轻松搭建起与 MySQL 的桥梁，实现数据的稳定入库。无论是开发智能照明系统，还是搭建环境监测平台，掌握这些代码，就能解锁数据从采集到存储的全链路高效处理！

pyhton自动化采集数据脚本：

1.引入包

# serial_to_mysql.py
import serial
import pymysql
from datetime import datetime
import time
import logging
from typing import Optional
import re

使用以下 pip 命令安装：

pip install pyserial pymysql python-dotenv

各依赖说明：

pyserial - 用于串口通信（对应代码中的serial）
pymysql - 用于 MySQL 数据库连接（对应pymysql）
python-dotenv - 用于加载环境变量（对应dotenv）

2.数据库配置

CONFIG = {
    "serial": {
        "port": "COM5",
        "baudrate": 115200,
        "timeout": 1
    },
    "database": {
        "host": "localhost",
        "user": "root",
        "password": "对应的root用户密码",
        "db": "数据库名",
        "charset": "utf8mb4"
    },
    "device_id": "STM32_001"  # 设备标识
}

pyserial - 用于串口通信（对应代码中的serial）

这里面就需要指定：串口号和波特率以及相应的请求时间

database -用于指定相关的数据库信息

3.MySQL数据库管理类

class DatabaseManager:
    """MySQL数据库管理类"""
    def __init__(self):
        self.connection = None
        #初始化数据库
    def connect(self):
        try:
            self.connection = pymysql.connect(**CONFIG["database"])
            self._initialize_database()
            logging.info("数据库连接成功")
            return True
        except pymysql.Error as e:
            logging.error(f"数据库连接失败: {e}")
            return False
    connect() 方法：建立数据库连接
    使用 pymysql.connect() 连接数据库，参数从 CONFIG["database"] 字典解包
    调用私有方法 _initialize_database() 创建数据表（如果不存在）
    连接成功返回 True，失败则记录错误并返回 False
    
    def _initialize_database(self):
        """初始化数据表（如果不存在）"""
        create_table_sql = """
        CREATE TABLE IF NOT EXISTS light_sensor (
            id INT AUTO_INCREMENT PRIMARY KEY,
            device_id VARCHAR(50) NOT NULL,
            light_value FLOAT NOT NULL,
            timestamp DATETIME NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            INDEX (device_id),
            INDEX (timestamp)
        )
        """
        with self.connection.cursor() as cursor:
            cursor.execute(create_table_sql)
        self.connection.commit()
    
    def insert_data(self, device_id: str, light_value: float, timestamp: datetime) -> bool:
        """插入传感器数据"""
        try:
            with self.connection.cursor() as cursor:
                sql = """
                INSERT INTO light_sensor 
                (device_id, light_value, timestamp) 
                VALUES (%s, %s, %s)
                """
                cursor.execute(sql, (device_id, light_value, timestamp))
            self.connection.commit()
            return True
        except pymysql.Error as e:
            logging.error(f"数据插入失败: {e}")
            return False
    
    def close(self):
        if self.connection:
            self.connection.close()
            logging.info("数据库连接已关闭")

这个类提供了完整的数据库操作封装：

连接管理：建立和关闭数据库连接
表结构管理：自动创建数据表
数据操作：插入传感器数据记录
错误处理：捕获数据库操作异常并记录日志
资源管理：确保连接正确关闭

4.解析串口数据

def parse_serial_data(line: str) -> Optional[tuple]:
    """
    解析串口数据
    期望格式: [时间戳] Actual light intensity: 数值% (如 "[1107000] Actual light intensity: 50%")
    """
    try:
        # 使用正则表达式匹配时间戳和光照值
        match = re.search(r'\[(\d+)\]\s+Actual\s+light\s+intensity:\s+(\d+\.?\d*)%', line)
        if not match:
            raise ValueError("数据格式不匹配")
            
        timestamp_ms = match.group(1)
        light_value = float(match.group(2))
        
        # 将毫秒时间戳转换为datetime对象
        timestamp = datetime.fromtimestamp(int(timestamp_ms) / 1000)
        
        return timestamp, light_value
        
    except (ValueError, IndexError) as e:
        logging.warning(f"数据解析失败: {e} | 原始数据: {line}")
        return None

功能：解析串口传输的光照强度数据行
参数：line（字符串类型），表示从串口读取的一行数据
返回值：成功时返回包含 (datetime对象, 光照值) 的元组，失败时返回 None
正则表达式解析：
- \[(\d+)\]：匹配方括号内的数字时间戳（如 [1107000]）
- \s+Actual\s+light\s+intensity:：匹配固定文本部分
- (\d+\.?\d*)%：匹配浮点数格式的光照值（如 50% 或 50.5%）
提取匹配结果：
- match.group(1)：获取第一个捕获组（时间戳字符串）
- match.group(2)：获取第二个捕获组（光照值字符串），并转换为浮点数
时间戳转换
```
    # 将毫秒时间戳转换为datetime对象
    timestamp = datetime.fromtimestamp(int(timestamp_ms) / 1000)
    
    return timestamp, light_value
```
1. 时间戳转换逻辑：
  - int(timestamp_ms)：将时间戳字符串转为整数（单位：毫秒）
  - / 1000：将毫秒转换为秒
  - datetime.fromtimestamp()：将 Unix 时间戳转换为本地 datetime 对象
2. 返回结果：
  - 返回包含 (datetime对象, 光照值) 的元组

5.主函数

def main():
    # 初始化日志
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('sensor.log'),
            logging.StreamHandler()
        ]
    )
    
    db = DatabaseManager()
    ser = None
    
    try:
        # 初始化数据库连接
        if not db.connect():
            return
            
        # 初始化串口
        ser = serial.Serial(**CONFIG["serial"])
        logging.info(f"串口监听启动: {CONFIG['serial']['port']}")
        
        # 主循环
        while True:
            try:
                line = ser.readline().decode('utf-8')
                if not line.strip():
                    continue
                    
                # 解析数据
                parsed_data = parse_serial_data(line)
                if not parsed_data:
                    continue
                    
                timestamp, light_value = parsed_data
                
                # 存储到数据库
                success = db.insert_data(
                    device_id=CONFIG["device_id"],
                    light_value=light_value,
                    timestamp=timestamp
                )
                
                if success:
                    logging.info(f"数据存储成功: {timestamp} | {light_value} lux")
                else:
                    logging.warning("数据存储失败")

            except ValueError as e:
                logging.error(f"数据处理错误: {e}")
            except UnicodeDecodeError:
                logging.warning("串口数据解码错误")
            except Exception as e:
                logging.error(f"意外错误: {e}")
                time.sleep(1)  # 防止错误循环

    except KeyboardInterrupt:
        logging.info("用户终止程序")
    finally:
        if ser and ser.is_open:
            ser.close()
            logging.info("串口已关闭")
        db.close()

if __name__ == "__main__":
    main()

6.实验效果

PS C:\Users\Lenovo\Desktop\Meteorological_monitoring\Python脚本> python .\serial_to_springboot.py
2025-06-26 09:57:38,098 - INFO - 数据库连接成功
2025-06-26 09:57:38,123 - INFO - 串口监听启动: COM5
2025-06-26 09:57:38,285 - INFO - 数据存储成功: 1970-01-01 08:22:19 | 45.0 lux
2025-06-26 09:57:39,293 - INFO - 数据存储成功: 1970-01-01 08:22:20 | 43.0 lux
2025-06-26 09:57:40,284 - INFO - 数据存储成功: 1970-01-01 08:22:21 | 44.0 lux
2025-06-26 09:57:41,284 - INFO - 数据存储成功: 1970-01-01 08:22:22 | 47.0 lux
2025-06-26 09:57:42,285 - INFO - 数据存储成功: 1970-01-01 08:22:23 | 49.0 lux
2025-06-26 09:57:43,293 - INFO - 数据存储成功: 1970-01-01 08:22:24 | 49.0 lux
2025-06-26 09:57:44,284 - INFO - 数据存储成功: 1970-01-01 08:22:25 | 50.0 lux
2025-06-26 09:57:45,284 - INFO - 数据存储成功: 1970-01-01 08:22:26 | 50.0 lux
2025-06-26 09:57:46,284 - INFO - 数据存储成功: 1970-01-01 08:22:27 | 50.0 lux
2025-06-26 09:57:47,284 - INFO - 数据存储成功: 1970-01-01 08:22:28 | 51.0 lux
2025-06-26 09:57:48,291 - INFO - 数据存储成功: 1970-01-01 08:22:29 | 50.0 lux
2025-06-26 09:57:49,292 - INFO - 数据存储成功: 1970-01-01 08:22:30 | 47.0 lux
2025-06-26 09:57:50,293 - INFO - 数据存储成功: 1970-01-01 08:22:31 | 45.0 lux
2025-06-26 09:57:51,293 - INFO - 数据存储成功: 1970-01-01 08:22:32 | 45.0 lux
2025-06-26 09:57:52,284 - INFO - 数据存储成功: 1970-01-01 08:22:33 | 45.0 lux
2025-06-26 09:57:53,293 - INFO - 数据存储成功: 1970-01-01 08:22:34 | 44.0 lux
2025-06-26 09:57:54,293 - INFO - 数据存储成功: 1970-01-01 08:22:35 | 43.0 lux
2025-06-26 09:57:55,285 - INFO - 数据存储成功: 1970-01-01 08:22:36 | 48.0 lux
2025-06-26 09:57:56,293 - INFO - 数据存储成功: 1970-01-01 08:22:37 | 49.0 lux
2025-06-26 09:57:57,284 - INFO - 数据存储成功: 1970-01-01 08:22:38 | 50.0 lux
2025-06-26 09:57:58,284 - INFO - 数据存储成功: 1970-01-01 08:22:39 | 51.0 lux
2025-06-26 09:57:59,284 - INFO - 数据存储成功: 1970-01-01 08:22:40 | 51.0 lux
2025-06-26 09:58:00,293 - INFO - 数据存储成功: 1970-01-01 08:22:41 | 51.0 lux
2025-06-26 09:58:01,293 - INFO - 数据存储成功: 1970-01-01 08:22:42 | 48.0 lux
2025-06-26 09:58:02,284 - INFO - 数据存储成功: 1970-01-01 08:22:43 | 45.0 lux
2025-06-26 09:58:03,293 - INFO - 数据存储成功: 1970-01-01 08:22:44 | 48.0 lux
2025-06-26 09:58:04,292 - INFO - 数据存储成功: 1970-01-01 08:22:45 | 47.0 lux
2025-06-26 09:58:05,293 - INFO - 数据存储成功: 1970-01-01 08:22:46 | 45.0 lux
2025-06-26 09:58:06,285 - INFO - 数据存储成功: 1970-01-01 08:22:47 | 43.0 lux
2025-06-26 09:58:07,283 - INFO - 数据存储成功: 1970-01-01 08:22:48 | 47.0 lux
2025-06-26 09:58:08,293 - INFO - 数据存储成功: 1970-01-01 08:22:49 | 48.0 lux
2025-06-26 09:58:09,293 - INFO - 数据存储成功: 1970-01-01 08:22:50 | 49.0 lux
2025-06-26 09:58:10,284 - INFO - 数据存储成功: 1970-01-01 08:22:51 | 49.0 lux
2025-06-26 09:58:11,284 - INFO - 数据存储成功: 1970-01-01 08:22:52 | 50.0 lux
2025-06-26 09:58:12,283 - INFO - 数据存储成功: 1970-01-01 08:22:53 | 51.0 lux
2025-06-26 09:58:13,293 - INFO - 数据存储成功: 1970-01-01 08:22:54 | 51.0 lux
2025-06-26 09:58:14,283 - INFO - 数据存储成功: 1970-01-01 08:22:55 | 50.0 lux
2025-06-26 09:58:15,293 - INFO - 数据存储成功: 1970-01-01 08:22:56 | 46.0 lux
2025-06-26 09:58:16,293 - INFO - 数据存储成功: 1970-01-01 08:22:57 | 44.0 lux
2025-06-26 09:58:17,290 - INFO - 数据存储成功: 1970-01-01 08:22:58 | 47.0 lux
2025-06-26 09:58:18,286 - INFO - 数据存储成功: 1970-01-01 08:22:59 | 45.0 lux
2025-06-26 09:58:19,294 - INFO - 数据存储成功: 1970-01-01 08:23:00 | 44.0 lux
2025-06-26 09:58:20,292 - INFO - 数据存储成功: 1970-01-01 08:23:01 | 46.0 lux
2025-06-26 09:58:21,293 - INFO - 数据存储成功: 1970-01-01 08:23:02 | 49.0 lux
2025-06-26 09:58:22,293 - INFO - 数据存储成功: 1970-01-01 08:23:03 | 50.0 lux
2025-06-26 09:58:23,295 - INFO - 数据存储成功: 1970-01-01 08:23:04 | 45.0 lux
2025-06-26 09:58:24,293 - INFO - 数据存储成功: 1970-01-01 08:23:05 | 46.0 lux
2025-06-26 09:58:37,396 - INFO - 用户终止程序
2025-06-26 09:58:37,397 - INFO - 串口已关闭
2025-06-26 09:58:37,397 - INFO - 数据库连接已关闭

pyhton自动化采集数据脚本

pyhton自动化采集数据脚本：

1.引入包

2.数据库配置

3.MySQL数据库管理类

4.解析串口数据

时间戳转换

5.主函数

6.实验效果

网站公告

今日签到

热门文章

最新发布