import sys
# Async worker thread
-class AsyncWorker(QRunnable):
+class AsyncWorker(QRunnable, QObject):
+ def __init__(self):
+ QRunnable.__init__(self)
+ QObject.__init__(self)
+
+ exception_signal = Signal(Exception)
+
@Slot()
def run(self):
self.loop = asyncio.new_event_loop()
async def _stop_internal(self):
self.loop.stop()
+ async def dispatch_and_check_exception(coro):
+ try:
+ await coro
+ except Exception as e:
+ AsyncWorker.worker.exception_happened.emit(e)
+
def dispatch(coro):
- asyncio.run_coroutine_threadsafe(coro, AsyncWorker.worker.loop)
+ asyncio.run_coroutine_threadsafe(AsyncWorker.dispatch_and_check_exception(coro), AsyncWorker.worker.loop)
if not hasattr(AsyncWorker, "worker"):
AsyncWorker.worker = AsyncWorker()
-class MainWindow(QMainWindow):
-
- start_signal = Signal()
- done_signal = Signal()
+class InitialLayout(QWidget):
+ connected_signal = Signal(BIRD)
def __init__(self):
super().__init__()
- self.bird = BIRD("/run/bird/bird.ctl")
- self.initial_layout()
-
- def initial_layout(self):
- widget = QWidget()
- self.setCentralWidget(widget)
-
- layout = QVBoxLayout(widget)
+ self.layout = QVBoxLayout(self)
self.text = QLabel("No connection to BIRD.")
- layout.addWidget(self.text, alignment=Qt.AlignmentFlag.AlignCenter)
+ self.layout.addWidget(self.text, alignment=Qt.AlignmentFlag.AlignCenter)
async_trigger = QPushButton(text="Connect")
- async_trigger.clicked.connect(self.connect)
- layout.addWidget(async_trigger, alignment=Qt.AlignmentFlag.AlignCenter)
+ async_trigger.clicked.connect(self.connect_slot)
+ self.layout.addWidget(async_trigger, alignment=Qt.AlignmentFlag.AlignCenter)
@Slot()
- def connect(self):
+ def connect_slot(self):
+ self.bird = BIRD("/run/bird/bird.ctl")
+
async def f():
async with self.bird as b:
await b.version.update()
await b.status.update()
- self.text.setText(f"Connected to {b.version.name} {b.version.version}")
+ self.connected_signal.emit(b)
AsyncWorker.dispatch(f())
+
+class ConnectedLayout(QWidget):
+ def __init__(self, bird):
+ super().__init__()
+ self.bird = bird
+
+ self.layout = QVBoxLayout(self)
+
+ self.main_info = QLabel(f"Connected to {bird.version.name} {bird.version.version}")
+ self.layout.addWidget(self.main_info)
+
+ self.status = QLabel(f"Status: {bird.status.status}")
+ self.layout.addWidget(self.status)
+
+class MainWindow(QMainWindow):
+ def __init__(self):
+ super().__init__()
+
+ AsyncWorker.worker.exception_signal.connect(self.exception_slot)
+
+ self.set_layout(InitialLayout())
+ self.layout.connected_signal.connect(self.connected_slot)
+
+ def set_layout(self, layout):
+ self.setCentralWidget(layout)
+ self.layout = layout
+
+ @Slot(Exception)
+ def exception_slot(self, e):
+ print("got exception")
+ raise Exception() from e
+
+ @Slot(BIRD)
+ def connected_slot(self, bird):
+ self.set_layout(ConnectedLayout(bird))
+
if __name__ == "__main__":
app = QApplication(sys.argv)
threadpool = QThreadPool()