DelphiVCL and asyncio

[shineworld 73]

Hi all,

it is possible to use the asyncio with a DelphiVCL ui-based program?

 

I've implemented an asyncua Client to get data from an OPC UA Server.
Here is a simple program that prints some values collected from an OPC UA server.

"""CNC OPC UA Client."""
#-------------------------------------------------------------------------------
# Name:         cnc_opcua_client
#
# Purpose:      CNC OPC UA Client
#
# Note          Checked with Python 3.11.3
#
# Coding Style  https://www.python.org/dev/peps/pep-0008/
#-------------------------------------------------------------------------------
import time

import asyncio
import logging

from asyncua import Client

# == opc ua client settings ==
OPC_UA_SERVER_URL       = "opc.tcp://localhost:4840/"
OPC_UA_SERVER_URI       = "https://www.someone.com/opc-ua/server"
OPC_UA_ASYNC_SLEEP_TIME = 0.01

# == deveopment settings ==
DEBUG_LEVEL             = logging.CRITICAL      # avoid any log not critical
DEBUG_ENABLED           = False

# creates a logger
logger = logging.getLogger(__name__)

async def main():
    """Main entry point."""

    # creats opc ua client in with-exception-handled block
    async with Client(url=OPC_UA_SERVER_URL) as client:

        # gets namespace index
        idx = await client.get_namespace_index(OPC_UA_SERVER_URI)

        # gets program position node
        program_position_node = client.get_node("ns=2;s=CNC.ProgramPosition")

        # server updating loop
        while True:
            # does nothing, just process relax
            await asyncio.sleep(OPC_UA_ASYNC_SLEEP_TIME)

            # reads and prints program positions
            value = await program_position_node.read_value()
            print('X:{:10.3f}, Y:{:10.3f}, Z:{:10.3f}'. format(value[0], value[1], value[2]))

# checks if this module is running as the main program
if __name__ == "__main__":
    logging.basicConfig(level=DEBUG_LEVEL)
    asyncio.run(main(), debug=DEBUG_ENABLED)

Getting:

*** Remote Interpreter Reinitialized ***
X:    51.206, Y:     0.000, Z:     0.000
X:    51.206, Y:     0.000, Z:     0.000
X:    51.206, Y:     0.000, Z:     0.000
X:    51.206, Y:     0.000, Z:     0.000

At this point I would like to create a fashioned UI using DelphiVCL, but I'm unable to understand where to place the

asyncio.run(main(), debug=DEBUG_ENABLED)

because to get DelphiVCL UI the main loop is already done by Application.Run() call:

    # initializes GUI Application

    Application.Initialize()

    Application.Title = "OPC UA Client Demo"
 

    # creates main application form

    app = DesktopView(Application)

    app.Show()

    FreeConsole()
 

    # enters in vcl main loop

    Application.Run()
 

    # frees main application form

    app.Destroy()
 

Thanks in advance for any suggestion.
Best Regards
Silverio

[shineworld 73]

A silly solution was to place DelphiVCL in a thread, but I don't know if it is the better way...
Code presents a lot of global values, is just a test in waiting for a better idea from Python/DelphiVCL gurus 🙂
 

import time
import asyncio
import logging
import threading

from asyncua import Client

from delphivcl import *

# == opc ua client settings ==
OPC_UA_SERVER_URL       = "opc.tcp://localhost:4840/"
OPC_UA_SERVER_URI       = "https://someone.com/opc-ua/server"
OPC_UA_ASYNC_SLEEP_TIME = 0.01

# == deveopment settings ==
DEBUG_LEVEL             = logging.CRITICAL      # avoid any log not critical
DEBUG_ENABLED           = False

# creates a logger
logger = logging.getLogger(__name__)

app = None
opcua_main_exit = False
program_position = []

async def opcua_main(form):

    global opcua_main_exit
    global program_position

    async with Client(url=OPC_UA_SERVER_URL) as client:
        # gets namespace index
        idx = await client.get_namespace_index(OPC_UA_SERVER_URI)

        # gets program position node
        program_position_node = client.get_node("ns=2;s=CNC.ProgramPosition")

        while not opcua_main_exit:
            await asyncio.sleep(OPC_UA_ASYNC_SLEEP_TIME)

            # reads program positions
            value = await program_position_node.read_value()
            program_position = list(value)

