If you need to provide an asynchronous-like interface while maintaining Python 2.7 compatibility (where async/await syntax is unavailable), you cannot use coroutines. Instead, you must use a callback-based pattern or a custom Future implementation that wraps a Qt timer or signal to simulate asynchronous completion.
In PyRPL, the goal is to allow Python 3 users to use await even if the underlying implementation uses Python 2-compatible callback logic. This is achieved by returning objects that behave like asyncio.Future or concurrent.futures.Future.
try:
from asyncio import Future, ensure_future
except ImportError:
from concurrent.futures import Future
class MyFuture(Future):
def __init__(self):
super(MyFuture, self).__init__()
# Use Qt timers to simulate async completion
self.timer = QtCore.QTimer()
self.timer.timeout.connect(lambda : self.set_result(rand(100)))
self.timer.setSingleShot(True)
self.timer.setInterval(1000)
self.timer.start()
def result(self):
# If not done, run a local event loop to wait
if not self.done():
self.loop = QtCore.QEventLoop()
self.add_done_callback(lambda x: self.loop.quit())
self.loop.exec_()
return super(MyFuture, self).result()
