Spectra

On-prem vision compliance for regulated industries.

Spectra is a two-stage vision pipeline that watches video footage and flags compliance issues. It uses a vision-language model to reason about scenes and a text-prompted segmentation model to outline every finding at pixel level — both running entirely on-prem. No cloud, no per-frame API costs, full audit trail.

You already have cameras everywhere for security. Spectra makes those cameras understand your SOPs.

This repo is the headless FastAPI backend. The web UI lives in a separate repo: spectra-web.

Scenarios

Each scenario is just a different system prompt — the pipeline is the same. Add new scenarios by editing src/spectra/vlm.py.

How it works

video frame
    │
    ▼
[ VLM ]  reasons about the scene. What should be checked?
    │   returns: SceneAssessment(summary, [concept phrases + severity])
    ▼
[ SEGMENTER ]  text-prompted instance segmentation per concept
    │   returns: [DetectedInstance(mask, box, score, track_id)]
    ▼
[ render ]   overlay masks + boxes + scene summary banner
    │
    ▼
annotated MP4  +  typed audit-trail JSON

Two execution modes for video:

• Per-frame — image segmentation runs on every frame independently. Fast, no cross-frame identity.
• Tracked — video session with shared memory. Each detected instance gets a stable track_id so the audit log can say "Worker #2 was non-compliant from 0:03 to 0:09".

Both produce the same RunReport audit-trail JSON.

API

GET    /api/health                   liveness + observability status
GET    /api/scenarios                scenario registry
GET    /api/footage                  local sample clips with metadata
POST   /api/runs                     start an async pipeline job
GET    /api/runs                     list recent jobs
GET    /api/runs/{id}                full job state + collapsed audit table
GET    /api/runs/{id}/events         Server-Sent Events progress stream
GET    /api/runs/{id}/video          annotated MP4 (range requests for scrub)
GET    /api/runs/{id}/source         original MP4 (range requests)
GET    /api/runs/{id}/report         raw audit-trail JSON
DELETE /api/runs/{id}                clean up
POST   /api/instrument               single-image instrument reading

See src/spectra/api.py for the full OpenAPI schema — or just hit http://127.0.0.1:8000/docs when the server is running.

Install

Requires uv and Python 3.12. On macOS:

brew install uv python@3.12 yt-dlp

Then from the repo root:

uv sync                                  # creates .venv, installs deps
bash scripts/fetch_footage.sh            # ~25 MB of royalty-free sample clips
uv run python -m spectra.server          # API at http://127.0.0.1:8000

First run downloads foundation-model weights (~6 GB) into ~/.cache/huggingface/. Subsequent runs are instant.

Observability (optional)

Spectra integrates with Pydantic Logfire for local or cloud observability. Configured via either the Logfire CLI:

uv run logfire auth
uv run logfire projects use <your-project>

…or a LOGFIRE_TOKEN in .env (see .env.example). When neither is set, every observability call is a no-op — zero overhead, zero data egress. This is the on-prem privacy story: instrument dev/internal builds, leave client deployments unconfigured.

Audit trail

Every run produces a typed RunReport JSON in output/ that records every frame's scene assessment, every detection's mask + confidence + track ID, and a timestamp for each. The API's _collapse_report helper compresses consecutive frame-level findings into one row per continuous event, which is what the UI's audit table displays.

The JSON format is stable and designed to be:

• POSTable to a LIMS / eQMS without transformation
• Queryable with any JSON-aware tool
• 21 CFR Part 11-compatible as a contemporaneous, attributable record

CLI usage

from pathlib import Path
from spectra.pipeline import PipelineConfig, run_dispatch

cfg = PipelineConfig(
    scenario="gowning",
    output_video_path=Path("output/demo.mp4"),
    output_report_path=Path("output/demo.json"),
    max_frames=300,           # ~10 seconds at 30fps for fast iteration
    use_video_tracker=False,  # True for stable cross-frame track IDs
)
report = run_dispatch(
    "footage/efficient-pharmaceutical-storage-in-cleanroom-31522472.mp4",
    cfg,
)
print(f"{report.violations} violations across {report.frames_processed} frames")

License

MIT. See LICENSE.