Amazon Rekognition Custom Labels SDK を利用した折損電柱の自動検知

def create_project(project_name):

    client=boto3.client('rekognition')

    #Create a project
    print('Creating project:' + project_name)
    response=client.create_project(ProjectName=project_name)
    arn = response['ProjectArn']
    print('project ARN: ' + response['ProjectArn'])
    return arn

# Train model helper method
import json

def train_model(project_arn, version_name, output_config, training_dataset,testing_dataset):

    client=boto3.client('rekognition')

    print('Starting training of: ' + version_name)

    try:
        response=client.create_project_version(ProjectArn=project_arn, 
            VersionName=version_name,
            OutputConfig=output_config,
            TrainingData=training_dataset,
            TestingData=testing_dataset)

        # Wait for the project version training to complete
        project_version_training_completed_waiter = client.get_waiter('project_version_training_completed')
        project_version_training_completed_waiter.wait(ProjectArn=project_arn,
        VersionNames=[version_name])

        #Get the completion status
        describe_response=client.describe_project_versions(ProjectArn=project_arn,
            VersionNames=[version_name])
        for model in describe_response['ProjectVersionDescriptions']:
            print('Project Version ARN: ' + model['ProjectVersionArn'])
            print("Status: " + model['Status'])
            print("Message: " + model['StatusMessage']) 
    except Exception as e:
        print(e)

    print('Done...')

project_arn='arn:aws:rekognition:us-east-1:<12桁のAWSアカウント>:project/project_utility_pole_blog/<表示されたARNと同じ値>'

version_name='v1'

output_config = json.loads('{"S3Bucket":"my-bucket", "S3KeyPrefix":"blog/REK-CL/utilitypoles/"}')

training_dataset= json.loads('{"Assets": [{ "GroundTruthManifest": { "S3Object": { "Bucket": "my-bucket", "Name": "datasets/cables-ds/manifests/output/output.manifest" } } } ] }')

testing_dataset= json.loads('{"AutoCreate":true}')

train_model(project_arn, version_name, output_config, training_dataset, testing_dataset)

def describe_model(project_arn, version_name):

    client=boto3.client('rekognition')

    response=client.describe_project_versions(ProjectArn=project_arn,
        VersionNames=[version_name])

    for model in response['ProjectVersionDescriptions']:
        print(json.dumps(model,indent=4,default=str))

import random
import numpy as np
from scipy import ndarray
import skimage as sk
from skimage import transform
from skimage import util

def random_rotation(image_array: ndarray):
    # pick a random degree of rotation between 25% on the left and 25% on the right
    random_degree = random.uniform(-25, 25)
    return sk.transform.rotate(image_array, random_degree, preserve_range=True)

def horizontal_flip(image_array: ndarray):
    # horizontal flip doesn't need skimage, it's easy as flipping the image array of pixels !
    return image_array[:, ::-1]

# Helper method that makes a inference on a Custom Labels trained model for a binary classification problem
import boto3
import io

from botocore.exceptions import ClientError

def getLabelAndConfidenceBinary(bucket, image, model):
    """
    :param bucket: The name of the S3 bucket that contains the image that you want to analyze.
    :param image: The name of the image that you want to analyze.
    :param model: The ARN of the Amazon Rekognition Custom Labels model that you want to use.
    """

    rek_client=boto3.client('rekognition')
    s3_connection = boto3.resource('s3')

    # set minimum confidence to 0 as we are interested in results for all condidence levels
    min_confidence = 0

    try:
        #Get image from S3 bucket.
        s3_object = s3_connection.Object(bucket, image)
        s3_response = s3_object.get()

        #Call DetectCustomLabels 
        response = rek_client.detect_custom_labels(Image={'S3Object': {'Bucket': bucket, 'Name': image}},
            MinConfidence=min_confidence,
            ProjectVersionArn=model)

    except ClientError as err:
        print("Exception in get_custom_labels()")
        raise

    if response['CustomLabels'][0]["Confidence"] >= response['CustomLabels'][1]["Confidence"]:
        return response['CustomLabels'][0]["Name"], response['CustomLabels'][0]["Confidence"]
