asynchronous grab and PyQt

[GuillaumeBeaudin]

Hello,
I'm trying to make a GUI for a scientific setup. When I want to put my camera in streaming mode I get all sorts of errors. Here is a basic example:

import sys

from PyQt6.QtCore import QThread
from PyQt6.QtWidgets import QApplication, QWidget
from vmbpy import Camera, Frame, Stream, VmbSystem, __version__


def handler(cam: Camera, stream: Stream, frame: Frame):
    print(f"{cam} acquired {frame}", flush=True)

    cam.queue_frame(frame)


class WorkerThread(QThread):
    def __init__(self, cam):
        super().__init__()
        self.cam = cam

    def run(self):
        self.cam.start_streaming(handler=handler)
        self.msleep(10000)
        self.cam.stop_streaming()


class ExampleApp(QWidget):
    def __init__(self, cam):
        super().__init__()

        self.cam = cam
        self.worker = WorkerThread(self.cam)
        self.worker.start()


if __name__ == "__main__":
    app = QApplication(sys.argv)
    with VmbSystem.get_instance() as vmb:
        print(
            "Python Version: "
            f"{sys.version_info.major}."
            f"{sys.version_info.minor}."
            f"{sys.version_info.micro}"
        )
        print(f"vmbpy version: {__version__}")
        cams = vmb.get_all_cameras()
        with cams[0] as cam:
            example = ExampleApp(cam)
            example.show()
    sys.exit(app.exec())

If I run this I get this output:

Python Version: 3.11.5
vmbpy version: 1.0.4
...
RuntimeError: Called 'FeatureContainer.get_all_features()' outside of Cameras 'with' context.

I have tried a lot of variation with QThread, treading.Thread and QRunnable, but I always have issues that seems to be related to the fact that the cam object exit the context at some point or another.

I'm out of idea on how to make this work. Is there a way to start the streaming in a QThread and still be able to change the cam parameters from the main window?

Thanks for your help!

[arunprakash-avt]

Please refer to the mutlithreading example code . Where it is explained how to adjust the cam parameter outside the main thread and exaplain the usage of multithreading.

[GuillaumeBeaudin]

Thanks for your suggestion.

In the mean time I managed to find a way to make my program do what I wanted with QThread. To avoid the problem with the context being exited, I put the context manager inside the thread.

Here is a working example, feel free to modify it and put it in your examples if you want (or not):

import sys
from queue import Queue

import numpy as np
from PyQt6.QtCore import QThread
from PyQt6.QtGui import QImage, QPixmap
from PyQt6.QtWidgets import (
    QApplication,
    QDoubleSpinBox,
    QLabel,
    QPushButton,
    QVBoxLayout,
    QWidget,
)
from vmbpy import Camera, Frame, PixelFormat, Stream, VmbSystem


class Handler:
    def __init__(self, label):
        self.label: QLabel = label

    def __call__(self, cam: Camera, stream: Stream, frame: Frame):
        frame_mono8: Frame = frame.convert_pixel_format(PixelFormat.Mono8)
        frame_array: np.ndarray = np.squeeze(frame_mono8.as_numpy_ndarray())
        height: int
        width: int
        height, width = frame_array.shape
        bytes_per_line: int = width
        q_image: QImage = QImage(
            frame_array,
            width,
            height,
            bytes_per_line,
            QImage.Format.Format_Grayscale8,
        )
        pixmap: QPixmap = QPixmap.fromImage(q_image)
        self.label.setPixmap(pixmap)

        cam.queue_frame(frame)


class CameraThread(QThread):
    def __init__(self, label: QLabel, queue: Queue):
        super().__init__()

        self.label: QLabel = label
        self.queue: Queue = queue

        self.handler: Handler = Handler(label=self.label)
        self.stop: bool = False

    def run(self):
        with VmbSystem.get_instance() as vmb:
            cam: Camera = vmb.get_all_cameras()[0]
            with cam as cam:
                cam.start_streaming(handler=self.handler)
                while True:
                    self.msleep(100)
                    if not self.queue.empty():
                        feature: str
                        value: bool | float | str
                        feature, value = self.queue.get()
                        print(f"{feature}: {value}", flush=True)
                        cam.get_feature_by_name(feature).set(value)
                    if self.stop:
                        cam.stop_streaming()
                        break


class MainWindow(QWidget):
    def __init__(self):
        super().__init__()

        self.queue: Queue = Queue()

        self.initUI()
        self.start_streaming()

    def initUI(self):
        self.label: QLabel = QLabel()
        self.label.setFixedSize(616 * 2, 514 * 2)

        self.start_button: QPushButton = QPushButton("Start")
        self.start_button.clicked.connect(self.start_streaming)

        self.stop_button: QPushButton = QPushButton("Stop")
        self.stop_button.clicked.connect(self.stop_streaming)
        self.stop_button.setEnabled(False)

        self.spin_box: QDoubleSpinBox = QDoubleSpinBox()
        self.spin_box.setDecimals(0)
        self.spin_box.setRange(64, 9000000)
        self.spin_box.setValue(4992)
        self.spin_box.setSuffix(" µs")
        self.spin_box.setSingleStep(10000)
        self.spin_box.valueChanged.connect(self.update_exposure_time)
        self.spin_box.editingFinished.connect(self.update_exposure_time)

        layout: QVBoxLayout = QVBoxLayout()
        layout.addWidget(self.label)
        layout.addWidget(self.spin_box)
        layout.addWidget(self.start_button)
        layout.addWidget(self.stop_button)
        self.setLayout(layout)

    def start_streaming(self):
        self.camera_thread: CameraThread = CameraThread(
            label=self.label, queue=self.queue
        )
        self.camera_thread.start()
        self.start_button.setEnabled(False)
        self.stop_button.setEnabled(True)

    def stop_streaming(self):
        self.camera_thread.stop = True
        self.camera_thread.quit()
        self.camera_thread.wait()
        self.start_button.setEnabled(True)
        self.stop_button.setEnabled(False)

    def update_exposure_time(self):
        if not self.camera_thread.stop:
            self.queue.put(("ExposureTime", self.spin_box.value()))

    def closeEvent(self, event):
        self.stop_streaming()
        super().closeEvent(event)


if __name__ == "__main__":
    app = QApplication(sys.argv)
    main_window: MainWindow = MainWindow()
    main_window.show()
    sys.exit(app.exec())

Thanks again!