detector関係の追加

Author: terapyon

.dockerignore

@@ -0,0 +1,6 @@
+Dockerfile
+README.md
+*.pyc
+*.pyo
+*.pyd
+__pycache__

mahjong_sample_web_app/detector/__init__.py

@@ -0,0 +1,77 @@
+from pathlib import Path
+import numpy as np
+
+# from scipy.misc import imresize
+from PIL import Image as pil_image
+from keras.preprocessing import image
+from .postprocess import PostProcess
+from .ssd.ssd import SingleShotMultiBoxDetector
+
+
+model_file = Path(__file__).parent / "models" / "weights.25-0.05.hdf5"
+param_file = (
+    Path(__file__).parent / "models" / "ssd300_params_mahjong_vgg16_train_2.json"
+)
+
+
+def model_build(model_file, param_file):
+    ssd = SingleShotMultiBoxDetector(
+        overlap_threshold=0.5, nms_threshold=0.45, max_output_size=400
+    )
+    ssd.load_parameters(param_file)
+    ssd.build(init_weight=model_file)
+    return ssd
+
+
+def _add_margin(img):
+    img_shape = list(img.shape)
+    if img_shape[0] == img_shape[1]:
+        return img
+    if img_shape[0] < img_shape[1]:
+        min_arg = 0
+        max_arg = 1
+    else:
+        min_arg = 1
+        max_arg = 0
+    margin_shape = img_shape
+    margin_shape[min_arg] = int((img_shape[max_arg] - img_shape[min_arg]) / 2.0)
+    margin = np.tile([0.0], margin_shape)
+    new_img = np.concatenate([margin, img], axis=min_arg)
+    new_img = np.concatenate([new_img, margin], axis=min_arg)
+    return new_img
+
+
+def pred(ssd, img):
+    inputs = np.array([img.copy()])
+    results = ssd.detect(inputs, batch_size=1, verbose=1, do_preprocess=True)
+    return results
+
+
+def load_image(img_obj, input_shape=(512, 512)):
+    # img = image.load_img(img_path)
+    img = pil_image.open(img_obj)
+    if img.mode != "RGB":
+        img = img.convert("RGB")
+
+    img_array = image.img_to_array(img)
+    new_img = _add_margin(img_array)
+    new_img_float = np.array(
+        pil_image.fromarray(new_img.astype("uint8")).resize(
+            size=input_shape  # , resample=pil_image.BICUBIC
+        )
+    ).astype("float32")
+    # new_img_float = imresize(new_img, input_shape).astype("float32")
+    return new_img_float
+
+
+def detect(img_obj):
+    img = load_image(img_obj)
+    ssd = model_build(model_file, param_file)
+    pred_result = pred(ssd, img)
+
+    pp = PostProcess(ssd.class_names, pred_threshold=0.9)
+    pp.set_top_score(pred_result)
+    list_label = pp.get_list_pi()
+    # pp.save_image(img, pred_result, savepath)
+    # print(list_label)
+    return list_label

mahjong_sample_web_app/detector/detector.py

@@ -0,0 +1,71 @@
+# import matplotlib.pyplot as plt
+import numpy as np
+
+
+class PostProcess:
+    def __init__(self, class_names, pred_threshold=0.9):
+        self.pred_threshold = pred_threshold
+        self.class_names = class_names
+
+    def set_top_score(self, pred_result):
+        # Parse the outputs.
+        det_label = pred_result[0][:, 0]
+        det_conf = pred_result[0][:, 1]
+
+        # get top score result
+        self.top_indices = [
+            i for i, conf in enumerate(det_conf) if conf >= self.pred_threshold
+        ]
+        self.top_conf = det_conf[self.top_indices]
+        self.top_label_indices = det_label[self.top_indices].tolist()
+
+    def get_list_pi(self):
+        list_label = []
+        for i in range(self.top_conf.shape[0]):
+            label = int(self.top_label_indices[i])
+            label_name = self.class_names[label]
+            list_label.append(label_name)
+
+        return list_label
+
+    # def save_image(self, img, pred_result, savepath):
+    #     colors = plt.cm.hsv(np.linspace(0, 1, 35)).tolist()
+    #     plt.tick_params(
+    #         labelbottom=False, labelleft=False, labelright=False, labeltop=False
+    #     )
+    #     plt.tick_params(bottom=False, left=False, right=False, top=False)
+
+    #     plt.imshow(img / 255.0)
+    #     currentAxis = plt.gca()
+
+    #     det_xmin = pred_result[0][:, 2]
+    #     det_ymin = pred_result[0][:, 3]
+    #     det_xmax = pred_result[0][:, 4]
+    #     det_ymax = pred_result[0][:, 5]
+
+    #     top_xmin = det_xmin[self.top_indices]
+    #     top_ymin = det_ymin[self.top_indices]
+    #     top_xmax = det_xmax[self.top_indices]
+    #     top_ymax = det_ymax[self.top_indices]
+
+    #     for i in range(self.top_conf.shape[0]):
+    #         xmin = int(round(top_xmin[i] * img.shape[1]))
+    #         ymin = int(round(top_ymin[i] * img.shape[0]))
+    #         xmax = int(round(top_xmax[i] * img.shape[1]))
+    #         ymax = int(round(top_ymax[i] * img.shape[0]))
+
+    #         label = int(self.top_label_indices[i])
+    #         score = self.top_conf[i]
+    #         label_name = self.class_names[label]
+    #         display_txt = "{:0.2f}, {}".format(score, label_name)
+    #         coords = (xmin, ymin), xmax - xmin + 1, ymax - ymin + 1
+    #         color = colors[label]
+
+    #         currentAxis.add_patch(
+    #             plt.Rectangle(*coords, fill=False, edgecolor=color, linewidth=2)
+    #         )
+    #         currentAxis.text(
+    #             xmin, ymin, display_txt, bbox={"facecolor": color, "alpha": 1.0}
+    #         )
+
+    #     plt.savefig(savepath)

