How can I preview streaming images in tkinter with Allied Vision camera that uses Vimba SDK?

Question

I want to display images from an Allied Vision camera inside a tkinter frame using OpenCV and the SDK for the camera, VimbaPython.

The only possible way to initialize the camera is with a Python with statement:

with Vimba.get_instance() as vimba:
    cams = vimba.get_all_cameras()
    with cams[0] as camera:
        camera.get_frame()
        # Convert frame to opencv image, then use Image.fromarray and ImageTk.PhotoImage to
        # display it on the tkinter GUI

Everything works fine so far. But I don't only need a single frame. Instead, I need to continuously get frames and display them on the screen so that it is streaming. I found that one way to do it is to call the .after(delay, function) method from a tkinter Label widget. So, after obtaining one frame, I want to call the same function to get a new frame and display it again. The code would look like that:

with Vimba.get_instance() as vimba:
    cams = vimba.get_all_cameras()
    with cams[0] as camera:

        def show_frame():
            frame = camera.get_frame()
            frame = frame.as_opencv_image()
            im = Image.fromarray(frame)
            img = Image.PhotoImage(im)
            lblVideo.configure(image=img)   # this is the Tkinter Label Widget
            lblVideo.image = img

        show_frame()
        lblVideo.after(20, show_frame)

Then this shows the first frame and stops, throwing an error saying that Vimba needs to be initialized with a with statement. I don't know much about Python, but it looks like when I call the function with the .after() method it ends the with statement.

I would like to know if it is possible to execute this show_frame() function without ending the with. Also, I can't initialize the camera every time because the program goes really slow. Thank you

Answer 1

You need to use thread to run the capture code and pass the frames read via queue. Then the main tkinter application reads the queue and show the frames periodically using .after().

Below is an example based on your posted code:

import threading
from queue import SimpleQueue
import tkinter as tk
from PIL import Image, ImageTk
from vimba import Vimba

def camera_streaming(queue):
    global is_streaming
    is_streaming = True
    print("streaming started")
    with Vimba.get_instance() as vimba:
        with vimba.get_all_cameras()[0] as camera:
            while is_streaming:
                frame = camera.get_frame()
                frame = frame.as_opencv_image()
                im = Image.fromarray(frame)
                img = ImageTk.PhotoImage(im)
                queue.put(img) # put the capture image into queue
    print("streaming stopped")

def start_streaming():
    start_btn["state"] = "disabled" # disable start button to avoid running the threaded task more than once
    stop_btn["state"] = "normal"    # enable stop button to allow user to stop the threaded task
    show_streaming()
    threading.Thread(target=camera_streaming, args=(queue,), daemon=True).start()

def stop_streaming():
    global is_streaming, after_id
    is_streaming = False  # terminate the streaming thread
    if after_id:
        lblVideo.after_cancel(after_id) # cancel the showing task
        after_id = None
    stop_btn["state"] = "disabled" # disable stop button
    start_btn["state"] = "normal"  # enable start button

# periodical task to show frames in queue
def show_streaming():
    global after_id
    if not queue.empty():
        image = queue.get()
        lblVideo.config(image=image)
        lblVideo.image = image
    after_id = lblVideo.after(20, show_streaming)

queue = SimpleQueue() # queue for video frames
after_id = None

root = tk.Tk()

lblVideo = tk.Label(root, image=tk.PhotoImage(), width=640, height=480)
lblVideo.grid(row=0, column=0, columnspan=2)

start_btn = tk.Button(root, text="Start", width=10, command=start_streaming)
start_btn.grid(row=1, column=0)

stop_btn = tk.Button(root, text="Stop", width=10, command=stop_streaming, state="disabled")
stop_btn.grid(row=1, column=1)

root.mainloop()

Note that I don't have the camera and the SDK installed, the above code may not work for you. I just demonstrate how to use thread, queue and .after().

Below is a testing vimba module (saved as vimba.py) I use to simulate VimbaPython module using OpenCV and a webcam:

import cv2

class Frame:
    def __init__(self, frame):
        self.frame = frame

    def as_opencv_image(self):
        return self.frame

class Camera:
    def __init__(self, cam_id=0):
        self.cap = cv2.VideoCapture(cam_id, cv2.CAP_DSHOW)

    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.cap.release()
        return self

    def get_frame(self):
        ret, frame = self.cap.read()
        if ret:
            return Frame(frame)

class Vimba:
    _instance = None
    
    @classmethod
    def get_instance(self):
        if self._instance is None:
            self._instance = Vimba()
        return self._instance

    def __enter__(self):
        return self

    def __exit__(self, *args):
        return self

    def get_all_cameras(self):
        return (Camera(),)

Answer 2

I tried to read the frames in openCV and display them in tkinter label. I was able to do so using the below code:

import tkinter as tk
import cv2
from PIL import ImageTk, Image

video_path = "SAMPLE/STORED_VIDEO/PATH"

