Possibility Separate Function Blocks for Connect and Read/Write Data

[Keni9292]

Hello, I have a problem converting my program from the OPCUA library to the Asyncua library.

My program is designed so that it establishes the connection in one Function and in separate Functions for
Read or Write Data

Is there a way to keep this in this library?

Currently I'm getting this error
return await future
^^^^^^^^^^^^
asyncio.exceptions.CancelledError

Current test program for test implementation:

async def asyncconnect(devicename, url, opcname, opcpassword, timeout):
    global client
    connect = False
    absolutpfad = os.getcwd()
    if opcname == "" and opcpassword == "":
        client = Client(url)
    elif opcname != "" and opcpassword != "":
        absolutpfad = os.getcwd()
        client = Client(url)
        client.set_user(opcname)
        client.set_password(opcpassword)
        devicename2 = devicename
        devicename2 = devicename2.replace(":", ".")
        await client.set_security_string(
            "Basic256Sha256,SignAndEncrypt,./OCV_Data/Certificates/" + devicename2 + ".pem,.\OCV_Data\Certificates\key_" + devicename[
                                                                                                                           12:17] + ".pem")
        client.application_uri = "urn:" + devicename

        await client.connect()
async def read(nodeid_string):
    global client

    node_list = []
    for nodeid in nodeid_string:
        node_list.append(client.get_node(nodeid))
    values = await client.read_values(node_list)
    print(values)




nodeid_string=['ns=3;s="USER_OPCUA_OCV_DATA_DB"."CellData"[1]."Celltype"','ns=3;s="USER_OPCUA_OCV_DATA_DB"."CellData"[1]."Partnumber_AI"']
asyncio.run(asyncconnect(devicename, url, opcname, opcpassword, timeout))
asyncio.run(read(nodeid_string))

[schroeder-]

This is not in the scope of the library. Its more of how to use asyncio.

You can use a real async task an wait for it, for example like this:

import asyncio

class OPCUAClient:
    def __init__(self) -> None:
        self.client = None
        self.connect_task = None
        self.connect = False

    async def asyncconnect(self, devicename, url, opcname, opcpassword, timeout):
        self.connect = False
        absolutpfad = os.getcwd()
        if opcname == "" and opcpassword == "":
            self.client = Client(url)
        elif opcname != "" and opcpassword != "":
            absolutpfad = os.getcwd()
            self.client = Client(url)
            self.client.set_user(opcname)
            self.client.set_password(opcpassword)
            devicename2 = devicename
            devicename2 = devicename2.replace(":", ".")
            await self.client.set_security_string(
            "Basic256Sha256,SignAndEncrypt,./OCV_Data/Certificates/" + devicename2 + ".pem,.\OCV_Data\Certificates\key_" + devicename[
                                                                                                                           12:17] + ".pem")
            self.client.application_uri = "urn:" + devicename
        self.connect_task = asyncio.create_task(self.do_connect)
        
    async def do_connect(self):
        await self.client.connect()
        self.connect = True

    async def read(nodeid_string):
        if not self.connected:
            asyncio.wait_for(self.connect_task)
        try:
            node_list = []
            for nodeid in nodeid_string:
                node_list.append(self.client.get_node(nodeid))
            values = await self.client.read_values(node_list)
            print(values)
        except:
            connected = False
            raise



cl = OPCUAClient()
nodeid_string=['ns=3;s="USER_OPCUA_OCV_DATA_DB"."CellData"[1]."Celltype"','ns=3;s="USER_OPCUA_OCV_DATA_DB"."CellData"[1]."Partnumber_AI"']
asyncio.run(cl.asyncconnect(devicename, url, opcname, opcpassword, timeout))
asyncio.run(cl.read(nodeid_string))