🚦 Event-Driven Traffic Alert System

A real-time traffic monitoring system built on Event-Driven Architecture (EDA) principles. Traffic cameras publish events (vehicle detections, speed violations, congestion alerts, clearings) onto a central in-process EventBus, which fans them out to multiple independent subscriber services. A live React dashboard visualises every event, alert, queue metric and simulator control in real time.

This project was built as a Complex Engineering Problem (CEP) for the Software Design & Architecture course and demonstrates seven well-known design patterns end-to-end:

Pattern	Where it lives
Publish–Subscribe	EventBus.publish() / EventBus.subscribe()
Observer	IEventSubscriber interface
Event Envelope	EventEnvelope Pydantic model
Idempotent Receiver	AlertService (dedupe by event_id)
Bounded Queue + Priority Eviction	EventBus._evict_lower_priority()
Outbox Pattern	Documented in backend/OUTBOX_PATTERN.md
Background Worker	asyncio task inside EventBus

---

📑 Table of Contents

1. Features
2. Tech Stack
3. System Architecture
4. Event Types
5. Project Structure
6. Prerequisites
7. Installation
8. Configuration
9. Running the Application
10. Using the Dashboard
11. REST API Reference
12. WebSocket Reference
13. Design Patterns — Deep Dive
14. Testing
15. Troubleshooting
16. CEP Mark Allocation Coverage
17. Roadmap / Future Work
18. License & Credits

---

✨ Features

• Asynchronous EventBus — fully in-process pub/sub built on asyncio. Publishers and subscribers never reference each other.
• Four event types out of the box: VehicleDetectedEvent, SpeedViolationEvent, CongestionAlertEvent, TrafficClearedEvent.
• EventEnvelope wrapping every message with event_id, correlation_id, schema_version, source_id, timestamp, event_type, priority, payload.
• Four loosely-coupled subscriber services:
  • AlertService — fires alerts on violations/congestion; deduplicates by event_id.
  • LoggingService — structured audit log of every event.
  • DashboardService — fans events out to attached WebSocket clients.
  • ReportingService — per-type and per-source counters + per-second rates.
• Bounded queue with priority-based eviction — under pressure, low-priority events (vehicle detections) are evicted to make room for critical ones (congestion alerts).
• Idempotent Receiver — same event_id published N times produces exactly one alert.
• Live React dashboard — NOC-style control-room aesthetic with simulated traffic feed, real-time event stream (WebSocket), alert log, charts (Recharts), bus metrics, and a one-click "DUP TEST" button to demo idempotency.
• Configurable traffic simulator — adjustable event rate, violation probability, congestion probability.
• Automated test — pytest verifying the idempotent receiver behaviour.
• Outbox Pattern documentation — explains how the system would extend to solve the Dual-Write problem.

---

🛠 Tech Stack

Layer	Technology
Backend language	Python 3.11
Backend framework	FastAPI
Async runtime	asyncio + Uvicorn
Validation	Pydantic v2
Frontend	React 19
Styling	Tailwind CSS + shadcn/ui
Charts	Recharts
Real-time stream	Native WebSocket
Toasts	Sonner
Tests	pytest + pytest-asyncio
Package manager (FE)	Yarn

---

🏗 System Architecture

                ┌────────────────┐         ┌────────────────────────┐
                │  CAM-101-NORTH │         │     AlertService       │  ← Idempotent Receiver
                │  CAM-202-SOUTH │  publish ─→  LoggingService       │  ← Audit log
PUBLISHERS  ───→│  CAM-303-EAST  ├────────►│     DashboardService   │  ← WebSocket fan-out
                │  CAM-404-WEST  │         │     ReportingService   │  ← Counters / rates
                └────────────────┘         └────────┬───────────────┘
                                                    │
                                                    │ WebSocket /api/ws/events
                                                    ▼
                                          ┌──────────────────────┐
                                          │   React Dashboard    │
                                          │  (live event feed,   │
                                          │   alerts, charts,    │
                                          │   simulator panel)   │
                                          └──────────────────────┘

The EventBus in the middle is the central artery:

• It owns a bounded deque of envelopes plus a list of subscribers.
• A single async worker task pulls envelopes off the deque and dispatches them concurrently to every subscriber via asyncio.gather.
• When the deque is full, it tries to evict a lower-priority queued event before dropping the incoming one.

---

📦 Event Types

Event Type	Priority	Published When	Subscribers That React
VehicleDetectedEvent	LOW (1)	Any vehicle is spotted	Dashboard, Reporting
TrafficClearedEvent	MEDIUM (2)	Congestion clears on a segment	Dashboard
SpeedViolationEvent	HIGH (3)	Vehicle exceeds the legal speed limit	Alert, Logging, Reporting
CongestionAlertEvent	CRITICAL (4)	Vehicle density at a segment exceeds the threshold	Dashboard, Logging, Alert

