I’m trying to execute code inside a jupyter kernel in a Qt application. I have the below snipplet that is supposed to asynchronously run the code and then print the result
import sys import asyncio import qasync from qasync import QApplication from PySide6.QtWidgets import QWidget from jupyter_client import AsyncKernelManager CODE = """print('test')""" class Test(): def __init__(self): kernel_manager = AsyncKernelManager() kernel_manager.start_kernel() self.client = kernel_manager.client() self.client.start_channels() def run(self): loop = asyncio.get_event_loop() asyncio.ensure_future(self.execute(), loop=loop) async def execute(self): self.client.execute(CODE) response: Coroutine = self.client.get_shell_msg() print('Before') res = await response print('After') def main(): app = QApplication(sys.argv) test = Test() test.run() sys.exit(app.exec()) main()
With the above I get the following output
/tmp/test/test.py:16: RuntimeWarning: coroutine 'KernelManager._async_start_kernel' was never awaited kernel_manager.start_kernel() RuntimeWarning: Enable tracemalloc to get the object allocation traceback /tmp/test/test.py:22: DeprecationWarning: There is no current event loop loop = asyncio.get_event_loop()
so trying to adjust the code according to an example from qasync to something like
async def main(): app = QApplication(sys.argv) test = Test() test.run() sys.exit(app.exec()) qasync.run(main())
will result in the following exception
Traceback (most recent call last): File "/tmp/test/test.py", line 40, in <module> qasync.run(main()) File "/tmp/test/.venv/lib/python3.10/site-packages/qasync/__init__.py", line 821, in run return asyncio.run(*args, **kwargs) File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run return loop.run_until_complete(main) File "/tmp/test/.venv/lib/python3.10/site-packages/qasync/__init__.py", line 409, in run_until_complete return future.result() File "/tmp/test/test.py", line 34, in main app = QApplication(sys.argv) RuntimeError: Please destroy the QApplication singleton before creating a new QApplication instance.
I’m pretty at lost at this point, does anyone know how to get this to work?
Advertisement
Answer
You have to create a QEventLoop, also start_kernel must use await. On the other hand it first imports PySide6 and then the other libraries that depend on PySide6 like qasync so that it can deduce the correct Qt binding.
import sys import asyncio from functools import cached_property from PySide6.QtWidgets import QApplication import qasync from jupyter_client import AsyncKernelManager CODE = """print('test')""" class Test: @cached_property def kernel_manager(self): return AsyncKernelManager() @cached_property def client(self): return self.kernel_manager.client() async def start(self): await self.kernel_manager.start_kernel() self.client.start_channels() asyncio.ensure_future(self.execute()) async def execute(self): self.client.execute(CODE) response = self.client.get_shell_msg() print("Before") res = await response print("After", res) def main(): app = QApplication(sys.argv) loop = qasync.QEventLoop(app) asyncio.set_event_loop(loop) test = Test() asyncio.ensure_future(test.start()) with loop: loop.run_forever() if __name__ == "__main__": main()