Workshop guard: cameras that warn first, then sound the siren

Frigate spots a person after hours, a horn tells them to leave, the siren follows if they stay, and two machines referee which one is on duty.

Difficulty
Advanced
Parts cost
A Pi Zero 2 W, a small amp and a horn speaker, plus a PC that's always on
Build time
Built in stages, April to September
Skills
Python, Linux services, Frigate, MQTT, Raspberry Pi

The problem

A small repair workshop I ran had cameras and an NVR (the box that records them), like most shops. The NVR is great for working out what happened the next morning. It does nothing to stop it happening. I wanted something that notices a person on the property after hours and does something about it, straight away: make a lot of noise, tell them to leave, and ring my phone.

I didn’t want to pay for a monitored alarm, and I already had most of the bits. This guide is how the “workshop guard” ended up, after six months of things going wrong in ways I didn’t expect. The most useful part is probably the lessons at the bottom, because most of what broke was the backup, not the alarm.

How it works

How the workshop guard works The cameras feed the NVR and Frigate. Frigate spots a person in the zone and tells the bridge over MQTT. The bridge asks for a spoken warning the first time, and the siren if the same person is still there 10 seconds later. The guard box re-checks that it is armed and the cameras are on, then plays the sound through the horn and has the phones rung. A standby machine runs a second Frigate and bridge, and a referee decides which bridge is on duty. Cameras and NVR the NVR records everything, all day Frigate on a mini PC, detection on the CPU "a person walked into the zone" The bridge new person → spoken warning same person 10 s later → siren they leave → stays quiet Guard box (Pi Zero 2 W) armed? cameras switched on? if not, nothing happens Horn speaker warning, then siren Phone call two phones ring Standby machine second Frigate and bridge, idle a referee picks which bridge is on duty Only one bridge runs at a time.

There are three parts, on purpose on different machines:

  • Frigate (free camera software that spots people, cars and so on) watches the cameras. It runs on a small PC at the workshop. When a person walks into a zone I’ve drawn on the picture, Frigate publishes a message over MQTT (a simple message bus that lots of home and camera software speaks).
  • The bridge is a short Python script on the same PC. It listens for those messages and decides what to do. The first time it sees a person in the zone, it asks for a spoken warning: “You are on private property. This area is monitored and recorded. Please leave now.” If the same person is still there 10 seconds later, it asks for the siren. If they leave, it stays quiet.
  • The guard box is a Raspberry Pi Zero 2 W with a USB sound card, an amplifier and a horn speaker. It plays the warning or the siren, and asks my home server to phone me and my business partner. The guard box is the final gate: it re-checks for every single request that the system is armed and that the cameras are switched on. The bridge can ask all day, and in daytime it does. The guard box just answers “suppressed” and nothing happens.

On top of that sits the part I’m proudest of: a referee. My home server runs a second copy of Frigate and the bridge, on standby. Every minute it checks whether the workshop PC is fit for duty. If the workshop PC goes sick, the home server takes over by itself, and hands back when the workshop PC is healthy again. It emails me every time duty changes hands. Only one bridge is ever running, so a person never gets warned twice.

Workshop alerts deliberately don’t go through Home Assistant. The house and the workshop are separate systems, and I didn’t want the workshop alarm to depend on a box at home.

What you need

The guard box

  • Raspberry Pi Zero 2 WH (the WH has the header pins already soldered), running Raspberry Pi OS Lite.
  • A USB audio adapter. The Pi Zero has no audio out of its own.
  • PAM8610 amplifier board, a 15 W class-D amp that runs off 12 V.
  • AS3180 horn speaker. Ordinary PC speakers are fine on the bench while you’re testing.
  • A 12 V supply for the amp.
  • Optional: a 3.2 inch 480×800 HDMI screen for a status display (armed or not, last trigger, network).
  • Optional: SR602 PIR motion sensors. I planned the box around these, then found the cameras did the job better (see the lessons). If you’ve got no cameras, they’re a cheap way in.

The watching side

  • IP cameras and an NVR that give out RTSP streams (RTSP is the standard way to pull a live video stream off a camera). Mine is a Uniview NVR. The camera that does the detecting is 6 MP (3072×2048).
  • An always-on PC for Frigate. Mine is a mini PC with a Ryzen 5 7640HS. No Coral, no graphics card doing the detecting. See the lessons for why.
  • Frigate 0.18 and Mosquitto (an MQTT broker), both in Docker.
  • Python 3 with paho-mqtt and requests for the bridge.