root = tk.Tk()
base_img = Image.open("PATH/TO/DEFAULT/LABLE/IMAGE")
img_obj = ImageTk.PhotoImage(base_img)
lblVideo = tk.Label(root, image=img_obj)
lblVideo.pack()
cap = cv2.VideoCapture(video_path)

if cap.isOpened():
    def show_frame():
        _, frame = cap.read()
        im = Image.fromarray(frame)
        img = ImageTk.PhotoImage(im)
        lblVideo.configure(image=img)
        lblVideo.image = img
        lblVideo.after(1, show_frame)  # Need to create callback here   
    show_frame()
root.mainloop()

Although this doesnot contain the with statement, you can try replacing the after() callback inside the show_frame function itself.

Answer 3

I know this question is pretty old, but I ran into a similar problem with the Allied Vision cameras and found the solution to be relatively robust. So I hope this helps someone, even if not the OP.

An alternative to using with statements is using __enter__ and __exit__ (see sample here). With this, I created a class for the Vimba camera and during the __init__ I used these functions twice: once to initialize the Vimba instance, and once to open the camera itself. An example as follows...

vimba_handle = Vimba.get_instance().__enter__()

camera = vimba_handle.get_all_cameras()[0].__enter__()

I'll include a longer snippet as code as well, but please note my purpose was slightly different the OP's intent. Hopefully, it is still useful.

class VimbaCam:

    def __init__(self, device_id=0):
        # Variables
        self.current_frame = np.array([])
        self.device = None
        self.device_id = device_id
        self.vimba_handle = Vimba.get_instance().__enter__()
        self.is_streaming = False
        self.scale_window = 4
        self.stream_thread = threading.Thread(target=self.thread_stream, daemon=True)

        # Default settings
        self.auto_exposure = "Off"
        self.auto_gain = "Off"
        self.acquisition = "Continuous"
        self.exposure_us = 200000
        self.fps = 6.763
        self.gain = 0
        self.gamma = 1

        self.open()

    def close(self):
        if self.device is not None:
            if self.is_streaming:
                self.stop_stream()
                time.sleep(1)
            self.device.__exit__(None, None, None)
            self.vimba_handle.__exit__(None, None, None)

    def open(self):
        cams = self.vimba_handle.get_all_cameras()
        if not cams:
            error_check(151, currentframe())
        else:
            self.device = cams[self.device_id].__enter__()

        self.set_defaults()
        self.start_stream()

    def start_stream(self):
        if self.device is not None:
            self.is_streaming = True
            self.stream_thread.start()
            time.sleep(1)

    def thread_stream(self):
        while self.is_streaming:
            current_frame = self.device.get_frame().as_opencv_image()
            h, w, _ = current_frame.shape
            self.current_frame = current_frame.reshape((h, w))
        self.stream_thread = threading.Thread(target=self.thread_stream, daemon=True)

    def stop_stream(self):
        if self.device is not None:
            self.is_streaming = False

    def live_video(self):
        if self.device is not None:
            window_name = "Allied Vision"
            h, w = self.current_frame.shape
            w = int(w / self.scale_window)
            h = int(h / self.scale_window)
            cv2.namedWindow(window_name, cv2.WINDOW_NORMAL)
            cv2.resizeWindow(window_name, w, h)
            while 1:
                cv2.imshow(window_name, self.current_frame)
                cv2.waitKey(1)
                if cv2.getWindowProperty(window_name, cv2.WND_PROP_VISIBLE) < 1:
                    break
            cv2.destroyAllWindows()

    def set_defaults(self):
        if self.device is not None:
            # Exposure time settings
            self.device.ExposureAuto.set(self.auto_exposure)
            self.device.ExposureTimeAbs.set(self.exposure_us)

            # Gain settings
            self.device.GainAuto.set(self.auto_gain)
            self.device.Gain.set(self.gain)

            # Gamma settings
            self.device.Gamma.set(self.gamma)
            self.device.AcquisitionMode.set(self.acquisition)
            self.device.AcquisitionFrameRateAbs.set(self.fps)

            # Try to adjust GeV packet size (available for GigE only)
            try:
                self.device.GVSPAdjustPacketSize.run()
                while not self.device.GVSPAdjustPacketSize.is_done():
                    pass
            except (AttributeError, VimbaFeatureError):
                pass

            # Color formatting (tries mono first, then color)
            cv_formats = intersect_pixel_formats(self.device.get_pixel_formats(), OPENCV_PIXEL_FORMATS)
            mono_formats = intersect_pixel_formats(cv_formats, MONO_PIXEL_FORMATS)
            color_formats = intersect_pixel_formats(cv_formats, COLOR_PIXEL_FORMATS)
            if mono_formats:
                self.device.set_pixel_format(mono_formats[0])
            elif color_formats:
                self.device.set_pixel_format(color_formats[0])

if __name__ == "__main__":

    dev = VimbaCam()
    dev.live_video()
    dev.close()