基于opcua-asyncio的Python客户端读取OPC UA Byte/UInt16数组值异常

问题 #

基于GitHub的opcua-asyncio库开发Python OPC UA客户端，读取仪器服务器数据时，Byte或UInt16类型的数组返回值全为0，但UaExpert/DataFeed能正常显示预期值。已尝试read_value、read_data_value等多种读取方法，均无效。

测试代码 #

import asyncio
import logging
import sys
import common
from asyncua import ua

sys.path.insert(0, "..")

async def main():
    client = await common.connectViaUsername()
    try:
        async with client:
            while True:
                # 获取变量节点
                pi_node = client.get_node("ns=1;i=57")

                val = await pi_node.read_value()

                print("val: ", val)
                print("type of val: ", type(val))

                att = await pi_node.read_attribute(ua.AttributeIds.DataType)
                print("att: ", att)
                print("type of att: ", type(att))
                print("att value value: ", att.Value.Value)

                dataval = await pi_node.read_data_value()
                print("dataval: ", dataval)
                print("variant type: ", dataval.Value.Value[0].VariantType)

                rv = ua.ReadValueId()
                rv.NodeId = pi_node.nodeid
                rv.AttributeId = ua.AttributeIds.DataType
                params = ua.ReadParameters()
                params.NodesToRead.append(rv)
                result = await pi_node.session.read(params)
                print("result: ", result)
                print(type(result[0]))
                print(type(result[0].Value))
                print(type(result[0].data_type))
                print(type(result[0].Encoding))
                print((result[0].Value))
                print((result[0].data_type))
                print((result[0].Encoding))

                print("node_class: ", await pi_node.read_node_class())
                print("att.value: ", await pi_node.read_attribute(ua.AttributeIds.Value))
                print("data type: ", await pi_node.read_data_type())
                print("array dim: ", await pi_node.read_array_dimensions())
                print("data values: ", await pi_node.read_data_value())
                print("variant type: ", await pi_node.read_data_type_as_variant_type())
                print("_______")

                await asyncio.sleep(100)
    except ua.UaError as err:
        _logger.error(err)
    finally:
        await client.disconnect()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)  # 日志级别设为INFO
    _logger = logging.getLogger(__name__)

    # 屏蔽asyncua库的日志
    logging.getLogger('asyncua').setLevel(logging.CRITICAL + 1)

    # 配置日志文件路径，不需要可注释
    fh = common.configureLogs("logs/OPC Status - InstrumentSensor (Pulse Sensors).log")
    _logger.addHandler(fh)

    asyncio.run(main())

输出结果 #

val:  [Variant(Value=0, VariantType=<VariantType.Byte: 3>, Dimensions=None, is_array=False), Variant(Value=0, VariantType=<VariantType.Byte: 3>, Dimensions=None, is_array=False), ...]
type of val:  <class 'list'>
att:  DataValue(Value=Variant(Value=NodeId(Identifier=3, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), VariantType=<VariantType.NodeId: 17>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), ...)
type of att:  <class 'asyncua.ua.uatypes.DataValue'>
att value value:  NodeId(Identifier=3, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>)
dataval:  DataValue(Value=Variant(Value=[Variant(Value=0, VariantType=<VariantType.Byte: 3>, Dimensions=None, is_array=False), ...], VariantType=<VariantType.Variant: 24>, Dimensions=None, is_array=True), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2023, 10, 27, 1, 48, 35, 270000), ...)
variant type:  VariantType.Byte
...
data type:  NodeId(Identifier=3, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>)
array dim:  [10]
variant type:  VariantType.Byte

对比情况 #

字符串、浮点型数组可正常读取，示例：

[Variant(Value=1002.32, VariantType=<VariantType.Double: 11>, Dimensions=None, is_array=False), Variant(Value=1002.01, VariantType=<VariantType.Double: 11>, Dimensions=None, is_array=False), ...]

解决方案 #

针对Byte/UInt16数组读取异常的问题，可尝试以下几种方法：

1. 手动解析原始二进制数据 #

绕过库的自动解码，直接读取原始字节流后手动解析：

async def read_raw_byte_array(node):
    # 读取原始数据的Value属性
    dv = await node.read_data_value()
    # 获取原始二进制数据
    raw_data = dv.Value._data
    # 解析Byte数组：每个元素占1字节
    byte_array = list(raw_data)
    # 如果是UInt16数组：每个元素占2字节，大端/小端根据服务器配置调整
    # uint16_array = [int.from_bytes(raw_data[i:i+2], byteorder='little') for i in range(0, len(raw_data), 2)]
    return byte_array

2. 指定读取时的Variant类型 #

强制指定读取的类型，避免库自动识别错误：

from asyncua import ua

async def read_with_specified_type(node, var_type):
    rv = ua.ReadValueId()
    rv.NodeId = node.nodeid
    rv.AttributeId = ua.AttributeIds.Value
    rv.DataEncoding = ua.QualifiedName("", 0)  # 使用默认编码
    params = ua.ReadParameters()
    params.NodesToRead.append(rv)
    # 发送读取请求，获取原始DataValue
    results = await node.session.read(params)
    dv = results[0]
    # 手动转换为指定类型的数组
    return [v.Value for v in dv.Value.Value]

# 使用示例：
byte_array = await read_with_specified_type(pi_node, ua.VariantType.Byte)
uint16_array = await read_with_specified_type(pi_node, ua.VariantType.UInt16)

3. 升级opcua-asyncio版本 #

旧版本库可能存在基础类型数组的解码bug，升级到最新稳定版：

pip install --upgrade opcua-asyncio

4. 验证节点类型定义 #

通过UaExpert查看节点的DataType属性，确认是标准的Byte（NodeId=ns=0;i=3）或UInt16（NodeId=ns=0;i=5），排除自定义类型与标准类型冲突的情况。

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