如何通过Python REST API读写OPC UA服务器值？附代码改造需求

实现思路 #

• 引入Python OPC UA客户端库：使用python-opcua库实现与OPC UA服务器的交互，先通过pip安装依赖
• 封装OPC UA写入逻辑：把连接服务器、写入节点值、断开连接的逻辑封装成独立函数，便于复用和异常处理
• 嵌入API流程：在本地模型保存完成后，调用封装的写入函数，将车速参数传递给OPC UA服务器
• 异常处理与日志：写入过程中捕获异常并记录日志，避免OPC UA操作失败影响API的正常响应

步骤与代码示例 #

1. 安装依赖 #

pip install opcua

2. 封装OPC UA写入函数 #

from opcua import Client
import logging

# 配置日志，用于排查问题
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def write_speed_to_opcua(opcua_server_url, speed_node_id, speed_value):
    """
    向指定OPC UA节点写入车速值
    :param opcua_server_url: OPC UA服务器地址（如 opc.tcp://localhost:4840/freeopcua/server/）
    :param speed_node_id: 车速对应的节点ID（如 ns=2;s=Car.Speed）
    :param speed_value: 待写入的车速数值
    """
    client = None
    try:
        # 建立服务器连接
        client = Client(opcua_server_url)
        client.connect()
        logger.info(f"已连接OPC UA服务器: {opcua_server_url}")
        
        # 获取目标节点并写入值
        speed_node = client.get_node(speed_node_id)
        speed_node.set_value(speed_value)
        logger.info(f"成功写入车速 {speed_value} 到节点 {speed_node_id}")
    except Exception as e:
        logger.error(f"OPC UA写入失败: {str(e)}")
    finally:
        # 确保断开连接，避免资源泄漏
        if client:
            client.disconnect()
            logger.info("已断开OPC UA服务器连接")

3. 改造原有API代码 #

@requires_login
def post_car_setting(api_key, id: str, car_settings_schema: CarSettingsSchema, login=None) -> Response:
    carModel: CarModel = CarModel.fetch_by_id(id)
    if carModel is not None:
        car_settings = CarSettingsModel.from_schema(car_settings_schema)
        carModel.car_settings = car_settings
        carModel.modified_by = login
        carModel.modified_on = datetime.now()
        carModel.update()
        
        # --- 新增OPC UA写入逻辑 ---
        # 建议将以下配置移到环境变量或配置文件中，避免硬编码
        OPCUA_SERVER_URL = "opc.tcp://localhost:4840/freeopcua/server/"
        SPEED_NODE_ID = "ns=2;s=Car.Speed"
        # 从car_settings中提取车速参数（假设字段名为speed）
        target_speed = car_settings.speed
        # 调用写入函数
        write_speed_to_opcua(OPCUA_SERVER_URL, SPEED_NODE_ID, target_speed)
        
        return Response(body=carModel.car_settings).send_response()
    else:
        return Response(False, ErrorCodes.NOT_FOUND)

注意事项 #

• 配置解耦：不要硬编码OPC UA服务器地址和节点ID，建议使用环境变量或配置文件（如config.py）管理，提升可维护性
• 权限验证：确保OPC UA服务器的目标节点具备写入权限，避免因权限不足导致写入失败
• 异步优化：如果OPC UA写入耗时较长，建议使用线程、进程或异步任务框架（如Celery）异步处理，避免阻塞API响应
• 错误处理：可以根据业务需求，在写入失败时选择是否返回API错误，或仅记录日志不影响主流程

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