Tutorial single loop async

 1'''
 2Tutorial - single_loop_async.py with asynchronous mode.
 3
 4This example project demonstrates how to use async thread to get RTC from EthanEXD.
 5
 6-------------------------------------------------------------------------------------
 7Please change correct serial number or IP and port number BEFORE you run example code.
 8
 9For other examples please check:
10    https://github.com/WPC-Systems-Ltd/WPC_Python_driver_release/tree/main/examples
11See README.md file to get detailed usage of this example.
12
13Copyright (c) 2022-2025 WPC Systems Ltd. All rights reserved.
14'''
15
16## WPC
17from wpcsys import pywpc
18
19## Python
20import asyncio
21import sys
22sys.path.insert(0, 'src/')
23
24
25async def loop_func(handle, timeout=3, delay=1):
26    t = 0
27    while t < timeout:
28        data = await handle.Sys_getRTC_async()
29        print("RTC Time:" + str(data))
30        await asyncio.sleep(delay)  ## delay [sec]
31        t += delay
32
33
34async def main():
35    ## Get Python driver version
36    print(f'{pywpc.PKG_FULL_NAME} - Version {pywpc.__version__}')
37
38    ## Create device handle
39    dev = pywpc.EthanEXD()
40
41    ## Connect to device
42    try:
43        dev.connect("192.168.1.110")  ## Depend on your device
44    except Exception as err:
45        pywpc.printGenericError(err)
46        ## Release device handle
47        dev.close()
48        return
49
50    ## Perform async thread to query data
51    try:
52        ## Get firmware model & version
53        driver_info = await dev.Sys_getDriverInfo_async()
54        print(f"Model name: {driver_info[0]}, Firmware version: {driver_info[-1]} ")
55
56        await loop_func(dev, timeout=3, delay=1)
57    except Exception as err:
58        pywpc.printGenericError(err)
59
60    ## This part never execute because the async thread
61    finally:
62        ## Disconnect device
63        dev.disconnect()
64
65        ## Release device handle
66        dev.close()
67
68
69def main_for_spyder(*args):
70    if asyncio.get_event_loop().is_running():
71        return asyncio.create_task(main(*args)).result()
72    else:
73        return asyncio.run(main(*args))
74
75
76if __name__ == '__main__':
77    asyncio.run(main())  ## Use terminal
78    # await main()  ## Use Jupyter or IPython(>=7.0)
79    # main_for_spyder()  ## Use Spyder