如何连接Flask服务器与OPC Python脚本？能否用Sqlite实现双向数据交互？

Hey there! Let's tackle your problem head-on—you absolutely can use SQLite to bridge your OPC data collector and Flask server, and we'll also cover how to set up full-duplex communication for real-time visualization.

1. 核心问题：SQLite支持多进程读写吗？ #

Short answer: Yes, with the right configuration. SQLite is a file-based database, which means it's not a client-server DB like PostgreSQL, but it does support concurrent read/write operations when you enable the Write-Ahead Logging (WAL) mode. Here's what you need to know:

• Enable WAL mode with PRAGMA journal_mode=WAL;—this allows multiple readers to access the database while a single writer is active.
• Set a reasonable timeout value when connecting (e.g., timeout=10) to handle cases where multiple processes try to write at the same time.
• Avoid long-running transactions in either the OPC collector or Flask server to prevent locking up the database.

For most PLC data visualization use cases (where data update rates aren't in the hundreds per second), SQLite will work perfectly fine.

2. 架构设计思路 #

Here's a straightforward setup to connect everything:

• OPC Data Collector Script: Runs continuously, fetches data from your PLC via OPC (e.g., OPC UA), and writes updates to SQLite.
• Flask Server: Exposes REST APIs for historical data queries, and uses Flask-SocketIO to push real-time updates to your mobile/web app (full-duplex communication).
• SQLite Database: Acts as the shared data layer between the two processes.

3. 代码示例 #

3.1 OPC数据采集脚本（写入SQLite） #

First, install dependencies:

pip install opcua python-dotenv sqlite3

import time
import sqlite3
from opcua import Client

# 初始化SQLite连接（配置WAL模式和超时）
def init_db():
    conn = sqlite3.connect("plc_data.db", timeout=10)
    cursor = conn.cursor()
    # 启用WAL模式
    cursor.execute("PRAGMA journal_mode=WAL;")
    # 创建PLC数据表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS plc_metrics (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            tag_name TEXT NOT NULL,
            value REAL NOT NULL
        )
    ''')
    conn.commit()
    return conn

def fetch_and_write_plc_data():
    conn = init_db()
    cursor = conn.cursor()
    
    # 连接OPC服务器（替换成你的PLC OPC地址）
    opc_client = Client("opc.tcp://your-plc-opc-server:4840/")
    opc_client.connect()
    
    try:
        while True:
            # 读取PLC标签数据（替换成你的实际标签路径）
            temperature_tag = opc_client.get_node("ns=2;s=Temperature")
            pressure_tag = opc_client.get_node("ns=2;s=Pressure")
            
            temperature = temperature_tag.get_value()
            pressure = pressure_tag.get_value()
            
            # 写入SQLite
            cursor.execute('''
                INSERT INTO plc_metrics (tag_name, value)
                VALUES (?, ?), (?, ?)
            ''', ("Temperature", temperature, "Pressure", pressure))
            conn.commit()
            
            print(f"Updated data: Temp={temperature}, Pressure={pressure}")
            time.sleep(1)  # 每秒更新一次，根据实际需求调整
    except Exception as e:
        print(f"Error: {e}")
    finally:
        opc_client.disconnect()
        conn.close()

if __name__ == "__main__":
    fetch_and_write_plc_data()

3.2 Flask服务器（读取SQLite + 实时推送） #

Install Flask-SocketIO:

pip install flask flask-socketio

from flask import Flask, jsonify
from flask_socketio import SocketIO, emit
import sqlite3
import threading
import time

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key-here'
socketio = SocketIO(app, cors_allowed_origins="*")

# 从SQLite获取最新数据
def get_latest_plc_data():
    conn = sqlite3.connect("plc_data.db", timeout=10)
    cursor = conn.cursor()
    cursor.execute('''
        SELECT tag_name, value, timestamp
        FROM plc_metrics
        ORDER BY timestamp DESC
        LIMIT 2
    ''')
    data = cursor.fetchall()
    conn.close()
    return [{"tag": tag, "value": value, "time": ts} for tag, value, ts in data]

# 后台线程：定时查询数据库并推送更新
def background_thread():
    while True:
        latest_data = get_latest_plc_data()
        socketio.emit('plc_update', {'data': latest_data})
        time.sleep(1)

# REST API：获取历史数据
@app.route('/api/plc/history', methods=['GET'])
def get_history():
    conn = sqlite3.connect("plc_data.db", timeout=10)
    cursor = conn.cursor()
    cursor.execute('''
        SELECT tag_name, value, timestamp
        FROM plc_metrics
        ORDER BY timestamp DESC
        LIMIT 100
    ''')
    data = cursor.fetchall()
    conn.close()
    return jsonify([{"tag": tag, "value": value, "time": ts} for tag, value, ts in data])

# SocketIO连接事件
@socketio.on('connect')
def handle_connect():
    print('Client connected')
    # 推送初始数据
    emit('plc_update', {'data': get_latest_plc_data()})

if __name__ == '__main__':
    # 启动后台推送线程
    threading.Thread(target=background_thread, daemon=True).start()
    socketio.run(app, host='0.0.0.0', port=5000, debug=True)

4. 全双工交互的优化 #

The above setup uses a polling thread in Flask to check for new data, which works for most cases. If you want more efficient real-time updates:

• Modify the OPC collector script to send a message to Flask's SocketIO server directly when it writes new data (instead of polling). You can use socketio_client in the OPC script to emit events to the Flask server.
• For higher-scale scenarios, consider replacing SQLite with Redis (using its Pub/Sub feature) as the data broker. Redis handles concurrent operations much better and is designed for real-time data streaming.

5. 注意事项 #

• Always close database connections properly in both scripts to avoid locks.
• For production, avoid using debug=True in Flask, and consider using a process manager like Gunicorn.
• Test the concurrency: run both the OPC collector and Flask server, and check if data is being written and read without errors.

内容的提问来源于stack exchange，提问作者Damián Lukáč