Python中OPC UA全VariantType值类型转换方案咨询

问题描述 #

我正在开发一款Python程序，用于读取CSV文件并将选定的值写入目标PLC。目前已实现常见OPC UA Variant类型的转换，但像ExtensionObject、Guid这类少见类型不知道怎么处理，希望程序能支持OPC UA架构的所有类型，请问如何为每种值做类型转换？

当前代码现状 #

import csv
from xmlrpc.client import Boolean
from opcua import Client, ua, uamethod

# Libraries I have created
from Definitions import *
from Connexion import *

def default_f(x: str) -> str:
    """Returns x as-is, a string; used when type is not declared in Transformer."""
    return x


Transformer = {
    "VariantType.String": lambda x: x,
    "VariantType.StatusCode": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.StatusCode)),
    "VariantType.Boolean": lambda x: x=='True',
    "VariantType.SByte": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.SByte)),
    "VariantType.Byte": lambda x: str.encode(x),
    "VariantType.ByteString": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.ByteString)),
    ...........
    "VariantType.ExpandedNodeId": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.ExpandedNodeId)),
    "VariantType.ExtensionObject": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.ExtensionObject)),
    "VariantType.Guid": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.Guid)),
    ...........
    "VariantType.UInt64": lambda x: ua.DataValue(ua.Variant(
        x, ua.VariantType.UInt64)),
}


try:
    connection(client)

    with open("Variables.csv", "r") as file:
        reader = csv.reader(file)
        for line in reader:
            try:
                # Writing
                this_node, this_type, this_val = line[1], line[2], line[3]
                transformer_func = Transformer.get(this_type, default_f)
                val_transformed = transformer_func(this_val)
                if transformer_func == default_f:
                    print(f"warning: found undeclared type {this_type} with value {this_val}")
                else:
                    write_value(this_node, val_transformed) # A function I created which uses set_value()
            
finally:
    disconnection(client)

完整解决方案：覆盖所有OPC UA Variant类型的转换 #

下面是针对所有OPC UA Variant类型的转换实现，同时优化了代码结构以提升可维护性：

1. 拆分转换函数 #

将每个类型的转换逻辑拆分为独立函数，替代复杂的lambda表达式，便于调试和扩展：

import csv
import json
from opcua import Client, ua, uamethod

# 自定义模块
from Definitions import *
from Connexion import *

def default_f(x: str) -> str:
    """默认转换：原样返回字符串，用于未声明的类型"""
    return x

# -------------------------- 基础类型转换 --------------------------
def transform_null(x):
    return ua.DataValue(ua.Variant(None, ua.VariantType.Null))

def transform_boolean(x):
    # 大小写不敏感判断，兼容'True'/'true'/'TRUE'
    return ua.DataValue(ua.Variant(x.strip().lower() == 'true', ua.VariantType.Boolean))

def transform_sbyte(x):
    return ua.DataValue(ua.Variant(int(x.strip()), ua.VariantType.SByte))

def transform_byte(x):
    return ua.DataValue(ua.Variant(int(x.strip()), ua.VariantType.Byte))

def transform_int16(x):
    return ua.DataValue(ua.Variant(int(x.strip()), ua.VariantType.Int16))

def transform_uint16(x):
    val = int(x.strip())
    if val < 0:
        raise ValueError(f"UInt16值必须非负，当前值：{val}")
    return ua.DataValue(ua.Variant(val, ua.VariantType.UInt16))

def transform_int32(x):
    return ua.DataValue(ua.Variant(int(x.strip()), ua.VariantType.Int32))

def transform_uint32(x):
    val = int(x.strip())
    if val < 0:
        raise ValueError(f"UInt32值必须非负，当前值：{val}")
    return ua.DataValue(ua.Variant(val, ua.VariantType.UInt32))

def transform_int64(x):
    return ua.DataValue(ua.Variant(int(x.strip()), ua.VariantType.Int64))

def transform_uint64(x):
    val = int(x.strip())
    if val < 0:
        raise ValueError(f"UInt64值必须非负，当前值：{val}")
    return ua.DataValue(ua.Variant(val, ua.VariantType.UInt64))

