<jupyter_start><jupyter_text>Test inference pipleTest running images through Megadetector API to obtain object bounding boxes, then on to the MIRA API for species classification.Adapted from http://dolphinvm.westus2.cloudapp.azure.com/ai4e/notebooks/cameratrap-sync-api-demo.htmlCamera-trap-detection-API-demo  Setup 1) Make sure virtual env is running and requirements.txt are installed (see README.md) 2) Load Megadetector API key from .env fileYou'll need to have a .env file somewhere in the project directory with the API key set to "MEGADETECTOR_API_KEY"<jupyter_code>%load_ext dotenv
%dotenv

import os
MEGADETECTOR_API_KEY = os.getenv('MEGADETECTOR_API_KEY')
import requests
import json
import matplotlib.pyplot as plt
from requests_toolbelt.multipart.encoder import MultipartEncoder

from io import BytesIO
from PIL import Image,ImageDraw
from azure.storage.blob import ContainerClient,BlobClient

%autosave 0

# Microsoft demo images:
demo_image_account_name = 'cameratrapblobs'
demo_image_container_name = 'cameratrapblobcontainer'
demo_image_account_url = 'https://' + demo_image_account_name + '.blob.core.windows.net/'
demo_image_blob_root = demo_image_account_url + demo_image_container_name
demo_image_container_client = ContainerClient(
    account_url=demo_image_account_url, 
    container_name=demo_image_container_name,
    credential=None)

# TNC test images (remote):
test_images_remote = []
test_images_root = 'https://animl-sample-images.s3-us-west-1.amazonaws.com/'
remote_image_files = [
    'sample-img-skunk-large.jpg',
    'p_001205.jpg',
    'p_001218.jpg',
]
for fil in remote_image_files:
    test_images_remote.append(test_images_root + fil)

# TNC test images (local):
test_images_local = []
test_images_dir = os.path.abspath(os.path.join(os.path.abspath(''), '..', 'input'))
local_image_files = [
    'sample-img-empty.jpg',
    'sample-img.jpg',
    'sample-img-skunk-large.jpg',
    'sample-img-rodent.jpg',
    'sample-img-fox.jpg',
    'sample-img-fox-2.jpg',
]
for fil in local_image_files:
    test_images_local.append(os.path.join(test_images_dir, fil))

# Megadetector API config
md_api_subscription_key = MEGADETECTOR_API_KEY
md_api_base_url = 'https://aiforearth.azure-api.net/api/v1/camera-trap/sync/'
md_api_detection_url = md_api_base_url + '/detect'
md_api_version_url = md_api_base_url + '/detector_model_version'
min_confidence_to_retrieve = 0.5
min_confidence_to_display = 0.8

# MIRA API config
mira_api_url = 'https://9q9u0ns6ul.execute-api.us-west-2.amazonaws.com/dev/classify'<jupyter_output><empty_output><jupyter_text>Megadetector API health/version check¶<jupyter_code>headers = { 'Ocp-Apim-Subscription-Key': md_api_subscription_key }
version_info = requests.get(md_api_version_url,headers=headers)
version_info.text<jupyter_output><empty_output><jupyter_text>Functions<jupyter_code>import urllib

def get_msft_demo_images():
    
    images_data = []
    generator = demo_image_container_client.list_blobs()

    for blob in generator:                  
        blob_client = BlobClient(demo_image_account_url,demo_image_container_name,blob.name)
        download_stream = blob_client.download_blob()
        images_data.append({'name' : blob.name, 'data': download_stream.readall()})
        # print('Read {} bytes'.format(len(images_data[-1]['data'])))
    
    return images_data   


def get_tnc_images_remote():

    images_data = []
    for url in test_images_remote:
        with urllib.request.urlopen(url) as img:
            images_data.append({'name': url, 'data': BytesIO(img.read()).read()})
            
    return images_data


def get_tnc_images_local():

    images_data = []
    for fil in test_images_local:
        with open(fil, "rb") as img:
            images_data.append({'name': fil, 'data': img.read()})
            
    return images_data


def draw_bounding_box_on_image(image,ymin,xmin,ymax,xmax):

    color = 'red'
    draw = ImageDraw.Draw(image)
    im_width, im_height = image.size
    (left, right, top, bottom) = (xmin * im_width, xmax * im_width,
                                  ymin * im_height, ymax * im_height)
    draw.line([(left, top), (left, bottom), (right, bottom),
               (right, top), (left, top)], width=4, fill=color)
    

    
