rbaleksandar
Reputation: 9701
Specifying which item is to be consumed by which consumer for a Python multiprocessing Queue
The following code is work in progress. It is an upgrade from the one I posted here regarding another issue. It uses a pipe for communicating (currently only the close event) data between the GUI (producer) and the controller (consumer) that controls the termination of the application (just like in the previous version of the code). The essential upgrade is the replacement of the pipe (for communicating data between the GUI and the image processing process) with a queue.
The reason I have picked a queue is because it is thread-safe and in the future I would like to move each independent image processing task to a separate thread (if avaialble). For example after generating a grayscale image from the original one there are multiple independent steps that I can undertakes using it - Gaussian and the Harris corner detection (1), Canny edge detection (2, currently an issue with displaying the resulting image using ImageTk so ignore) etc. The GUI is there to just visualize so the order it receives the separate items is not important (hence the queue).
from multiprocessing import Process, Pipe, Queue
from threading import Thread
from concurrent.futures import ThreadPoolExecutor # For future use
import tkinter as tk
from tkinter import filedialog, ttk
from PIL import ImageTk, Image
import atexit
import cv2
import os
class ProcessGUI(object):
def __init__(self, queue_gui_ip, pipe_gui_co_out):
print("[ProcessGUI]: PID = " + str(os.getpid()))
self.queue_gui_ip = queue_gui_ip
self.pipe_gui_co_out = pipe_gui_co_out
self.poll_image_data_thread = Thread(target=self.poll_image_data)
self.setup_gui()
self.run()
def setup_gui(self):
self.app = tk.Tk()
self.app.protocol("WM_DELETE_WINDOW", self.close_event)
fr_browse_controls = ttk.Frame(self.app)
fr_browse_controls.pack(side=tk.TOP, expand=True, fill="x")
lb_browse = ttk.Label(fr_browse_controls, text="Path to image file:")
lb_browse.pack(side=tk.LEFT)
self.ent_browse = ttk.Entry(fr_browse_controls)
self.ent_browse.pack(side=tk.LEFT, expand=True, fill="x")
btn_browse = ttk.Button(fr_browse_controls, text="Browse", command=self.btn_browse_clicked)
btn_browse.pack(side=tk.RIGHT)
self.edge_algs = tk.StringVar(self.app)
om_edge_algs = ttk.OptionMenu(self.app, self.edge_algs, "canny", "canny", "sobelx", "sobely", "sobelxy")
om_edge_algs.pack(side=tk.TOP, expand=True, fill="x")
self.pb_load = ttk.Progressbar(self.app, orient=tk.HORIZONTAL, mode="determinate")
self.pb_load.pack(side=tk.TOP, expand=True, fill="x")
tw_images = ttk.Notebook(self.app)
tw_images.pack(side=tk.BOTTOM, expand=True, fill="both")
tb_original = ttk.Frame(tw_images)
self.image_original = None
tw_images.add(tb_original, text="RGB")
self.lb_image_original = ttk.Label(tb_original, image=None)
self.lb_image_original.pack(expand=True, fill="both")
tb_gray = ttk.Frame(tw_images)
self.image_gray = None
tw_images.add(tb_gray, text="Grayscale")
self.lb_image_gray = ttk.Label(tb_gray, image=None)
self.lb_image_gray.pack(expand=True, fill="both")
tb_gaussian = ttk.Frame(tw_images)
self.image_gaussian = None
tw_images.add(tb_gaussian, text="Gaussian")
self.lb_image_gaussian = ttk.Label(tb_gaussian, image=None)
self.lb_image_gaussian.pack(expand=True, fill="both")
tb_edges = ttk.Frame(tw_images)
self.image_edges = None
tw_images.add(tb_edges, text="Edges")
self.lb_image_edges = ttk.Label(tb_edges, image=None)
self.lb_image_edges.pack(expand=True, fill="both")
tb_corners = ttk.Frame(tw_images)
self.image_corners = None
tw_images.add(tb_corners, text="Harris corners")
self.lb_image_corners = ttk.Label(tb_corners, image=None)
self.lb_image_corners.pack(expand=True, fill="both")
def run(self):
try:
self.poll_image_data_thread.start()
self.app.mainloop()
except KeyboardInterrupt:
self.close_event()
def poll_image_data(self):
while True:
request = self.queue_gui_ip.get()
method = request[0]
args = request[1:]
print("------------[ProcessGUI]------------")
print("Method: " + method)
print("------------------------------------")
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
def display_image_dims_callback(self, height, width, channels):
print("[ProcessGUI]")
print("Height: " + str(height))
print("Width: " + str(width))
print("Channels: " + str(channels))
def display_image_processing_progress_callback(self, progress):
progress = 0 + ((100 - 0) / (len(ProcessImageProcessing.ProcessingSteps) - 0)) * (progress - 0)
print("[ProcessGUI]: Updating progress to " + str(progress) + "%")
self.pb_load["value"] = progress
def display_image_original_gray_callback(self, image_original, image_gray):
self.image_original = ImageTk.PhotoImage(Image.fromarray(image_original))
self.lb_image_original.configure(image=self.image_original)
self.image_gray = ImageTk.PhotoImage(Image.fromarray(image_gray))
self.lb_image_gray.configure(image=self.image_gray)
def display_image_gaussian_callback(self, image_gaussian):
self.image_gaussian = ImageTk.PhotoImage(Image.fromarray(image_gaussian))
self.lb_image_gaussian.configure(image=self.image_gaussian)
def display_image_edges_callback(self, image_edges):
self.image_edges = ImageTk.PhotoImage(Image.fromarray(image_edges))
self.lb_image_edges.configure(image=self.image_edges)
def display_image_corners_callback(self, image_corners):
self.image_corners = ImageTk.PhotoImage(Image.fromarray(image_corners))
self.lb_image_corners.configure(image=self.image_corners)
def btn_browse_clicked(self):
filename = tk.filedialog.askopenfilename(initialdir=".",
title="Select image",
filetypes=(
("Portable Network graphics", "*.png"),
("All files", "*.*")))
self.ent_browse.delete(0, tk.END)
self.ent_browse.insert(0, filename)
edge_alg = self.edge_algs.get()
self.queue_gui_ip.put(["process_image", filename, edge_alg])
def close_event(self):
print("[ProcessGUI]: Shutting down")
self.pipe_gui_co_out.send(["close"])
if self.poll_image_data_thread.is_alive():
self.poll_image_data_thread.join()
self.pipe_gui_co_out.close()
self.app.destroy()
class ProcessImageProcessing(object):
ProcessingSteps = [
"rgb",
"gray",
"gaussian",
"edges",
"corners"
]
def __init__(self, queue_gui_ip):
print("[ProcessImageProcessing]: PID = " + str(os.getpid()))
self.queue_gui_ip = queue_gui_ip
# atexit.register(self.close)
self.run()
def run(self):
while True:
request = self.queue_gui_ip.get()
method = request[0]
args = request[1:]
print("------[ProcessImageProcessing]------")
print("Method: " + method)
print("------------------------------------")
if "display_" in method:
# Skip incoming requests that contain methods meant for ProcessGUI class
continue
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
def process_image_callback(self, image_path, edge_alg):
print("[ProcessImageProcessing]: Received file \"" + image_path + "\"")
try:
progress = 0
original_bgr = cv2.imread(image_path)
original_rgb = cv2.cvtColor(original_bgr, cv2.COLOR_BGR2RGB)
(height, width, channels) = original_bgr.shape[:3]
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_dims", height, width, channels])
gray = cv2.cvtColor(original_bgr, cv2.COLOR_BGR2GRAY)
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_original_gray", original_rgb, gray])
gaussian = self.process_image_gaussian(gray)
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_gaussian", gaussian])
try:
edges = self.process_image_edges(gaussian, edge_alg)
if not edges:
print("Unknown edge detection algorithm")
else:
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_edges", edges])
except:
pass
corners = self.process_image_corners(original_rgb, gray)
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_corners", corners])
except Exception as ex:
print("Exception: " + str(ex))
def process_image_gaussian(self, image_gray):
gaussian = cv2.GaussianBlur(image_gray, (3, 3), cv2.BORDER_DEFAULT)
return gaussian
def process_image_edges(self, image_gaussian, edge_alg):
edges = None
if edge_alg not in ["canny", "sobelx", "sobely", "sobelxy"]:
return edges
# Sobel edge detection
# Sobel edge detection on the X axis
if edge_alg == "sobelx":
print("Sobel X")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=1, dy=0, ksize=5)
# Sobel edge detection on the Y axis
elif edge_alg == "sobely":
print("Sobel Y")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=0, dy=1, ksize=5)
# Combined X and Y Sobel edge detection
elif edge_alg == "sobelxy":
print("Sobel XY")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=1, dy=1, ksize=5)
# Canny edge detection
elif edge_alg == "canny":
print("Canny")
edges = cv2.Canny(image=image_gaussian, threshold1=100, threshold2=200)
return edges
def process_image_corners(self, image_original, image_gray):
original_with_corners = image_original
corners = cv2.cornerHarris(image_gray, 2, 3, 0.04)
# result is dilated for marking the corners, not important
corners = cv2.dilate(corners, None)
# Threshold for an optimal value, it may vary depending on the image.
original_with_corners[corners > 0.01 * corners.max()] = [0, 0, 255]
return original_with_corners
def close(self):
print("[ProcessImageProcessing]: Shutting down")
class Controller(object):
def __init__(self):
print("[Controller]: PID = " + str(os.getpid()))
queue_gui_ip = Queue()
pipe_gui_co_out, self.pipe_co_in = Pipe()
self.gui = Process(
target=ProcessGUI,
args=(queue_gui_ip, pipe_gui_co_out)
)
self.ip = Process(
target=ProcessImageProcessing,
args=(queue_gui_ip,)
)
def run(self):
try:
self.gui.start()
self.ip.start()
while self.pipe_co_in.poll(None):
try:
request = self.pipe_co_in.recv()
method = request[0]
args = request[1:]
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
except EOFError:
# Raised when nothing to receive from pipe
pass
except KeyboardInterrupt:
self.close_callback()
except BrokenPipeError:
self.close_callback()
def close_callback(self):
print("Quitting processes...")
self.gui.join(1)
if self.gui.is_alive():
self.gui.terminate()
self.ip.join(1)
if self.ip.is_alive():
self.ip.terminate()
print("Finished")
def main():
c = Controller()
c.run()
if __name__ == "__main__":
main()
My problem is in the way a Queue works. Unlike a duplex Pipe, where I have never encountered this issue, a Queue offers its items to both processes - ProcessGUI and ProcessImageProcessing. After an item is retrieved it is (naturally) consumed. The problem is that my main producer - the ProcessImageProcessing - received a large amount of the items it has put into the queue for the lesser producer - the ProcessGUI. This leads to items being consumed by the wrong consumer.
I am looking for a way to solve this problem. At first I thought that I can consume an item and, if it is not meant for the consumer that consumed it, put it back into the queue (hence the print statements and the
if "display_" in method:
continue
inside ProcessImageProcessing, where I was thinking of bouncing back the item that is meant for ProcessGUI. This clearly has many flaws including checking every item for validity and more importantly the absence of a guarantee that an item, that was put back into the queue, will next time be consumed by the correct consumer. This can potentially lead to a scenario, where the item is continuously consumed by the wrong consumer and bounced back indefinitely.
My next thought is to perhaps add a Pipe for incoming data (ProcessGUI to ProcessImageProcessing) and have the Queue be used for transferring the image data from a single producer (ProcessImageProcessing) and a single consumer (ProcessGUI). This seems like the right way to go but I am curious if there is another way so that I learn more about IPC and Pipe/Queue in Python.
Here is the updated code (far from perfect :D) with the additional Pipe and a "single producer, single consumer" Queue:
from multiprocessing import Process, Pipe, Queue
from threading import Thread
from concurrent.futures import ThreadPoolExecutor
import tkinter as tk
from tkinter import filedialog, ttk
from PIL import ImageTk, Image
import atexit
import cv2
import os
class ProcessGUI(object):
def __init__(self, queue_ip_gui, pipe_gui_ip_in, pipe_gui_co_out):
print("[ProcessGUI]: PID = " + str(os.getpid()))
self.queue_ip_gui = queue_ip_gui
self.pipe_gui_ip_in = pipe_gui_ip_in
self.pipe_gui_co_out = pipe_gui_co_out
self.poll_image_data_thread = Thread(target=self.poll_image_data)
self.setup_gui()
self.run()
def setup_gui(self):
# SAME AS ABOVE
# ...
def run(self):
try:
self.poll_image_data_thread.start()
self.app.mainloop()
except KeyboardInterrupt:
self.close_event()
def poll_image_data(self):
while True:
request = self.queue_ip_gui.get()
method = request[0]
args = request[1:]
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
def display_image_dims_callback(self, height, width, channels):
print("[ProcessGUI]")
print("Height: " + str(height))
print("Width: " + str(width))
print("Channels: " + str(channels))
def display_image_processing_progress_callback(self, progress):
progress = 0 + ((100 - 0) / (len(ProcessImageProcessing.ProcessingSteps) - 0)) * (progress - 0)
print("[ProcessGUI]: Updating progress to " + str(progress) + "%")
self.pb_load["value"] = progress
def display_image_original_gray_callback(self, image_original, image_gray):
self.image_original = ImageTk.PhotoImage(Image.fromarray(image_original))
self.lb_image_original.configure(image=self.image_original)
self.image_gray = ImageTk.PhotoImage(Image.fromarray(image_gray))
self.lb_image_gray.configure(image=self.image_gray)
def display_image_gaussian_callback(self, image_gaussian):
self.image_gaussian = ImageTk.PhotoImage(Image.fromarray(image_gaussian))
self.lb_image_gaussian.configure(image=self.image_gaussian)
def display_image_edges_callback(self, image_edges):
self.image_edges = ImageTk.PhotoImage(Image.fromarray(image_edges))
self.lb_image_edges.configure(image=self.image_edges)
def display_image_corners_callback(self, image_corners):
self.image_corners = ImageTk.PhotoImage(Image.fromarray(image_corners))
self.lb_image_corners.configure(image=self.image_corners)
def btn_browse_clicked(self):
filename = tk.filedialog.askopenfilename(initialdir=".",
title="Select image",
filetypes=(
("Portable Network graphics", "*.png"),
("All files", "*.*")))
self.ent_browse.delete(0, tk.END)
self.ent_browse.insert(0, filename)
edge_alg = self.edge_algs.get()
self.pipe_gui_ip_in.send(["process_image", filename, edge_alg])
def close_event(self):
print("[ProcessGUI]: Shutting down")
self.pipe_gui_co_out.send(["close"])
if self.poll_image_data_thread.is_alive():
self.poll_image_data_thread.join()
self.pipe_gui_co_out.close()
self.app.destroy()
class ProcessImageProcessing(object):
ProcessingSteps = [
"rgb",
"gray",
"gaussian",
"edges",
"corners"
]
def __init__(self, queue_gui_ip, pipe_gui_ip_out):
print("[ProcessImageProcessing]: PID = " + str(os.getpid()))
self.queue_gui_ip = queue_gui_ip
self.pipe_gui_ip_out = pipe_gui_ip_out
self.run()
def run(self):
try:
while self.pipe_gui_ip_out.poll(None):
try:
request = self.pipe_gui_ip_out.recv()
method = request[0]
args = request[1:]
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
except EOFError:
# Raised when nothing to receive from pipe
pass
except KeyboardInterrupt:
self.close()
except BrokenPipeError:
# Raised when exiting the process
self.close()
def process_image_callback(self, image_path, edge_alg):
print("[ProcessImageProcessing]: Received file \"" + image_path + "\"")
try:
progress = 0
original_bgr = cv2.imread(image_path)
original_rgb = cv2.cvtColor(original_bgr, cv2.COLOR_BGR2RGB)
(height, width, channels) = original_bgr.shape[:3]
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_dims", height, width, channels])
gray = cv2.cvtColor(original_bgr, cv2.COLOR_BGR2GRAY)
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_original_gray", original_rgb, gray])
gaussian = self.process_image_gaussian(gray)
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_gaussian", gaussian])
try:
edges = self.process_image_edges(gaussian, edge_alg)
if not edges:
print("Unknown edge detection algorithm")
else:
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_edges", edges])
except:
pass
corners = self.process_image_corners(original_rgb, gray)
progress += 1
self.queue_gui_ip.put(["display_image_processing_progress", progress])
self.queue_gui_ip.put(["display_image_corners", corners])
except Exception as ex:
print("Exception: " + str(ex))
def process_image_gaussian(self, image_gray):
gaussian = cv2.GaussianBlur(image_gray, (3, 3), cv2.BORDER_DEFAULT)
return gaussian
def process_image_edges(self, image_gaussian, edge_alg):
edges = None
if edge_alg not in ["canny", "sobelx", "sobely", "sobelxy"]:
return edges
if edge_alg == "sobelx":
print("Sobel X")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=1, dy=0, ksize=5)
elif edge_alg == "sobely":
print("Sobel Y")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=0, dy=1, ksize=5)
elif edge_alg == "sobelxy":
print("Sobel XY")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=1, dy=1, ksize=5)
elif edge_alg == "canny":
print("Canny")
edges = cv2.Canny(image=image_gaussian, threshold1=100, threshold2=200)
return edges
def process_image_corners(self, image_original, image_gray):
original_with_corners = image_original
corners = cv2.cornerHarris(image_gray, 2, 3, 0.04)
corners = cv2.dilate(corners, None)
original_with_corners[corners > 0.01 * corners.max()] = [0, 0, 255]
return original_with_corners
def close(self):
print("[ProcessImageProcessing]: Shutting down")
class Controller(object):
def __init__(self):
print("[Controller]: PID = " + str(os.getpid()))
queue_ip_gui = Queue()
pipe_gui_ip_in, pipe_gui_ip_out = Pipe()
pipe_gui_co_out, self.pipe_co_in = Pipe()
self.gui = Process(
target=ProcessGUI,
args=(queue_ip_gui, pipe_gui_ip_in, pipe_gui_co_out)
)
self.ip = Process(
target=ProcessImageProcessing,
args=(queue_ip_gui, pipe_gui_ip_out)
)
def run(self):
try:
self.gui.start()
self.ip.start()
while self.pipe_co_in.poll(None):
try:
request = self.pipe_co_in.recv()
method = request[0]
args = request[1:]
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
except EOFError:
# Raised when nothing to receive from pipe
pass
except KeyboardInterrupt:
self.close_callback()
except BrokenPipeError:
self.close_callback()
def close_callback(self):
print("Quitting processes...")
self.gui.join(1)
if self.gui.is_alive():
self.gui.terminate()
self.ip.join(1)
if self.ip.is_alive():
self.ip.terminate()
print("Finished")
def pipes():
c = Controller()
c.run()
if __name__ == "__main__":
pipes()
Upvotes: 0
Views: 84
Answers (1)
Booboo
Reputation: 44283
I believe the best way to go is to have an input queue for the ProcessImageProcessing process and a output queue that this process will write its results to and from which ProcessGUI will read from.
I would also make ProcessGui a thread launched from the main process and get rid of using a Pipe to communicate the close event and use instead an actual threading.Event instance. This allows the main thread to simply wait for the GUI to close without burning CPU cycles in a polling loop. Finally, by using daemon thread for doing the queue get requests in ProcessGui and making the ProcessImageProcessing process a daemon process, the logic becomes somewhat simplified.
from multiprocessing import Process, Pipe, Queue
from threading import Thread, Event
import tkinter as tk
from tkinter import filedialog, ttk
from PIL import ImageTk, Image
import cv2
import os
class ProcessGUI(object):
def __init__(self, queue_gui_ip, queue_gui_op, termination_event):
print("[ProcessGUI]: PID = " + str(os.getpid()))
self.queue_gui_ip = queue_gui_ip
self.queue_gui_op = queue_gui_op
self.termination_event = termination_event
self.poll_image_data_thread = Thread(target=self.poll_image_data, daemon=True)
self.setup_gui()
self.run()
def setup_gui(self):
self.app = tk.Tk()
self.app.protocol("WM_DELETE_WINDOW", self.close_event)
fr_browse_controls = ttk.Frame(self.app)
fr_browse_controls.pack(side=tk.TOP, expand=True, fill="x")
lb_browse = ttk.Label(fr_browse_controls, text="Path to image file:")
lb_browse.pack(side=tk.LEFT)
self.ent_browse = ttk.Entry(fr_browse_controls)
self.ent_browse.pack(side=tk.LEFT, expand=True, fill="x")
btn_browse = ttk.Button(fr_browse_controls, text="Browse", command=self.btn_browse_clicked)
btn_browse.pack(side=tk.RIGHT)
self.edge_algs = tk.StringVar(self.app)
om_edge_algs = ttk.OptionMenu(self.app, self.edge_algs, "canny", "canny", "sobelx", "sobely", "sobelxy")
om_edge_algs.pack(side=tk.TOP, expand=True, fill="x")
self.pb_load = ttk.Progressbar(self.app, orient=tk.HORIZONTAL, mode="determinate")
self.pb_load.pack(side=tk.TOP, expand=True, fill="x")
tw_images = ttk.Notebook(self.app)
tw_images.pack(side=tk.BOTTOM, expand=True, fill="both")
tb_original = ttk.Frame(tw_images)
self.image_original = None
tw_images.add(tb_original, text="RGB")
self.lb_image_original = ttk.Label(tb_original, image=None)
self.lb_image_original.pack(expand=True, fill="both")
tb_gray = ttk.Frame(tw_images)
self.image_gray = None
tw_images.add(tb_gray, text="Grayscale")
self.lb_image_gray = ttk.Label(tb_gray, image=None)
self.lb_image_gray.pack(expand=True, fill="both")
tb_gaussian = ttk.Frame(tw_images)
self.image_gaussian = None
tw_images.add(tb_gaussian, text="Gaussian")
self.lb_image_gaussian = ttk.Label(tb_gaussian, image=None)
self.lb_image_gaussian.pack(expand=True, fill="both")
tb_edges = ttk.Frame(tw_images)
self.image_edges = None
tw_images.add(tb_edges, text="Edges")
self.lb_image_edges = ttk.Label(tb_edges, image=None)
self.lb_image_edges.pack(expand=True, fill="both")
tb_corners = ttk.Frame(tw_images)
self.image_corners = None
tw_images.add(tb_corners, text="Harris corners")
self.lb_image_corners = ttk.Label(tb_corners, image=None)
self.lb_image_corners.pack(expand=True, fill="both")
def run(self):
try:
self.poll_image_data_thread.start()
self.app.mainloop()
except KeyboardInterrupt:
self.close_event()
def poll_image_data(self):
while True:
request = self.queue_gui_op.get()
method = request[0]
args = request[1:]
print("------------[ProcessGUI]------------")
print("Method: " + method)
print("------------------------------------")
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
def display_image_dims_callback(self, height, width, channels):
print("[ProcessGUI]")
print("Height: " + str(height))
print("Width: " + str(width))
print("Channels: " + str(channels))
def display_image_processing_progress_callback(self, progress):
progress = 0 + ((100 - 0) / (len(ProcessImageProcessing.ProcessingSteps) - 0)) * (progress - 0)
print("[ProcessGUI]: Updating progress to " + str(progress) + "%")
self.pb_load["value"] = progress
def display_image_original_gray_callback(self, image_original, image_gray):
self.image_original = ImageTk.PhotoImage(Image.fromarray(image_original))
self.lb_image_original.configure(image=self.image_original)
self.image_gray = ImageTk.PhotoImage(Image.fromarray(image_gray))
self.lb_image_gray.configure(image=self.image_gray)
def display_image_gaussian_callback(self, image_gaussian):
self.image_gaussian = ImageTk.PhotoImage(Image.fromarray(image_gaussian))
self.lb_image_gaussian.configure(image=self.image_gaussian)
def display_image_edges_callback(self, image_edges):
self.image_edges = ImageTk.PhotoImage(Image.fromarray(image_edges))
self.lb_image_edges.configure(image=self.image_edges)
def display_image_corners_callback(self, image_corners):
self.image_corners = ImageTk.PhotoImage(Image.fromarray(image_corners))
self.lb_image_corners.configure(image=self.image_corners)
def btn_browse_clicked(self):
filename = tk.filedialog.askopenfilename(initialdir=".",
title="Select image",
filetypes=(
("Portable Network graphics", "*.png"),
("All files", "*.*")))
self.ent_browse.delete(0, tk.END)
self.ent_browse.insert(0, filename)
edge_alg = self.edge_algs.get()
self.queue_gui_ip.put(["process_image", filename, edge_alg])
def close_event(self):
print("[ProcessGUI]: Shutting down")
self.app.destroy()
self.termination_event.set() # show were are terminating
class ProcessImageProcessing(object):
ProcessingSteps = [
"rgb",
"gray",
"gaussian",
"edges",
"corners"
]
def __init__(self, queue_gui_ip, queue_gui_op):
print("[ProcessImageProcessing]: PID = " + str(os.getpid()))
self.queue_gui_ip = queue_gui_ip
self.queue_gui_op = queue_gui_op
self.run()
def run(self):
while True:
request = self.queue_gui_ip.get()
method = request[0]
args = request[1:]
print("------[ProcessImageProcessing]------")
print("Method: " + method)
print("------------------------------------")
if "display_" in method:
# Skip incoming requests that contain methods meant for ProcessGUI class
continue
try:
getattr(self, method + "_callback")(*args)
except AttributeError as ae:
print("Unknown callback received from pipe", str(ae))
def process_image_callback(self, image_path, edge_alg):
print("[ProcessImageProcessing]: Received file \"" + image_path + "\"")
try:
progress = 0
original_bgr = cv2.imread(image_path)
original_rgb = cv2.cvtColor(original_bgr, cv2.COLOR_BGR2RGB)
(height, width, channels) = original_bgr.shape[:3]
progress += 1
self.queue_gui_op.put(["display_image_processing_progress", progress])
self.queue_gui_op.put(["display_image_dims", height, width, channels])
gray = cv2.cvtColor(original_bgr, cv2.COLOR_BGR2GRAY)
progress += 1
self.queue_gui_op.put(["display_image_processing_progress", progress])
self.queue_gui_op.put(["display_image_original_gray", original_rgb, gray])
gaussian = self.process_image_gaussian(gray)
progress += 1
self.queue_gui_op.put(["display_image_processing_progress", progress])
self.queue_gui_op.put(["display_image_gaussian", gaussian])
try:
edges = self.process_image_edges(gaussian, edge_alg)
if not edges:
print("Unknown edge detection algorithm")
else:
progress += 1
self.queue_gui_op.put(["display_image_processing_progress", progress])
self.queue_gui_op.put(["display_image_edges", edges])
except:
pass
corners = self.process_image_corners(original_rgb, gray)
progress += 1
self.queue_gui_op.put(["display_image_processing_progress", progress])
self.queue_gui_op.put(["display_image_corners", corners])
except Exception as ex:
print("Exception: " + str(ex))
def process_image_gaussian(self, image_gray):
gaussian = cv2.GaussianBlur(image_gray, (3, 3), cv2.BORDER_DEFAULT)
return gaussian
def process_image_edges(self, image_gaussian, edge_alg):
edges = None
if edge_alg not in ["canny", "sobelx", "sobely", "sobelxy"]:
return edges
# Sobel edge detection
# Sobel edge detection on the X axis
if edge_alg == "sobelx":
print("Sobel X")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=1, dy=0, ksize=5)
# Sobel edge detection on the Y axis
elif edge_alg == "sobely":
print("Sobel Y")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=0, dy=1, ksize=5)
# Combined X and Y Sobel edge detection
elif edge_alg == "sobelxy":
print("Sobel XY")
edges = cv2.Sobel(src=image_gaussian, ddepth=cv2.CV_64F, dx=1, dy=1, ksize=5)
# Canny edge detection
elif edge_alg == "canny":
print("Canny")
edges = cv2.Canny(image=image_gaussian, threshold1=100, threshold2=200)
return edges
def process_image_corners(self, image_original, image_gray):
original_with_corners = image_original
corners = cv2.cornerHarris(image_gray, 2, 3, 0.04)
# result is dilated for marking the corners, not important
corners = cv2.dilate(corners, None)
# Threshold for an optimal value, it may vary depending on the image.
original_with_corners[corners > 0.01 * corners.max()] = [0, 0, 255]
return original_with_corners
def close(self):
print("[ProcessImageProcessing]: Shutting down")
class Controller(object):
def __init__(self):
print("[Controller]: PID = " + str(os.getpid()))
queue_gui_ip = Queue()
queue_gui_op = Queue()
# A Thread!
self.termination_event = Event()
self.gui = Thread(
target=ProcessGUI,
args=(queue_gui_ip, queue_gui_op, self.termination_event)
)
# Make it a daemon process:
self.ip = Process(
target=ProcessImageProcessing,
args=(queue_gui_ip, queue_gui_op),
daemon=True
)
def run(self):
try:
self.gui.start()
self.ip.start()
self.termination_event.wait() # wait for termination
self.close_callback()
except KeyboardInterrupt:
self.close_callback()
def close_callback(self):
print("Quitting processes...")
self.gui.join(1)
if self.gui.is_alive():
self.gui.terminate()
print("Finished")
def main():
c = Controller()
c.run()
if __name__ == "__main__":
main()
Upvotes: 1
Related Questions
- Single Producer Multiple Consumer
- How can I share a queue among three or more processes?
- Separate process sharing queue with main process (producer/consumer)
- Implementation of multiprocessing.Queue and queue.Queue
- How to set up a pipeline using Queues in multiprocessing
- Data type for a "closable" queue to handle a stream of items for multiple producers and consumers
- Python multiprocessing in objects
- Tell consumers to stop waiting for queue elements
- How can I share a Queue between processes in Python?
- Shared queue across multiple processes