Crack Segmentation

__init__.py

import azure.functions as func
import json
from ..shared import logger
from . import detect_object
import sys

async def main(segmentBatch: func.QueueMessage, res: func.Out[func.QueueMessage]) -> None:
    await queue_segmentation_results(segmentBatch, res)
async def queue_segmentation_results(segmentBatch: func.QueueMessage, res: func.Out[func.QueueMessage]) -> None:
    # Get Message Args
    infer_batch_message = segmentBatch.get_json()
    jwt = infer_batch_message.get('Jwt')
    image_uris = infer_batch_message.get('BlobUris')
    activity_id = infer_batch_message.get('ActivityId')
    file_names = infer_batch_message.get('FileNames')
    # Init Dependencies
    log = logger.Logger(activity_id)
    log.log_info("Python Segmentation Queue Triggered")
    detector = detect_object.ObjectDetector(log)
    detector.initialize()
    log.log_info("Getting Predictions for Images")
    inference_results = []
    try:
        predictions = detector.get_images_predictions(image_uris)
        for i, prediction in enumerate(predictions):
            inference_result = {
                "ImageId": file_names[i],
                "ClassificationResults": [],
                "SegmentationResults": [prediction]
            }
            inference_results.append(inference_result)
    except:
        log.log_error("Error: {}".format(sys.exc_info()[0]))
    
    inference_results = {
        "Results": inference_results
    }
    result_json = json.dumps(inference_results)
    log.log_info('Obtained results: {}'.format(result_json))
    log.dispatch_logs(jwt)
    res.set(result_json)

deployment.yaml

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  labels:
    app: crack-detection-segmentation-deployment
  name: crack-detection-segmentation-deployment
  namespace: azure-functions
  selfLink: /apis/extensions/v1beta1/namespaces/azure-functions/deployments/crack-detection-segmentation-deployment
spec:
  progressDeadlineSeconds: 600
  replicas: 1
  revisionHistoryLimit: 2
  selector:
    matchLabels:
      app: crack-detection-segmentation-deployment
  strategy:
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 25%
    type: RollingUpdate
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: crack-detection-segmentation-deployment
    spec:
      affinity:
        nodeAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 33
            preference:
              matchExpressions:
              - key: kubernetes.io/hostname
                operator: In 
                values:
                - aks-nodepool0
          - weight: 33
            preference:
              matchExpressions:
              - key: kubernetes.io/hostname
                operator: In 
                values:
                - aks-nodepool1
          - weight: 33
            preference:
              matchExpressions:
              - key: kubernetes.io/hostname
                operator: In 
                values:
                - aks-nodepool2
              
      containers:
      - image: nickjpurcell/crackdetectionsegmentation
        imagePullPolicy: Always
        name: crack-detection-segmentation-deployment
        env:
          - name: Key
            value: Value

        ports:
        - containerPort: 80
          protocol: TCP
        resources:
          requests:
            cpu: 450m
            memory: 2100Mi
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30
      tolerations:
      - effect: NoSchedule
        key: azure.com/aci

detect_object.py

from PIL import Image
from urllib.request import urlopen
import numpy as np
import tensorflow as tf
from urllib.request import urlopen
import gc

# Synchronized Globals 
seg_model_graph = None
seg_model_lock = None

