Object detection
Want to run this code for yourself?
You can find the interactive ipython notebook where you can run all the steps listed here at
PS: it will take a couple of minutes for the mybinder instance to boot up and be ready for use.
Imports
import json
import os
import re
import requests
import shutil
import tempfile
from multiprocessing import Pool
# Helper methods for creating, uploading data and training an object detection model.
def create_new_model(base_url, auth_key, categories):
"""
function to create a new model for training
Args:
base_url: url to nanonets endpoint which will decide what type of model to create
auth_key: authentication key provided by https://app.nanonets.com/#/keys
categories: List of labels you want to predict/ detect
return:
model_id: a unique reference to new created model
"""
payload = json.dumps({"categories" : categories})
headers = {
'Content-Type': "application/json",
}
response = requests.request(
"POST",
base_url,
headers=headers,
auth=requests.auth.HTTPBasicAuth(auth_key, ''),
data=payload,
)
result = json.loads(response.text)
print("Model Information: ", result)
model_id, model_type, categories = (result["model_id"], result["model_type"], result["categories"])
return model_id
def get_model_info(base_url, auth_key, model_id):
"""
function to get/ print information about model at any time
Args:
base_url: url to nanonets endpoint which will decide what type of model to create
auth_key: authentication key provided by https://app.nanonets.com/#/keys
model_id: unique model_id generated at model creation time
"""
print('%s%s'%(base_url, model_id))
response = requests.request(
'GET',
'%s%s'%(base_url, model_id),
auth=requests.auth.HTTPBasicAuth(auth_key, '')
)
print(response.text)
result = json.loads(response.text)
model_id, model_type, categories, state = (result["model_id"], result["model_type"], result["categories"], result["state"])
return model_id, model_type, categories, state
def generate_upload_data(image_file, annotation_info, model_id):
"""
function to translate image and annotation info into format suitable for upload
Args:
image_file[str]: full path to where the image is located
annotation_info[str]: json formatted string of the object annotations in the image
eg. '[{"name": "object_1", "bndbox": {"xmin": 50, "ymin": 50, "xmax": 100, "ymax": 100}}, ...]'
model_id[str]: model id for which data needs to be uploaded
Returns:
data[Dict[str, Any]]: data that can be passed onto to the upload data method
"""
data = {
'file' : open(image_file, 'rb'),
'data' :('', '[{"filename":"%s", "object": %s}]' % (image_file.rsplit('/', 1)[1], annotation_info)),
'modelId' :('', '%s'% model_id),
}
return data
def upload_data(base_url, model_id, auth_key, data):
"""
function to upload data for a model that has been created
Args:
base_url[str]: nanonets endpoint to which the model upload request will be sent
eg. https://app.nanonets.com/api/v2/ObjectDetection/Model/
model_id[str]: model id of the model for which data is being uploaded generated by calling the create_model method
auth_key[str]: authentication key provided by https://app.nanonets.com/#/keys
data[Dict[str, Any]]: dictionary recieved from the generate_upload_data method
"""
response = requests.post(
'%s%s/UploadFile/'% (base_url, model_id),
auth=requests.auth.HTTPBasicAuth(auth_key, ''),
files=data,
)
print(response.text)
def train_model(base_url, auth_key, model_id):
headers = {'authorization': 'Basic %s'%auth_key}
querystring = {'modelId': model_id}
response = requests.request(
'POST',
'%s%s/Train/'%(base_url, model_id),
headers=headers,
auth=requests.auth.HTTPBasicAuth(AUTH_KEY, ''),
params=querystring,
)
print("training started .... ")
print(json.loads(response.text))
Putting it all together
Constants
Some universal constants that we will need for creating new models, uploading data and launching training
BASE_MODEL_URL = "https://app.nanonets.com/api/v2/ObjectDetection/Model/"
CATEGORIES = ['TieFighter', 'MillenniumFalcon']
AUTH_KEY = "<AUTH_KEY_FROM_NANONETS_APP>" ## can be foung https://app.nanonets.com/#/keys
Use Data directly from github?
If you already have a copy of the images and annotations required, you can set the following variable to False and update the image_directory and annotation_directory with the values to the local paths. Else you can directly use the image and annotations available on github to train a new object detection model.
use_github_data = True
Getting the Data
If you already have a local copy of the data available you can skip the next few cells and directly update the image_directory and annotation_directory values with the location of the images and annotations respectively.
If you do not have a local copy of the data, the next 3 cells, will download the images and annotations from the object-detection-sample github repo store them in a local directory which will then be used to launch an object detection job.
PS: The directories will be deleted as soon as the job has been launched.
git_repo_url = "https://github.com/NanoNets/object-detection-sample-python/tree/master"
git_repo_images = os.path.join(git_repo_url, "images")
git_repo_annotations = os.path.join(git_repo_url, "annotations/json/")
raw_github_url = 'https://raw.githubusercontent.com/NanoNets/object-detection-sample-python/master/'
# Download images to temp folder
def download_file_to(source_url, destination_location):
f = open(destination_location, 'wb')
f.write(requests.get(source_url).content)
f.close()
def download_file(file_name, base_source_url, base_destination_location):
source_url = os.path.join(base_source_url, file_name)
destination_location = os.path.join(base_destination_location, file_name)
download_file_to(source_url, destination_location)
def download_file_multiprocess(download_information):
file_name, base_source_url, base_destination_location = download_information
download_file(file_name, base_source_url, base_destination_location)
# Download annotations to temp folder
if use_github_data:
p = Pool(5)
page = requests.get(git_repo_images)
image_files = set()
pattern = re.compile("(videoplayback[\d]*\.jpg)")
for line in page.iter_lines():
if not line:
continue
matches = pattern.findall(str(line))
if not matches:
continue
for i in matches:
image_files.add(i)
github_images_url = os.path.join(raw_github_url, "images")
temp_images_folder = tempfile.mkdtemp(suffix="images")
p.map(
download_file_multiprocess,
[(file_name, github_images_url, temp_images_folder) for file_name in image_files]
)
page = requests.get(git_repo_annotations)
annotation_files = set()
pattern = re.compile("(videoplayback[\d]*\.json)")
for line in page.iter_lines():
if not line:
continue
matches = pattern.findall(str(line))
if not matches:
continue
for i in matches:
annotation_files.add(i)
github_annotations_url = os.path.join(raw_github_url, "annotations/json/")
temp_annotations_folder = tempfile.mkdtemp(suffix="annotations")
p.map(
download_file_multiprocess,
[(file_name, github_annotations_url, temp_annotations_folder) for file_name in annotation_files]
)
# create_model
model_id = create_new_model(base_url=BASE_MODEL_URL, auth_key=AUTH_KEY, categories=CATEGORIES)
print("New model created: ", model_id)
# generate and upload_data
# change current working directory to location where object-detection-sample-python repo is cloned
image_directory = temp_images_folder # REPLACE WITH LOCAL FOLDER IF ALREADY EXISTS
annotation_directory = temp_annotations_folder # REPLACE WITH LOCAL FOLDER IF ALREADY EXISTS
annotation_files = os.listdir(annotation_directory)
# get the image and annotation info in format easy to upload
image_and_annotations = []
for annotation_file in annotation_files:
with open(os.path.join(annotation_directory, annotation_file)) as f:
# check corresponding image file exists
image_path = os.path.join(image_directory, os.path.basename(annotation_file).replace("json", "jpg"))
if not os.path.exists(image_path):
# skipping annotation as image does not exist
continue
annotation_info = f.readline().strip()
image_and_annotations.append((image_path, annotation_info, model_id))
Upload image and annotation using the previously built helper methods
# # This process will however be very slow as it will upload each image and annotation information serially.
# # We can speed this process up significantly by using the python Multiprocessing module.
# total_images = len(image_and_annotations)
# for i, (image_file, annotation_info, model_id) in enumerate(image_and_annotations):
# print("Processing Image %d / %d " % (i, total_images))
# data = generate_upload_data(image_file, annotation_info, model_id)
# upload_data(BASE_MODEL_URL, model_id, AUTH_KEY, data)
# i = i + 1
# upload the data for the model to be trained, using multiprocessig to make data upload faster.
def upload_data_multiprocessing(image_and_annotation):
image_file, annotation_info, model_id = image_and_annotation
data = generate_upload_data(image_file, annotation_info, model_id)
upload_data(BASE_MODEL_URL, model_id, AUTH_KEY, data)
p = Pool(4)
p.map(upload_data_multiprocessing, image_and_annotations)
# get model info
get_model_info(BASE_MODEL_URL, AUTH_KEY, model_id)
Train
# launch training for model once all data has been uploaded
train_model(BASE_MODEL_URL, AUTH_KEY, model_id)
Check model status
get_model_info(BASE_MODEL_URL, AUTH_KEY, model_id)
# delete temp folders
if use_github_data:
# delete temp images folder
shutil.rmtree(temp_images_folder)
# delete temp annotations folder
shutil.rmtree(temp_annotations_folder)