What is this app?

It is a clone of a city's public traffic-camera portal, modeled on Semarang, Indonesia. You get a live map of the city, and each colored dot is a real street camera.

Click a dot and live video plays with green boxes drawn around every vehicle, plus a running count of how busy that road is right now.

Think of it as a tireless traffic control room

Picture a traffic control room: a wall of monitors showing far more cameras than any human could ever watch at once. A real operator can glance at maybe one or two feeds at a time, and only when they happen to be looking.

This app is the operator that never blinks. It watches every monitor simultaneously, recognizes the vehicles on each one, and turns what it sees into numbers and colors on the map.

Trace one click, end to end

You already know what happens on screen — a dot, a click, a video. Now let's follow the journey behind it. Three characters do the work: your browser, the server, and the AI detector.

Here is the same journey as cards you can scan:

That very first step — your browser visiting the home address — runs one short function on the server. Here it is, in plain English:

@app.route('/')
def index():
    """Main page - Map with CCTV locations"""
    return render_template('map_dashboard.html')

The route @app.route('/') is the front door. render_template is the act of assembling an HTML page and handing it back so the browser can render it.

What's behind the green box?

Behind each green box, the AI is examining the video one picture at a time — every still picture is called a frame — finding vehicles and tallying them as they cross an invisible line on the road.

That tally is what becomes the count, and the count is what becomes the color. We'll open up that machinery in Module 3.

Colors aren't decoration — each congestion state maps to one exact color, defined once and reused everywhere:

const colors = {
    'FREE_FLOW': '#4CAF50',
    'MODERATE': '#FFC107',
    'CONGESTED': '#FF9800',
    'SEVERE': '#f44336',
    'UNKNOWN': '#9E9E9E'
};

Check your understanding

These aren't memory questions — each one asks you to use the journey you just traced. Take a moment to think before you check.

A Live TV News Crew

Picture a live news broadcast: camera crews in the field, an editor in the booth, an archive of old tapes, and the screen you watch at home. This app works exactly the same way — seven roles, each with one job.

The Brain That Holds It Together

The producer's clipboard, TrafficSystem, is one Python class. The app creates a single instance of it and keeps that one object alive the entire time the app runs.

Inside it are the cameras, the live numbers, the worker crews, the archive connection, and the AI model — all in one place. Here is the moment it is born:

class TrafficSystem:
    def __init__(self):
        self.cctvs = {}  # CCTV configurations
        self.streams = {}  # Active stream handlers
        self.detectors = {}  # Detection workers
        self.db = DatabaseManager()
        self.model = None
        self.running = False

The real brain also keeps a traffic_data notebook — the live readings (vehicle counts, congestion level) for every camera. That is the page workers scribble fresh numbers onto.

One Crew Per Camera

Every camera gets its very own background worker — a daemon crew that watches that one road and nothing else. This is why self.detectors is a roster: one entry per camera.

Here is the studio floor plan — where each character lives in the project:

Match the Character to the Job

Drag each crew member onto the job they own. One symptom, one responsible character.

Think It Through

No definitions here — just situations. Pick the character you would actually talk to.

The person at the stadium door

Picture someone standing in a stadium doorway holding a hand clicker. Every time a person crosses the threshold, they click. Simple.

The hard part is not double-counting. If someone shuffles back and forth in the doorway, you must recognize them as the same person so you only click once. That "recognize the same one again" job is the secret sauce of this whole module.

Step 1 — Read the video, but not every frame

A camera streams roughly 30 frames per second. Examining all 30 with an AI every second is too slow to keep up live, so the worker analyzes only every 3rd frame — fast enough to stay in real time, accurate enough to count cars.

# Process every 3rd frame for performance (10fps processing)
if self.frame_count % 3 == 0:
    self.process_frame(frame)

Step 2 — YOLO finds the vehicles (the eyes)

For each frame it does inspect, the system hands the picture to YOLO, an AI model that returns a bounding box around every vehicle and what type it is.

# Run YOLO detection
results = self.model(frame, conf=0.3)
det = results[0].boxes

Steps 3 & 4 — Remember each vehicle, then click

YOLO has no memory — to it, every frame is a brand-new scene, so the same car looks "new" 10 times a second. DeepSORT fixes that: it tracks each vehicle and gives it a sticky ID.