The phone call and the backup

  • Something that can place a phone call when a web address is hit. Mine is a small web app on my home server that uses Twilio (a pay-as-you-go phone API) to ring two phones. Any service that can make a call from a webhook will do.
  • Optional but worth it: a second always-on machine for the standby Frigate and bridge. Mine is my home server, a Ryzen 7 mini PC, which reaches the workshop camera over a private link between the two sites.

Wiring

The guard box is simple:

From To Notes
Pi Zero USB port (via an OTG adapter) USB audio adapter Set it as the default sound card in /etc/asound.conf
USB audio adapter line out PAM8610 input
PAM8610 output Horn speaker
12 V supply PAM8610 power
Pi Zero mini-HDMI Status screen Optional
GPIO17, GPIO27, GPIO22 PIR sensor outputs Optional. SR602s also need 3.3 V and ground

Turn the USB adapter all the way up in software, then set the loudness on the amp’s own gain knob:

amixer -c 1 sset Speaker 100% unmute
sudo alsactl store   # keeps the setting after a reboot

Card 1 is the USB adapter on mine. Check yours with aplay -l.

Everything else (cameras, NVR, PCs) talks over the network. Put anything that matters on ethernet, not Wi-Fi. Two of my worst outages were the Wi-Fi chip on a Raspberry Pi 5 getting itself stuck (see the lessons). And give every box a fixed address in your router. Mine drifted more than once.

Firmware and config

There’s no firmware. It’s a Frigate config, two short Python programs, and some service files.

Frigate

This is a tidied version of the workshop PC’s setup, not a straight copy. The RTSP password lives in an env file next to the compose file, not in the config:

# docker-compose.yml
services:
  mosquitto:
    image: eclipse-mosquitto:2
    restart: unless-stopped
    ports:
      - "127.0.0.1:1883:1883"   # only this PC can reach the broker
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro

  frigate:
    image: ghcr.io/blakeblackshear/frigate:0.18.0
    restart: unless-stopped
    depends_on: [mosquitto]
    env_file: frigate.env            # FRIGATE_RTSP_PASSWORD=...  (chmod 600)
    environment:
      LIBVA_DRIVER_NAME: radeonsi    # AMD graphics for video decoding
    devices:
      - /dev/dri/renderD128
    ports:
      - "5000:5000"
      - "8971:8971"
    volumes:
      - ./config:/config
# mosquitto.conf
listener 1883
allow_anonymous true

The broker is only published on 127.0.0.1, so nothing off the PC can reach it.

# config/config.yml
mqtt:
  host: mosquitto
  port: 1883

# Two OpenVINO detectors, both on the CPU.
detectors:
  ov_0:
    type: openvino
    device: CPU
  ov_1:
    type: openvino
    device: CPU

# The model block goes at the TOP LEVEL, not under the detector. See the lessons.
model:
  model_type: yolo-generic
  width: 640
  height: 640
  input_tensor: nchw
  input_dtype: float
  path: /config/model_cache/yolov9-s-640.onnx
  labelmap_path: /labelmap/coco-80.txt

ffmpeg:
  hwaccel_args: preset-vaapi

record:
  enabled: false     # the NVR records everything. Frigate's only job is spotting people.

snapshots:
  enabled: true

cameras:
  camera_1:          # the one camera that detects
    ffmpeg:
      inputs:
        - path: rtsp://admin:{FRIGATE_RTSP_PASSWORD}@NVR_IP:554/MAIN_STREAM_PATH
          roles: [detect]
    detect:
      width: 3072    # the camera's real resolution
      height: 2048
      fps: 5
    objects:
      track: [person]
    # Draw the zone in Frigate's own editor and it writes this block for you,
    # with the coordinates filled in. The name must match ZONES_JSON in the bridge.
    # zones:
    #   watch_zone:
    #     coordinates: (written by Frigate)
    #     objects: [person]

  camera_2:          # live view only
    ffmpeg:
      inputs:
        - path: rtsp://admin:{FRIGATE_RTSP_PASSWORD}@NVR_IP:554/SUB_STREAM_PATH
          roles: [detect]
    detect:
      enabled: false

Only one camera detects. The others are in Frigate so I can look at them from one place, with detection off. Draw the zone on the part of the picture where nobody has any business being after hours, not the footpath out the front.