def transform_float(x):
    return ua.DataValue(ua.Variant(float(x.strip()), ua.VariantType.Float))

def transform_double(x):
    return ua.DataValue(ua.Variant(float(x.strip()), ua.VariantType.Double))

def transform_string(x):
    return ua.DataValue(ua.Variant(x.strip(), ua.VariantType.String))

# -------------------------- 时间与GUID类型 --------------------------
def transform_datetime(x):
    # 支持ISO格式字符串，如"2024-05-20T12:34:56"
    dt = ua.DateTime.from_string(x.strip())
    return ua.DataValue(ua.Variant(dt, ua.VariantType.DateTime))

def transform_guid(x):
    # 支持标准GUID格式，如"12345678-1234-1234-1234-1234567890AB"
    guid = ua.Guid(x.strip())
    return ua.DataValue(ua.Variant(guid, ua.VariantType.Guid))

# -------------------------- 二进制与XML类型 --------------------------
def transform_byte_string(x):
    # 假设CSV中存储十六进制字符串，如"48656C6C6F"对应"Hello"
    byte_data = bytes.fromhex(x.strip())
    return ua.DataValue(ua.Variant(byte_data, ua.VariantType.ByteString))

def transform_xml_element(x):
    xml_elem = ua.XmlElement(x.strip())
    return ua.DataValue(ua.Variant(xml_elem, ua.VariantType.XmlElement))

# -------------------------- OPC UA节点与名称类型 --------------------------
def transform_node_id(x):
    # 支持NodeId字符串格式，如"ns=1;s=MyNode"
    node_id = ua.NodeId.from_string(x.strip())
    return ua.DataValue(ua.Variant(node_id, ua.VariantType.NodeId))

def transform_expanded_node_id(x):
    exp_node_id = ua.ExpandedNodeId.from_string(x.strip())
    return ua.DataValue(ua.Variant(exp_node_id, ua.VariantType.ExpandedNodeId))

def transform_qualified_name(x):
    # 支持格式如"1:MyName"（命名空间索引:名称）
    qname = ua.QualifiedName.from_string(x.strip())
    return ua.DataValue(ua.Variant(qname, ua.VariantType.QualifiedName))

def transform_localized_text(x):
    # 支持格式如"en:Hello"（语言代码:文本）
    loc_text = ua.LocalizedText.from_string(x.strip())
    return ua.DataValue(ua.Variant(loc_text, ua.VariantType.LocalizedText))

# -------------------------- 状态与诊断类型 --------------------------
def transform_status_code(x):
    # 支持两种输入：状态码数值（如"0"）或状态码名称（如"Good"）
    try:
        sc = ua.StatusCode(int(x.strip()))
    except ValueError:
        sc = ua.StatusCode.from_name(x.strip())
    return ua.DataValue(ua.Variant(sc, ua.VariantType.StatusCode))

def transform_diagnostic_info(x):
    # CSV中存储JSON格式的诊断信息，示例：{"symbolic_id": "ns=1;s=DiagId", "namespace_uri": "http://example.com"}
    data = json.loads(x.strip())
    diag_info = ua.DiagnosticInfo()
    if 'symbolic_id' in data:
        diag_info.SymbolicId = ua.NodeId.from_string(data['symbolic_id'])
    if 'namespace_uri' in data:
        diag_info.NamespaceUri = data['namespace_uri']
    # 按需添加其他DiagnosticInfo属性的设置逻辑
    return ua.DataValue(ua.Variant(diag_info, ua.VariantType.DiagnosticInfo))

# -------------------------- 复杂类型 --------------------------
def transform_extension_object(x):
    # CSV中存储JSON，示例：{"type_id": "ns=1;s=MyExtensionType", "body": {"key": "value"}}
    data = json.loads(x.strip())
    type_id = ua.NodeId.from_string(data['type_id'])
    body = data['body']
    ext_obj = ua.ExtensionObject(type_id, body)
    return ua.DataValue(ua.Variant(ext_obj, ua.VariantType.ExtensionObject))