---

📁 Project Structure

Traffic-Light-System/
├── README.md                          ← you are here
│
├── backend/
│   ├── server.py                      FastAPI app + WebSocket
│   ├── simulator.py                   TrafficSimulator (generates events)
│   ├── requirements.txt               Python dependencies
│   ├── .env                           Backend environment variables
│   ├── OUTBOX_PATTERN.md              Explanation of Outbox / Dual-Write
│   │
│   ├── events/
│   │   ├── envelope.py                EventEnvelope + EventPriority enum
│   │   ├── types.py                   Payload schemas + PRIORITY_BY_TYPE
│   │   └── event_bus.py               async EventBus (bounded + priority)
│   │
│   ├── services/
│   │   ├── base.py                    IEventSubscriber interface
│   │   ├── alert_service.py           Idempotent Receiver
│   │   ├── logging_service.py         Audit log
│   │   ├── dashboard_service.py       WebSocket fan-out
│   │   └── reporting_service.py       Counters / rates
│   │
│   └── tests/
│       └── test_idempotent.py         pytest: same event_id → 1 alert
│
└── frontend/
    ├── package.json                   Node dependencies
    ├── .env                           Frontend environment variables
    ├── public/
    │   └── index.html                 React entry HTML
    │
    └── src/
        ├── App.js                     Dashboard layout + WebSocket plumbing
        ├── index.js                   React entry point
        ├── index.css                  Global styles + tailwind tokens
        │
        ├── lib/
        │   ├── api.js                 axios + WebSocket helpers
        │   └── eventStyles.js         event-type → colour / icon map
        │
        └── components/
            ├── Header.jsx             top bar (queue, published, WS status)
            ├── SimulatorControls.jsx  start / stop / sliders / DUP TEST
            ├── BusInspector.jsx       queue depth, subscribers, metrics
            ├── CameraGrid.jsx         4 cameras with live overlays
            ├── EventStream.jsx        live monospaced event log
            ├── AlertLog.jsx           fired / deduped alerts
            └── StatsCharts.jsx        Recharts line + bar

---

🧰 Prerequisites

Install these once on your machine before running the project:

Tool	Version	Download
Python	3.10 or newer	https://www.python.org/downloads/
Node.js	18 LTS or newer	https://nodejs.org/
Yarn	1.22+	After Node: npm install -g yarn
Git	any recent	https://git-scm.com/download/win

During Python installation on Windows, CHECK the "Add python.exe to PATH" box. This single click avoids 90% of "python is not recognized" errors later.

You do not need MongoDB. The whole project runs in-memory.

---

📥 Installation

1. Clone the repository

git clone https://github.com/<your-username>/Traffic-Light-System.git
cd Traffic-Light-System

2. Backend setup (one-time)

cd backend
python -m venv venv

# Windows PowerShell:
.\venv\Scripts\Activate.ps1
# macOS/Linux:
source venv/bin/activate

pip install -r requirements.txt

If PowerShell blocks script execution, run once as Admin: Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned -Force

3. Frontend setup (one-time)

cd ../frontend
yarn install

---

⚙️ Configuration

