Support for OPC UA Programs

[CraigBuilds]

OPC UA Programs can be thought of like OPC UA Methods, used when it would not make sense to block control flow after invoking a long running task. For example, something like "TurnLightOn" could be a method while something like "MoveJointToHomePosition" could be a program. Like methods, they can be invoked by a client, but unlike methods they can also be managed by a client, e.g, to suspend it or monitor progress.

Unified Automation defines them here: https://documentation.unified-automation.com/uasdknet/2.5.7/html/L2UaPrograms.html

Their behaviour is defined by a Program Finite State Machine (PFSM). This is not yet implemented in asyncua (see below).

https://github.com/FreeOpcUa/opcua-asyncio/blob/bb03fed7eaef4eedb43bc078b72e740c7982e653/asyncua/common/statemachine.py#L9

Is there any work being done to implement PFSM in asyncua? And would it be possible to add an api to the Nodes to easily create programs, in a similar way to how we create methods?

Maybe something like (using rclpy ROS2 Actions as an API inspiration)

#server.py
async def move_joint_to_home_position(ctx: ServerProgramHandle, limit_speed: bool) -> bool:
    while true:
         #interact with hardware to invoke behaviour
         #read back from hardware to see if done or fault
         #check ctx to see if client has requested suspend or hault
         #send feedback back to client using ctx
         #break from the loop if done
    return true

server = Server()
#init and setup server and namespace
...
objects = server.nodes.objects
await objects.add_folder(idx, "myEmptyFolder")
myobj = await objects.add_object(idx, "MyObject")
await myobj.add_program(idx, "move_home", move_joint_to_home_position, [ua.VariantType.Bool], [ua.VariantType.Bool])

This will work in a similar way to creating a method with an async function, but the server owns a state machine that represents the methods state.

The client could work in a similar way to invoking methods, but with some extra functionality

#client.py
#Setup client
...
handle: ClientProgramHandle= await client.nodes.objects.start_program(f"{nsidx}:myEmptyFolder:MyObject:move_home", true)
while not handle.is_in_terminal_state():
    print(handle.read_feedback())
if handle.has_result()
    res: bool = handle.result()

[AndreasHeine]

hey @CraigPersonal100,

currently i stopped at finitestatemachine!

the correct reference would by the official one -> https://reference.opcfoundation.org/v104/Core/docs/Part10/

A program is a abstraction of "long running logic" which can not return within a few seconds, means a method should return always within the clients timeout limit (<5s)! a program can be invoked via methods and creates after some time (>5s) a result (long processing time e.g. imageprocessing).

Part 10 is more then just adding a method, you need to create a class which installs a instance of the program objecttype to the addressspace (in different configurations according to the optional stuff) and creates all the required bindings to the controlmethods inkl. behaviour to some class methods/properties, the objecttype contains a lot of other stuff which needs to be implemented:

the plan in the statemachine section was to manage most behavior stuff with inheritance because the objecttypes inherit aswell (less code duplication) so currently you only have the fsm-type stuff and a programstatemachinetype is a concrete instance of a fsm-type (which is a abstract type) currently i have some open fixes in the finitstatemachine class : -manage optionals (not just all or nothing) -some improvements for statemachines which got imported via XML which need to be "detected/explored" and linked to the python-class correctly -subtype checking which is not yet committed but WIP (there was a issue with the server api pre 1.0.0 which has been resolved with the ABC for client and server api):

    async def install(self, optionals: bool=False):
        if await self._is_subtype_of_finitestatemachine():
            super().install(optionals)
            pass
        else:
            raise ua.UaError(f"NodeId: {self._state_machine_type} is not a subtype of FiniteStateMachine!")

    async def _is_subtype_of_finitestatemachine(self):
        result = False
        type_node = Node(self._server, self._state_machine_type)
        parent = await type_node.get_parent() # pre v1.0.0 crashing here because the server api was not complete and in sync with the client api so node-class was not consistent
        if not parent:
            raise ua.UaError("Node does not have a Parent!")
        if parent.nodeid == self._finitestatemachine_nodeid:
            result = True
        else:
            while not parent.nodeid == self._finitestatemachine_nodeid:
                parent = await parent.get_parent()
                if not parent:
                    result = False
                    break
                if parent.nodeid == self._finitestatemachine_nodeid:
                    result = True
                    break
                if parent.nodeid == self._server.nodes.root.nodeid:
                    result = False
                    break
        return result

if all that is DONE, we can talk about Part 10 ;)

[PhilWillecke]

Is the link to the XML imported statemachine working? Or is this still an open TO DO?

[PhilWillecke]

@CraigPersonal100 Have you found a solution for your approach?

[CraigBuilds]

@CraigPersonal100 Have you found a solution for your approach?

I haven't been working on this to be honest. I wanted programs because they are a good abstraction for the machine I'm trying to represent with an opcua interface, but variables and methods also work fine to be honest.

But I'll likely use this in a future project if it does become a feature.