Detect Anomalies with Amazon Lookout for Vision and publish to Amazon Kinesis Data Firehose

raspberry-pi-with-amazon-lookout-for-vision.py

from time import sleep, time
from picamera import PiCamera
from datetime import datetime, timedelta
from boto3 import client
import json
import os

firehose = client('firehose')
lookout_vision = client('lookoutvision')


def detect_anomaly(img_file_path):
    """
    Function will detect anomaly with Amazon Lookout for vision
    and publish inference details to Amazon Kinesis Data Firehose.
    """
    inference_response = lookout_vision.detect_anomalies(
        ProjectName='my-cassava-project-name',
        ModelVersion='4',
        Body=open(img_file_path, 'rb'),
        ContentType='image/jpeg'
    )
    inference_metadata = {
        'InferenceTime': int(time()*1000.0),
        'InferenceImageSize': os.stat(img_file_path).st_size,
        # You may use PIL to get the image dimensions, this may be good for analytics later
    }
    
    payload = json.dumps({**inference_response, **inference_metadata})
    
    firehose.put_record(
        DeliveryStreamName='<my-delivery-stream-name>',
        Record={
            'Data': payload
        }
    )
    # You may delete the file once inference is completed to converse space 
    # on your Raspberry Pi microSD

def wait():
    # Calculate the delay to the start of the work day
    start_of_work_day = datetime.today().replace(hour=8, minute=0, second=0, microsecond=0)
    daily_work_hours = 9
    end_of_work_day = (start_of_work_day + timedelta(hours=daily_work_hours)).replace(minute=0, second=0, microsecond=0)
    delay = (end_of_work_day - start_of_work_day).seconds
    sleep(delay)

camera = PiCamera()
camera.start_preview()
sleep(5)
for filename in camera.capture_continuous('img{timestamp:%Y-%m-%d-%H-%M}.jpg'):
    print('Captured %s' % filename)
    detect_anomaly(filename)
    wait()