import numpy as np
import cv2
import pandas as pd
import operator
import matplotlib.pyplot as plt
import os
from sklearn.model_selection import train_test_split
from tensorflow.keras.utils import Sequence
from config import yolo_config
def load_weights(model, weights_file_path):
conv_layer_size = 110
conv_output_idxs = [93, 101, 109]
with open(weights_file_path, 'rb') as file:
major, minor, revision, seen, _ = np.fromfile(file, dtype=np.int32, count=5)
bn_idx = 0
for conv_idx in range(conv_layer_size):
conv_layer_name = f'conv2d_{conv_idx}' if conv_idx > 0 else 'conv2d'
bn_layer_name = f'batch_normalization_{bn_idx}' if bn_idx > 0 else 'batch_normalization'
conv_layer = model.get_layer(conv_layer_name)
filters = conv_layer.filters
kernel_size = conv_layer.kernel_size[0]
input_dims = conv_layer.input_shape[-1]
if conv_idx not in conv_output_idxs:
# darknet bn layer weights: [beta, gamma, mean, variance]
bn_weights = np.fromfile(file, dtype=np.float32, count=4 * filters)
# tf bn layer weights: [gamma, beta, mean, variance]
bn_weights = bn_weights.reshape((4, filters))[[1, 0, 2, 3]]
bn_layer = model.get_layer(bn_layer_name)
bn_idx += 1
else:
conv_bias = np.fromfile(file, dtype=np.float32, count=filters)
# darknet shape: (out_dim, input_dims, height, width)
# tf shape: (height, width, input_dims, out_dim)
conv_shape = (filters, input_dims, kernel_size, kernel_size)
conv_weights = np.fromfile(file, dtype=np.float32, count=np.product(conv_shape))
conv_weights = conv_weights.reshape(conv_shape).transpose([2, 3, 1, 0])
if conv_idx not in conv_output_idxs:
conv_layer.set_weights([conv_weights])
bn_layer.set_weights(bn_weights)
else:
conv_layer.set_weights([conv_weights, conv_bias])
if len(file.read()) == 0:
print('all weights read')
else:
print(f'failed to read all weights, # of unread weights: {len(file.read())}')
def get_detection_data(img, model_outputs, class_names):
"""
:param img: target raw image
:param model_outputs: outputs from inference_model
:param class_names: list of object class names
:return:
"""
num_bboxes = model_outputs[-1][0]
boxes, scores, classes = [output[0][:num_bboxes] for output in model_outputs[:-1]]
h, w = img.shape[:2]
df = pd.DataFrame(boxes, columns=['x1', 'y1', 'x2', 'y2'])
df[['x1', 'x2']] = (df[['x1', 'x2']] * w).astype('int64')
df[['y1', 'y2']] = (df[['y1', 'y2']] * h).astype('int64')
df['class_name'] = np.array(class_names)[classes.astype('int64')]
df['score'] = scores
df['w'] = df['x2'] - df['x1']
df['h'] = df['y2'] - df['y1']
print(f'# of bboxes: {num_bboxes}')
return df
def read_annotation_lines(annotation_path, test_size=None, random_seed=5566):
with open(annotation_path) as f:
lines = f.readlines()
if test_size:
return train_test_split(lines, test_size=test_size, random_state=random_seed)
else:
return lines
def draw_bbox(img, detections, cmap, random_color=True, figsize=(10, 10), show_img=True, show_text=True):
"""
Draw bounding boxes on the img.
:param img: BGR img.
:param detections: pandas DataFrame containing detections
:param random_color: assign random color for each objects
:param cmap: object colormap
:param plot_img: if plot img with bboxes
:return: None
"""
img = np.array(img)
scale = max(img.shape[0:2]) / 416
line_width = int(2 * scale)
for _, row in detections.iterrows():
x1, y1, x2, y2, cls, score, w, h = row.values
color = list(np.random.random(size=3) * 255) if random_color else cmap[cls]
cv2.rectangle(img, (x1, y1), (x2, y2), color, line_width)
if show_text:
text = f'{cls} {score:.2f}'
font = cv2.FONT_HERSHEY_DUPLEX
font_scale = max(0.3 * scale, 0.3)
thickness = max(int(1 * scale), 1)
(text_width, text_height) = cv2.getTextSize(text, font, fontScale=font_scale, thickness=thickness)[0]
cv2.rectangle(img, (x1 - line_width//2, y1 - text_height), (x1 + text_width, y1), color, cv2.FILLED)
cv2.putText(img, text, (x1, y1), font, font_scale, (255, 255, 255), thickness, cv2.LINE_AA)
if show_img:
plt.figure(figsize=figsize)
plt.imshow(img)
plt.show()
return img
class DataGenerator(Sequence):
"""
Generates data for Keras
ref: https://stanford.edu/~shervine/blog/keras-how-to-generate-data-on-the-fly
"""
def __init__(self,
annotation_lines,
class_name_path,
folder_path,
max_boxes=100,
shuffle=True):
self.annotation_lines = annotation_lines
self.class_name_path = class_name_path
self.num_classes = len([line.strip() for line in open(class_name_path).readlines()])
self.num_gpu = yolo_config['num_gpu']
self.batch_size = yolo_config['batch_size'] * self.num_gpu
self.target_img_size = yolo_config['img_size']
self.anchors = np.array(yolo_config['anchors']).reshape((9, 2))
self.shuffle = shuffle
self.indexes = np.arange(len(self.annotation_lines))
self.folder_path = folder_path
self.max_boxes = max_boxes
self.on_epoch_end()
def __len__(self):
'number of batches per epoch'
return int(np.ceil(len(self.annotation_lines) / self.batch_size))
def __getitem__(self, index):
'Generate one batch of data'
# Generate indexes of the batch
idxs = self.indexes[index * self.batch_size:(index + 1) * self.batch_size]
# Find list of IDs
lines = [self.annotation_lines[i] for i in idxs]
# Generate data
X, y_tensor, y_bbox = self.__data_generation(lines)
return [X, *y_tensor, y_bbox], np.zeros(len(lines))
def on_epoch_end(self):
'Updates indexes after each epoch'
if self.shuffle:
np.random.shuffle(self.indexes)
def __data_generation(self, annotation_lines):
"""
Generates data containing batch_size samples
:param annotation_lines:
:return:
"""
X = np.empty((len(annotation_lines), *self.target_img_size), dtype=np.float32)
y_bbox = np.empty((len(annotation_lines), self.max_boxes, 5), dtype=np.float32) # x1y1x2y2
for i, line in enumerate(annotation_lines):
img_data, box_data = self.get_data(line)
X[i] = img_data
y_bbox[i] = box_data
y_tensor, y_true_boxes_xywh = preprocess_true_boxes(y_bbox, self.target_img_size[:2], self.anchors, self.num_classes)
return X, y_tensor, y_true_boxes_xywh
def get_data(self, annotation_line):
line = annotation_line.split()
img_path = line[0]
img = cv2.imread(os.path.join(self.folder_path, img_path))[:, :, ::-1]
ih, iw = img.shape[:2]
h, w, c = self.target_img_size
boxes = np.array([np.array(list(map(float, box.split(',')))) for box in line[1:]], dtype=np.float32) # x1y1x2y2
scale_w, scale_h = w / iw, h / ih
img = cv2.resize(img, (w, h))
image_data = np.array(img) / 255.
# correct boxes coordinates
box_data = np.zeros((self.max_boxes, 5))
if len(boxes) > 0:
np.random.shuffle(boxes)
boxes = boxes[:self.max_boxes]
boxes[:, [0, 2]] = boxes[:, [0, 2]] * scale_w # + dx
boxes[:, [1, 3]] = boxes[:, [1, 3]] * scale_h # + dy
box_data[:len(boxes)] = boxes
return image_data, box_data
def preprocess_true_boxes(true_boxes, input_shape, anchors, num_classes):
'''Preprocess true boxes to training input format
Parameters
----------
true_boxes: array, shape=(bs, max boxes per img, 5)
Absolute x_min, y_min, x_max, y_max, class_id relative to input_shape.
input_shape: array-like, hw, multiples of 32
anchors: array, shape=(N, 2), (9, wh)
num_classes: int
Returns
-------
y_true: list of array, shape like yolo_outputs, xywh are reletive value
'''
num_stages = 3 # default setting for yolo, tiny yolo will be 2
anchor_mask = [[0, 1, 2], [3, 4, 5], [6, 7, 8]]
bbox_per_grid = 3
true_boxes = np.array(true_boxes, dtype='float32')
true_boxes_abs = np.array(true_boxes, dtype='float32')
input_shape = np.array(input_shape, dtype='int32')
true_boxes_xy = (true_boxes_abs[..., 0:2] + true_boxes_abs[..., 2:4]) // 2 # (100, 2)
true_boxes_wh = true_boxes_abs[..., 2:4] - true_boxes_abs[..., 0:2] # (100, 2)
# Normalize x,y,w, h, relative to img size -> (0~1)
true_boxes[..., 0:2] = true_boxes_xy/input_shape[::-1] # xy
true_boxes[..., 2:4] = true_boxes_wh/input_shape[::-1] # wh
bs = true_boxes.shape[0]
grid_sizes = [input_shape//{0:8, 1:16, 2:32}[stage] for stage in range(num_stages)]
y_true = [np.zeros((bs,
grid_sizes[s][0],
grid_sizes[s][1],
bbox_per_grid,
5+num_classes), dtype='float32')
for s in range(num_stages)]
# [(?, 52, 52, 3, 5+num_classes) (?, 26, 26, 3, 5+num_classes) (?, 13, 13, 3, 5+num_classes) ]
y_true_boxes_xywh = np.concatenate((true_boxes_xy, true_boxes_wh), axis=-1)
# Expand dim to apply broadcasting.
anchors = np.expand_dims(anchors, 0) # (1, 9 , 2)
anchor_maxes = anchors / 2. # (1, 9 , 2)
anchor_mins = -anchor_maxes # (1, 9 , 2)
valid_mask = true_boxes_wh[..., 0] > 0 # (1, 100)
for batch_idx in range(bs):
# Discard zero rows.
wh = true_boxes_wh[batch_idx, valid_mask[batch_idx]] # (# of bbox, 2)
num_boxes = len(wh)
if num_boxes == 0: continue
wh = np.expand_dims(wh, -2) # (# of bbox, 1, 2)
box_maxes = wh / 2. # (# of bbox, 1, 2)
box_mins = -box_maxes # (# of bbox, 1, 2)
# Compute IoU between each anchors and true boxes for responsibility assignment
intersect_mins = np.maximum(box_mins, anchor_mins) # (# of bbox, 9, 2)
intersect_maxes = np.minimum(box_maxes, anchor_maxes)
intersect_wh = np.maximum(intersect_maxes - intersect_mins, 0.)
intersect_area = np.prod(intersect_wh, axis=-1) # (9,)
box_area = wh[..., 0] * wh[..., 1] # (# of bbox, 1)
anchor_area = anchors[..., 0] * anchors[..., 1] # (1, 9)
iou = intersect_area / (box_area + anchor_area - intersect_area) # (# of bbox, 9)
# Find best anchor for each true box
best_anchors = np.argmax(iou, axis=-1) # (# of bbox,)
for box_idx in range(num_boxes):
best_anchor = best_anchors[box_idx]
for stage in range(num_stages):
if best_anchor in anchor_mask[stage]:
x_offset = true_boxes[batch_idx, box_idx, 0]*grid_sizes[stage][1]
y_offset = true_boxes[batch_idx, box_idx, 1]*grid_sizes[stage][0]
# Grid Index
grid_col = np.floor(x_offset).astype('int32')
grid_row = np.floor(y_offset).astype('int32')
anchor_idx = anchor_mask[stage].index(best_anchor)
class_idx = true_boxes[batch_idx, box_idx, 4].astype('int32')
# y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, 0] = x_offset - grid_col # x
# y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, 1] = y_offset - grid_row # y
# y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, :4] = true_boxes_abs[batch_idx, box_idx, :4] # abs xywh
y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, :2] = true_boxes_xy[batch_idx, box_idx, :] # abs xy
y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, 2:4] = true_boxes_wh[batch_idx, box_idx, :] # abs wh
y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, 4] = 1 # confidence
y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, 5+class_idx] = 1 # one-hot encoding
# smooth
# onehot = np.zeros(num_classes, dtype=np.float)
# onehot[class_idx] = 1.0
# uniform_distribution = np.full(num_classes, 1.0 / num_classes)
# delta = 0.01
# smooth_onehot = onehot * (1 - delta) + delta * uniform_distribution
# y_true[stage][batch_idx, grid_row, grid_col, anchor_idx, 5:] = smooth_onehot
return y_true, y_true_boxes_xywh
"""
Calculate the AP given the recall and precision array
1st) We compute a version of the measured precision/recall curve with
precision monotonically decreasing
2nd) We compute the AP as the area under this curve by numerical integration.
"""
def voc_ap(rec, prec):
"""
--- Official matlab code VOC2012---
mrec=[0 ; rec ; 1];
mpre=[0 ; prec ; 0];
for i=numel(mpre)-1:-1:1
mpre(i)=max(mpre(i),mpre(i+1));
end
i=find(mrec(2:end)~=mrec(1:end-1))+1;
ap=sum((mrec(i)-mrec(i-1)).*mpre(i));
"""
rec.insert(0, 0.0) # insert 0.0 at begining of list
rec.append(1.0) # insert 1.0 at end of list
mrec = rec[:]
prec.insert(0, 0.0) # insert 0.0 at begining of list
prec.append(0.0) # insert 0.0 at end of list
mpre = prec[:]
"""
This part makes the precision monotonically decreasing
(goes from the end to the beginning)
matlab: for i=numel(mpre)-1:-1:1
mpre(i)=max(mpre(i),mpre(i+1));
"""
# matlab indexes start in 1 but python in 0, so I have to do:
# range(start=(len(mpre) - 2), end=0, step=-1)
# also the python function range excludes the end, resulting in:
# range(start=(len(mpre) - 2), end=-1, step=-1)
for i in range(len(mpre)-2, -1, -1):
mpre[i] = max(mpre[i], mpre[i+1])
"""
This part creates a list of indexes where the recall changes
matlab: i=find(mrec(2:end)~=mrec(1:end-1))+1;
"""
i_list = []
for i in range(1, len(mrec)):
if mrec[i] != mrec[i-1]:
i_list.append(i) # if it was matlab would be i + 1
"""
The Average Precision (AP) is the area under the curve
(numerical integration)
matlab: ap=sum((mrec(i)-mrec(i-1)).*mpre(i));
"""
ap = 0.0
for i in i_list:
ap += ((mrec[i]-mrec[i-1])*mpre[i])
return ap, mrec, mpre
"""
Draw plot using Matplotlib
"""
def draw_plot_func(dictionary, n_classes, window_title, plot_title, x_label, output_path, to_show, plot_color, true_p_bar):
# sort the dictionary by decreasing value, into a list of tuples
sorted_dic_by_value = sorted(dictionary.items(), key=operator.itemgetter(1))
print(sorted_dic_by_value)
# unpacking the list of tuples into two lists
sorted_keys, sorted_values = zip(*sorted_dic_by_value)
#
if true_p_bar != "":
"""
Special case to draw in:
- green -> TP: True Positives (object detected and matches ground-truth)
- red -> FP: False Positives (object detected but does not match ground-truth)
- pink -> FN: False Negatives (object not detected but present in the ground-truth)
"""
fp_sorted = []
tp_sorted = []
for key in sorted_keys:
fp_sorted.append(dictionary[key] - true_p_bar[key])
tp_sorted.append(true_p_bar[key])
plt.barh(range(n_classes), fp_sorted, align='center', color='crimson', label='False Positive')
plt.barh(range(n_classes), tp_sorted, align='center', color='forestgreen', label='True Positive', left=fp_sorted)
# add legend
plt.legend(loc='lower right')
"""
Write number on side of bar
"""
fig = plt.gcf() # gcf - get current figure
axes = plt.gca()
r = fig.canvas.get_renderer()
for i, val in enumerate(sorted_values):
fp_val = fp_sorted[i]
tp_val = tp_sorted[i]
fp_str_val = " " + str(fp_val)
tp_str_val = fp_str_val + " " + str(tp_val)
# trick to paint multicolor with offset:
# first paint everything and then repaint the first number
t = plt.text(val, i, tp_str_val, color='forestgreen', va='center', fontweight='bold')
plt.text(val, i, fp_str_val, color='crimson', va='center', fontweight='bold')
if i == (len(sorted_values)-1): # largest bar
adjust_axes(r, t, fig, axes)
else:
plt.barh(range(n_classes), sorted_values, color=plot_color)
"""
Write number on side of bar
"""
fig = plt.gcf() # gcf - get current figure
axes = plt.gca()
r = fig.canvas.get_renderer()
for i, val in enumerate(sorted_values):
str_val = " " + str(val) # add a space before
if val < 1.0:
str_val = " {0:.2f}".format(val)
t = plt.text(val, i, str_val, color=plot_color, va='center', fontweight='bold')
# re-set axes to show number inside the figure
if i == (len(sorted_values)-1): # largest bar
adjust_axes(r, t, fig, axes)
# set window title
fig.canvas.set_window_title(window_title)
# write classes in y axis
tick_font_size = 12
plt.yticks(range(n_classes), sorted_keys, fontsize=tick_font_size)
"""
Re-scale height accordingly
"""
init_height = fig.get_figheight()
# comput the matrix height in points and inches
dpi = fig.dpi
height_pt = n_classes * (tick_font_size * 1.4) # 1.4 (some spacing)
height_in = height_pt / dpi
# compute the required figure height
top_margin = 0.15 # in percentage of the figure height
bottom_margin = 0.05 # in percentage of the figure height
figure_height = height_in / (1 - top_margin - bottom_margin)
# set new height
if figure_height > init_height:
fig.set_figheight(figure_height)
# set plot title
plt.title(plot_title, fontsize=14)
# set axis titles
# plt.xlabel('classes')
plt.xlabel(x_label, fontsize='large')
# adjust size of window
fig.tight_layout()
# save the plot
fig.savefig(output_path)
# show image
# if to_show:
plt.show()
# close the plot
# plt.close()
"""
Plot - adjust axes
"""
def adjust_axes(r, t, fig, axes):
# get text width for re-scaling
bb = t.get_window_extent(renderer=r)
text_width_inches = bb.width / fig.dpi
# get axis width in inches
current_fig_width = fig.get_figwidth()
new_fig_width = current_fig_width + text_width_inches
propotion = new_fig_width / current_fig_width
# get axis limit
x_lim = axes.get_xlim()
axes.set_xlim([x_lim[0], x_lim[1]*propotion])
def read_txt_to_list(path):
# open txt file lines to a list
with open(path) as f:
content = f.readlines()
# remove whitespace characters like `\n` at the end of each line
content = [x.strip() for x in content]
return content