class DesktopView(Form):

    def __init__(self, owner):
        self.SetProps(Caption = "Welcome")

        # creates labels for program position axes values
        self.label_x = Label(self)
        self.label_x.SetProps(Parent=self, Top=40, Left=20)
        self.label_x.Caption = 'Z : {:15.3f}'.format(0.0)

        self.label_y = Label(self)
        self.label_y.SetProps(Parent=self, Top=60, Left=20)
        self.label_y.Caption = 'Y : {:15.3f}'.format(0.0)

        self.label_z = Label(self)
        self.label_z.SetProps(Parent=self, Top=80, Left=20)
        self.label_z.Caption = 'X : {:15.3f}'.format(0.0)

        # create and set update timer
        self.tmrUpdate = Timer(self)
        self.tmrUpdate.Enabled = False
        self.tmrUpdate.Interval = 1
        self.tmrUpdate.OnTimer = self.__on_timer_update
        self.tmrUpdate.Enabled = True

    def __on_timer_update(self, sender):
        global program_position
        if len(program_position) == 6:
            self.label_x.Caption = str(program_position[0])
            self.label_y.Caption = str(program_position[1])
            self.label_z.Caption = str(program_position[2])

def app_main():
    # initializes GUI Application
    Application.Initialize()
    Application.Title = "OPC UA Client Demo"

    # creates main application form
    app = DesktopView(Application)
    app.Show()
    FreeConsole()

    # enters in vcl main loop
    Application.Run()

    global opcua_main_exit
    opcua_main_exit = True

    # frees main application form
    app.Destroy()

if __name__ == "__main__":
    app_thread = threading.Thread(target=app_main)
    app_thread.start()

    logging.basicConfig(level=DEBUG_LEVEL)
    asyncio.run(opcua_main(app))

 

[pyscripter 694]

Not too bad an idea.  For instance this appears to run OK.

 

import threading
import time
from delphivcl import *

class DesktopView(Form):

    def __init__(self, owner):
        self.SetProps(Caption = "Welcome")
        # create and set update timer
        self.tmrUpdate = Timer(self)
        self.tmrUpdate.Enabled = False
        self.tmrUpdate.Interval = 1
        self.tmrUpdate.OnTimer = self.__on_timer_update
        self.tmrUpdate.Enabled = True

    def __on_timer_update(self, sender):
        time.sleep(1)

def app_main():
    # initializes GUI Application
    Application.Initialize()
    Application.Title = "OPC UA Client Demo"
    # creates main application form
    app = DesktopView(Application)
    app.Show()
    FreeConsole()

    # enters in vcl main loop
    Application.Run()
    Application.Free()

def inthread():
    for i in range(20):
        print(i)
        time.sleep(1)

if __name__ == "__main__":
    threading.Thread(target=inthread).start()
    threading.Thread(target=app_main).start()

You could run the asyncio in a thread as in python - Running asyncio loop and tkinter gui - Stack Overflow.

Also please submit an issue to the delphivcl project to expose the Application.OnIdle event.   That would make the use of timer unnecessary.

[shineworld 73]

3 minutes ago, pyscripter said:

You could run the asyncio in a thread as in python - Running asyncio loop and tkinter gui - Stack Overflow.

Also please submit an issue to the delphivcl project to expose the Application.OnIdle event.   That would make the use of timer unnecessary.

In the first experiment, I tried to attach a method to the Application.OnIdle but, visible onto dir(Application) Python notice that is not published.
What is the right python4delphi repo from https://github.com/Embarcadero/python4delphi and https://github.com/pyscripter/python4delphi ?

Sincerely I don't know what to prefer and there are enought of differences between them.

[pyscripter 694]

The Embarcadero fork of P4D, is often ahead of the PyScripter repo, but it is currently behind.   Its main focus is the generation of the delphivcl and delphifmx  extension modules as well as Android support.  The two repos are synced regularly.  I am only responsible for the pyscripter P4D home repo.   In most cases it does not matter which one you use.

Edited November 10, 2023 by pyscripter