Here is the doorway click in code. It fires only once per vehicle, the instant its tracked position moves from one side of the line to the other:

if len(positions) >= 2 and not self.counted_vehicles[track_id]['counted']:
    if (positions[0] < line_y and positions[-1] >= line_y) or \
       (positions[0] > line_y and positions[-1] <= line_y):
        self.counted_vehicles[track_id]['counted'] = True
        self.vehicle_count['in'] += 1

Step 5 — From count to congestion (the math)

Every 10 seconds, the system takes its tally and turns it into a verdict. First it scales the 10-second count up to a full minute (multiply by 6), then sorts that rate into one of four buckets with a Level-of-Service letter grade. This is the hero of the whole module.

vehicles_per_minute = self.vehicle_count['in'] * 6  # Scale 10s to 1min

# Determine congestion level based on vehicle count
if vehicles_per_minute < 10:
    congestion = 'FREE_FLOW'
    level = 'A'
elif vehicles_per_minute < 30:
    congestion = 'MODERATE'
    level = 'C'
elif vehicles_per_minute < 60:
    congestion = 'CONGESTED'
    level = 'D'
else:
    congestion = 'SEVERE'
    level = 'F'

Watch one frame travel through the pipeline

The four components are like coworkers passing a clipboard down the line. Step through their conversation as a single frame gets processed end to end.

Think it through

You've seen the full pipeline. Now use it to reason about what could go wrong.

No more hitting refresh

Old websites were like a printed newspaper: to see anything new, you had to fetch a fresh copy yourself. This dashboard is different — the moment traffic changes at a camera, your screen updates on its own.

Think of it like a live sports broadcast. You sit back and watch; the picture and the scoreboard come to you. That magic rests on two ideas.

One worker per camera

Imagine 50 cameras handled by a single worker who must visit each in turn. If camera #12 hangs, everyone behind it waits. So each camera gets its own dedicated worker running in the background.

One reading, three destinations

Here is the hero moment. When a worker computes a fresh reading, the brain does three things at once: it updates its live memory, saves a permanent copy, and pushes the news to every open browser.

def update_traffic_data(self, cctv_id, data):
    """Update traffic data and emit to clients"""
    self.traffic_data[cctv_id].update(data)
    self.traffic_data[cctv_id]['last_updated'] = datetime.now().isoformat()

    # Save to database
    self.db.add_traffic_data(cctv_id, data)

    # Emit to connected clients (with error handling)
    try:
        socketio.emit('traffic_update', {
            'cctv_id': cctv_id,
            'data': data
        })
    except Exception as e:
        print(f'[Socket] Emit error: {e}')
            

Two channels to your browser

The numbers and the video do not travel together — they ride two separate pipes that happen to arrive at the same screen, just like the broadcast and its scoreboard ticker.

Here is the video pipe. It loops forever, grabbing the latest processed image (boxes already drawn) and sending it down the wire as one more JPEG.

def generate_stream(cctv_id):
    """Generate MJPEG stream for a CCTV"""
    while True:
        if cctv_id in traffic_system.detectors:
            frame = traffic_system.detectors[cctv_id].get_frame()
            if frame:
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
        time.sleep(0.033)  # ~30fps
            

Trace a single reading

Watch one brand-new reading travel from a camera worker, through the brain, splitting to both the database and a live push, until a dot on your map quietly changes color. Step through it.

On the browser side, the receiving end is just three short lines: store the new data, refresh the sidebar, and recolor the map markers.

socket.on('traffic_update', (data) => {
    trafficData[data.cctv_id] = data.data;
    updateSidebar();
    updateMarkers();
});
            

Check your understanding

No definitions to recite here — just real situations. Pick what you would conclude, then check.

Why bother remembering?

The live counts you saw in the last module live in the app's memory — and memory is forgetful. The moment the app restarts, those numbers vanish to zero.

So the system writes every reading into a database. That permanent record is what gives us history, trends over time, and a full road-network view.

Three drawers in the filing cabinet

The database is split into three tables, each holding one kind of thing.

What one reading looks like on disk

Before you can file cards, you write a blueprint describing what each card holds. That blueprint is the schema. Here is the exact blueprint for the readings drawer.