The detection model

Frigate’s OpenVINO detector comes with a small model, which is fine to start with. I moved to YOLOv9-s at 640×640, which I exported myself. This builds the file in Docker and drops yolov9-s-640.onnx in the current folder. Copy it into config/model_cache/:

docker build . --build-arg MODEL_SIZE=s --build-arg IMG_SIZE=640 --output . -f- <<'EOF'
FROM python:3.11 AS build
RUN apt-get update && apt-get install --no-install-recommends -y cmake libgl1 && rm -rf /var/lib/apt/lists/*
COPY --from=ghcr.io/astral-sh/uv:0.10.4 /uv /bin/
WORKDIR /yolov9
ADD https://github.com/WongKinYiu/yolov9.git .
RUN uv pip install --system -r requirements.txt
RUN uv pip install --system onnx==1.18.0 onnxruntime onnx-simplifier==0.4.* onnxscript
ARG MODEL_SIZE
ARG IMG_SIZE
ADD https://github.com/WongKinYiu/yolov9/releases/download/v0.1/yolov9-${MODEL_SIZE}-converted.pt yolov9-${MODEL_SIZE}.pt
RUN sed -i "s/ckpt = torch.load(attempt_download(w), map_location='cpu')/ckpt = torch.load(attempt_download(w), map_location='cpu', weights_only=False)/g" models/experimental.py
RUN python3 export.py --weights ./yolov9-${MODEL_SIZE}.pt --imgsz ${IMG_SIZE} --simplify --include onnx
FROM scratch
ARG MODEL_SIZE
ARG IMG_SIZE
COPY --from=build /yolov9/yolov9-${MODEL_SIZE}.onnx /yolov9-${MODEL_SIZE}-${IMG_SIZE}.onnx
EOF

Be ready for a big download. The build pulls in a lot of machine-learning libraries, about 11 GB of build cache on mine. Clear it out afterwards with docker builder prune.

Why 640 and not the smaller 320? A wide 6 MP view of a yard makes people small. I tested both on 59 snapshots that had a person in them: 640 found the person in 54 of them, 320 in 44. Neither made up a person on empty-yard frames. 640 is slower (about 24 ms per look against about 7 ms), which is why there are two detectors sharing the work. On a close-up camera, 320 would probably be enough.

The bridge

Two files. The decision logic is kept separate from the MQTT and web plumbing so it can be tested without a camera. This is the code from my workshop PC, lightly tidied.

# bridge_logic.py: pure decision logic, no network, easy to test.

class EscalationState:
    """Follows each person Frigate is tracking (by its tracking id) through
    warn -> still here? -> siren."""
    def __init__(self, escalate_after_s=10, warn_cooldown_s=60):
        self.escalate_after_s = escalate_after_s
        self.warn_cooldown_s = warn_cooldown_s
        self.people = {}      # tracking id -> {"camera", "warned_at", "escalated", "last_seen"}
        self.last_warn = {}   # camera -> when it last warned


def _in_zone(after, zones):
    camera = after.get("camera")
    if camera not in zones:
        return (False, camera)
    wanted = zones[camera]
    entered = set(after.get("entered_zones") or [])
    return ((not wanted) or bool(entered & set(wanted)), camera)


def decide(ev, zones, state, now):
    """Return (action, camera). action is "warn", "siren" or None.
    warn  = first time we see this person in a zone (and the camera hasn't just warned)
    siren = the same person is still there escalate_after_s later
    None  = nothing to do (duplicate, too soon, they left, not a person...)"""
    after = ev.get("after") or {}
    kind = ev.get("type")
    oid = after.get("id")
    if oid is None:
        return (None, None)

    if kind == "end":                       # Frigate lost them: forget them
        state.people.pop(oid, None)
        return (None, None)
    if kind not in ("new", "update"):
        return (None, None)
    if after.get("label") != "person":
        return (None, None)
    in_zone, camera = _in_zone(after, zones)
    if not in_zone:
        return (None, None)

    rec = state.people.get(oid)
    if rec is None:
        # A new person in the zone: warn, unless this camera warned very recently.
        if (now - state.last_warn.get(camera, 0)) < state.warn_cooldown_s:
            return (None, None)
        state.people[oid] = {"camera": camera, "warned_at": now,
                             "escalated": False, "last_seen": now}
        state.last_warn[camera] = now
        return ("warn", camera)

    rec["last_seen"] = now
    if not rec["escalated"] and (now - rec["warned_at"]) >= state.escalate_after_s:
        rec["escalated"] = True
        return ("siren", rec["camera"])
    return (None, None)


def prune(state, now, ttl_s):
    """Drop people not seen for ttl_s, in case Frigate never sent an "end"."""
    stale = [oid for oid, r in state.people.items() if (now - r["last_seen"]) > ttl_s]
    for oid in stale:
        state.people.pop(oid, None)
    return len(stale)

It works on Frigate’s tracking id, not just “a person is there”. That’s what makes “still there 10 seconds later” mean the same person, and why a person standing still in the zone escalates without having to walk out and back in.

#!/usr/bin/env python3
# frigate_siren_bridge.py: Frigate MQTT events -> the guard box, plus a heartbeat.
import json, logging, os, threading, time
import requests
import paho.mqtt.client as mqtt
from bridge_logic import decide, prune, EscalationState

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

GUARD_URL = os.environ["GUARD_URL"]          # the guard box's API
TOKEN = os.environ["TRIGGER_TOKEN"]          # shared secret, also set on the guard box
ZONES = json.loads(os.environ.get("ZONES_JSON", "{}"))
MQTT_HOST = os.environ.get("MQTT_HOST", "localhost")
HEARTBEAT_S = int(os.environ.get("HEARTBEAT_S", "30"))
ESCALATE_AFTER_S = int(os.environ.get("ESCALATE_AFTER_S", "10"))
WARN_COOLDOWN_S = int(os.environ.get("WARN_COOLDOWN_S", "60"))
ID_TTL_S = int(os.environ.get("ID_TTL_S", "120"))

state = EscalationState(escalate_after_s=ESCALATE_AFTER_S, warn_cooldown_s=WARN_COOLDOWN_S)


def post(path, camera, message):
    try:
        r = requests.post(GUARD_URL + path, timeout=8, headers={"X-Token": TOKEN},
                          json={"sensor": "camera_" + camera, "message": message})
        log.info("%s %s -> %s %s", path, camera, r.status_code, r.text[:120])
    except Exception as e:
        log.warning("%s POST failed: %s", path, e)   # logged, never fatal


def heartbeat_loop():
    """Tells the guard box the camera path is alive, so it can report when it isn't."""
    while True:
        try:
            requests.post(GUARD_URL + "/heartbeat", timeout=5,
                          headers={"X-Token": TOKEN}, json={"source": "frigate-bridge"})
        except Exception as e:
            log.debug("heartbeat failed: %s", e)
        time.sleep(HEARTBEAT_S)


def on_connect(c, userdata, flags, rc, *args):
    log.info("mqtt connected rc=%s", rc)   # grep for this line. See the lessons.
    c.subscribe("frigate/events")


def on_message(c, userdata, msg):
    try:
        ev = json.loads(msg.payload.decode())
    except Exception:
        return
    now = time.time()
    action, camera = decide(ev, ZONES, state, now=now)
    if action == "warn":
        log.info("WARN for %s", camera)
        post("/warn", camera, "Camera person detected (warning): " + camera)
    elif action == "siren":
        log.info("SIREN (still there) for %s", camera)
        post("/trigger", camera, "Camera person detected: " + camera)
    prune(state, now, ID_TTL_S)


def main():
    log.info("bridge starting; guard=%s zones=%s", GUARD_URL, ZONES)
    threading.Thread(target=heartbeat_loop, daemon=True).start()
    c = mqtt.Client()      # written for paho-mqtt 1.6
    c.on_connect = on_connect
    c.on_message = on_message
    while True:
        try:
            c.connect(MQTT_HOST, 1883, 60)
            c.loop_forever()
        except Exception as e:
            log.warning("mqtt loop error: %s; retry in 5s", e)
            time.sleep(5)


if __name__ == "__main__":
    main()

The bridge’s settings file. Make it readable only by you (chmod 600):

# bridge.env
GUARD_URL=http://GUARD_BOX_IP:8090
TRIGGER_TOKEN=a-long-random-string
MQTT_HOST=127.0.0.1
ZONES_JSON={"camera_1": ["watch_zone"]}
ESCALATE_AFTER_S=10
WARN_COOLDOWN_S=60
ID_TTL_S=120
HEARTBEAT_S=30

A camera only goes live when its zone is drawn in Frigate and named in ZONES_JSON. No matching zone, no triggers. That made it safe to add cameras to Frigate without them setting anything off.

# /etc/systemd/system/frigate-siren-bridge.service
[Unit]
Description=Frigate -> workshop guard bridge
After=network-online.target docker.service
Wants=network-online.target

[Service]
User=YOUR_USER
WorkingDirectory=/home/YOUR_USER/frigate-siren/bridge
EnvironmentFile=/home/YOUR_USER/frigate-siren/bridge/bridge.env
ExecStart=/home/YOUR_USER/frigate-siren/bridge/venv/bin/python -u frigate_siren_bridge.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

If you’re running a referee, don’t enable this service at boot. The referee starts and stops it, so that only one bridge is ever on duty.

The guard box

My guard box does more than this (PIR inputs, push-to-talk from my phone out the horn, the status screen). This is the camera side, cut down to the parts you need. The alarm window comes from the settings file, so pick your own hours.

#!/usr/bin/env python3
"""guard_api.py: the camera side of the guard box, cut down.

The bridge POSTs /warn (say the warning once) and /trigger (sound the siren).
This box is the final gate: every request is re-checked against "armed" and
"cameras on", whatever the bridge thinks.
"""
import json, os, subprocess, threading, time, urllib.request
from datetime import datetime, time as dtime
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler

TOKEN       = os.environ["TRIGGER_TOKEN"]                   # same secret as the bridge
CALL_URL    = os.environ["CALL_WEBHOOK_URL"]                # whatever places your phone calls
START       = dtime.fromisoformat(os.environ["ALARM_START"])  # "HH:MM"
END         = dtime.fromisoformat(os.environ["ALARM_END"])
DISARM_S    = int(os.environ.get("DISARM_S", "900"))
SIREN_WAV   = os.environ.get("SIREN_WAV", "warning.wav")        # siren + "leave now"
VOICE_WAV   = os.environ.get("VOICE_WAV", "warning_voice.wav")  # the first, calmer warning
SIREN_MAX_S = 60
CAM_COOLDOWN_S = 60
HEARTBEAT_TTL_S = 90

lock = threading.Lock()
st = {"force_armed": False, "disarm_until": 0.0, "cameras_on": True,
      "last_fire": 0.0, "last_beat": 0.0}


def in_window():
    t = datetime.now().time()
    return (t >= START or t < END) if START > END else (START <= t < END)


def armed():
    with lock:
        if st["force_armed"]:
            return True
        if time.time() < st["disarm_until"]:
            return False
    return in_window()


def call_phones(sensor, message):
    body = json.dumps({"sensor": sensor, "message": message,
                       "timestamp": datetime.now().isoformat()}).encode()
    req = urllib.request.Request(CALL_URL, data=body, method="POST",
                                 headers={"Content-Type": "application/json"})
    try:
        urllib.request.urlopen(req, timeout=10).read()
    except Exception as e:
        print("call webhook failed:", e, flush=True)


def play(wav, max_s, loop=False):
    deadline = time.time() + max_s
    while time.time() < deadline:
        p = subprocess.Popen(["aplay", wav], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        while p.poll() is None:
            if time.time() >= deadline:
                p.terminate(); p.wait(timeout=2); return
            time.sleep(0.5)
        if not loop:
            return


def bg(fn, *args):
    threading.Thread(target=fn, args=args, daemon=True).start()


def handle(kind, sensor, message):
    if not armed():
        return "suppressed: disarmed/off-hours"
    if not st["cameras_on"]:
        return "suppressed: cameras disabled"
    if kind == "warn":
        bg(call_phones, sensor, message)
        bg(play, VOICE_WAV, SIREN_MAX_S)
        return "warned"
    with lock:
        if time.time() - st["last_fire"] < CAM_COOLDOWN_S:
            return "suppressed: cooldown"
        st["last_fire"] = time.time()
    bg(call_phones, sensor, message)
    bg(play, SIREN_WAV, SIREN_MAX_S, True)
    return "fired"


def status():
    return {"armed": armed(), "in_window": in_window(), "force_armed": st["force_armed"],
            "cameras_on": st["cameras_on"],
            "camera_path_healthy": time.time() - st["last_beat"] < HEARTBEAT_TTL_S}


class Api(BaseHTTPRequestHandler):
    def _reply(self, code, data):
        out = json.dumps(data).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(out)

    def do_GET(self):
        if self.path == "/status":
            return self._reply(200, status())
        self._reply(404, {"error": "not found"})

    def do_POST(self):
        if self.headers.get("X-Token") != TOKEN:
            return self._reply(403, {"ok": False})
        n = int(self.headers.get("Content-Length") or 0)
        body = json.loads(self.rfile.read(n) or b"{}")
        sensor, message = body.get("sensor", "camera"), body.get("message", "")
        if self.path == "/warn":
            return self._reply(200, {"ok": True, "result": handle("warn", sensor, message)})
        if self.path == "/trigger":
            return self._reply(200, {"ok": True, "result": handle("siren", sensor, message)})
        if self.path == "/heartbeat":
            st["last_beat"] = time.time()
            return self._reply(200, {"ok": True})
        if self.path in ("/cameras/enable", "/cameras/disable"):
            st["cameras_on"] = self.path.endswith("enable")
            return self._reply(200, status())
        if self.path == "/disarm":
            with lock:
                st["force_armed"] = False
                st["disarm_until"] = time.time() + DISARM_S
            return self._reply(200, status())
        if self.path == "/arm":
            with lock:
                st["disarm_until"] = 0.0
                st["force_armed"] = not in_window()   # arming out of hours stays on until disarmed
            return self._reply(200, status())
        self._reply(404, {"error": "not found"})

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    # Threading server: one slow request must never block /status or /disarm. See the lessons.
    ThreadingHTTPServer(("0.0.0.0", 8090), Api).serve_forever()

Two things in there matter more than they look:

  • The guard box checks everything again. The bridge on the workshop PC runs around the clock, so in daytime it happily sends /warn whenever someone walks past. The guard box answers suppressed: disarmed/off-hours and nothing happens. That one check is what makes it safe to leave the bridge running all day.
  • Disarming is a short pause, not “off”. Disarm during alarm hours and it re-arms itself after DISARM_S. I disarm and re-arm by text message from my phone: the server that does my phone calls also takes a few text commands and passes them on to /disarm, /arm and /cameras/disable. The last one is a kill switch for just the camera side.

The warning sounds

The siren file is a siren with a recorded voice over it (“Intruder detected. Leave the premises now. Security has been notified.”), three times through. The calmer first warning is made with Piper (a free, local text-to-speech program), then converted to a plain WAV the Pi plays cleanly:

echo "You are on private property. This area is monitored and recorded. Please leave now." \
  | piper --model en_US-amy-medium.onnx --output_file warn.wav
ffmpeg -y -i warn.wav -ar 22050 -ac 1 -sample_fmt s16 warning_voice.wav

To change the wording, run it again and copy the new file to the guard box. Nothing is generated on the live path.

The referee

The referee is a shell script running as a service on my standby machine (the home server). I haven’t tidied it up for sharing, so here are its rules instead. They’re the part worth copying:

  1. Every 60 seconds, ask the primary two questions. Is Frigate’s web API answering? And has its container been running for more than 12 minutes? The 12 minutes is longer than a Frigate that’s crash-looping ever stays up, so a Frigate that keeps falling over never looks healthy just because it happened to be up when asked.
  2. Five answers in a row the same way flips duty. One bad check doesn’t. Flipping means stopping one bridge and starting the other. Only one is ever running.
  3. Email on every flip, both directions.
  4. The standby’s own camera feed is off unless it has duty. Frigate has a switch for this over MQTT: publish ON or OFF to frigate/<camera>/enabled/set. The referee switches the standby camera on before starting the standby bridge, and re-sends the right state every loop, because a Frigate restart puts it back to whatever the config file says. Keep the camera enabled: true in the config, or Frigate refuses the ON.
  5. If the referee itself dies, a small service on the primary covers it: if the standby machine hasn’t answered a ping for 10 minutes and the local Frigate is healthy, the primary starts its own bridge. When the standby comes back, the referee sorts out who has duty.
  6. Pause the referee before a planned Frigate restart on the primary. Even Frigate’s own restart button restarts the container, which resets the 12-minute clock, and you’ll get a false takeover and two emails.

Step by step

  1. Build the guard box: Pi OS Lite, USB audio as the default card, amp and horn. Play a WAV with aplay to check the sound path. Do this on the bench with PC speakers, not the horn.
  2. Make the two WAV files.
  3. Put guard_api.py on the guard box with a settings file (TRIGGER_TOKEN, CALL_WEBHOOK_URL, ALARM_START, ALARM_END) and run it as a service, the same way as the bridge.
  4. Set up Frigate and Mosquitto on the always-on PC. Get the camera detecting people in the Frigate web page before going any further.
  5. Draw the zone in Frigate and give it the same name you’ll use in ZONES_JSON.
  6. Install the bridge in a Python venv (pip install paho-mqtt==1.6.1 requests), fill in bridge.env, and start it. Its log should say mqtt connected rc=0 within a few seconds.
  7. Test it while the guard box is disarmed (below).
  8. Only once that’s clean, test it for real on site.
  9. If you have a second machine, set up the standby Frigate and bridge, then the referee. Drill both directions on purpose before trusting it.

Testing it

  • Unit-test the decisions. Because decide() has no network in it, you can test leave-versus-stay without anyone walking anywhere. Mine has 14 tests. The two that matter most:

    # test_bridge_logic.py: run with pytest
from bridge_logic import decide, EscalationState

ZONES = {"camera_1": ["watch_zone"]}
T0 = 1_000_000.0   # a realistic clock; time.time() is never near 0

def ev(kind, oid="p1"):
    return {"type": kind, "after": {"id": oid, "label": "person", "camera": "camera_1",
                                    "entered_zones": ["watch_zone"]}}

def test_person_who_stays_gets_the_siren():
    s = EscalationState(escalate_after_s=10)
    assert decide(ev("new"), ZONES, s, now=T0) == ("warn", "camera_1")
    assert decide(ev("update"), ZONES, s, now=T0 + 5) == (None, None)
    assert decide(ev("update"), ZONES, s, now=T0 + 11) == ("siren", "camera_1")

def test_person_who_leaves_only_gets_the_warning():
    s = EscalationState(escalate_after_s=10)
    assert decide(ev("new"), ZONES, s, now=T0) == ("warn", "camera_1")
    assert decide(ev("end"), ZONES, s, now=T0 + 6) == (None, None)
    assert decide(ev("update"), ZONES, s, now=T0 + 12)[0] != "siren"

  • Push fake Frigate events through the real chain while disarmed. This is how I proved the move to the workshop PC without making a sound. Publish a made-up person, then an update 11 seconds later, then an end:

    P='{"type":"%s","after":{"id":"test-1","label":"person","camera":"camera_1","entered_zones":["watch_zone"]}}'
mosquitto_pub -h 127.0.0.1 -t frigate/events -m "$(printf "$P" new)"
sleep 11
mosquitto_pub -h 127.0.0.1 -t frigate/events -m "$(printf "$P" update)"
mosquitto_pub -h 127.0.0.1 -t frigate/events -m "$(printf "$P" end)"

    The bridge log should show WARN and then SIREN, and the guard box should answer both with suppressed: disarmed/off-hours. The whole chain ran, and nothing played and nobody got rung. Check /status says disarmed before you do this. If it’s armed, this will ring real phones.

  • Then walk it, once, on site. My first camera version passed a real walk test end to end. For the two-stage version, test both cases: walk in and leave inside 10 seconds (warning only, no siren), then wait out the 60-second cooldown and walk in and stay (warning, then the siren).

  • Watch /status from time to time. camera_path_healthy goes false when the bridge’s heartbeat stops, which is the quickest way to see that the camera side is down.

  • Drill the referee. I forced it both ways on purpose (a pretend-healthy primary for a handback, a real failed check for a takeover) and confirmed both emails arrived. Its first real, fully automatic handback came the night a dead detector came back to life.

Lessons learnt

The PIRs never got wired, and that was the right call. I planned the guard box around PIR motion sensors and left GPIO pins for three of them. By the time I got to wiring them, the cameras were already doing the same job better: they know a person from a cat, a moving shadow or the heater ticking, and they keep a clip of what they saw. A PIR only knows “something warm moved”. They became redundant before they were ever connected. Build the thing that tells you the most, and drop what it makes unnecessary.

The backup was dead for weeks, and every check said it was fine. The standby bridge on my home server was set to use an MQTT broker that lived on my Home Assistant box. When Home Assistant moved to a new machine, that old address became a different device, with no broker on it. The service was “running” and the referee was handing it duty for a few minutes every evening, and every time it logged mqtt loop error: Connection refused until it handed back. There wasn’t a single mqtt connected line in the whole log. If the primary had died overnight, there would have been no siren. A month earlier I’d found four monitors on my home server that were checking something next to what they were meant to check. My rule now: a monitor isn’t trusted until it’s been run against a real failure and a known-good period. For a bridge, that means “it connected”, not “the service is up”.

The referee caused a false takeover every single night. Its watch window opened at the same moment the primary’s Frigate was started for the night, so the primary hadn’t passed the 12-minute age check yet. So every evening the standby took over, emailed me, then handed back and emailed me again. The fix was opening the referee’s window a little later, once the primary was old enough. Make sure your health checks and your schedules agree with each other.

The Coral got stuck after power cuts. The first version ran Frigate on a Raspberry Pi 5 with a Coral USB accelerator (a plug-in chip that runs the detection model). After power events, the Coral sometimes came up stuck in its boot mode. Frigate restart-looped on “No EdgeTPU detected”, 55 times in one go. A software unplug and replug didn’t fix it, and neither did resetting the USB controller. A full reboot eventually brought it back. It had hiccuped at power events twice before, too. That outage is why the standby and the referee exist at all.

I blamed the CPU, and it was the Coral. The same Pi 5 also played the workshop radio to a Bluetooth speaker, and the radio was choppy. Turning Frigate’s priority down made it “a lot better”, so I put Frigate on a timer to only run in alarm hours. Then I found the radio was still choppy with Frigate stopped. A Bluetooth trace showed 31 stalls over 100 ms in 20 seconds. The cause was the Coral on a USB 3 port right next to the Pi’s Bluetooth antenna: USB 3 throws off radio noise. Letting the Coral’s USB link go to sleep took the stalls from 31 to 0. If you need a Coral near a radio, use a USB 2 port.

Then I didn’t need the Coral at all. When I moved Frigate to a Ryzen mini PC, I expected to move the Coral too. OpenVINO (Intel’s free AI toolkit, which also runs on AMD processors) on the plain CPU was faster than the Coral had been on the Pi: about 7.7 ms per look with YOLOv9-s at 320. A Coral would have been a downgrade: it only runs its own model format, so it locks you out of the better YOLOv9 models. And the AMD graphics chip can’t help with detection here, because OpenVINO’s graphics support is Intel-only. It still decodes the video, which is what preset-vaapi is doing.

The OpenVINO config passed and then crashed. I put the model: block under the detector, where it looks like it belongs. Frigate loaded the config without complaint, then crashed at runtime with a TypeError about a path being None. The model: block has to sit at the top level of the config. That cost me one failed change and a rollback.

Talking out the horn set off the siren. I added push-to-talk, so I can speak out the horn from my phone, and the siren pauses while I talk. When I let go, it resumed the siren if the system was armed. But “armed” isn’t the same as “the siren was going”. Talk to someone in alarm hours and it started a siren that had never been sounding. Now it records whether a siren was actually playing when I started talking, and only resumes if one was. Separately, my first push-to-talk version used Python’s single-threaded web server, and a half-open talk session froze the whole API, disarm included. Use ThreadingHTTPServer and put timeouts on reads.

A false warning that nobody could check. One night it warned and rang us for what turned out to be a truck’s headlights, and none of us could see from the call what had set it off. Now the bridge grabs a snapshot from Frigate on every warning and siren and emails it, in a background thread so it never slows the alarm down. My first version of that emailed snapshots all day, because the bridge reacts around the clock. It now asks the guard box whether it’s armed first. If the guard box doesn’t answer, it treats that as armed and sends the photo anyway.

Wi-Fi took the Pi 5 down twice. Once, an old Wi-Fi hotspot I’d set up on it and forgotten about was sharing the one Wi-Fi chip with the normal connection, and the chip’s firmware locked up and froze the whole Pi. The second time, my router’s band steering (nudging devices between 2.4 and 5 GHz) left the Wi-Fi “connected” but passing no traffic for most of a day. Ethernet for anything that matters, and a hardware watchdog so a frozen Pi reboots itself.

The NVR’s smart codec broke the live view. The NVR’s “smart” compression stretched the gap between full frames to 6 seconds, so the live view in a browser came up green and smeared. Turning it off fixed it. And Frigate’s built-in restreamer could only get the audio from this NVR, never the video, so I had to pull those streams through its ffmpeg: source instead. Detection kept reading straight from the NVR the whole time.