RPi Camera Ensemble

Central orchestration of synchronized video acquisition across multiple Raspberry Pi cameras.

Runs a Conductor on the acquisition machine that controls one or more Agents: each Agent is a FastAPI server running on an individual Raspberry Pi, managing local camera acquisition and GPIO-based TTL synchronization.

Features

• Multi-camera orchestration: start/stop all cameras with a single call; UUID-guarded sessions prevent accidental stops
• TTL synchronization: per-frame GPIO pulses (emitter) and edge-detection timestamps (receiver) via lgpio (RPi 4 + RPi 5) or legacy pigpio
• PiCamera2 backend: uses the modern libcamera stack; preview and recording states
• REST API on each agent: GET /status, POST /agent/acquisition/init, POST /agent/acquisition/record, POST /agent/acquisition/stop
• Heartbeat monitoring: background health checks with configurable interval
• Disk space check: pre-acquisition guard based on estimated bitrate and free space
• CLI: rce-agent run/config entrypoints for systemd or docker-compose deployment

Architecture

Conductor (acquisition machine)
    │  HTTP via aiohttp
    ├── AgentClient → Agent (RPi cam01)  →  Acquisition → PiCamera2
    ├── AgentClient → Agent (RPi cam02)  →  Acquisition → PiCamera2
    └── AgentClient → Agent (RPi cam03)  →  Acquisition → PiCamera2
                                                    │
                                              TTL out pin (lgpio)
                                              TTL in  pin (lgpio)

Requirements

• Python ≥ 3.10
• On each Raspberry Pi: picamera2 (system package), lgpio (pip install lgpio), libgpiod2 (system package)
• On the acquisition machine (Conductor): aiohttp

Installation

# Acquisition machine (Conductor only)
pip install "rpi-camera-ensemble[conductor]"

# On each Raspberry Pi (Agent + TTL sync)
pip install "rpi-camera-ensemble[agent,lgpio]"

Install from source:

git clone https://github.com/MurineShiftWork/rpi-camera-ensemble.git
cd rpi-camera-ensemble
pip install -e ".[agent,lgpio,dev]"

Quick start

Agent (on each Raspberry Pi)

Create a config file (agent.yaml):

instance_name: cam01
api_host: 0.0.0.0
api_port: 8000
data_dir: /data/recordings
log_dir: /data/logs

Start the agent:

rce-agent run --config agent.yaml

Generate a template config:

rce-agent config --to-template agent.template.yaml

Conductor (on the acquisition machine)

from rpi_camera_ensemble.conductor.conductor import Conductor
from rpi_camera_ensemble.config.acquisition import EnsembleAcquisitionConfig
from rpi_camera_ensemble.config.conductor import ConductorConfig

ensemble_cfg = EnsembleAcquisitionConfig.from_yaml("ensemble.yaml")
conductor_cfg = ConductorConfig(log_dir="/data/logs/conductor")

with Conductor(config=conductor_cfg, ensemble_config=ensemble_cfg) as conductor:
    conductor.setup_agents()
    conductor.initialize_acquisition(
        acquisition_path="/data/recordings/experiment_001",
        acquisition_name="experiment_001",
    )
    conductor.start_preview()
    conductor.start_recording()
    # ... recording ...
    conductor.stop_acquisition()

Ensemble config (ensemble.yaml):

agents:
  cam01:
    api_url: http://192.168.1.10:8000
  cam02:
    api_url: http://192.168.1.11:8000

TTL synchronization

TTL is configured per agent inside ensemble.yaml under camera_config.ttl. Default backend is lgpio (recommended):

agents:
  cam01:
    api_url: http://192.168.1.10:8000
    camera_config:
      ttl:
        library: lgpio
        gpiochip: 0       # RPi 4: 0, RPi 5: 4
        out_pin: 25       # emits 1 ms pulse on each encoded frame
        in_pin: 23        # records edge timestamps from external trigger
        out_duration_us: 1000

lgpio requires libgpiod2 on the Pi:

sudo apt install libgpiod2
pip install lgpio

RPi 5 note: set gpiochip: 4 (RP1 southbridge). RPi 4 uses gpiochip: 0.

Output structure

Each agent writes files named {basename}.rce.{instance}.{timestamp}.{type}.{ext}:

/data/recordings/experiment_001/
├── session.rce.cam01.20260101_120000_000000.video.h264
├── session.rce.cam01.20260101_120000_000000.ttl_out.npz
├── session.rce.cam01.20260101_120000_000000.ttl_in.npz
├── session.rce.cam02.20260101_120000_000000.video.h264
├── session.rce.cam02.20260101_120000_000000.ttl_out.npz
└── session.rce.cam02.20260101_120000_000000.ttl_in.npz

Load and validate a session from the acquisition machine:

from rpi_camera_ensemble.io import RCESession, validate_session

session = RCESession.from_directory("/data/recordings/experiment_001/")
qc = validate_session(session)
qc.print_report()

Deployment

See deploy/ for Ansible playbooks to provision Raspberry Pis (apt packages, Python venv, systemd service).

Development

uv sync --extra dev
uv run pre-commit install
uv run pytest

License

Copyright (c) 2025 Lars B. Rollik. All rights reserved. See LICENSE.