CREATE TABLE IF NOT EXISTS traffic_data (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    cctv_id TEXT NOT NULL,
    timestamp TEXT DEFAULT CURRENT_TIMESTAMP,
    vehicle_count INTEGER DEFAULT 0,
    vehicles_per_minute REAL DEFAULT 0,
    cars INTEGER DEFAULT 0,
    motorcycles INTEGER DEFAULT 0,
    buses INTEGER DEFAULT 0,
    trucks INTEGER DEFAULT 0,
    congestion_level TEXT DEFAULT 'UNKNOWN',
    los TEXT DEFAULT '-'
)

Joining the dots

To color a whole road on the map (not just a camera dot), the system links each road to its camera and pulls that camera's latest reading. Stitching two drawers together like this is called a JOIN.

Pinning cameras to real streets

A camera knows its coordinates, but not which street it overlooks. So the system acts like a surveyor: it walks out to each camera and pins it to the exact road on the map.

It asks OpenStreetMap (through the Overpass API) "which roads are near this point?", then measures the straight-line distance to each with the haversine formula and snaps the camera to the nearest one.

def haversine_distance(lat1, lon1, lat2, lon2):
    """Calculate distance in meters between two lat/lon points."""
    R = 6371000  # Earth radius in meters
    phi1 = math.radians(lat1)
    phi2 = math.radians(lat2)
    dphi = math.radians(lat2 - lat1)
    dlam = math.radians(lon2 - lon1)
    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2
    return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

The matching process, step by step

Check your understanding

No memorizing — each question asks you to reason about how memory and mapping really work here.

Software Meets the Messy Real World

Camera feeds drop. Files go missing. A library quietly changes its rules in a new version. Robust software does not hope this won't happen — it expects it.

Symptom: The Camera Stream Won't Open

The app tries to open a feed with its primary video engine (FFmpeg). If that fails, it tries a different engine before giving up. A backup plan beats surrender.

self.cap = cv2.VideoCapture(self.stream_url, cv2.CAP_FFMPEG)

# Set buffer size to reduce latency
self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)

if not self.cap.isOpened():
    print(f"[Detector {self.cctv_id}] Failed to open stream, retrying with HTTP...")
    # Try with GStreamer as fallback
    gst_pipeline = f'souphttpsrc location={self.stream_url} ! hlsdemux ! decodebin ! videoconvert ! appsink'
    self.cap = cv2.VideoCapture(gst_pipeline, cv2.CAP_GSTREAMER)

    if not self.cap.isOpened():
        print(f"[Detector {self.cctv_id}] Failed to open stream completely")
        self.system.cctvs[self.cctv_id]['status'] = 'error'
        return

The lesson: try a backup before giving up. And notice — only this camera is flagged. Its neighbors keep streaming.

Symptom: The AI Tracker File Is Missing

DeepSORT — the continuity assistant from Module 2 — needs a downloaded checkpoint file to do its smart tracking. If that file is missing, the app does not crash. It quietly falls back to simpler counting and keeps running.

if os.path.exists(cfg_deep.DEEPSORT.REID_CKPT):
    self.deepsort = DeepSort(
        cfg_deep.DEEPSORT.REID_CKPT,
        ...
        use_cuda=torch.cuda.is_available()
    )
    self.use_deepsort = True
    print(f"[Detector {self.cctv_id}] DeepSORT initialized")
else:
    print(f"[Detector {self.cctv_id}] DeepSORT checkpoint not found, using simple detection")

Two Gotchas Worth Knowing

Two failure stories from this real project — the kind that teach you how to read symptoms.

The version-trap fix — a small wrapper that quietly restores the old loading rule:

def safe_torch_load(*args, **kwargs):
    """Wrapper for torch.load that sets weights_only=False"""
    if 'weights_only' not in kwargs:
        kwargs['weights_only'] = False
    return original_torch_load(*args, **kwargs)

The thread-safety move is a single, deliberate line inside each worker:

# Load YOLO model (each thread needs its own)
self.model = YOLO('yolov8n.pt')

The Big Picture

Here is the whole system, internal and external. Click any box to see what it does — this is everything you have learned, on one map.

Read it left to right: the messy outside world feeds in, the workers run the AI pipeline, the brain holds the live numbers and saves history, and the server pushes it all to your browser.

Diagnose From the Symptoms

The finale: four real debugging situations. Read each symptom and reason your way to the cause — just like a doctor.