Smart Traffic Management System – Code Explanation

 Introduction

This project is a Smart Traffic Management System built using Python, Flask, OpenCV, SQLAlchemy, and multithreading.

The system processes video feeds from multiple cameras, detects vehicle density, dynamically controls traffic signals, and logs traffic data into a database.

---

Project Flow

1. Video Input – Multiple cameras (or video files) provide live feeds.

2. Processing – Each frame is analyzed using OpenCV to count vehicles.

3. Signal Logic – Based on vehicle density, traffic lights are dynamically switched.

4. Database Logging – Vehicle counts are stored every minute in SQLite.

5. Web Dashboard – Flask renders live video, API endpoints, and a dashboard for monitoring.

---

Code Explanation

1. Importing Libraries

import cv2, time, threading, numpy as np

from datetime import datetime, timedelta

from flask import Flask, render_template, Response, jsonify

from flask_sqlalchemy import SQLAlchemy

from sqlalchemy import UniqueConstraint

• OpenCV (cv2) – Video frame processing.

• Flask – Web server and API routes.

• SQLAlchemy – Database ORM for storing traffic logs.

• Threading – For handling video capture and logging simultaneously.

---

2. Flask + Database Setup

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///traffic.db'

db = SQLAlchemy(app)

• SQLite database is used to store traffic logs.

• A VehicleLog model records lane number, vehicle count, and timestamp.

---

3. Camera Configuration

camera_sources = {

    0: 'video1.mp4',

    1: 'video2.mp4',

    2: 'video3.mp4',

    3: 'video4.mp4'

}

caps = {k: cv2.VideoCapture(v) for k, v in camera_sources.items()}

• Multiple video sources are mapped (lanes).

• Each lane is processed separately.

---

4. Vehicle Detection

def process_frame_simple(frame):

    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

    blur = cv2.GaussianBlur(gray, (3, 3), 0)

    _, th1 = cv2.threshold(blur, 50, 255, cv2.THRESH_BINARY_INV)

    _, th2 = cv2.threshold(blur, 100, 255, cv2.THRESH_BINARY_INV)

    thresh = cv2.bitwise_or(th1, th2)

    kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (5, 5))

    thresh = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel)

    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

    

    count = 0

    for cnt in contours:

        x, y, w, h = cv2.boundingRect(cnt)

        area = w * h

        ar = w / h if h > 0 else 0

        if 200 < area < 80000 and 0.3 < ar < 8.0:

            count += 1

    return count

• Each frame is converted to grayscale, blurred, and thresholded.

• Contours are detected to approximate vehicles.

• Counts are filtered by size and aspect ratio.

---

5. Traffic Signal Control

• Signals alternate between two groups of lanes:

  • Group 1 → Lanes 1 & 2

  • Group 2 → Lanes 3 & 4

• The active group is set to GREEN, the other to RED.

• Adaptive timing: if traffic density difference > 1.25%, extra green time is added.
  
  

if diff >= 0.0125:

    signal_timer = base_time + 15

else:

    signal_timer = base_time

---

6. Drawing Traffic Lights on Frames

def draw_traffic_light(frame, state):

    colors = {"RED": (0, 0, 255), "YELLOW": (0, 255, 255), "GREEN": (0, 255, 0)}

    cv2.circle(frame, (50, 50), 15, colors[state], -1)

    return frame

• Each frame shows a visual traffic light (red, yellow, green).

---

7. Logging Vehicle Data

def log_vehicle_counts():

    while True:

        ts = datetime.now().replace(second=0, microsecond=0)

        for lane, count in vehicle_counts.items():

            log = VehicleLog(lane=lane+1, vehicles=int(count), timestamp=ts)

            db.session.add(log)

            db.session.commit()

• Every minute, vehicle counts per lane are stored in the database.
  
  

---

8. Video Streaming

@app.route('/video_feed/<int:cam_id>')

def video_feed(cam_id):

    return Response(gen_video(cam_id), mimetype='multipart/x-mixed-replace; boundary=frame')

• Streams real-time lane feeds to the browser.

---

9. API Endpoints

• /api/signals → Returns signal status, remaining time, and vehicle counts.

• /api/logs → Returns last 100 logs from the database.

• /api/stats → Provides system-level statistics (total intersections, efficiency, avg wait time).

---

10. Dashboard

@app.route("/dashboard")

def dashboard():

    logs = VehicleLog.query.order_by(VehicleLog.timestamp.desc()).limit(50).all()

    return render_template("dashboard.html", logs=logs)

• Renders a dashboard with traffic history logs.

---

Conclusion

This system integrates computer vision, real-time processing, adaptive signal control, database logging, and web APIs. It is scalable for multiple intersections and can be enhanced with advanced ML models like YOLO or DeepSORT for precise vehicle detection.