MagPieCam

Introduction

MagPieCam lets users choose what they want to be notified about using rules they specify in natural language. The camera uses an object tracker (ByteTrack) to recognize moving objects and then sends an images to the rules engine (Gemma4 / Gemini API) that decides if the image should trigger a push-notification. The camera itself is built using a Raspberry Pi Zero 2W and the Raspberry Pi AI Camera. On this page I will walk through the architecture of the system and some of the more interesting parts of building it.

Why I Built This

I wanted to build a system that is centered around an AI model, and does something useful. Not just a dataset and a model with performance metrics but a full system in the cloud that works and scales. I also have two cats at home named Zia and Luna! They are 3 years old and very curious, so I need to keep an eye on them sometimes. Luna for example, has some health problems, and so sometimes I need to track when she is eating or using the bathroom. So I tried setting up a Ring cam near their litter box or food bowls for example. But I get many notifications for events that are not what I am looking for. For example, every time a cat or person walks by, I get an event. Additionally my girlfriend and I live in an area that has a decent amount of foot traffic. Similarly we tend to get a lot of notifications that are just people walking by. Ring now has unusual event detection but the user doesn't determine what is unusual. So I built MagPieCam.

Key Features

• Users can add new cameras to their account seamlessly via a qr code
• View the live video
• Users can add rules using natural language for notifications
• Each notifications includes a 20 second video clip that includes the 10 seconds leading up to the event.

Demo

App Screenshots

Architecture Features

Architecture Overview

I think the most digestible way to look at the system is by looking at it feature by feature in the order that a user might encounter each feature.

1

Creating an Account

User account are pretty straightforward and implemented using Django's User model with the simple JWT plugin. When a user creates an account, the app receives a short lived access token and a refresh token that can be used to retrieve a new access token. All other endpoints besides the device provision ones require JWTs.

2

Adding a camera

Generating the Claim Token

First to create a camera in the system I create a QR code using a claim_token which is generated by the cameras/provision endpoint. The cameras/provision endpoint takes a device id (CPU serial number on the Raspberry Pi) and generates a claim_token by appending a uuid to the device id. The claim_token is then hashed using SHA-256, stored in a Camera model in Postgres along with the device id. Then the text token is returned to the caller.

Why the Claim Token Is Single-Use

This process basically simulates what would usually take place when the cameras are assembled. The claim_token can only be used once to pair a camera to a user's app. This design ensures that if the camera is stolen it can't be claimed. But before the user can claim the camera it must register with the backend at start up.

Registering the Camera

At startup, the camera calls the cameras/register endpoint passing it's device id and the claim_token. The backend core then queries the DB for the device id and compares the stored hashed claim_tokens to the hashed input claim_token. If a matching camera is found in the DB, a flag is set on the camera model and another uuid is generated and attached to the device id, creating the device_token. This token is saved to disk in the camera and is used to request JWTs going forward. This system enables the device token to be revoked in the case that the camera is lost or stolen.

Claiming the Camera

Once the camera has been registered the app can scan the QR code which calls cameras/claim and sets the claimed.

3

Detecting Moving Objects

The edge agent on the camera kind of operates outside of what the user does. It tracks moving objects and uploads detection images regardless of what is going on. The camera is built using a Raspberry Pi Zero 2W and the Raspberry Pi AI Camera. The AI camera is running a quantized YOLOv11n model and passes the bounding boxes to the ByteTrack tracker. Each track has a it's own exponential backoff, so that you don't get bursts of images of the same thing uploaded to S3. I used this approach because an object may enter the view of the camera but it may take a while for it to do something that triggers a notification. For example my cat can walk into view but not eat for 30 seconds. Each time an object is detected the camera requests a short-lived (currently good for 5 minutes) presigned url from /cameras/presigned_upload in order to upload the image to S3. Using a presigned url enable the camera to upload images without permanently storing AWS credentials. The image_detection S3 bucket then enqueues a message in SQS on upload events. (I am going to write an article that goes into more depth on the camera soon)

4

Creating & Evaluating Rules

Adding Rules