mahjong_sample_web_app/detector/postprocess.py

@@ -0,0 +1,33 @@
+import numpy as np
+import keras.backend as K
+from keras.engine.topology import Layer
+from keras.engine.topology import InputSpec
+
+
+class L2Normalization(Layer):
+    """
+    """
+
+    def __init__(self, scale, **kwargs):
+        self.scale = scale
+        self.gamma = None
+        self.axis = None
+        # if K.image_dim_ordering() == "tf":
+        if K.image_data_format() == "tf":
+            self.axis = 3
+        else:
+            self.axis = 1
+        super(L2Normalization, self).__init__(**kwargs)
+
+    def build(self, input_shape):
+        self.input_spec = [InputSpec(shape=input_shape)]
+        shape = (input_shape[self.axis],)
+        self.gamma = K.variable(
+            self.scale * np.ones(shape), name="{}_gamma".format(self.name)
+        )
+        self.trainable_weights = [self.gamma]
+
+    def call(self, x, mask=None):
+        output = K.l2_normalize(x, self.axis)
+        output *= self.gamma
+        return output

mahjong_sample_web_app/detector/ssd/__init__.py

@@ -0,0 +1,139 @@
+import tensorflow as tf
+from keras.losses import categorical_crossentropy
+
+
+class MultiBoxLoss:
+    """
+    """
+    def __init__(self, n_classes, alpha=1.0, neg_pos_ratio=3.0,
+                 negatives_for_hard=100):
+        self.n_classes = n_classes
+        self.alpha = alpha
+        self.neg_pos_ratio = neg_pos_ratio
+        self.negatives_for_hard = negatives_for_hard
+
+    def _softmax_loss(self, y_true, y_pred):
+        """
+        """
+        softmax_loss = categorical_crossentropy(y_true, y_pred)
+        # y_pred = tf.maximum(tf.minimum(y_pred, 1 - 1e-15), 1e-15)
+        # softmax_loss = -tf.reduce_sum(y_true * tf.log(y_pred), axis=-1)
+        return softmax_loss
+
+    def _l1_smooth_loss(self, y_true, y_pred):
+        """
+        """
+        abs_loss = tf.abs(y_true - y_pred)
+        sq_loss = 0.5 * (y_true - y_pred)**2
+        l1_loss = tf.where(tf.less(abs_loss, 1.0), sq_loss, abs_loss - 0.5)
+        return tf.reduce_sum(l1_loss, -1)
+
+    def compute_loss_old(self, y_true, y_pred):
+        """ compute loss
+        """
+        batch_size = tf.shape(y_true)[0]
+        num_boxes = tf.to_float(tf.shape(y_true)[1])
+
+        # loss for all default boxes
+        conf_loss = self._softmax_loss(y_true[:, :, 4:],
+                                       y_pred[:, :, 4:])
+        loc_loss = self._l1_smooth_loss(y_true[:, :, :4],
+                                        y_pred[:, :, :4])
+
+        # positives loss
+        num_pos = num_boxes - tf.reduce_sum(y_true[:, :, 4], axis=-1)
+        fpmask = 1 - y_true[:, :, 4]
+        pos_loc_loss = tf.reduce_sum(loc_loss * fpmask, axis=1)
+        pos_conf_loss = tf.reduce_sum(conf_loss * fpmask, axis=1)
+
+        # negatives loss
+        num_neg = tf.minimum(self.neg_pos_ratio * num_pos,
+                             num_boxes - num_pos)
+        pos_num_neg_mask = tf.greater(num_neg, 0)
+        has_min = tf.to_float(tf.reduce_any(pos_num_neg_mask))
+        num_neg = tf.concat(axis=0,
+                            values=[num_neg,
+                                    [(1 - has_min) * self.negatives_for_hard]])
+        num_neg_batch = tf.reduce_min(tf.boolean_mask(num_neg,
+                                                      tf.greater(num_neg, 0)))
+        num_neg_batch = tf.to_int32(num_neg_batch)
+        confs_start = 4 + 1
+        confs_end = confs_start + self.n_classes - 1
+        max_confs = tf.reduce_max(y_pred[:, :, confs_start:confs_end],
+                                  axis=2)
+
+        nvalues, indices = tf.nn.top_k(max_confs * y_true[:, :, 4],
+                                       k=num_neg_batch)
+
+        batch_idx = tf.expand_dims(tf.range(0, batch_size), 1)
+        batch_idx = tf.tile(batch_idx, (1, num_neg_batch))
+        full_indices = (tf.reshape(batch_idx, [-1]) * tf.to_int32(num_boxes) +
+                        tf.reshape(indices, [-1]))
+
+        neg_conf_loss = tf.gather(tf.reshape(conf_loss, [-1]),
+                                  full_indices)
+        neg_conf_loss = tf.reshape(neg_conf_loss,
+                                   [batch_size, num_neg_batch])
+        neg_conf_loss = tf.reduce_sum(neg_conf_loss, axis=1)
+
+        # loss is sum of positives and negatives
+        total_loss = pos_conf_loss + neg_conf_loss
+        total_loss /= (num_pos + tf.to_float(num_neg_batch))
+        num_pos = tf.where(tf.not_equal(num_pos, 0), num_pos,
+                           tf.ones_like(num_pos))
+        total_loss += (self.alpha * pos_loc_loss) / num_pos
+        return total_loss
+
+    def compute_loss(self, y_true, y_pred):
+        """ compute loss
+        """
+        batch_size = tf.shape(y_true)[0]
+        num_boxes = tf.to_float(tf.shape(y_true)[1])
+
+        # loss for all default boxes
+        conf_loss = self._softmax_loss(y_true[:, :, 4:],
+                                       y_pred[:, :, 4:])
+        loc_loss = self._l1_smooth_loss(y_true[:, :, :4],
+                                        y_pred[:, :, :4])
+
+        # positives loss
+        num_pos = num_boxes - tf.reduce_sum(y_true[:, :, 4], axis=-1)
+        fpmask = 1 - y_true[:, :, 4]
+        pos_loc_loss = tf.reduce_sum(loc_loss * fpmask, axis=1)
+        pos_conf_loss = tf.reduce_sum(conf_loss * fpmask, axis=1)
+
+        # negatives loss
+        num_neg = tf.minimum(self.neg_pos_ratio * num_pos,
+                             num_boxes - num_pos)
+        pos_num_neg_mask = tf.greater(num_neg, 0)
+        has_min = tf.to_float(tf.reduce_any(pos_num_neg_mask))
+        num_neg = tf.concat(axis=0,
+                            values=[num_neg,
+                                    [(1 - has_min) * self.negatives_for_hard]])
+        num_neg_batch = tf.reduce_min(tf.boolean_mask(num_neg,
+                                                      tf.greater(num_neg, 0)))
+        num_neg_batch = tf.to_int32(num_neg_batch)
+        confs_start = 4 + 1
+        confs_end = confs_start + self.n_classes - 1
+        max_confs = tf.reduce_max(y_pred[:, :, confs_start:confs_end],
+                                  axis=2)
+
+        nvalues, indices = tf.nn.top_k(max_confs * y_true[:, :, 4],
+                                       k=num_neg_batch)
+        min_nvalues = nvalues[:, -1]
+        min_nvalues = tf.expand_dims(min_nvalues, 1)
+        min_nvalues = tf.tile(min_nvalues, (1, tf.shape(max_confs)[1]))
+        nmask = tf.logical_not(tf.cast(fpmask, tf.bool))
+        nmask = tf.logical_and(nmask,
+                               tf.greater_equal(max_confs, min_nvalues))
+        fnmask = tf.to_float(nmask)
+
+        neg_conf_loss = tf.reduce_sum(conf_loss * fnmask, axis=1)
+
+        # loss is sum of positives and negatives
+        total_loss = pos_conf_loss + neg_conf_loss
+        total_loss /= (num_pos + tf.to_float(num_neg_batch))
+        num_pos = tf.where(tf.not_equal(num_pos, 0), num_pos,
+                           tf.ones_like(num_pos))
+        total_loss += (self.alpha * pos_loc_loss) / num_pos
+        return total_loss