求助：如何将Python通过OPC UA读取的PLC变量存入PostgreSQL数据库

1. 确认数据表结构 #

先确保你的数据表包含必要字段，比如记录时间、变量名、变量值。如果还没创建表，可以用这段SQL：

CREATE TABLE plc_data (
    id SERIAL PRIMARY KEY,
    capture_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    var_name VARCHAR(100) NOT NULL,
    var_value NUMERIC NOT NULL
);

2. 核心代码实现 #

结合OPC UA读取逻辑与PostgreSQL插入操作，以下是可直接复用的代码：

import time
from opcua import Client
import psycopg2
from psycopg2 import OperationalError

# 数据库配置（替换成你的实际信息）
DB_CONFIG = {
    'dbname': 'your_db',
    'user': 'your_user',
    'password': 'your_pwd',
    'host': 'localhost',
    'port': '5432'
}

# OPC UA服务器地址（替换成你的PLC OPC UA地址）
OPC_SERVER_URL = "opc.tcp://your_plc_ip:4840"

def get_db_connection():
    """获取数据库连接，处理自动重连"""
    while True:
        try:
            conn = psycopg2.connect(**DB_CONFIG)
            conn.autocommit = True
            return conn
        except OperationalError as e:
            print(f"数据库连接失败: {e}，5秒后重试...")
            time.sleep(5)

def insert_plc_data(conn, var_name, var_value):
    """插入单条PLC数据"""
    insert_sql = """
        INSERT INTO plc_data (var_name, var_value)
        VALUES (%s, %s);
    """
    try:
        with conn.cursor() as cur:
            cur.execute(insert_sql, (var_name, var_value))
    except Exception as e:
        print(f"插入失败: {e}")
        conn.close()
        return get_db_connection()
    return conn

def read_plc_var(client, node_id):
    """读取指定OPC UA变量值"""
    try:
        var_node = client.get_node(node_id)
        return var_node.get_value()
    except Exception as e:
        print(f"读取PLC变量失败: {e}")
        return None

def main():
    opc_client = Client(OPC_SERVER_URL)
    db_conn = get_db_connection()

    try:
        opc_client.connect()
        print("已连接OPC UA服务器")

        # 替换成你要读取的变量Node ID
        target_var_id = "ns=2;s=PLC_Temperature"

        while True:
            var_val = read_plc_var(opc_client, target_var_id)
            if var_val is not None:
                db_conn = insert_plc_data(db_conn, "Temperature", var_val)
                print(f"已存储数据: {var_val}")
            
            time.sleep(1)

    except KeyboardInterrupt:
        print("程序终止")
    finally:
        opc_client.disconnect()
        db_conn.close()

if __name__ == "__main__":
    main()

3. 关键注意事项 #

• 参数化查询：必须用%s作为占位符，禁止直接拼接字符串，避免SQL注入风险。
• 连接可靠性：get_db_connection函数处理了数据库连接断开后的自动重连，适合长期运行的服务。
• 异常捕获：读取OPC UA或插入数据库时的异常被捕获，防止单次失败导致整个程序崩溃。

4. 优化方向 #

• 多变量批量读取：如果需要读取多个PLC变量，可扩展读取逻辑一次性获取多个值，再批量插入数据库，减少IO次数。
• 批量插入：对于高频率数据，可积累N条数据后用executemany批量插入，提升效率：

  def batch_insert(conn, data_list):
      insert_sql = """INSERT INTO plc_data (var_name, var_value) VALUES (%s, %s);"""
      try:
          with conn.cursor() as cur:
              cur.executemany(insert_sql, data_list)
      except Exception as e:
          print(f"批量插入失败: {e}")
          conn.close()
          return get_db_connection()
      return conn
  

• 连接池：使用psycopg2.pool创建连接池，避免频繁创建销毁数据库连接，进一步优化性能。

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