def draw_raw_images():  
    
    fig = plt.figure(figsize=(20,35))

    num_images = len(images)
    
    columns = 2

    rows = (num_images // 2) + (num_images % 2)
    
    for i in range(len(images)):
        
        image = Image.open(BytesIO(images[i]['data']))        
        axis = plt.subplot(rows,columns, i + 1)     
        axis.imshow(image)
        plt.axis('off')
        plt.axis('tight')
            
    plt.show()


def response_to_json(r):
    
    from requests_toolbelt.multipart import decoder
    res = decoder.MultipartDecoder.from_response(r)
    results = {}
    images = {}

    for part in res.parts:
        
        # 'part' is a BodyPart object with b'Content-Type', and b'Content-Disposition';
        # the latter includes 'name' and 'filename' info

        headers = {}
        for k, v in part.headers.items():
            headers[k.decode(part.encoding)] = v.decode(part.encoding)

        if headers.get('Content-Type', None) == 'image/jpeg':
            c = headers.get('Content-Disposition')
            image_name = c.split('name="')[1].split('"')[0]
            image = Image.open(io.BytesIO(part.content))
            images[image_name] = image
        elif headers.get('Content-Type', None) == 'application/json':
            content_disposition = headers.get('Content-Disposition', '')
            if 'detection_result' in content_disposition:
                results['detection_result'] = json.loads(part.content.decode())
            elif 'classification_result' in content_disposition:
                results['classification_result'] = json.loads(part.content.decode())
                
    # ...for each part
    
    return results,images


def call_megadetector(image_info):
    
    image_bytes = image_info['data']
    image_name = image_info['name']
    
    assert isinstance(image_bytes,bytes)
    file = BytesIO(image_bytes)
    files = {}
    files[image_name] = (image_name, file, 'application/octet-stream')
    
    headers = { 'Ocp-Apim-Subscription-Key': md_api_subscription_key }
    params = {
        'confidence': min_confidence_to_retrieve,
        'render': False
    }

    r = requests.post(md_api_detection_url, params=params, headers=headers, files=files)
    
    return r


def call_mira(image_info):

    image_bytes = image_info.get('data')
    image_name = image_info.get('name')
    image_bbox = image_info.get('bbox')

    fields = {}
    fields['image'] = (image_name, image_bytes, 'image/jpeg')
    fields['bbox'] = json.dumps(image_bbox)
    
    multipart_data = MultipartEncoder(fields = fields)
    r = requests.post(mira_api_url,
                      data = multipart_data,
                      headers = {'Content-Type': multipart_data.content_type})
    
    return r<jupyter_output><empty_output><jupyter_text>Retrieve sample imagesUncomment the image fetching function you want & comment out the rest.<jupyter_code># images = get_msft_demo_images()
# images = get_tnc_images_local()
images = get_tnc_images_remote()

draw_raw_images()<jupyter_output><empty_output><jupyter_text>Select an image<jupyter_code>i_image = 2<jupyter_output><empty_output><jupyter_text>Run inference<jupyter_code># Call Megadetector 

# Each detection is: [ymin, xmin, ymax, xmax, confidence, class]
# Image coordinates are normalized, with the origin in the upper-left

r = call_megadetector(images[i_image])

# print('megadetector response code: {}'.format(r))
results,_ = response_to_json(r)
image_name = images[i_image]['name']
detections = results['detection_result'][image_name]
print('Megadetector found {} object(s) in {}'.format(len(detections), image_name))
print('{}\n'.format(detections))

# Add bounding box to image dict and call MIRA
images[i_image]['bbox'] = detections[0][:4]
r = call_mira(images[i_image])

# print('mira response code: {}'.format(r))
# print('response: {}'.format(r.json()))
for key, value in r.json().items():
    print('{} predictions: '.format(key))
    for classification, pred in value.get('predictions').items():
        print('{}: {}'.format(classification, pred))
    print('\n')<jupyter_output><empty_output><jupyter_text>Show results<jupyter_code>from PIL import Image
image = Image.open(BytesIO(images[i_image]['data']))

for detection in detections:
    box = detection[0:4]
    confidence = detection[4]
    clss = detection[5]
    if (confidence >= min_confidence_to_display):
        draw_bounding_box_on_image(image, box[0], box[1], box[2], box[3])
image<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12236046113457368, 'fraction_numerical': 0.006049537724004109, 'mean_word_length': 3.809001097694841, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 30, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '332984', 'license': 'MIT', 'n_tokens_mistral': 2975, 'n_tokens_neox': 2910, 'n_words': 554, 'path': 'notebooks/test-inference-pipeline.ipynb', 'repo_name': 'tnc-ca-geo/animl-ml', 'size': '8761'}