I am working on a camera applicaiton where I need to get images from camera buffer and display it in a TK label live:
def display_thread_run(self):
while self.is_acquiring:
image_data = self.camera.get_next_image()
# --- some image conversion methods omitted ---
pil_image = Image.fromarray(image_data)
photo = ImageTk.PhotoImage(image=pil_image)
self.image_frame.after(0, self._create_photo_and_update_gui, photo)
time.sleep(0.01)
When the application exits, I need to close the camera stream, join the thread and at the very end, release camera related resources:
def stop_and_cleanup(self):
camera.stop_acquisition()
self.is_acquiring = False
if self.display_thread and self.display_thread.is_alive():
self.display_thread.join()
camera.release()
logger.info("Display resources released")
However, after I click the close window button (cross button), photo = ImageTk.PhotoImage(image=pil_image) causes deadlock and the thread won't join.
Setting thread.daemon=True and removing display_thread.join() won't work because camera.release() must be called after the thread is properly cleaned.
I'm using:
A minimum code example:
import tkinter as tk
from PIL import Image, ImageTk
import threading
import time
IMG_PATH = 'test_img.jpg'
class App:
def __init__(self, root):
self.root = root
self.label = tk.Label(root)
self.label.pack()
self.running = True
self.img = Image.open(IMG_PATH)
self.display_thread= threading.Thread(target=self.update_image_loop, daemon=True)
self.display_thread.start()
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
def update_image_loop(self):
while self.running:
tk_img = ImageTk.PhotoImage(image=self.img)
self.label.after(0, self.set_image, tk_img)
time.sleep(0.001)
def set_image(self, tk_img):
self.label.img = tk_img # keep reference
self.label.config(image=tk_img)
def on_close(self):
print("on_close")
self.running = False
self.display_thread.join()
self.root.destroy()
if __name__ == '__main__':
root = tk.Tk()
app = App(root)
root.mainloop()
tkinter is not thread-safe and it can make problem.
It works for me if I remove after() from thread and run directly set_image() but this make fickering image.
So I would use thread only to get (and process) image from camera,
def update_image_loop(self):
"""Run in other thread."""
while self.running:
self.frame = Image.open(IMG_PATH)
#self.frame = ... get image from camera ...
#... processing image ...
time.sleep(0.001)
and in main thread I would use after() to replace image on canvas.
def __init__(self, root):
# ... code ...
# start in main thread (after 1ms)
self.root.after(1, self.set_image)
def set_image(self):
"""Run in main thread."""
self.image = ImageTk.PhotoImage(image=self.frame)
self.label.config(image=self.image)
if self.running:
self.root.after(1, self.set_image) # repeat after 1ms
As for me it works fast enough.
My code used for tests:
I used image Lenna from Wikipedia
import tkinter as tk
from PIL import Image, ImageTk
import threading
import time
IMG_PATH = 'test_img.jpg'
IMG_PATH = 'lenna.png'
class App:
def __init__(self, root):
self.root = root
self.label = tk.Label(root)
self.label.pack()
self.running = True
# run in other thread
self.display_thread = threading.Thread(target=self.update_image_loop, daemon=True)
self.display_thread.start()
self.root.protocol("WM_DELETE_WINDOW", self.on_close)
# start in main thread (after 1ms)
self.root.after(1, self.set_image)
def update_image_loop(self):
"""Run in other thread."""
while self.running:
#print('[DEBUG] get image')
self.frame = Image.open(IMG_PATH)
#self.frame = ... get image from camera ...
#... processing image ...
#print('[DEBUG] sleep')
time.sleep(0.001)
#print('[DEBUG] exit loop')
def set_image(self):
"""Run in main thread."""
#print('[DEBUG] set_image')
self.image = ImageTk.PhotoImage(image=self.frame)
self.label.config(image=self.image)
#self.label['image'] = self.img
if self.running:
self.root.after(1, self.set_image) # repeat after 1ms
def on_close(self):
#print("[DEBUG] on_close")
self.running = False
#print('[DEBUG] join')
self.display_thread.join()
#print('[DEBUG] destroy')
self.root.destroy()
if __name__ == '__main__':
root = tk.Tk()
app = App(root)
root.mainloop()