def transform_data_value(x):
    # CSV中存储JSON，示例：{"value": "123", "value_type": "VariantType.Int32", "status_code": "0", "source_timestamp": "2024-05-20T12:34:56"}
    data = json.loads(x.strip())
    dv = ua.DataValue()
    if 'value' in data:
        val_type = data.get('value_type', 'VariantType.String')
        transform_func = Transformer.get(val_type, default_f)
        dv.Value = transform_func(data['value']).Value
    if 'status_code' in data:
        dv.StatusCode = ua.StatusCode(int(data['status_code']))
    if 'source_timestamp' in data:
        dv.SourceTimestamp = ua.DateTime.from_string(data['source_timestamp'])
    return ua.DataValue(ua.Variant(dv, ua.VariantType.DataValue))

def transform_variant(x):
    # CSV中存储JSON，示例：{"type": "VariantType.Int32", "value": "123"}
    data = json.loads(x.strip())
    val_type = data['type']
    transform_func = Transformer.get(val_type, default_f)
    transformed_val = transform_func(data['value'])
    variant_type = getattr(ua.VariantType, val_type.split('.')[1])
    return ua.DataValue(ua.Variant(transformed_val.Value, variant_type))

2. 构建完整Transformer字典 #

将所有转换函数映射到对应的Variant类型：

Transformer = {
    "VariantType.Null": transform_null,
    "VariantType.Boolean": transform_boolean,
    "VariantType.SByte": transform_sbyte,
    "VariantType.Byte": transform_byte,
    "VariantType.Int16": transform_int16,
    "VariantType.UInt16": transform_uint16,
    "VariantType.Int32": transform_int32,
    "VariantType.UInt32": transform_uint32,
    "VariantType.Int64": transform_int64,
    "VariantType.UInt64": transform_uint64,
    "VariantType.Float": transform_float,
    "VariantType.Double": transform_double,
    "VariantType.String": transform_string,
    "VariantType.DateTime": transform_datetime,
    "VariantType.Guid": transform_guid,
    "VariantType.ByteString": transform_byte_string,
    "VariantType.XmlElement": transform_xml_element,
    "VariantType.NodeId": transform_node_id,
    "VariantType.ExpandedNodeId": transform_expanded_node_id,
    "VariantType.StatusCode": transform_status_code,
    "VariantType.QualifiedName": transform_qualified_name,
    "VariantType.LocalizedText": transform_localized_text,
    "VariantType.ExtensionObject": transform_extension_object,
    "VariantType.DataValue": transform_data_value,
    "VariantType.Variant": transform_variant,
    "VariantType.DiagnosticInfo": transform_diagnostic_info,
}

3. 优化主逻辑的错误处理 #

添加行号追踪和异常捕获，方便定位CSV中的错误：

try:
    connection(client)

    with open("Variables.csv", "r") as file:
        reader = csv.reader(file)
        next(reader)  # 跳过表头（CSV第一行为表头时启用）
        for line_num, line in enumerate(reader, start=2):  # 从第2行开始计数
            try:
                if len(line) < 4:
                    print(f"警告：第{line_num}行列数不足，跳过")
                    continue
                this_node, this_type, this_val = line[1], line[2], line[3]
                transformer_func = Transformer.get(this_type, default_f)
                val_transformed = transformer_func(this_val)
                if transformer_func == default_f:
                    print(f"警告：第{line_num}行发现未声明类型 {this_type}，值为 {this_val}")
                else:
                    write_value(this_node, val_transformed)
            except Exception as e:
                print(f"第{line_num}行处理失败：{str(e)}")
            
finally:
    disconnection(client)

关键注意事项 #

• CSV格式约定：对于复杂类型（如ExtensionObject、DataValue），建议在CSV单元格中使用JSON字符串存储结构化数据，确保转换函数能正确解析
• 类型验证：无符号整数类型添加了非负检查，避免写入PLC无效值
• 鲁棒性优化：布尔类型支持大小写不敏感判断，状态码支持数值和名称两种输入格式
• 错误定位：通过行号追踪和异常捕获，快速定位CSV中的错误行

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