Motion Sensing with Raspberry Pi Camera and Cat Face Scanning with AWS Lambda + OpenCV


I wrote a Python script that detects camera motion with a Raspberry Pi camera and uploads photos to S3. The frames are then searched for cat faces using OpenCV running on AWS for alerting of heightened 'cattivity' in the house.

OpenCV Haar\-Cascade cat face detection

This could also be done using Amazon’s very cool new image ‘Rekognition’ service but would cost more if you are processing thousands of frames per day.

I wanted to know how often my flatmates cat was attempting to attack his mouse enclosure. Cat detection events are sent to DataDog using the statsd protocol for alerting!

Raspberry Pi Agent Code

I am monitoring the occurrence of motion events using DataDog and uploading the frames to S3 for further detection of objects in the frame such as cats or people.

Full source code is on GitHub: timatooth/catscanface

Haar-Cascade Cat Detection with AWS Lambda

You will need to package compiled Numpy & OpenCV libraries prepared for running inside the AWS Lambda functions thanks to aeddi

Update: 2019 I Believen numpy & OpenCV may work out-of-box using AWS SAM to package Lambda functions.

from __future__ import print_function import logging logger = logging.getLogger() logger.setLevel(logging.INFO) import os import os.path import numpy import cv2 import boto3 # Config s3_prefix = 'catscanface/' s3_bucket = 'timatooth' cat_cascade = cv2.CascadeClassifier('haarcascade_frontalcatface_extended.xml') s3_client = boto3.client('s3') def scan_frame(image_data): img_array = numpy.asarray(bytearray(image_data), dtype=numpy.uint8) frame = cv2.imdecode(img_array, 0) cat_faces = cat_cascade.detectMultiScale(frame, 1.3, 5) # Draw rectangle onto region where cat face is found cat_detected = False for (x, y, w, h) in cat_faces: cat_detected = True cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 255, 0), 2)"Found cat") if cat_detected: _, buf = cv2.imencode(".jpg", frame) return buf.tobytes() def get_image(object_key): response = s3_client.get_object(Bucket=s3_bucket, Key=object_key) image_bytes = response['Body'].read()'Got {} bytes'.format(len(image_bytes))) return image_bytes def lambda_handler(event, context): object_key = event['Records'][0]['s3']['object']['key'] image_bytes = get_image(object_key) cat_image = scan_frame(image_bytes) if cat_image is not None: key = s3_prefix + 'detections/' + os.path.basename(object_key)'Saving cat detection image: {}'.format(key)) s3_client.put_object( Bucket=s3_bucket, Key=key, Body=cat_image )


--- AWSTemplateFormatVersion: '2010-09-09' Description: Cat facial recognition OpenCV lambda Parameters: S3BucketName: Type: String Description: The S3 bucket which contains cat pictures S3BucketPrefix: Type: String Description: A bucket prefix where the cat pictures are found Default: /catscanface Resources: LambdaRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Principal: Service: - Action: - sts:AssumeRole Path: "/" Policies: - PolicyName: CloudwatchLog PolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Action: - logs:CreateLogGroup - logs:CreateLogStream - logs:PutLogEvents Resource: arn:aws:logs:*:*:* - PolicyName: AccessCatPictureBucket PolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Action: - s3:GetObject Resource: !Join ['', ['arn:aws:s3:::', Ref: S3BucketName, Ref: S3BucketPrefix, '/motion/*']] - PolicyName: WriteDetectedCatFrames PolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Action: - s3:PutObject Resource: !Join ['', ['arn:aws:s3:::', Ref: S3BucketName, Ref: S3BucketPrefix, '/detections/*']] Lambda: Type: AWS::Lambda::Function Properties: Code: ZipFile: 'print("hello world")' Description: Scan images uploaded to S3 for cats FunctionName: Ref: AWS::StackName Handler: catscan.lambda_handler MemorySize: 256 Role: !GetAtt LambdaRole.Arn Runtime: python2.7 Timeout: '4' Outputs: LambdaRole: Description: IAM Role for LambdaRole Value: Ref: LambdaRole Lambda: Value: !GetAtt Lambda.Arn