Python读取OPC UA ExtensionObject时无法解析字节数组求助

你遇到的问题是调用load_type_definitions()后，扩展对象仍以字节数组形式返回，无法解析为实际的自定义类型。以下是针对性的解决步骤：

1. 显式加载目标命名空间的类型定义 #

默认的client.load_type_definitions()可能未正确加载变量所在命名空间的自定义类型，需指定命名空间索引加载：

from opcua import Client
from opcua import ua

connessione= 'opc.tcp://10.1.17.21:4840/'
client = Client(connessione)

try:
    client.connect()
    # 查看服务器所有命名空间，找到ns=1对应的URI
    print("服务器命名空间列表:", client.get_namespace_array())
    # 替换为ns=1对应的实际URI
    ns_uri = "http://your-server-namespace-uri"
    ns_idx = client.get_namespace_index(ns_uri)
    # 加载指定命名空间的类型定义
    client.load_type_definitions(namespace=ns_idx)
    
    var = client.get_node('ns=1;s=VARIABLE_OPC_first_coiling_machine_production_list')
    parsed_value = var.get_value()
    print("解析后的值:", parsed_value)
    client.disconnect()
except Exception as e:
    print(str(e))

2. 手动注册自定义类型编码器（自动加载失效时） #

如果服务器的自定义类型无法被自动识别，需要手动定义对应类并注册编码器：

步骤1：获取自定义类型结构 #

先用UaExpert等OPC UA客户端工具，查看ENC_DATATYPE_OPC_coiling_machine_production_list的字段结构（比如包含ProductionId、Quantity等属性）。

步骤2：定义类型类并注册 #

from opcua import Client
from opcua import ua
from opcua.common import extension_objects

# 根据服务器类型结构定义类
class CoilingMachineProductionList(ua.ExtensionObject):
    def __init__(self):
        super().__init__()
        # 匹配服务器的类型ID
        self.TypeId = ua.StringNodeId(ns=1, s="ENC_DATATYPE_OPC_coiling_machine_production_list")
        # 添加类型的具体字段
        self.ProductionId = ua.UInt32()
        self.Quantity = ua.UInt16()
        # 补充其他字段...

# 注册类型编码器
extension_objects.register_type(CoilingMachineProductionList, CoilingMachineProductionList.TypeId)

# 连接并读取变量
connessione= 'opc.tcp://10.1.17.21:4840/'
client = Client(connessione)
try:
    client.connect()
    var = client.get_node('ns=1;s=VARIABLE_OPC_first_coiling_machine_production_list')
    parsed_value = var.get_value()
    print("解析后的值:", parsed_value)
    client.disconnect()
except Exception as e:
    print(str(e))

3. opcua-asyncio 适配方案 #

异步库的解决逻辑一致，只需调整为异步语法：

from asyncua import Client
from asyncua import ua
import asyncio

async def main():
    connessione= 'opc.tcp://10.1.17.21:4840/'
    async with Client(url=connessione) as client:
        # 获取命名空间索引
        ns_uri = "http://your-server-namespace-uri"
        ns_idx = await client.get_namespace_index(ns_uri)
        # 加载类型定义
        await client.load_type_definitions(namespace=ns_idx)
        
        var = await client.get_node('ns=1;s=VARIABLE_OPC_first_coiling_machine_production_list')
        parsed_value = await var.get_value()
        print("解析后的值:", parsed_value)

if __name__ == "__main__":
    asyncio.run(main())

关键注意事项 #

• 确保服务器已将自定义DataType节点暴露在Objects/Server/DataTypeDictionary路径下，否则客户端无法获取类型定义。
• 命名空间URI必须与服务器端一致，不能直接硬编码索引（索引可能动态变化）。

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