Historize ExtensionObjects or custom Structures?

[dakotahorstman]

Good day.

Is there a documented or native way to serialize data changes for ExtensionObjects, or custom Structures? For "normal" tags, I use asyncua's built-in string_to_val and val_to_string to serialize data, but those aren't compatible with ExtensionObjects nor have I been able to find a "proper" way to historize these values without writing my own historizer. I'm fine doing that, but I'd rather explore any native possibilities

Consider the following example to understand my approach (as an MRE; slightly modified version of the server-custom-structures-and-enums.py example):

import asyncio
import logging

from asyncua import Server, ua
from asyncua.common.structures104 import new_enum, new_struct, new_struct_field
from asyncua.common.ua_utils import string_to_val, val_to_string
from asyncua.server.history import HistoryStorageInterface
from asyncua.ua import DataValue, NodeId
from loguru import logger


class MyHistoryStorage(HistoryStorageInterface):
    async def init(self) -> None:
        pass

    async def new_historized_event(self, source_id, evtypes, period, count=0) -> None:
        pass

    async def new_historized_node(self, node_id: NodeId, period, count=0) -> None:
        pass

    async def save_event(self, event) -> None:
        pass

    async def read_event_history(
        self, source_id, start, end, nb_values, evfilter
    ) -> None:
        pass

    async def read_node_history(self, node_id: NodeId, start, end, nb_values) -> None:
        pass

    async def stop(self) -> None:
        pass

    @logger.catch
    async def save_node_value(self, node_id: NodeId, datavalue: DataValue) -> None:
        print(node_id, datavalue)
        print("As string:", val_to_string(datavalue))
        # Comment this print statement as it will throw errors due to ExtensionObject being unsupported in `string_to_val`
        print(
            "As object from string:",
            string_to_val(val_to_string(datavalue), datavalue.Value.VariantType),
        )
        print("-" * 200)


logging.basicConfig(level=logging.INFO)
_logger = logging.getLogger("asyncua")


async def main():
    # setup our server
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/freeopcua/server/")
    server.iserver.history_manager.set_storage(MyHistoryStorage())

    # setup our own namespace, not really necessary but should as spec
    uri = "http://examples.freeopcua.github.io"
    idx = await server.register_namespace(uri)

    snode1, _ = await new_struct(
        server,
        idx,
        "MyStruct",
        [
            new_struct_field("MyBool", ua.VariantType.Boolean),
            new_struct_field("MyUInt32List", ua.VariantType.UInt32, array=True),
        ],
    )
    snode2, _ = await new_struct(
        server,
        idx,
        "MyOptionalStruct",
        [
            new_struct_field("MyBool", ua.VariantType.Boolean),
            new_struct_field("MyUInt32List", ua.VariantType.UInt32, array=True),
            new_struct_field("MyInt64", ua.VariantType.Int64, optional=True),
        ],
    )
    snode3, _ = await new_struct(
        server,
        idx,
        "MyNestedStruct",
        [
            new_struct_field("MyStructArray", snode1, array=True),
        ],
    )
    enode = await new_enum(
        server,
        idx,
        "MyEnum",
        [
            "titi",
            "toto",
            "tutu",
        ],
    )

    custom_objs = await server.load_data_type_definitions()
    print("Custom objects on server")
    for name, obj in custom_objs.items():
        print("    ", obj)

    valnode = await server.nodes.objects.add_variable(
        idx, "my_enum", ua.MyEnum.toto, datatype=enode.nodeid
    )
    my_struct = await server.nodes.objects.add_variable(
        idx, "my_struct", ua.Variant(ua.MyStruct(), ua.VariantType.ExtensionObject)
    )
    my_struct_optional = await server.nodes.objects.add_variable(
        idx,
        "my_struct_optional",
        ua.Variant(ua.MyOptionalStruct(), ua.VariantType.ExtensionObject),
    )
    t1 = await server.nodes.objects.add_variable(
        idx, "t1", ua.Variant(ua.MyNestedStruct(), ua.VariantType.ExtensionObject)
    )
    await valnode.set_writable(True)
    await my_struct.set_writable(True)
    await my_struct_optional.set_writable(True)
    await t1.set_writable(True)
    await server.historize_node_data_change(valnode, period=None)
    await server.historize_node_data_change(my_struct, period=None)
    await server.historize_node_data_change(my_struct_optional, period=None)
    await server.historize_node_data_change(t1, period=None)
    await server.export_xml(
        [
            server.nodes.objects,
            server.nodes.root,
            snode1,
            snode2,
            snode3,
            enode,
            valnode,
        ],
        "structs_and_enum.xml",
    )

    async with server:
        while True:
            await asyncio.sleep(1)


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

Print out in terminal for convenience. Note that this assumes the "As object from string" print statement is commented as it will raise errors. The other three print statements can stay:

NodeId(Identifier=8, NamespaceIndex=2, NodeIdType=<NodeIdType.FourByte: 1>) DataValue(Value=Variant(Value=<MyEnum.toto: 1>, VariantType=<VariantType.Int32: 6>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 222902, tzinfo=datetime.timezone.utc), ServerTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 222903, tzinfo=datetime.timezone.utc), SourcePicoseconds=None, ServerPicoseconds=None)
As string: toto
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
NodeId(Identifier=9, NamespaceIndex=2, NodeIdType=<NodeIdType.FourByte: 1>) DataValue(Value=Variant(Value=MyStruct(MyBool=True, MyUInt32List=[]), VariantType=<VariantType.ExtensionObject: 22>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 223004, tzinfo=datetime.timezone.utc), ServerTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 223005, tzinfo=datetime.timezone.utc), SourcePicoseconds=None, ServerPicoseconds=None)
As string: MyStruct(MyBool=True, MyUInt32List=[])
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
NodeId(Identifier=10, NamespaceIndex=2, NodeIdType=<NodeIdType.FourByte: 1>) DataValue(Value=Variant(Value=MyOptionalStruct(MyBool=True, MyUInt32List=[], MyInt64=None), VariantType=<VariantType.ExtensionObject: 22>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 223089, tzinfo=datetime.timezone.utc), ServerTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 223090, tzinfo=datetime.timezone.utc), SourcePicoseconds=None, ServerPicoseconds=None)
As string: MyOptionalStruct(MyBool=True, MyUInt32List=[], MyInt64=None)
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
NodeId(Identifier=11, NamespaceIndex=2, NodeIdType=<NodeIdType.FourByte: 1>) DataValue(Value=Variant(Value=MyNestedStruct(MyStructArray=[]), VariantType=<VariantType.ExtensionObject: 22>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 223166, tzinfo=datetime.timezone.utc), ServerTimestamp=datetime.datetime(2025, 2, 21, 21, 57, 37, 223166, tzinfo=datetime.timezone.utc), SourcePicoseconds=None, ServerPicoseconds=None)
As string: MyNestedStruct(MyStructArray=[])
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

[dakotahorstman]

Found the solution after some more digging. asyncua provides the functions struct_to_binary and struct_from_binary. These will work perfectly for my needs for historization. Here's an updated snippet for the above:

from asyncua.ua.ua_binary import struct_from_binary, struct_to_binary

...

@logger.catch
async def save_node_value(self, node_id: NodeId, datavalue: DataValue) -> None:
    print(f"Original: {node_id} -> {datavalue}")
    binary = struct_to_binary(datavalue.Value.Value)
    binary_stream = io.BytesIO(binary)
    print(f"Serialized: {binary}")
    print(
        f"Deserialized: {struct_from_binary(datavalue.Value.Value.__class__, binary_stream)}"
    )
    print("-" * 200)