else:
        return response['CustomLabels'][1]["Name"], response['CustomLabels'][1]["Confidence"]


# Helper method that iterates through an S3 bucket and retrieves image labels
def getImageNames(bucket, prefix):
    import boto3
    client = boto3.client('s3')
    paginator = client.get_paginator('list_objects_v2')
    result = paginator.paginate(Bucket=bucket, Prefix=prefix)
    images = []
for page in result:
        if "Contents" in page:
            for key in page[ "Contents" ]:
                keyString = key[ "Key" ]
                name = keyString.split("/")[-1]
iclass = name.split("-")[0]
if len(iclass)>0:
                    images.append([keyString, iclass])
    return images

# This code loops through all images and creates a y_true list objects based on image labels
image_folder = "blog/REK-CL/utilitypoles/new_test_images/"
y_true = []
names = []
test_images = getImageNames('my-bucket', image_folder)
for image in test_images:
    iclass = image[1]
name = image[0]
names.append(name)
    if iclass=="bad":
        y_true.append(1)
    else:
        y_true.append(0)

# The helper method below makes predictions on a deploy Custom Labels model and returns predictions
# Custom Labels associates confidence level to the label it predicts
# In the code below, we are turning that to a [0 to 1] probability scale, where 1 indicates a broken pole 
def getPredictions(model, bucket, images):
    y_pred = []
y_prob = []
for image in images:
        labelconf = getLabelAndConfidenceBinary(bucket, image[0], model)
        if labelconf[0]=="broken":
            y_pred.append(1)
            prob = labelconf[1]/100.0
            y_prob.append(prob)
        if labelconf[0]=="good":
            y_pred.append(0)
            prob = 1.0 - labelconf[1]/100.0
            y_prob.append(prob)
        if labelconf[0]=="":
            raise Exception("Invalid label")
    return y_pred, y_prob

bucket = "my-bucket"
# Assign Project Version ARN returned by train_model() below for both the models
original_model = "arn:aws:rekognition:us-east-1:xxxxxxxxxxxx:project/upole-new-aug14/version/upole-new-aug14.2021-08-14T17.20.06/1628976006624"
enhanced_model = "arn:aws:rekognition:us-east-1: xxxxxxxxxxxx:project/upole-new-aug14/version/upole-new-aug14.2021-08-14T18.36.45/1628980605880"
y_pred_original, y_prob_original = getPredictions(original_model, bucket, test_images)
y_pred_enhanced, y_prob_enhanced = getPredictions(enhanced_model, bucket, test_images)


from sklearn.metrics import accuracy_score
print("Original model accuracy = ", round(accuracy_score(y_true, y_pred_original), 2))
print("Enhanced model accuracy = ", round(accuracy_score(y_true, y_pred_enhanced), 2))

------ 上記コードの出力結果 -------
Original model accuracy =  0.79
Enhanced model accuracy =  0.89
--------------------------------
import numpy as np
from sklearn import metrics
def calculateAUC(y_true, y_prob, pos_label):
    fpr, tpr, thresholds = metrics.roc_curve(y_true, np.array(y_prob), pos_label=pos_label)
    return metrics.auc(fpr, tpr)

print("Original model AUC = ", round(calculateAUC(y_true, y_prob_original, 1),2))
print("Enhanced model AUC = ", round(calculateAUC(y_true, y_prob_enhanced, 1),2))

---- 上記コードの出力結果 ----
Original model AUC =  0.92
Enhanced model AUC =  0.96
---------------------------

def delete_project(project_arn):

	client = boto3.client(‘rekognition’)

	print(‘Deleting project: ’ + project_arn)
	response = client.delete_project(ProjectArn=project_arn)
	print(‘Status: ’ + response[‘Status’])
	print(‘Done…’)