Real-time monitoring of San Francisco's Muni Metro subway using OpenCV-based computer vision to detect system status from internal SFMTA status images.
URL: https://munimet.ro
Current status:
This project was "vibe coded" using Anthropic's Claude Code. Uses deterministic computer vision analysis; no ML models or external AI services required.
First time setup? See the comprehensive Setup Guide (SETUP.md) for detailed installation instructions for macOS, Linux, and Windows.
Quick checklist:
- Python 3.13+
- Git
- Google Cloud SDK with gsutil (for accessing reference data and cloud deployment)
# Clone repository
git clone https://github.com/MrEricSir/munimet.ro.git
cd munimet.ro
# Download reference data from GCS (optional - for development/testing)
./scripts/sync-artifacts.sh download # macOS/Linux
# or
.\scripts\sync-artifacts.ps1 download # WindowsThe sync script will download:
- Reference dataset (~270MB) - labeled images for testing
Note: Requires gcloud authentication. Run gcloud auth login first.
Need help with installation? See SETUP.md for platform-specific installation instructions.
# Local deployment
cd deploy/local
./setup.sh # Creates venv, installs dependencies
./start.sh # Starts cache writer + API server
# Access services
open http://localhost:8000 # Landing page
open http://localhost:8000/dashboard # Status dashboard
# Cloud deployment (Google Cloud Run)
cd deploy/cloud
./setup-infrastructure.sh
./deploy-services.sh
./setup-scheduler.shSee deploy/README.md for detailed deployment instructions.
To enable social media posting and webhook notifications:
# Social media credentials (Bluesky, Mastodon)
python3 scripts/setup/setup-credentials.py # Local (.env file)
python3 scripts/setup/setup-credentials.py --cloud # Google Cloud Secret Manager
# Webhook URLs (Slack, Discord, Teams, etc.)
python3 scripts/setup/manage-webhooks.py # Local (.env file)
python3 scripts/setup/manage-webhooks.py --cloud # Google Cloud Secret ManagerCredentials and webhooks are optional - the app works without them, but won't post status updates or send notifications.
The reference dataset (2,666 labeled images, ~270MB) is stored in Google Cloud Storage and synced via rsync scripts.
-
Follow SETUP.md to install base dependencies (gcloud CLI, etc.)
-
Authenticate with Google Cloud:
gcloud auth login gcloud config set project munimetro -
Download reference data using sync scripts:
./scripts/sync-artifacts.sh download # macOS/Linux .\scripts\sync-artifacts.ps1 download # Windows # Or download reference data only: ./scripts/sync-reference-data.sh download # Reference data only (~270MB)
The sync scripts use gsutil rsync to efficiently download only changed files.
Contributors can collect their own test images:
-
Collect status images:
python scripts/download_muni_image.py # Run periodically to build dataset -
Analyze a single image:
python scripts/analyze.py path/to/image.jpg # Pretty output python scripts/analyze.py path/to/image.jpg --json # JSON output python scripts/analyze.py path/to/image.jpg --verbose # Full details
-
Visualize detection interactively:
python scripts/detection_viewer.py # Interactive detection viewer -
Run tests:
pytest tests/test_system_status.py -v # Verify detection accuracy
munimet.ro/
├── lib/ # Shared library code
│ ├── muni_lib.py # Core download & detection functions
│ ├── detection.py # OpenCV-based status detection
│ ├── station_detector.py # Station position detection
│ ├── train_detector.py # Train ID detection (OCR)
│ ├── station_constants.py # Station definitions
│ ├── analytics.py # SQLite-based delay analytics
│ ├── badge.py # SVG status badge generator
│ └── notifiers/ # Notification channels (Bluesky, Mastodon, RSS, webhooks)
│
├── scripts/ # Development and utility scripts
│ ├── analyze.py # CLI tool for image analysis
│ ├── detect_stations.py # Station detection CLI
│ ├── detection_viewer.py # Interactive detection viewer
│ ├── validate.sh # Local validation (lint + tests)
│ ├── install-hooks.sh # Git hooks installer
│ └── setup/ # Setup and configuration scripts
│ ├── setup-credentials.py # Social media credentials
│ └── manage-webhooks.py # Webhook URL manager
│
├── api/ # Production web API
│ ├── api.py # Falcon web server
│ ├── check_status.py # Status checker
│ ├── check_status_job.py # Cloud Run Job entry point
│ ├── html/ # Frontend files
│ │ ├── index.html # Landing page
│ │ └── dashboard.html # Status dashboard
│ └── requirements.txt # API dependencies
│
├── deploy/ # Deployment configuration
│ ├── local/ # Local development scripts
│ └── cloud/ # Google Cloud Run deployment
│
├── tests/ # Test suite
│ ├── images/ # Test images (~2MB)
│ ├── test_system_status.py # Status detection tests
│ ├── test_train_detection.py # Train detection tests
│ ├── test_api.py # API endpoint tests
│ ├── test_analytics.py # Analytics module tests
│ ├── test_notifiers.py # Notification system tests
│ ├── test_check_status.py # Status checker tests
│ ├── test_circuit_breaker.py # Circuit breaker tests
│ └── test_frontend.py # Frontend integration tests
│
└── artifacts/ # Generated data (synced via GCS)
├── reference_data/ # Reference images (~270MB)
│ ├── images/ # Labeled snapshots
│ └── labels.json # Training labels
└── runtime/ # Transient runtime data (gitignored)
├── cache/ # API response cache
└── downloads/ # Recent snapshots
- Setup Guide - Comprehensive installation guide for macOS, Linux, and Windows
- Deployment Guide - Local and cloud deployment instructions
- API Documentation - API endpoints and configuration
- Testing - Automated test suite
The system uses OpenCV-based computer vision to analyze SFMTA status images:
- Image Download - Fetches real-time status image from SFMTA internal system
- Station Detection - Identifies platform colors (blue=normal, yellow=hold)
- Track Analysis - Detects track segment status (cyan=normal, red=disabled)
- Train Detection - Locates trains and reads IDs via OCR (optional)
- Status Calculation - Determines system status (green/yellow/red) based on:
- Platforms in hold mode
- Disabled track segments
- Train bunching (4+ trains clustered together)
Cache Writer (background process)
↓ downloads image every 60s
↓ runs OpenCV detection
↓ writes JSON to local disk
Local Cache File
↑ reads JSON (~30ms)
API Server (gunicorn)
↓ serves dashboard & endpoints
Browser
Cloud Scheduler (every 3 min)
↓ triggers via OAuth
Cloud Run Job (munimetro-checker)
↓ downloads image + detects status
↓ writes JSON + exits
Cloud Storage (gs://munimetro-cache/)
↑ reads JSON (~100-200ms)
Cloud Run Service (munimetro-api)
↓ serves dashboard & endpoints
Users
- Computer Vision Detection - OpenCV-based analysis for deterministic status classification
- Rich Detection Data - Detects trains, platform holds, disabled tracks, and bunching
- Production API - Falcon web framework with health checks, caching, and graceful degradation
- Lightweight Frontend - Vanilla JavaScript with zero runtime dependencies
- Containerized Deployment - Multi-stage Docker builds with security best practices
- Smart Caching - Best-of-three smoothing reduces false positives (~30ms local response time)
- Cloud Native - Serverless deployment on Google Cloud Run with automatic scaling
- No ML Dependencies - No PyTorch or large model files required
- Multi-Channel Notifications - Status updates via Bluesky, Mastodon, RSS, and webhooks (Slack, Discord, Teams)
- Delay Analytics - SQLite-based tracking with visual dashboard for delay patterns
Embed a live status badge on any site or README:
[](https://munimet.ro)The badge updates automatically and shows the current system status (green/yellow/red).
Get notified on Slack, Discord, Microsoft Teams, or any HTTP endpoint when the system status changes.
# Interactive manager — add, remove, test, list webhooks
python scripts/setup/manage-webhooks.py # Local (.env)
python scripts/setup/manage-webhooks.py --cloud # Google Cloud Secret Manager
# Non-interactive
python scripts/setup/manage-webhooks.py --add https://hooks.slack.com/services/T00/B00/xxx
python scripts/setup/manage-webhooks.py --remove https://hooks.slack.com/services/T00/B00/xxx
python scripts/setup/manage-webhooks.py --list
python scripts/setup/manage-webhooks.py --test # Send a test notificationSlack, Discord, and Teams URLs are auto-detected and receive platform-native payloads (rich embeds, action buttons, etc.). All other URLs receive a generic JSON payload:
{
"status": "yellow",
"previous_status": "green",
"description": "Uh oh: Muni's not feeling well",
"delay_summaries": ["Westbound delay at Powell"],
"timestamp": "2026-03-01T12:00:00",
"url": "https://munimet.ro",
"badge_url": "https://munimet.ro/badge.svg"
}Subscribe to status changes via RSS at https://munimet.ro/feed.xml. Works with any RSS reader, and can be bridged to Slack or Discord using their built-in RSS integrations.
Follow @munimetro.bsky.social on Bluesky or @MuniMetro@mastodon.social on Mastodon for status updates.
- Collect Data - Run
download_muni_image.pyperiodically - Test Detection - Use
scripts/detection_viewer.pyto visualize detection - Run Tests - Execute
pytest tests/to verify detection accuracy - Test Locally - Deploy with
./deploy/local/setup.sh && ./deploy/local/start.sh - Deploy Cloud - Deploy to Cloud Run with
./deploy/cloud/deploy-services.sh
- Computer Vision: OpenCV for image analysis
- OCR: Tesseract for train ID recognition (optional)
- Web Framework: Falcon (async-ready, production WSGI)
- Frontend: Vanilla JavaScript (no build step)
- Deployment: Docker, Google Cloud Run, Cloud Scheduler
- Storage: Google Cloud Storage (reference data, cache)
- Python 3.13+
- OpenCV (opencv-python-headless)
- Falcon, Gunicorn
- Optional: Tesseract OCR for train ID detection
- Google Cloud Storage client (for cloud deployment)
- Docker & Docker Compose
- Google Cloud SDK with gsutil (for accessing reference data and cloud deployment)
- pytest for running tests
MIT