Spaces:
Runtime error
Runtime error
File size: 2,606 Bytes
e08f02e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
import os
import sys
import threading
import time
import tkinter as tk
from multiprocessing import Queue
from typing import List
from PIL import Image, ImageTk
from streamv2v.image_utils import postprocess_image
sys.path.append(os.path.join(os.path.dirname(__file__), "..", ".."))
def update_image(image_data: Image.Image, label: tk.Label) -> None:
"""
Update the image displayed on a Tkinter label.
Parameters
----------
image_data : Image.Image
The image to be displayed.
label : tk.Label
The labels where the image will be updated.
"""
width = 512
height = 512
tk_image = ImageTk.PhotoImage(image_data, size=width)
label.configure(image=tk_image, width=width, height=height)
label.image = tk_image # keep a reference
def _receive_images(
queue: Queue, fps_queue: Queue, label: tk.Label, fps_label: tk.Label
) -> None:
"""
Continuously receive images from a queue and update the labels.
Parameters
----------
queue : Queue
The queue to receive images from.
fps_queue : Queue
The queue to put the calculated fps.
label : tk.Label
The label to update with images.
fps_label : tk.Label
The label to show fps.
"""
while True:
try:
if not queue.empty():
label.after(
0,
update_image,
postprocess_image(queue.get(block=False), output_type="pil")[0],
label,
)
if not fps_queue.empty():
fps_label.config(text=f"FPS: {fps_queue.get(block=False):.2f}")
time.sleep(0.0005)
except KeyboardInterrupt:
return
def receive_images(queue: Queue, fps_queue: Queue) -> None:
"""
Setup the Tkinter window and start the thread to receive images.
Parameters
----------
queue : Queue
The queue to receive images from.
fps_queue : Queue
The queue to put the calculated fps.
"""
root = tk.Tk()
root.title("Image Viewer")
label = tk.Label(root)
fps_label = tk.Label(root, text="FPS: 0")
label.grid(column=0)
fps_label.grid(column=1)
def on_closing():
print("window closed")
root.quit() # stop event loop
return
thread = threading.Thread(
target=_receive_images, args=(queue, fps_queue, label, fps_label), daemon=True
)
thread.start()
try:
root.protocol("WM_DELETE_WINDOW", on_closing)
root.mainloop()
except KeyboardInterrupt:
return
|