# Constants
PATH_TO_CKPT = "segment/crack_inception_resnet.pb"
PATH_TO_LABELS = "segment/crack_label_map.pbtxt"
NUM_CLASSES = 1
BBOX_SCORE_THRESHOLD = 0.5
MASK_SCORE_THRESHOLD = 0.3
class ObjectDetector:
    def __init__(self, log):
        self.log = log
        self.gpu_options = tf.GPUOptions(allow_growth=True)
    # Initialize Model for inference
    def initialize(self):
        global seg_model_graph
        global seg_model_lock
        self.log.log_info("Initializing Object Detector")
        gc.collect()
        # Locking mechanism for deciding to load the model into memory
        # Prevents erroneous memory exaustion within the pods
        i_locked = False
        if seg_model_lock is None:
            seg_model_lock = True
            i_locked = True
        
        if seg_model_lock and not i_locked and seg_model_graph is None:
            t = 0
            sleepy_time = .1
            while seg_model_graph is None and t < 10:               
                time.sleep(sleepy_time)
                t = t + sleepy_time

        # If Model is not cached, initialize it from disk
        if seg_model_graph is None:
            self.log.log_info("Loading segmentation model from disk")
            
            # LOAD FROZEN TENSORFLOW MODEL INTO MEMORY
            seg_model_graph = tf.Graph()
            with seg_model_graph.as_default():
                od_graph_def = tf.GraphDef()
                with tf.gfile.GFile(PATH_TO_CKPT, 'rb') as fid:
                    serialized_graph = fid.read()
                    od_graph_def.ParseFromString(serialized_graph)
                    tf.import_graph_def(od_graph_def, name='')
        else:
            self.log.log_info("Model is cached")

    def run_inference_for_single_image(self, image):
        self.log.log_info("Running inference for single image")
        with seg_model_graph.as_default():
            with tf.Session(config=tf.ConfigProto(gpu_options=self.gpu_options)) as sess:
                # Get handles to input and output tensors
                ops = tf.get_default_graph().get_operations()
                all_tensor_names = {output.name for op in ops for output in op.outputs}
                tensor_dict = {}
                for key in [
                    'num_detections', 'detection_boxes', 'detection_scores',
                    'detection_classes', 'detection_masks'
                ]:
                    tensor_name = key + ':0'
                    if tensor_name in all_tensor_names:
                        tensor_dict[key] = tf.get_default_graph().get_tensor_by_name(
                            tensor_name)
                if 'detection_masks' in tensor_dict:
                    # The following processing is only for single image
                    detection_boxes = tf.squeeze(tensor_dict['detection_boxes'], [0])
                    detection_masks = tf.squeeze(tensor_dict['detection_masks'], [0])
                    # Reframe is required to translate mask from box coordinates to image coordinates and fit the image size.
                    real_num_detection = tf.cast(tensor_dict['num_detections'][0], tf.int32)
                    detection_boxes = tf.slice(detection_boxes, [0, 0], [real_num_detection, -1])
                    detection_masks = tf.slice(detection_masks, [0, 0, 0], [real_num_detection, -1, -1])
                    detection_masks_reframed = self.reframe_box_masks_to_image_masks(
                        detection_masks, detection_boxes, image.shape[1], image.shape[2])
                    detection_masks_reframed = tf.cast(
                        tf.greater(detection_masks_reframed, 0.5), tf.uint8)
                    # Follow the convention by adding back the batch dimension
                    tensor_dict['detection_masks'] = tf.expand_dims(
                        detection_masks_reframed, 0)
                image_tensor = tf.get_default_graph().get_tensor_by_name('image_tensor:0')
                self.log.log_info("Making forward pass now...")
                # Run inference
                output_dict = sess.run(tensor_dict,
                                    feed_dict={image_tensor: image})

                self.log.log_info("Completed forward pass with {} detections, now converting output types".format(output_dict['num_detections']))
                # all outputs are float32 numpy arrays, so convert types as appropriate
                output_dict['num_detections'] = int(output_dict['num_detections'][0])
                output_dict['detection_classes'] = output_dict[
                    'detection_classes'][0].astype(np.uint8)
                output_dict['detection_boxes'] = output_dict['detection_boxes'][0]
                output_dict['detection_scores'] = output_dict['detection_scores'][0]
                if 'detection_masks' in output_dict:
                    output_dict['detection_masks'] = output_dict['detection_masks'][0]
        self.log.log_info("Returning segmentation outputs")
        return output_dict

    def get_images_predictions(self, image_uris):
        predictions = []
        for image_uri in image_uris:
            bboxes = []
            scores = []
            print('Fetching: {}'.format(image_uri))
            with urlopen(image_uri) as image:
                pil_image = Image.open(image)
                image_np = self.load_image_into_numpy_array(pil_image)
                image_np_expanded = np.expand_dims(image_np, axis=0)
                output_dict = self.run_inference_for_single_image(image_np_expanded)
                box_idxs = []
                for idx, score in enumerate(output_dict['detection_scores']):
                    if score >= BBOX_SCORE_THRESHOLD:
                        box_idxs.append(idx)

                for idx in box_idxs:
                    bbox = output_dict['detection_boxes'][idx]
                    score = output_dict['detection_scores'][idx]
                    bboxes.append(bbox)
                    scores.append(score)
   
                if len(box_idxs) > 0:
                    non_suppressed_boxes = tf.image.non_max_suppression(tf.convert_to_tensor(np.array(bboxes)), tf.convert_to_tensor(np.array(scores)), 40, iou_threshold=0.25, score_threshold=float('-inf'))
                    with tf.Session(config=tf.ConfigProto(gpu_options=self.gpu_options)):
                        non_suppressed_boxes = non_suppressed_boxes.eval()
                else:
                    non_suppressed_boxes = box_idxs
                detections = []
                shape = image_np.shape
                h = shape[0]
                w = shape[1]
                for bbox_idx in non_suppressed_boxes:
                    bbox = bboxes[bbox_idx]
                    score = scores[bbox_idx]
                    p1 = (int(bbox[1] * w), int(bbox[0] * h))
                    p2 = (int(bbox[3] * w), int(bbox[2] * h))
                    detection = {
                        "BoundingBox": {
                            "P1": {
                                "X": int(bbox[1] * w), "Y": int(bbox[0] * h)
                            },
                            "P2": {
                                "X": int(bbox[3] * w), "Y": int(bbox[2] * h)
                            }
                        },
                        "Score": float(score),
                        "Class": "crack"
                    }
                    detections.append(detection)
                predictions.append({"Detections": detections})
                pil_image.close()
        return predictions
    def load_image_into_numpy_array(self, image):
        (im_width, im_height) = image.size
        return np.array(image.getdata()).reshape(
            (im_height, im_width, 3)).astype(np.uint8)
    """Transforms the box masks back to full image masks.
    Embeds masks in bounding boxes of larger masks whose shapes correspond to
    image shape.
    Args:
    box_masks: A tf.float32 tensor of size [num_masks, mask_height, mask_width].
    boxes: A tf.float32 tensor of size [num_masks, 4] containing the box
        corners. Row i contains [ymin, xmin, ymax, xmax] of the box
        corresponding to mask i. Note that the box corners are in
        normalized coordinates.
    image_height: Image height. The output mask will have the same height as
                the image height.
    image_width: Image width. The output mask will have the same width as the
                image width.
    Returns:
    A tf.float32 tensor of size [num_masks, image_height, image_width].
    """
    def reframe_box_masks_to_image_masks(self, box_masks, boxes, image_height, image_width):
        # TODO(rathodv): Make this a public function.
        def reframe_box_masks_to_image_masks_default():
            """The default function when there are more than 0 box masks."""
            def transform_boxes_relative_to_boxes(boxes, reference_boxes):
                boxes = tf.reshape(boxes, [-1, 2, 2])
                min_corner = tf.expand_dims(reference_boxes[:, 0:2], 1)
                max_corner = tf.expand_dims(reference_boxes[:, 2:4], 1)
                transformed_boxes = (boxes - min_corner) / (max_corner - min_corner)
                return tf.reshape(transformed_boxes, [-1, 4])
            box_masks_expanded = tf.expand_dims(box_masks, axis=3)
            num_boxes = tf.shape(box_masks_expanded)[0]
            unit_boxes = tf.concat(
                [tf.zeros([num_boxes, 2]), tf.ones([num_boxes, 2])], axis=1)
            reverse_boxes = transform_boxes_relative_to_boxes(unit_boxes, boxes)
            return tf.image.crop_and_resize(
                image=box_masks_expanded,
                boxes=reverse_boxes,
                box_ind=tf.range(num_boxes),
                crop_size=[image_height, image_width],
                extrapolation_value=0.0)
        image_masks = tf.cond(
            tf.shape(box_masks)[0] > 0,
            reframe_box_masks_to_image_masks_default,
            lambda: tf.zeros([0, image_height, image_width, 1], dtype=tf.float32))
        return tf.squeeze(image_masks, axis=3)

Dockerfile

    
FROM mcr.microsoft.com/azure-functions/python:2.0

COPY . /home/site/wwwroot

RUN cd /home/site/wwwroot && \
    pip install -r requirements.txt

host.json

{
    "version":  "2.0",
    "functionTimeout": "00:10:00",
    "extensions": {
        "queues": {
            "maxPollingInterval": "00:00:02",
            "visibilityTimeout" : "00:00:45",
            "batchSize": 1,
            "maxDequeueCount": 10
        }
    }
}

requirements.txt

azure-functions==1.0.0a5
azure-functions-worker==1.0.0a6
grpcio==1.14.0
grpcio-tools==1.14.0
numpy==1.15.4
pillow==5.4.1
protobuf==3.6.1
requests==2.20.1
tensorflow==1.12.0
flask
urllib3