Once a camera has been added to a user's account they can add rules for smart notifications to the camera. In the iOS app a prefix is hardcoded to give the rules a predictable structure. As it is now, all rules need to be prefixed with 'Tell me when you see:' and then the user fills in the rule. This and other constraints like structured output make the model's output more predictable. The rules model itself in Django is pretty straightforward: it stores the rule suffix, has it's own id public_rule_id that is a uuid7 like the other public ids, and has an field enabled for toggling the rule.

Evaluating Detection Images

The event processor picks up messages from SQS which have the S3 object key needed to download the image. The event processor then creates a task for Celery workers, and thats pretty much it for the event processor. It just runs in a loop consuming from SQS. One important thing it does NOT do is mark the message as delivered. That is done by the celery worker so that if something goes wrong (Gemini API is down etc.) the message can be delivered again. The celery worker then downloads the image from S3, builds a prompt from the user's rules and then runs inference either using the rules model.

Prompting

The full prompt is composed of:

System Instructions

SYSTEM_INSTRUCTIONS = 'You evaluate whether a condition is present in a security camera image. Answer only from what is visibly present. If the image is too dark, blurry, or ambiguous to tell, answer 'unsure' rather than guessing.'

Currently I am using gemma4-e2b-it as the hosted model in dev (more on this below) and I noticed that it tends to trigger false alerts in darker images. So the goal of the system prompt is to make the model lean toward false negatives over false positives. This idea here is, even though the instructions say the camera is a security camera, the use cases for the camera broader than security camera. For example going forward the plan is to add analytics so you can see how many events triggered for a rule and how often. So the idea is to focus on highly valuable notifications.

Rules Preamble

RULES_PREAMBLE = 'For each condition below, answer whether it is visible in the image.'

Rules list

The rules are then combined with the prefix 'Is there' because as seen above the format of the rules in the database is:

* a person eating
* a blue car in the driveway
* a cat on the counter

Each rule is then listed with it's public_rule_id, so the full rules list would look like:

'01a01b0f-2cf3-7110-8f89-9c59f02d2c51:Is there a person eating. 01a01b0f-2cf3-7110-8f89-9c59f02d2c52:Is there a car in the driveway. 01a01b0f-2cf3-7110-8f89-9c59f02d2c53:Is there a cat on the counter.' Structuring the rules like this enables the model to evaluate all the rules in a single prompt or inference run. This reduces the number of api calls made to the Gemini API in production. It also means the system instructions, preamble are only included once as opposed to multiple times if calling the API for each rule, which means less tokens, and less money spent. When using the Gemini API the prompt end here because it can force the model to adhere to a json schema using structured outputs. So I created a pydantic model for the output and pass that to the Gemini API. When hosting the model myself I am currently using a json response prompt along with some post-processing like removing a json code fence (```json) for example:

JSON_RESPONSE_INSTRUCTIONS = 'Reply with JSON and nothing else. No prose, no explanation, no code fences. Use exactly this shape: {"verdicts": [{"id": "", "verdict": "yes"}]} Each verdict must be one of "yes", "no", or "unsure". Include exactly one entry for every rule id listed, copying each id character for character.'

But the plan is to use lm-format-enforcer to actually force the output to adhere to the pydantic model. Speaking of the pydantic model, it mirrors the input but has a verdict (yes/no/unsure) for each public_rule_id:

{ '01a01b0f-2cf3-7110-8f89-9c59f02d2c51':'yes' '01a01b0f-2cf3-7110-8f89-9c59f02d2c52':'no' '01a01b0f-2cf3-7110-8f89-9c59f02d2c53':'unsure' } See prompt.py for reference.

If a verdict is yes, a push notification is sent for the whole bundle along with a link to a preview image.

Hosting a model vs Using an API

Coming Soon

5

Notifications

coming soon

6

Streaming Live Video

coming soon

My Workflow with Claude

When I started this project, I wanted to solve a problem and learn about AI Engineering by building a backend system that used an AI model. However, I wanted to build the backend in the context of a fully working system. If I built just the backend, it would be very hard to conceptualize some of the interesting problems that I encountered while connecting the iOS app to the edge device, like the streaming lifecycle for example. So I wrote most the MagPieCam-Core services myself and delegated the a lot of the iOS app development and the edge agent development to Claude Code. When designing the system, I would come up with a design myself and have Claude play systems architect and review the design for anything I was missing or help me understand how engineering teams typically solve the problem.

Related Repositories