Custom Data Object Detection (Azure Custom Vision)

In this post I am going to talk about the steps to train custom data on Azure Custom Vision.

Dataset

The dataset that is being used in this tutorial is a car object detection dataset from Kaggle.

The dataset consists of 5 columns: image filename, xmin, ymin, xmax, ymax

Preprocessing

First, the dataset has to be preprocessed into specific format:

The regions specify the bounding box in normalized coordinates, and the coordinates are given in the order: left, top, width, height.

The python below preprocess the csv into a dictionary that follows the required format:

def process_points(row):

    fn = row["image"]
    xmin = int(row["xmin"])
    ymin = int(row["ymin"])
    xmax = int(row["xmax"])
    ymax = int(row["ymax"])

    img = cv2.imread(f'C:projects/Images/car/{fn}')
    height, width, channels= img.shape


    car_image_regions[fn.strip('.jpg')] = [xmin/width,ymin/height,(xmax-
            xmin)/width,(ymax-ymin)/height]


if __name__ == '__main__':
  car_image_regions = {}
  df = pd.read_csv('car.csv')
  df.apply(process_points, axis=1)

Create env file

Create a .env file in the project directory and the file should contains the following:

VISION_TRAINING_ENDPOINT= YOUR_AZURE_TRAINING_ENDPOINT
VISION_TRAINING_KEY= YOUR_AZURE_TRAINING_KEY
VISION_PREDICTION_KEY= YOUR_AZURE_PREDICTION_KEY
VISION_PREDICTION_RESOURCE_ID= YOUR_AZURE_PREDICTION_RESOURCE_ID

You can find your training endpoint and key from the custom vision training resource on Azure platform. (create a resource if you haven’t)

Get Wei Wen’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Similarly, you can also find the prediction key from the custom vision prediction resource. However, the prediction resource id is in the properties tab under the prediction resource:

To load the credentials:

load_dotenv()
# Replace with valid values
ENDPOINT = os.environ["VISION_TRAINING_ENDPOINT"]
training_key = os.environ["VISION_TRAINING_KEY"]
prediction_key = os.environ["VISION_PREDICTION_KEY"]
prediction_resource_id = os.environ["VISION_PREDICTION_RESOURCE_ID"]

Project Creation

We can now create project on custom vision platform using Python:

credentials = ApiKeyCredentials(in_headers={"Training-key": training_key})
trainer = CustomVisionTrainingClient(ENDPOINT, credentials)
prediction_credentials = ApiKeyCredentials(in_headers={"Prediction-key": prediction_key})
predictor = CustomVisionPredictionClient(ENDPOINT, prediction_credentials)

publish_iteration_name = "cardetect"

# Find the object detection domain
obj_detection_domain = next(domain for domain in trainer.get_domains() if
                           domain.type == "ObjectDetection" and domain.name == "General")

# Create a new project
print("Creating project...")
# Use uuid to avoid project name collisions.
project = trainer.create_project(publish_iteration_name,
                                domain_id=obj_detection_domain.id)

# Make two tags in the new project
car_tag = trainer.create_tag(project.id, "car")

If you head to custom vision platform, you shall see a project is being created:

But it is empty, we will upload the images and coordinates data in the next section.

Upload Images

The code to upload images is simple, it just loop through all the data entry and call the azure api to upload the image:

def upload_images():
    base_image_location = os.path.join(os.path.dirname(__file__), "Images")

    # Go through the data table above and create the images
    print("Adding images...")
    tagged_images_with_regions = []

    for file_name in car_image_regions.keys():
        x, y, w, h = car_image_regions[file_name]
        regions = [
            Region(tag_id=car_tag.id, left=x, top=y, width=w, height=h)]

        with open(
                os.path.join(base_image_location, "car", file_name + ".jpg"),
                mode="rb") as image_contents:
            tagged_images_with_regions.append(
                ImageFileCreateEntry(name=file_name,
                                     contents=image_contents.read(),
                                     regions=regions))

    upload_result = trainer.create_images_from_files(project.id,
                                                     ImageFileCreateBatch(
                                                         images=tagged_images_with_regions))
    if not upload_result.is_batch_successful:
        print("Image batch upload failed.")
        for image in upload_result.images:
            print("Image status: ", image.status)
        exit(-1)



car_image_regions  = dict(itertools.islice(car_image_regions.items(),
                                                64))

upload_images()

You may notice that I am slicing the dictionary into a size of 64, this is because Azure can only upload 64 images at one time, you may upload multiple times if you have a larger dataset.

Run the upload code and now if you head to the custom vision project, you shall see the images have been uploaded and tagged:

Train

To train the model:

def train():
    print("Training...")
    iteration = trainer.train_project(project.id)
    while (iteration.status != "Completed"):
        iteration = trainer.get_iteration(project.id, iteration.id)
        print("Training status: " + iteration.status)
        time.sleep(1)

    return iteration


iteration = train()

Access the performance of the model:

If you feel that the model is good enough, you can publish the model with the following code:

def publish(iteration):
    trainer.publish_iteration(project.id, iteration.id, publish_iteration_name,
                              prediction_resource_id)
    print("Done!")

publish(iteration)

After it’s published, you will see the option became unpublish:

Prediction

To make a prediction with your test image:

def predict(project,publish_iteration_name):
    # Now there is a trained endpoint that can be used to make a prediction
    # Open the sample image and get back the prediction results.
    with open(os.path.join('C:/projects/Images', "test",
                           "vid_4_2360.jpg"),
              mode="rb") as test_data:
        results = predictor.detect_image(project.id, publish_iteration_name,
                                         test_data)

    # Display the results.
    for prediction in results.predictions:
        print(
            "\t" + prediction.tag_name + ": {0:.2f}% bbox.left = {1:.2f}, bbox.top = {2:.2f}, bbox.width = {3:.2f}, bbox.height = {4:.2f}".format(
                prediction.probability * 100, prediction.bounding_box.left,
                prediction.bounding_box.top, prediction.bounding_box.width,
                prediction.bounding_box.height))


predict(project,publish_iteration_name)

Putting everything together:

from azure.cognitiveservices.vision.customvision.training import CustomVisionTrainingClient
from azure.cognitiveservices.vision.customvision.prediction import CustomVisionPredictionClient
from azure.cognitiveservices.vision.customvision.training.models import ImageFileCreateBatch, ImageFileCreateEntry, Region
from msrest.authentication import ApiKeyCredentials
import os, uuid
from dotenv import load_dotenv
import pandas as pd
import itertools
import cv2
import time

def create_project(trainer):
    # Find the object detection domain
    obj_detection_domain = next(domain for domain in trainer.get_domains() if
                                domain.type == "ObjectDetection" and domain.name == "General")

    # Create a new project
    print("Creating project...")
    # Use uuid to avoid project name collisions.
    project = trainer.create_project(str(uuid.uuid4()),
                                     domain_id=obj_detection_domain.id)


    return project,trainer

def add_tags(project,trainer):
    print('adding tags...')
    face_tag = trainer.create_tag(project.id, "face")

    return face_tag

def process_points(row):

    fn = row["image"]
    xmin = int(row["xmin"])
    ymin = int(row["ymin"])
    xmax = int(row["xmax"])
    ymax = int(row["ymax"])

    img = cv2.imread(f'C:/projects/Images/car/{fn}')
    height, width, channels= img.shape


    car_image_regions[fn.strip('.jpg')] = [xmin/width,ymin/height,(xmax-
            xmin)/width,(ymax-ymin)/height]

def upload_images():
    base_image_location = os.path.join(os.path.dirname(__file__), "Images")

    # Go through the data table above and create the images
    print("Adding images...")
    tagged_images_with_regions = []

    for file_name in car_image_regions.keys():
        x, y, w, h = car_image_regions[file_name]
        regions = [
            Region(tag_id=car_tag.id, left=x, top=y, width=w, height=h)]

        with open(
                os.path.join(base_image_location, "car", file_name + ".jpg"),
                mode="rb") as image_contents:
            tagged_images_with_regions.append(
                ImageFileCreateEntry(name=file_name,
                                     contents=image_contents.read(),
                                     regions=regions))

    upload_result = trainer.create_images_from_files(project.id,
                                                     ImageFileCreateBatch(
                                                         images=tagged_images_with_regions))
    if not upload_result.is_batch_successful:
        print("Image batch upload failed.")
        for image in upload_result.images:
            print("Image status: ", image.status)
        exit(-1)


def train():
    print("Training...")
    iteration = trainer.train_project(project.id)
    while (iteration.status != "Completed"):
        iteration = trainer.get_iteration(project.id, iteration.id)
        print("Training status: " + iteration.status)
        time.sleep(1)

    return iteration

def publish(iteration):
    trainer.publish_iteration(project.id, iteration.id, publish_iteration_name,
                              prediction_resource_id)
    print("Done!")

def predict(project,publish_iteration_name):
    # Now there is a trained endpoint that can be used to make a prediction
    # Open the sample image and get back the prediction results.
    with open(os.path.join('C:/projects/Images', "test",
                           "vid_4_2360.jpg"),
              mode="rb") as test_data:
        results = predictor.detect_image(project.id, publish_iteration_name,
                                         test_data)

    # Display the results.
    for prediction in results.predictions:
        print(
            "\t" + prediction.tag_name + ": {0:.2f}% bbox.left = {1:.2f}, bbox.top = {2:.2f}, bbox.width = {3:.2f}, bbox.height = {4:.2f}".format(
                prediction.probability * 100, prediction.bounding_box.left,
                prediction.bounding_box.top, prediction.bounding_box.width,
                prediction.bounding_box.height))

if __name__ == '__main__':
    car_image_regions = {}
    #get coordinates ready
    df = pd.read_csv('car.csv')
    df.apply(process_points, axis=1)



    load_dotenv()
    # Replace with valid values
    ENDPOINT = os.environ["VISION_TRAINING_ENDPOINT"]
    training_key = os.environ["VISION_TRAINING_KEY"]
    prediction_key = os.environ["VISION_PREDICTION_KEY"]
    prediction_resource_id = os.environ["VISION_PREDICTION_RESOURCE_ID"]

    credentials = ApiKeyCredentials(in_headers={"Training-key": training_key})
    trainer = CustomVisionTrainingClient(ENDPOINT, credentials)
    prediction_credentials = ApiKeyCredentials(in_headers={"Prediction-key": prediction_key})
    predictor = CustomVisionPredictionClient(ENDPOINT, prediction_credentials)

    publish_iteration_name = "cardetect"

    # Find the object detection domain
    obj_detection_domain = next(domain for domain in trainer.get_domains() if
                                domain.type == "ObjectDetection" and domain.name == "General")

   # Create a new project
    print("Creating project...")
    # Use uuid to avoid project name collisions.
    project = trainer.create_project(publish_iteration_name,
                                     domain_id=obj_detection_domain.id)

    # Make two tags in the new project
    car_tag = trainer.create_tag(project.id, "car")

    car_image_regions  = dict(itertools.islice(car_image_regions.items(),
                                                64))

    upload_images()

    iteration = train()

    publish(iteration)

    predict(project,publish_iteration_name)