Python采集单片机生成Webserver网页中高频变化数据的方案咨询

首先明确：你的需求是每秒采集5次数据，Selenium完全不适合——它是模拟浏览器的工具，资源占用极高，高频采集会导致性能拉胯，直接排除。下面分场景给你适配Python的方案：

优先推荐的高效方案 #

1. 直接抓数据接口（最常用） #

单片机Webserver的表格每秒更新，大概率是通过AJAX请求后台接口获取数据，而非整页刷新。你可以用浏览器F12打开「网络」面板，刷新页面后观察那些每秒调用几次的请求（一般是XHR/Fetch类型），找到返回JSON/XML格式数据的接口。

这种情况下，直接用requests或者更高效的httpx（支持异步）请求接口，拿到原始数据后直接解析，不用处理HTML，速度最快、开销最小。

2. WebSocket实时接收（如果是主动推送） #

如果单片机是通过WebSocket向前端主动推送数据（比如页面没发请求就自动更新），那就用Python的websocket-client库连接WebSocket，实时接收数据，这比轮询请求更高效，不用反复发起HTTP请求。

只能解析HTML的备用方案 #

如果网页确实是整页刷新更新表格（这种情况很少，因为单片机带宽/性能有限），用requests + Beautiful Soup组合：

• 用requests.Session()保持连接，减少重复建立TCP连接的开销；
• 用Beautiful Soup精准定位表格元素，只提取需要的参数，避免解析整个HTML文档；
• 控制请求频率，尽量贴近每秒5次，但要扣除请求和解析的耗时，避免给单片机Webserver造成过大压力。

数据库写入优化 #

因为每秒5次写入，单条插入会有IO开销，建议：

• 攒N条数据（比如攒5条，1秒）再批量插入，用数据库的批量插入语句（比如SQLite的executemany）；
• 用轻量型数据库比如SQLite做本地存储，如果需要远程数据库，用pymysql/psycopg2配合批量插入，必要时用异步数据库驱动提升效率。

代码示例 #

接口采集+SQLite存储 #

import requests
import sqlite3
import time

# 初始化数据库
conn = sqlite3.connect('sensor_data.db')
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS params
                  (timestamp REAL, temperature REAL, humidity REAL)''')
conn.commit()

# 替换成你找到的API接口地址
API_URL = "http://192.168.1.100/api/get_data"

while True:
    start = time.time()
    try:
        resp = requests.get(API_URL)
        data = resp.json()
        # 假设接口返回{"temperature": 25.3, "humidity": 60.2}
        cursor.execute(
            "INSERT INTO params VALUES (?, ?, ?)",
            (time.time(), data['temperature'], data['humidity'])
        )
        conn.commit()
    except Exception as e:
        print(f"采集/存储出错: {str(e)}")
    # 控制采集间隔，确保每秒5次
    elapsed = time.time() - start
    if elapsed < 0.2:
        time.sleep(0.2 - elapsed)

HTML表格采集+SQLite存储 #

import requests
from bs4 import BeautifulSoup
import sqlite3
import time

conn = sqlite3.connect('sensor_data.db')
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS params
                  (timestamp REAL, param1 REAL, param2 REAL)''')
conn.commit()

WEB_URL = "http://192.168.1.100"
session = requests.Session()  # 保持会话

while True:
    start = time.time()
    try:
        resp = session.get(WEB_URL)
        soup = BeautifulSoup(resp.text, 'html.parser')
        # 定位表格，替换成你网页的表格选择器
        table = soup.find('table', id='data-table')
        rows = table.find_all('tr')[1:]  # 跳过表头行
        for row in rows:
            cols = row.find_all('td')
            param1 = float(cols[0].text.strip())
            param2 = float(cols[1].text.strip())
            cursor.execute(
                "INSERT INTO params VALUES (?, ?, ?)",
                (time.time(), param1, param2)
            )
        conn.commit()
    except Exception as e:
        print(f"采集/存储出错: {str(e)}")
    elapsed = time.time() - start
    if elapsed < 0.2:
        time.sleep(0.2 - elapsed)

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