|
|
|
@ -5,16 +5,26 @@ import logging
|
|
|
|
|
import argparse
|
|
|
|
|
import numpy as np
|
|
|
|
|
import time
|
|
|
|
|
import datetime
|
|
|
|
|
from PIL import ImageFont, ImageDraw, Image
|
|
|
|
|
import os
|
|
|
|
|
|
|
|
|
|
draw_colors = {
|
|
|
|
|
'hog': (255,0,0),
|
|
|
|
|
'haar': (0,255,0),
|
|
|
|
|
'dnn': (0,0,255),
|
|
|
|
|
'hog': (198,65,124),
|
|
|
|
|
'haar': (255,255,255),
|
|
|
|
|
'dnn': (251,212,36),
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
font = ImageFont.truetype("/home/ruben/Documents/Projecten/2018/PATH/presentation/lib/font/source-sans-pro/source-sans-pro-regular.ttf", 30)
|
|
|
|
|
titles = {
|
|
|
|
|
'hog' : "Histogram of oriented gradients",
|
|
|
|
|
'haar' : "Haar cascades",
|
|
|
|
|
'dnn' : "Neural network",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fontfile = "SourceSansPro-Regular.ttf"
|
|
|
|
|
|
|
|
|
|
font = ImageFont.truetype(fontfile, 30)
|
|
|
|
|
font_s = ImageFont.truetype(fontfile, 20)
|
|
|
|
|
|
|
|
|
|
class Result():
|
|
|
|
|
def __init__(self, algorithm, image, confidence_threshold = 0.5):
|
|
|
|
@ -33,18 +43,27 @@ class Result():
|
|
|
|
|
})
|
|
|
|
|
return self
|
|
|
|
|
|
|
|
|
|
def draw_detections(self):
|
|
|
|
|
color = draw_colors[self.algorithm]
|
|
|
|
|
def draw_detections(self, include_title = False):
|
|
|
|
|
cv2_im_rgb = cv2.cvtColor(self.visualisation,cv2.COLOR_BGR2RGB)
|
|
|
|
|
# Pass the image to PIL
|
|
|
|
|
pil_im = Image.fromarray(cv2_im_rgb)
|
|
|
|
|
draw = ImageDraw.Draw(pil_im, 'RGBA')
|
|
|
|
|
|
|
|
|
|
self.draw_detections_on(draw)
|
|
|
|
|
|
|
|
|
|
if include_title:
|
|
|
|
|
draw.text((10,10), titles[self.algorithm], fill=draw_colors[self.algorithm], font=font)
|
|
|
|
|
|
|
|
|
|
return cv2.cvtColor(np.array(pil_im), cv2.COLOR_RGB2BGR)
|
|
|
|
|
|
|
|
|
|
def draw_detections_on(self, draw: ImageDraw):
|
|
|
|
|
'''
|
|
|
|
|
Draw on a specified canvas
|
|
|
|
|
'''
|
|
|
|
|
color = draw_colors[self.algorithm]
|
|
|
|
|
for detection in self.detections:
|
|
|
|
|
self.draw_detection(draw, detection, color)
|
|
|
|
|
|
|
|
|
|
self.visualisation = cv2.cvtColor(np.array(pil_im), cv2.COLOR_RGB2BGR)
|
|
|
|
|
|
|
|
|
|
def draw_detection(self, draw: ImageDraw, detection: dict, color: tuple):
|
|
|
|
|
width = 2
|
|
|
|
|
|
|
|
|
@ -68,51 +87,8 @@ class Result():
|
|
|
|
|
color = tuple(color)
|
|
|
|
|
|
|
|
|
|
draw.rectangle((detection['startX'], detection['startY'], detection['endX'], detection['endY']), outline=color, width=width)
|
|
|
|
|
# cv2.rectangle(rect_img, (0, 0),
|
|
|
|
|
# (sub_img.shape[1]-int(width/2), sub_img.shape[0]-int(width/2)),
|
|
|
|
|
# color, width)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def draw_detections_cv2(self):
|
|
|
|
|
color = draw_colors[self.algorithm]
|
|
|
|
|
for detection in self.detections:
|
|
|
|
|
self.draw_detection(detection, color)
|
|
|
|
|
|
|
|
|
|
def draw_detection_cv2(self, detection, color=(0,0,255)):
|
|
|
|
|
|
|
|
|
|
# First we crop the sub-rect from the image
|
|
|
|
|
sub_img = self.visualisation[detection['startY']:detection['endY'], detection['startX']:detection['endX']]
|
|
|
|
|
rect_img = sub_img.copy()
|
|
|
|
|
width = 2
|
|
|
|
|
cv2.rectangle(rect_img, (0, 0),
|
|
|
|
|
(sub_img.shape[1]-int(width/2), sub_img.shape[0]-int(width/2)),
|
|
|
|
|
color, width)
|
|
|
|
|
# white_rect = np.ones(sub_img.shape, dtype=np.uint8) * 255
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# filter out weak detections by ensuring the `confidence` is
|
|
|
|
|
# greater than the minimum confidence
|
|
|
|
|
if detection['confidence'] > self.confidence_threshold:
|
|
|
|
|
# draw the bounding box of the face along with the associated
|
|
|
|
|
# probability
|
|
|
|
|
text = "{:.2f}%".format(detection['confidence'] * 100)
|
|
|
|
|
y = detection['startY'] - 10 if detection['startY'] - 10 > 10 else detection['startY'] + 10
|
|
|
|
|
# cv2.rectangle(image, (startX, startY), (endX, endY),
|
|
|
|
|
# color, 2)
|
|
|
|
|
cv2.putText(self.visualisation, text, (detection['startX'], y),
|
|
|
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.45, color, 2, lineType = cv2.LINE_AA)
|
|
|
|
|
|
|
|
|
|
alpha = 1
|
|
|
|
|
else:
|
|
|
|
|
# At least 10% opacity
|
|
|
|
|
alpha = max(.3, detection['confidence'])
|
|
|
|
|
|
|
|
|
|
res = cv2.addWeighted(sub_img, 1-alpha, rect_img, alpha, 1.0)
|
|
|
|
|
|
|
|
|
|
# Putting the image back to its position
|
|
|
|
|
self.visualisation[detection['startY']:detection['endY'], detection['startX']:detection['endX']] = res
|
|
|
|
|
|
|
|
|
|
def resize(self, width, height):
|
|
|
|
|
# TODO resize to new target incl all detections
|
|
|
|
|
img = self.visualisation
|
|
|
|
@ -131,6 +107,9 @@ class Result():
|
|
|
|
|
)
|
|
|
|
|
return result
|
|
|
|
|
|
|
|
|
|
def count_detections(self):
|
|
|
|
|
detections = [d for d in self.detections if d['confidence'] > self.confidence_threshold]
|
|
|
|
|
return len(detections)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def record(device_id, q1,q2, q3, q4, resolution, rotate):
|
|
|
|
@ -361,7 +340,17 @@ def process3_haar(in_q, out_q, cascade_file):
|
|
|
|
|
""")
|
|
|
|
|
|
|
|
|
|
dir_path = os.path.dirname(os.path.realpath(__file__))
|
|
|
|
|
C = ffi.dlopen(os.path.join(dir_path,"../visualhaar/target/debug/libvisual_haarcascades_lib.so"))
|
|
|
|
|
|
|
|
|
|
lib_path = os.path.join(dir_path, "..", "visualhaar", "target", "debug")
|
|
|
|
|
so_path = os.path.join(lib_path, "libvisual_haarcascades_lib.so")
|
|
|
|
|
dll_path = os.path.join(lib_path, "visual_haarcascades_lib.dll")
|
|
|
|
|
|
|
|
|
|
if os.path.exists(so_path):
|
|
|
|
|
C = ffi.dlopen(so_path)
|
|
|
|
|
elif os.path.exists(dll_path):
|
|
|
|
|
C = ffi.dlopen(dll_path)
|
|
|
|
|
else:
|
|
|
|
|
raise RuntimeException("Visual haarcascades library is not found")
|
|
|
|
|
|
|
|
|
|
# print(C.test(9))
|
|
|
|
|
# i = Image.open("Marjo.jpg")
|
|
|
|
@ -437,11 +426,33 @@ def process3_haar(in_q, out_q, cascade_file):
|
|
|
|
|
# print(img)
|
|
|
|
|
out_q.put(result)
|
|
|
|
|
|
|
|
|
|
def display(image_res, q1, q2, q3, q4, fullscreen = False):
|
|
|
|
|
prev_image1 = np.zeros((image_res[1],image_res[0],3), np.uint8)
|
|
|
|
|
prev_image2 = np.zeros((image_res[1],image_res[0],3), np.uint8)
|
|
|
|
|
prev_image3 = np.zeros((image_res[1],image_res[0],3), np.uint8)
|
|
|
|
|
prev_image4 = np.zeros((image_res[1],image_res[0],3), np.uint8)
|
|
|
|
|
def draw_stats(image, results):
|
|
|
|
|
pil_im = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
|
|
|
|
|
draw = ImageDraw.Draw(pil_im, 'RGBA')
|
|
|
|
|
|
|
|
|
|
for i, result in enumerate(results):
|
|
|
|
|
if result is None:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
c = result.count_detections()
|
|
|
|
|
txt = "face" if c == 1 else "faces"
|
|
|
|
|
txt = f"{result.algorithm.ljust(5)} {c} {txt}"
|
|
|
|
|
draw.text((10, pil_im.size[1] - i*25 - 50), txt, fill=draw_colors[result.algorithm], font=font_s, stroke_width=1, stroke_fill=(0,0,0))
|
|
|
|
|
|
|
|
|
|
return cv2.cvtColor(np.array(pil_im), cv2.COLOR_RGB2BGR)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def display(image_res, q1, q2, q3, q4, fullscreen, output_dir):
|
|
|
|
|
logger = logging.getLogger('display')
|
|
|
|
|
|
|
|
|
|
empty_image = np.zeros((image_res[1],image_res[0],3), np.uint8)
|
|
|
|
|
|
|
|
|
|
results = [None, None, None]
|
|
|
|
|
result_queues = [q2, q3, q4]
|
|
|
|
|
images = [empty_image, empty_image, empty_image, empty_image]
|
|
|
|
|
|
|
|
|
|
override_image = None
|
|
|
|
|
override_until = None
|
|
|
|
|
|
|
|
|
|
if fullscreen:
|
|
|
|
|
cv2.namedWindow("output", cv2.WND_PROP_FULLSCREEN)
|
|
|
|
@ -450,57 +461,68 @@ def display(image_res, q1, q2, q3, q4, fullscreen = False):
|
|
|
|
|
while True:
|
|
|
|
|
logging.debug('r')
|
|
|
|
|
try:
|
|
|
|
|
image1 = q1.get_nowait()
|
|
|
|
|
image1 = cv2.resize(image1, (image_res[0], image_res[1]))
|
|
|
|
|
prev_image1 = image1
|
|
|
|
|
image = q1.get_nowait()
|
|
|
|
|
images[0] = cv2.resize(image, (image_res[0], image_res[1]))
|
|
|
|
|
except Empty as e:
|
|
|
|
|
image1 = prev_image1
|
|
|
|
|
try:
|
|
|
|
|
result2 = q2.get_nowait()
|
|
|
|
|
result2 = result2.resize(image_res[0], image_res[1])
|
|
|
|
|
result2.draw_detections()
|
|
|
|
|
image2 = result2.visualisation
|
|
|
|
|
# image2 = cv2.resize(image2, (image_res[0], image_res[1]))
|
|
|
|
|
prev_image2 = image2
|
|
|
|
|
except Empty as e:
|
|
|
|
|
image2 = prev_image2
|
|
|
|
|
try:
|
|
|
|
|
result3 = q3.get_nowait()
|
|
|
|
|
result3 = result3.resize(image_res[0], image_res[1])
|
|
|
|
|
result3.draw_detections()
|
|
|
|
|
image3 = result3.visualisation
|
|
|
|
|
# image3 = cv2.resize(image3, (image_res[0], image_res[1]))
|
|
|
|
|
prev_image3 = image3
|
|
|
|
|
except Empty as e:
|
|
|
|
|
image3 = prev_image3
|
|
|
|
|
try:
|
|
|
|
|
result4 = q4.get_nowait()
|
|
|
|
|
result4 = result4.resize(image_res[0], image_res[1])
|
|
|
|
|
result4.draw_detections()
|
|
|
|
|
image4 = result4.visualisation
|
|
|
|
|
# image4 = cv2.resize(image4, (image_res[0], image_res[1]))
|
|
|
|
|
prev_image4 = image4
|
|
|
|
|
except Empty as e:
|
|
|
|
|
image4 = prev_image4
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
for idx, queue in enumerate(result_queues):
|
|
|
|
|
try:
|
|
|
|
|
result = queue.get_nowait()
|
|
|
|
|
results[idx] = result.resize(image_res[0], image_res[1])
|
|
|
|
|
images[idx+1] = results[idx].draw_detections(include_title=True)
|
|
|
|
|
except Empty as e:
|
|
|
|
|
pass
|
|
|
|
|
finally:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
if override_image is not None and override_until > time.time():
|
|
|
|
|
cv2.imshow("output", override_image)
|
|
|
|
|
else:
|
|
|
|
|
override_image = None
|
|
|
|
|
|
|
|
|
|
img_concate_Verti1 = np.concatenate((image1,image2),axis=0)
|
|
|
|
|
img_concate_Verti2 = np.concatenate((image3,image4),axis=0)
|
|
|
|
|
grid_img = np.concatenate((img_concate_Verti1,img_concate_Verti2),axis=1)
|
|
|
|
|
cv2.imshow("output", grid_img)
|
|
|
|
|
images[0] = draw_stats(images[0], results)
|
|
|
|
|
|
|
|
|
|
img_concate_Verti1 = np.concatenate((images[0],images[1]),axis=0)
|
|
|
|
|
img_concate_Verti2 = np.concatenate((images[2],images[3]),axis=0)
|
|
|
|
|
grid_img = np.concatenate((img_concate_Verti1,img_concate_Verti2),axis=1)
|
|
|
|
|
cv2.imshow("output", grid_img)
|
|
|
|
|
|
|
|
|
|
# Hit 'q' on the keyboard to quit!
|
|
|
|
|
key = cv2.waitKey(1) & 0xFF
|
|
|
|
|
if key == ord('q'):
|
|
|
|
|
break
|
|
|
|
|
if key == ord(' '):
|
|
|
|
|
# TODO save frame
|
|
|
|
|
pass
|
|
|
|
|
# TODO wait for frame to be processed. Eg. if I move and make a pic, it should use the last frame...
|
|
|
|
|
output_res = (image_res[0] *2, image_res[1] * 2)
|
|
|
|
|
pil_im = Image.fromarray(cv2.cvtColor(images[0], cv2.COLOR_BGR2RGB))
|
|
|
|
|
pil_im = pil_im.resize(output_res)
|
|
|
|
|
draw = ImageDraw.Draw(pil_im, 'RGBA')
|
|
|
|
|
|
|
|
|
|
for result in results:
|
|
|
|
|
if result is None:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
result.resize(output_res[0], output_res[1]).draw_detections_on(draw)
|
|
|
|
|
|
|
|
|
|
override_image = cv2.cvtColor(np.array(pil_im), cv2.COLOR_RGB2BGR)
|
|
|
|
|
override_until = time.time() + 5
|
|
|
|
|
logger.info("Show frame until %f", override_until)
|
|
|
|
|
|
|
|
|
|
def main(camera_id, rotate, fullscreen, cascade_file):
|
|
|
|
|
# save images:
|
|
|
|
|
name = datetime.datetime.now().isoformat(timespec='seconds')
|
|
|
|
|
cv2.imwrite(os.path.join(output_dir, f'{name}.png'),override_image)
|
|
|
|
|
for result in results:
|
|
|
|
|
cv2.imwrite(os.path.join(output_dir, f'{name}-{result.algorithm}.png'),result.visualisation)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def main(camera_id, rotate, fullscreen, cascade_file, output_dir):
|
|
|
|
|
image_size = (int(1920/2), int(1080/2))
|
|
|
|
|
|
|
|
|
|
if not os.path.exists(cascade_file):
|
|
|
|
|
raise RuntimeError(f"Cannot load OpenCV haar-cascade file '{cascade_file}'")
|
|
|
|
|
if not os.path.isdir(output_dir):
|
|
|
|
|
raise RuntimeError(f"Non-existent directory to store files '{output_dir}'")
|
|
|
|
|
|
|
|
|
|
is_rotated_90 = rotate in [cv2.ROTATE_90_CLOCKWISE, cv2.ROTATE_90_COUNTERCLOCKWISE]
|
|
|
|
|
|
|
|
|
@ -519,7 +541,7 @@ def main(camera_id, rotate, fullscreen, cascade_file):
|
|
|
|
|
q_process3 = Queue(maxsize=1)
|
|
|
|
|
|
|
|
|
|
p1 = Process(target=record, args=(camera_id, q_webcam1, q_webcam2,q_webcam3,q_webcam4, image_size, rotate))
|
|
|
|
|
p2 = Process(target=display, args=(image_size, q_webcam1, q_process1, q_process2, q_process3, fullscreen ))
|
|
|
|
|
p2 = Process(target=display, args=(image_size, q_webcam1, q_process1, q_process2, q_process3, fullscreen, output_dir ))
|
|
|
|
|
p3 = Process(target=process1_hog, args=(q_webcam2, q_process1,))
|
|
|
|
|
p4 = Process(target=process2_dnn, args=(q_webcam3, q_process2,))
|
|
|
|
|
p5 = Process(target=process3_haar, args=(q_webcam4, q_process3,cascade_file))
|
|
|
|
|