Create two .env files (they're not committed to the repo).

backend/.env

MONGO_URL=mongodb://localhost:27017
DB_NAME=traffic_monitor
CORS_ORIGINS=http://localhost:3000,http://localhost:3001
BUS_CAPACITY=500

Variable	Meaning
MONGO_URL	Reserved for future Outbox Pattern use. Not currently read.
DB_NAME	Reserved for future use.
CORS_ORIGINS	Comma-separated list of allowed frontend origins.
BUS_CAPACITY	Maximum events the EventBus can buffer before eviction kicks in.

frontend/.env

REACT_APP_BACKEND_URL=http://localhost:8001

---

▶️ Running the Application

You need two terminals running at the same time.

Terminal 1 — Backend

cd backend
.\venv\Scripts\Activate.ps1     # Windows
# source venv/bin/activate      # macOS/Linux

uvicorn server:app --reload --port 8001

You should see:

INFO:     Subscriber registered: AlertService
INFO:     Subscriber registered: LoggingService
INFO:     Subscriber registered: DashboardService
INFO:     Subscriber registered: ReportingService
INFO:     EventBus worker started (capacity=500).
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8001

Test in your browser: http://localhost:8001/api/health → {"status":"ok","bus_running":true}

Terminal 2 — Frontend

cd frontend
yarn start

After ~30 seconds, your browser auto-opens at http://localhost:3000.

Stopping

Press Ctrl + C in each terminal.

---

🖥 Using the Dashboard

When you open http://localhost:3000, you'll see five major regions:

1. Header bar

• QUEUE — current queue depth / capacity
• PUBLISHED — total envelopes accepted by the bus
• DROPPED — total envelopes lost to back-pressure
• SIM — IDLE / RUNNING
• WS LIVE — green dot if WebSocket is connected

2. Simulator panel (left)

• START / STOP — toggle the traffic simulator
• Rate slider — events per second (1–30)
• Violation prob slider — chance any detection becomes a speed violation
• Congestion prob slider — chance per tick to trigger congestion on a segment
• DUP TEST — publishes the SAME SpeedViolationEvent 3 times. Watch the alert log: 1 alert fires; 2 duplicates are silently ignored. This is Task 4 of the CEP, proven live.
• RESET — clears all in-memory counters, alerts and logs

3. Event Bus inspector (left)

• Queue-depth bar (24 segments)
• Published / Processed / Dropped / Evicted counters
• Subscriber pills (proves Observer Pattern wiring)
• Idempotent Receiver block — Fired / Deduped / Memory-IDs

4. Camera grid + Charts (centre)

• 4 traffic-camera tiles with timestamps & last-known speed
• Events per second line chart (last 30s)
• Total events by type bar chart

5. Alert log + Live event stream

• Right column — every fired alert, with payload details
• Bottom — every envelope as it lands on the bus

---

🌐 REST API Reference

All endpoints are prefixed with /api.

Health & Stats

Method	Path	Purpose
GET	/api/health	Liveness probe
GET	/api/bus/stats	Full system snapshot: bus, all services, simulator
GET	/api/events/recent?limit=N	Latest N events from the dashboard buffer
GET	/api/alerts?limit=N	Recent alerts + AlertService stats
GET	/api/reporting/summary?seconds=N	Totals, by-source, per-second timeline

Mutations

Method	Path	Body	Purpose
POST	/api/publish	PublishRequest	Manually publish any envelope
POST	/api/simulator/start	–	Start the traffic simulator
POST	/api/simulator/stop	–	Stop the traffic simulator
POST	/api/simulator/config	SimulatorConfig	Update rate / probabilities
POST	/api/reset	–	Wipe all in-memory state
POST	/api/demo/duplicate-violation?times=N	–	Publish same event N times — demo idempotency

Schemas

PublishRequest:
  event_type: str
  source_id: str
  payload: dict
  event_id: str | None
  correlation_id: str | None

SimulatorConfig:
  rate_per_sec: float | None       # 0.1 .. 50.0
  violation_rate: float | None     # 0.0 .. 1.0
  congestion_rate: float | None    # 0.0 .. 1.0

Sample curl

# Start simulator
curl -X POST http://localhost:8001/api/simulator/start

# Demo idempotent receiver
curl -X POST "http://localhost:8001/api/demo/duplicate-violation?times=5"

# Reset everything
curl -X POST http://localhost:8001/api/reset

---

🔌 WebSocket Reference

Endpoint

ws://localhost:8001/api/ws/events

Behaviour

On connect, the DashboardService attaches a new per-client asyncio.Queue. Every envelope dispatched on the bus is pushed to every attached queue and forwarded to the client as JSON.

Message format

{
  "event_id": "35d7961a-53fb-4a92-9670-589498f88006",
  "correlation_id": "ea42eaa5-7cae-480c-87c2-bd37b8e563e3",
  "schema_version": "1.0",
  "source_id": "CAM-303-EAST",
  "timestamp": "2026-05-18T11:02:47.519268+00:00",
  "event_type": "VehicleDetectedEvent",
  "priority": 1,
  "payload": {
    "vehicle_id": "V-A3B7C8D2",
    "plate": "AB-1234",
    "speed_kmh": 72.5,
    "lane": 2
  }
}

---

🧠 Design Patterns — Deep Dive

Observer Pattern (Task 2)

Every subscriber implements IEventSubscriber:

class IEventSubscriber(ABC):
    name: str = "Subscriber"

    @abstractmethod
    async def handle(self, envelope: EventEnvelope) -> None: ...

The EventBus stores a List[IEventSubscriber] and never references concrete classes. Adding a new subscriber adds only a new realisation arrow — no bus change. This honours the Open/Closed Principle.

Event Envelope Pattern (Task 3)

The envelope carries seven required metadata fields (plus a non-mandatory priority used by the bus eviction policy). The bus inspects metadata only; payload is opaque to it. Subscribers all read metadata the same way.

Idempotent Receiver Pattern (Task 4)

AlertService keeps _seen_ids: set[str] plus a parallel _seen_order deque to enforce a bounded LRU window of 5 000 ids. On every handle():

1. If the type is not in ALERT_TYPES, return.
2. If event_id is already in _seen_ids, increment duplicates_ignored and return.
3. Otherwise add to the set, evict the oldest if the window is full, fire the alert.

The accompanying test (tests/test_idempotent.py) publishes the same envelope three times and asserts fired == 1 and duplicates_ignored == 2.

Bounded Queue + Priority Eviction (Scenario 2)

async def publish(self, envelope):
    async with self._lock:
        if len(self._buffer) >= self._capacity:
            if not self._evict_lower_priority(envelope.priority):
                self._dropped += 1
                return False
        self._buffer.append(envelope)
        self._has_items.set()
        return True

Eviction scans the queue for the oldest event whose priority is strictly lower than the incoming one. If none exists, the incoming event itself is dropped. Effect: a flood of LOW-priority detections can never starve out a CRITICAL congestion alert.

Outbox Pattern (Scenario 3)

Documented in backend/OUTBOX_PATTERN.md. The current implementation is in-memory only, but the document outlines how to extend it to MongoDB so that a fine row and an outbox row commit in the same transaction, eliminating the Dual-Write hazard. Combined with the existing Idempotent Receiver, this delivers effectively-once semantics.

---

🧪 Testing

The required CEP test is automated with pytest:

cd backend
.\venv\Scripts\Activate.ps1
pytest tests/test_idempotent.py -v

Expected output:

tests/test_idempotent.py::test_alert_service_is_idempotent PASSED [100%]
============================== 1 passed in 0.11s ==============================

You can also demonstrate the same behaviour interactively from the dashboard with the DUP TEST button or via curl:

curl -X POST "http://localhost:8001/api/demo/duplicate-violation?times=5"
curl http://localhost:8001/api/alerts

---

🛟 Troubleshooting

Symptom	Cause	Fix
python is not recognized	Python not on PATH	Reinstall Python and check "Add to PATH"
yarn is not recognized	Yarn not installed	npm install -g yarn, then restart VS Code
running scripts is disabled (PowerShell)	Execution policy	Set-ExecutionPolicy -Scope CurrentUser RemoteSigned -Force
(venv) prefix missing in terminal	venv not activated	cd backend && .\venv\Scripts\Activate.ps1
ERROR: No matching distribution found for emergentintegrations	Leftover Emergent-only package	Delete that line from requirements.txt, reinstall
Could not find a version that satisfies the requirement …	Wrong Python version	Use 3.10+
Frontend lands on :3001 instead of :3000	Port 3000 already taken	Add http://localhost:3001 to CORS_ORIGINS in backend/.env and restart backend
"Could not start simulator" toast	CORS blocked the response	Same fix as above — broaden CORS_ORIGINS
WS OFFLINE indicator in header	Backend not running or wrong URL	Check backend log, verify REACT_APP_BACKEND_URL
WebSocket error … 1012 service restart after Ctrl+C	Normal shutdown noise	Harmless — can be silenced with try/except asyncio.CancelledError
ModuleNotFoundError after installing	venv not activated	Re-run Activate.ps1
A tiny "e" badge in browser sidebar	Opera GX / Vivaldi side panel — NOT the app	Right-click the icon in browser sidebar → Remove

---

🎯 CEP Mark Allocation Coverage

Marks	Requirement	Where It's Implemented
10	Task 1 — Event Bus with publish/subscribe + zero-change extensibility	backend/events/event_bus.py
5	Task 2 — Observer Pattern via IEventSubscriber interface	backend/services/base.py + all four services
5	Task 3 — Event Envelope (7 fields)	backend/events/envelope.py
10	Task 4 — Idempotent Receiver in AlertService + passing test	backend/services/alert_service.py + tests/test_idempotent.py
10	Scenario 1 — Schema evolution + ADR	Report only
10	Scenario 2 — Event flood calculation + bounded queue eviction	EventBus._evict_lower_priority()
10	Scenario 3 — Outbox Pattern for Dual Write	backend/OUTBOX_PATTERN.md
60	Total

---

🚀 Roadmap / Future Work

• Persist alerts to MongoDB to demonstrate the Outbox Pattern end-to-end.
• Add an OutboxRelay background task that polls the outbox collection and republishes onto the bus.
• Add a geographic map view of cameras and segments.
• Multi-region: separate EventBus instances per traffic zone with cross-region replication.
• Replay buffer / event sourcing — re-derive state from the audit log.
• Role-based access (operator vs analyst) using JWT.
• Replace polling for stats with a stats WebSocket channel.
• Dockerise the backend + frontend with docker-compose up.

---

📝 License & Credits

This project was developed as a Complex Engineering Problem (CEP) for the Software Design & Architecture course.

Team members:

• [Muhmmed Omer ]

Institution: [Software Engineering / Bahria University H11]

Acknowledgements / References

• Gamma, Helm, Johnson, Vlissides — Design Patterns: Elements of Reusable Object-Oriented Software (Observer Pattern).
• Hohpe & Woolf — Enterprise Integration Patterns (Event Envelope, Idempotent Receiver, Bounded Queue).
• Chris Richardson — Microservices Patterns (Outbox Pattern).
• FastAPI, React, Tailwind CSS, shadcn/ui, Recharts.

---