kicap/traffic-monitoring-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
kicap
2025-07-16 07:13:43 +08:00
commit aa90d3a455
18 changed files with 1410 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored Normal file
View File

@ -0,0 +1,21 @@
# Virtual environments
env/
venv/
.venv/
# Python cache
__pycache__/
*.py[cod]
*.pyo
# Distribution / packaging
build/
dist/
*.egg-info/
# IDEs/editors (optional but common)
.vscode/
.idea/
# Jupyter Notebook checkpoints (optional)
.ipynb_checkpoints/

110
app.py Normal file
View File

@ -0,0 +1,110 @@
import cv2
import pymysql
from datetime import datetime
from ultralytics import YOLO
from deep_sort_realtime.deepsort_tracker import DeepSort
# -------------------------------
# CONFIG
# -------------------------------
RTSP_URL = "pagi.mp4"
LINE_A = 200
LINE_B = 500
DB_CONFIG = {
'host': 'localhost',
'user': 'root',
'password': '',
'database': 'db_traffic2'
}
# -------------------------------
# Setup
# -------------------------------
cap = cv2.VideoCapture(RTSP_URL)
model = YOLO("yolov8n.pt") # Downloads automatically
tracker = DeepSort(max_age=30)
# MySQL Connection
db = pymysql.connect(**DB_CONFIG)
cursor = db.cursor()
tracked_ids = {} # {track_id: {'y': prev_y, 'line': 'A' or 'B'}}
while cap.isOpened():
ret, frame = cap.read()
if not ret:
print("RTSP stream ended or cannot be read.")
break
# -------------------------------
# YOLO Detection
# -------------------------------
results = model(frame, verbose=False)[0]
detections = []
for box in results.boxes:
x1, y1, x2, y2 = box.xyxy[0]
conf = float(box.conf[0])
cls_id = int(box.cls[0])
class_name = model.names[cls_id]
if class_name in ["car", "bus", "truck", "motorbike"]:
bbox = [x1.item(), y1.item(), x2.item() - x1.item(), y2.item() - y1.item()]
detections.append((bbox, conf, class_name))
# -------------------------------
# DeepSORT Tracking
# -------------------------------
tracks = tracker.update_tracks(detections, frame=frame)
for track in tracks:
if not track.is_confirmed():
continue
track_id = track.track_id
ltrb = track.to_ltrb()
x1, y1, x2, y2 = map(int, ltrb)
center_y = (y1 + y2) // 2
prev = tracked_ids.get(track_id)
tracked_ids[track_id] = {'y': center_y}
if prev:
prev_y = prev['y']
# Determine direction and line crossing
if prev_y < LINE_A and center_y >= LINE_A:
direction = 'down'
now = datetime.now()
elif prev_y > LINE_B and center_y <= LINE_B:
direction = 'up'
now = datetime.now()
else:
continue
print(f"[{now}] Vehicle {track_id} crossed {direction}")
# Save to DB
sql = "INSERT INTO vehicle_crossings (vehicle_id, direction, crossing_time) VALUES (%s, %s, %s)"
cursor.execute(sql, (f"vehicle_{track_id}", direction, now))
db.commit()
# Draw box and label
cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
cv2.putText(frame, f'ID {track_id}', (x1, y1 - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# Draw reference lines
cv2.line(frame, (0, LINE_A), (frame.shape[1], LINE_A), (255, 0, 0), 2)
cv2.line(frame, (0, LINE_B), (frame.shape[1], LINE_B), (0, 0, 255), 2)
# Show frame
cv2.imshow("YOLO + DeepSORT", frame)
if cv2.waitKey(1) == 27: # ESC to quit
break
# Cleanup
cap.release()
cv2.destroyAllWindows()
cursor.close()
db.close()

131
app2.py Normal file
View File

@ -0,0 +1,131 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# Database connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP stream or test video
cap = cv2.VideoCapture('pagi.mp4') # Change to RTSP URL if needed
# Resize ratio
ratio = 0.5
ret, frame = cap.read()
image = cv2.resize(frame, (0, 0), None, ratio, ratio)
width2, height2, channels = image.shape
# Background subtractor
fgbg = cv2.createBackgroundSubtractorMOG2()
# Tracking + speed
vehicle_id_counter = 0
tracker_dict = {}
speed_line_y1 = 150
speed_line_y2 = 200
pixel_distance = abs(speed_line_y2 - speed_line_y1)
fps = cap.get(cv2.CAP_PROP_FPS) or 60 # fallback default
# Main loop
while True:
ret, frame = cap.read()
if not ret:
break
image = cv2.resize(frame, (0, 0), None, ratio, ratio)
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
hull = [cv2.convexHull(c) for c in contours]
cv2.drawContours(image, hull, -1, (0, 255, 0), 2)
min_area = 400
max_area = 40000
for cnt in contours:
area = cv2.contourArea(cnt)
if min_area < area < max_area:
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(image, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(image, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {'pos': (cx, cy), 't1': None, 't2': None, 'logged': False}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
if not vehicle['t1'] and cy >= speed_line_y1:
vehicle['t1'] = datetime.now()
elif not vehicle['t2'] and cy >= speed_line_y2:
vehicle['t2'] = datetime.now()
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
meters = 1 # realistic estimated distance in meters
speed_kmh = (meters / delta_time) * 3.6
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h")
# Log to database
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, 'down', datetime.now(), round(speed_kmh, 2))
)
conn.commit()
vehicle['logged'] = True
# Show ID and speed
cv2.putText(image, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (7.5 / delta_time) * 3.6
cv2.putText(image, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
# Draw full-width lines
cv2.line(image, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(image, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
# Display window
cv2.imshow("Vehicle Detection", image)
if cv2.waitKey(1) & 0xFF == 27: # ESC to quit
break
# Cleanup
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

131
app3.py Normal file
View File

@ -0,0 +1,131 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# Database connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP stream or test video
cap = cv2.VideoCapture('pagi.mp4') # Replace with RTSP URL when needed
# Get original width and height
ret, frame = cap.read()
if not ret:
print("Failed to read from video source.")
exit()
height2, width2, channels = frame.shape
# Background subtractor
fgbg = cv2.createBackgroundSubtractorMOG2()
# Vehicle tracking and speed
vehicle_id_counter = 0
tracker_dict = {}
speed_line_y1 = 225
speed_line_y2 = 365
pixel_distance = abs(speed_line_y2 - speed_line_y1)
fps = cap.get(cv2.CAP_PROP_FPS) or 60 # fallback
meters_travel_pixel = 0.2
while True:
ret, frame = cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
hull = [cv2.convexHull(c) for c in contours]
cv2.drawContours(frame, hull, -1, (0, 255, 0), 2)
min_area = 400
max_area = 40000
for cnt in contours:
area = cv2.contourArea(cnt)
if min_area < area < max_area:
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(frame, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(frame, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {'pos': (cx, cy), 't1': None, 't2': None, 'logged': False}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
if not vehicle['t1'] and cy >= speed_line_y1:
vehicle['t1'] = datetime.now()
elif not vehicle['t2'] and cy >= speed_line_y2:
vehicle['t2'] = datetime.now()
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
# realistic estimation
speed_kmh = (meters_travel_pixel / delta_time) * 3.6
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h")
# Insert into DB
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, 'down', datetime.now(), round(speed_kmh, 2))
)
conn.commit()
vehicle['logged'] = True
# Show ID and speed
cv2.putText(frame, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (meters_travel_pixel / delta_time) * 3.6
cv2.putText(frame, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
# Draw full-width lines
cv2.line(frame, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(frame, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
cv2.imshow("Vehicle Detection", frame)
if cv2.waitKey(1) & 0xFF == 27:
break
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

142
app4.py Normal file
View File

@ -0,0 +1,142 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# Database connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP stream or test video
# cap = cv2.VideoCapture('pagi.mp4') # Replace with RTSP URL when needed
cap = cv2.VideoCapture('rtsp://20.20.20.33:8554/') # Replace with RTSP URL when needed
# Get original width and height
ret, frame = cap.read()
if not ret:
print("Failed to read from video source.")
exit()
height2, width2, channels = frame.shape
# Background subtractor
fgbg = cv2.createBackgroundSubtractorMOG2()
# Vehicle tracking and speed
vehicle_id_counter = 0
tracker_dict = {}
# Line positions (adjust as needed)
speed_line_y1 = 255
speed_line_y2 = 390
pixel_distance = abs(speed_line_y2 - speed_line_y1)
# Real-world distance between lines in meters (adjust based on real road)
real_world_distance_m = 0.25 # ← adjust if your line spacing is closer/farther
while True:
ret, frame = cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
hull = [cv2.convexHull(c) for c in contours]
cv2.drawContours(frame, hull, -1, (0, 255, 0), 2)
min_area = 400
max_area = 40000
for cnt in contours:
area = cv2.contourArea(cnt)
if min_area < area < max_area:
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(frame, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(frame, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {'pos': (cx, cy), 't1': None, 't2': None, 'logged': False}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
if not vehicle['t1'] and cy >= speed_line_y1:
vehicle['t1'] = datetime.now()
elif not vehicle['t2'] and cy >= speed_line_y2:
vehicle['t2'] = datetime.now()
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
if speed_kmh >= 10:
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h")
# Insert into DB only if speed ≥ 10
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, 'down', datetime.now(), round(speed_kmh, 2))
)
conn.commit()
else:
print(f"ID {matched_id} IGNORED (too slow): {speed_kmh:.2f} km/h")
vehicle['logged'] = True
# Display ID
cv2.putText(frame, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
# Show speed on screen
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
cv2.putText(frame, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
# Draw lines
cv2.line(frame, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(frame, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
# Show output
cv2.imshow("Vehicle Detection", frame)
if cv2.waitKey(1) & 0xFF == 27:
break
# Cleanup
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

146
app5.py Normal file
View File

@ -0,0 +1,146 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# DB Connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP or video
cap = cv2.VideoCapture('rtsp://20.20.20.33:8554/')
ret, frame = cap.read()
if not ret:
print("Failed to read from source.")
exit()
height2, width2, channels = frame.shape
fgbg = cv2.createBackgroundSubtractorMOG2()
vehicle_id_counter = 0
tracker_dict = {}
# Speed lines
speed_line_y1 = 275
speed_line_y2 = 370
real_world_distance_m = 0.25
min_speed_kmh = 10 # Threshold
# Track last counted vehicle IDs to avoid spamming the total_vehicle table
last_vehicle_ids_in_zone = set()
while True:
ret, frame = cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
cv2.drawContours(frame, [cv2.convexHull(c) for c in contours], -1, (0, 255, 0), 2)
min_area = 400
max_area = 40000
vehicle_ids_in_zone = set()
for cnt in contours:
area = cv2.contourArea(cnt)
if not (min_area < area < max_area):
continue
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(frame, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(frame, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {'pos': (cx, cy), 't1': None, 't2': None, 'logged': False}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
if not vehicle['t1'] and cy >= speed_line_y1:
vehicle['t1'] = datetime.now()
elif not vehicle['t2'] and cy >= speed_line_y2:
vehicle['t2'] = datetime.now()
# Speed check and log
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
if speed_kmh >= min_speed_kmh:
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h")
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, 'down', datetime.now(), round(speed_kmh, 2))
)
conn.commit()
else:
print(f"ID {matched_id} IGNORED (too slow): {speed_kmh:.2f} km/h")
vehicle['logged'] = True
# Labeling
cv2.putText(frame, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
cv2.putText(frame, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
# Check if vehicle is between the lines
if speed_line_y1 <= cy <= speed_line_y2:
vehicle_ids_in_zone.add(matched_id)
# Save count only if it changes
if vehicle_ids_in_zone != last_vehicle_ids_in_zone:
last_vehicle_ids_in_zone = vehicle_ids_in_zone.copy()
count = len(vehicle_ids_in_zone)
cursor.execute("INSERT INTO total_vehicle (count, timestamp) VALUES (%s, %s)", (count, datetime.now()))
conn.commit()
print(f"🚗 Vehicles in zone: {count}")
# Draw detection lines
cv2.line(frame, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(frame, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
cv2.imshow("Vehicle Detection", frame)
if cv2.waitKey(1) & 0xFF == 27:
break
# Cleanup
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

164
app6.py Normal file
View File

@ -0,0 +1,164 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# DB Connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP or video
cap = cv2.VideoCapture('rtsp://20.20.20.33:8554/')
ret, frame = cap.read()
if not ret:
print("Failed to read from source.")
exit()
height2, width2, channels = frame.shape
fgbg = cv2.createBackgroundSubtractorMOG2()
vehicle_id_counter = 0
tracker_dict = {}
# Speed lines
speed_line_y1 = 230
speed_line_y2 = 340
real_world_distance_m = 0.25 # Distance between lines in meters
min_speed_kmh = 10 # Speed threshold
last_vehicle_ids_in_zone = set()
while True:
ret, frame = cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
cv2.drawContours(frame, [cv2.convexHull(c) for c in contours], -1, (0, 255, 0), 2)
min_area = 400
max_area = 40000
vehicle_ids_in_zone = set()
for cnt in contours:
area = cv2.contourArea(cnt)
if not (min_area < area < max_area):
continue
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(frame, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(frame, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {
'pos': (cx, cy),
'last_cy': cy,
't1': None,
't2': None,
'direction': None,
'logged': False
}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
last_cy = vehicle['last_cy']
# Detect downward movement
if not vehicle['t1'] and last_cy < speed_line_y1 <= cy:
vehicle['t1'] = datetime.now()
elif vehicle['t1'] and not vehicle['t2'] and last_cy < speed_line_y2 <= cy:
vehicle['t2'] = datetime.now()
vehicle['direction'] = 'down'
# Detect upward movement
elif not vehicle['t1'] and last_cy > speed_line_y2 >= cy:
vehicle['t1'] = datetime.now()
elif vehicle['t1'] and not vehicle['t2'] and last_cy > speed_line_y1 >= cy:
vehicle['t2'] = datetime.now()
vehicle['direction'] = 'up'
vehicle['last_cy'] = cy
# Speed and database log
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
if speed_kmh >= min_speed_kmh:
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h DIRECTION: {vehicle['direction']}")
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, vehicle['direction'], datetime.now(), round(speed_kmh, 2))
)
conn.commit()
else:
print(f"ID {matched_id} IGNORED (too slow): {speed_kmh:.2f} km/h")
vehicle['logged'] = True
# Label
cv2.putText(frame, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
cv2.putText(frame, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
if speed_line_y1 <= cy <= speed_line_y2:
vehicle_ids_in_zone.add(matched_id)
# Store vehicle count in zone
if vehicle_ids_in_zone != last_vehicle_ids_in_zone:
last_vehicle_ids_in_zone = vehicle_ids_in_zone.copy()
count = len(vehicle_ids_in_zone)
cursor.execute("INSERT INTO total_vehicle (count, timestamp) VALUES (%s, %s)", (count, datetime.now()))
conn.commit()
print(f"🚗 Vehicles in zone: {count}")
# Draw lines
cv2.line(frame, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(frame, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
cv2.imshow("Vehicle Detection", frame)
if cv2.waitKey(1) & 0xFF == 27:
break
# Cleanup
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

174
app7.py Normal file
View File

@ -0,0 +1,174 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# DB Connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP or video
cap = cv2.VideoCapture('rtsp://20.20.20.33:8554/')
ret, frame = cap.read()
if not ret:
print("Failed to read from source.")
exit()
height2, width2, _ = frame.shape
cv2.setUseOptimized(True)
fgbg = cv2.createBackgroundSubtractorMOG2()
vehicle_id_counter = 0
tracker_dict = {}
# Line positions & buffer
speed_line_y1 = 265
speed_line_y2 = 360
buffer = 10
# Real-world distance between lines (meters)
real_world_distance_m = 0.25
min_speed_kmh = 10 # Threshold
last_vehicle_ids_in_zone = set()
while True:
ret, frame = cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
cv2.drawContours(frame, [cv2.convexHull(c) for c in contours], -1, (0, 255, 0), 2)
min_area = 200
max_area = 40000
vehicle_ids_in_zone = set()
for cnt in contours:
area = cv2.contourArea(cnt)
if not (min_area < area < max_area):
continue
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(frame, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(frame, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {
'pos': (cx, cy),
'last_cy': cy,
't1': None,
't2': None,
'direction': None,
'logged': False
}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
last_cy = vehicle['last_cy']
# Detect downward direction
if not vehicle['t1'] and last_cy < (speed_line_y1 - buffer) <= cy:
vehicle['t1'] = datetime.now()
elif vehicle['t1'] and not vehicle['t2'] and last_cy < (speed_line_y2 - buffer) <= cy:
vehicle['t2'] = datetime.now()
vehicle['direction'] = 'down'
# Detect upward direction
elif not vehicle['t1'] and last_cy > (speed_line_y2 + buffer) >= cy:
vehicle['t1'] = datetime.now()
elif vehicle['t1'] and not vehicle['t2'] and last_cy > (speed_line_y1 + buffer) >= cy:
vehicle['t2'] = datetime.now()
vehicle['direction'] = 'up'
vehicle['last_cy'] = cy
# Visual indicators for debug
if vehicle['t1']:
cv2.circle(frame, (cx, cy), 5, (0, 255, 255), -1)
if vehicle['t2']:
cv2.circle(frame, (cx, cy), 5, (255, 0, 255), -1)
# Speed and database log
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
if speed_kmh >= min_speed_kmh:
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h DIRECTION: {vehicle['direction']}")
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, vehicle['direction'], datetime.now(), round(speed_kmh, 2))
)
conn.commit()
else:
print(f"ID {matched_id} IGNORED (too slow): {speed_kmh:.2f} km/h")
vehicle['logged'] = True
# Display ID & speed
cv2.putText(frame, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (real_world_distance_m / delta_time) * 3.6
cv2.putText(frame, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
# Zone counting
if speed_line_y1 <= cy <= speed_line_y2:
vehicle_ids_in_zone.add(matched_id)
# Store zone count if changed
if vehicle_ids_in_zone != last_vehicle_ids_in_zone:
last_vehicle_ids_in_zone = vehicle_ids_in_zone.copy()
count = len(vehicle_ids_in_zone)
cursor.execute("INSERT INTO total_vehicle (count, timestamp) VALUES (%s, %s)", (count, datetime.now()))
conn.commit()
print(f"🚗 Vehicles in zone: {count}")
# Draw detection lines
cv2.line(frame, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(frame, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
cv2.imshow("Vehicle Detection", frame)
if cv2.waitKey(1) & 0xFF == 27:
break
# Cleanup
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

BIN
image.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 754 KiB

BIN
image2.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 103 KiB

11
jalankan.py Normal file
View File

@ -0,0 +1,11 @@
import subprocess
# Jalankan app.py
p1 = subprocess.Popen(["python", "this.py"])
# Jalankan main.py
p2 = subprocess.Popen(["python", "main1.py"])
# Tunggu keduanya selesai (opsional, biasanya biarkan berjalan)
p1.wait()
p2.wait()

52
main1.py Normal file
View File

@ -0,0 +1,52 @@
from flask import Flask, jsonify, render_template
import pymysql
app = Flask(__name__)
# Database config
db_config = {
'host': 'localhost',
'user': 'root',
'password': '',
'database': 'db_traffic2'
}
@app.route('/')
def index():
return render_template('index.html')
@app.route('/get_video')
def get_video_data():
try:
conn = pymysql.connect(**db_config)
cursor = conn.cursor(pymysql.cursors.DictCursor)
# Total vehicle log entries
cursor.execute("SELECT COUNT(*) AS total FROM vehicle_log")
total = cursor.fetchone()['total']
# Last 10 vehicle log entries
cursor.execute("SELECT * FROM vehicle_log ORDER BY id DESC LIMIT 10")
last_records = cursor.fetchall()
# Get latest count from total_vehicle table
cursor.execute("SELECT count FROM total_vehicle ORDER BY id DESC LIMIT 1")
latest_vehicle_count = cursor.fetchone()
current_count = latest_vehicle_count['count'] if latest_vehicle_count else 0
return jsonify({
'total_entries': total,
'last_10_entries': last_records,
'vehicles_in_zone': current_count # ⬅️ NEW FIELD
})
except Exception as e:
return jsonify({'error': str(e)}), 500
finally:
cursor.close()
conn.close()
if __name__ == '__main__':
app.run(debug=True, port=5000)

BIN
pagi.mp4 Executable file
View File

Binary file not shown.

6
readme.md Normal file
View File

@ -0,0 +1,6 @@
# Traffic monitoring
## Traffic monitoring using opencv, monitoring vehicles in real time and save to database
## You can view the data on flask
![This is the monitoring](image.png)
![This is the result](image2.png)

44
requirements.txt Normal file
View File

@ -0,0 +1,44 @@
blinker==1.9.0
certifi==2025.7.14
charset-normalizer==3.4.2
click==8.2.1
contourpy==1.3.2
cycler==0.12.1
deep-sort-realtime==1.3.2
filelock==3.18.0
Flask==3.1.1
fonttools==4.58.5
fsspec==2025.5.1
idna==3.10
itsdangerous==2.2.0
Jinja2==3.1.6
kiwisolver==1.4.8
MarkupSafe==3.0.2
matplotlib==3.10.3
mpmath==1.3.0
networkx==3.4.2
numpy==1.24.4
opencv-python==4.12.0.88
packaging==25.0
pandas==2.3.1
pillow==11.3.0
psutil==7.0.0
py-cpuinfo==9.0.0
PyMySQL==1.1.1
pyparsing==3.2.3
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==6.0.2
requests==2.32.4
scipy==1.15.3
six==1.17.0
sympy==1.14.0
torch==2.2.2
torchvision==0.17.2
tqdm==4.67.1
typing_extensions==4.14.1
tzdata==2025.2
ultralytics==8.3.166
ultralytics-thop==2.0.14
urllib3==2.5.0
Werkzeug==3.1.3

114
templates/index.html Normal file
View File

@ -0,0 +1,114 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Traffic Monitor</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 30px;
background-color: #f4f4f4;
}
h1 {
color: #333;
}
#stats {
margin-bottom: 20px;
}
.stat {
font-size: 18px;
margin-bottom: 10px;
}
table {
border-collapse: collapse;
width: 100%;
background-color: white;
}
th, td {
border: 1px solid #ccc;
padding: 8px 12px;
text-align: center;
}
th {
background-color: #eee;
}
tr:nth-child(even) {
background-color: #f9f9f9;
}
.loading {
color: #888;
}
</style>
</head>
<body>
<h1>🚦 Traffic Logger Dashboard</h1>
<div id="stats">
<div class="stat">🔢 Total Vehicle Log Entries: <span id="total-entries" class="loading">Loading...</span></div>
<div class="stat">🚗 Vehicles Currently Between Lines: <span id="vehicle-zone" class="loading">Loading...</span></div>
</div>
<h2>📋 Last 10 Vehicle Entries</h2>
<table>
<thead>
<tr>
<th>ID</th>
<th>Vehicle ID</th>
<th>Direction</th>
<th>Timestamp</th>
<th>Speed (km/h)</th>
</tr>
</thead>
<tbody id="log-table-body">
<tr><td colspan="5" class="loading">Loading data...</td></tr>
</tbody>
</table>
<script>
async function loadData() {
try {
const response = await fetch('/get_video');
const data = await response.json();
// Update stats
document.getElementById('total-entries').textContent = data.total_entries;
document.getElementById('vehicle-zone').textContent = data.vehicles_in_zone;
// Populate table
const tbody = document.getElementById('log-table-body');
tbody.innerHTML = '';
data.last_10_entries.forEach(entry => {
const row = document.createElement('tr');
row.innerHTML = `
<td>${entry.id}</td>
<td>${entry.vehicle_id}</td>
<td>${entry.direction}</td>
<td>${entry.timestamp}</td>
<td>${entry.speed.toFixed(1)}</td>
`;
tbody.appendChild(row);
});
} catch (err) {
console.error('Error loading data:', err);
document.getElementById('log-table-body').innerHTML = `<tr><td colspan="5">Error loading data</td></tr>`;
}
}
// Initial load
loadData();
// Refresh every 10 seconds
setInterval(loadData, 5000);
</script>
</body>
</html>

164
this.py Normal file
View File

@ -0,0 +1,164 @@
import cv2
import numpy as np
import pymysql
from datetime import datetime
# Database connection
conn = pymysql.connect(
host='localhost',
user='root',
password='',
database='db_traffic2'
)
cursor = conn.cursor()
# RTSP stream or test video
cap = cv2.VideoCapture('pagi.mp4') # Replace with RTSP URL when needed
# Get original width and height
ret, frame = cap.read()
if not ret:
print("Failed to read from video source.")
exit()
height2, width2, channels = frame.shape
# Background subtractor
fgbg = cv2.createBackgroundSubtractorMOG2()
# Vehicle tracking and speed
vehicle_id_counter = 0
tracker_dict = {}
speed_line_y1 = 245
speed_line_y2 = 355
meters_travel_pixel = 0.2
last_vehicle_ids_in_zone = set()
while True:
ret, frame = cap.read()
if not ret:
break
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
fgmask = fgbg.apply(gray)
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
closing = cv2.morphologyEx(fgmask, cv2.MORPH_CLOSE, kernel)
opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)
dilation = cv2.dilate(opening, kernel)
_, bins = cv2.threshold(dilation, 220, 255, cv2.THRESH_BINARY)
contours, _ = cv2.findContours(bins, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)[-2:]
hull = [cv2.convexHull(c) for c in contours]
cv2.drawContours(frame, hull, -1, (0, 255, 0), 2)
min_area = 400
max_area = 40000
vehicle_ids_in_zone = set()
for cnt in contours:
area = cv2.contourArea(cnt)
if min_area < area < max_area:
M = cv2.moments(cnt)
if M['m00'] == 0:
continue
cx = int(M['m10'] / M['m00'])
cy = int(M['m01'] / M['m00'])
if speed_line_y1 <= cy <= speed_line_y2:
vehicle_ids_in_zone.add(matched_id)
x, y, w, h = cv2.boundingRect(cnt)
cv2.rectangle(frame, (x, y), (x + w, y + h), (255, 0, 0), 2)
cv2.drawMarker(frame, (cx, cy), (0, 0, 255), cv2.MARKER_CROSS, 10, 1)
matched_id = None
for vid, info in tracker_dict.items():
old_cx, old_cy = info['pos']
if abs(cx - old_cx) < 30 and abs(cy - old_cy) < 30:
matched_id = vid
break
if matched_id is None:
matched_id = vehicle_id_counter
vehicle_id_counter += 1
tracker_dict[matched_id] = {
'pos': (cx, cy),
't1': None,
't2': None,
'logged': False,
'last_cy': cy,
'direction': None
}
else:
tracker_dict[matched_id]['pos'] = (cx, cy)
vehicle = tracker_dict[matched_id]
last_cy = vehicle['last_cy']
vehicle['last_cy'] = cy
# Downward detection (your original logic)
if not vehicle['t1'] and cy >= speed_line_y1:
vehicle['t1'] = datetime.now()
elif not vehicle['t2'] and cy >= speed_line_y2:
vehicle['t2'] = datetime.now()
vehicle['direction'] = 'down'
# Upward detection (added logic)
if not vehicle['t1'] and cy <= speed_line_y2:
vehicle['t1'] = datetime.now()
elif not vehicle['t2'] and cy <= speed_line_y1:
vehicle['t2'] = datetime.now()
vehicle['direction'] = 'up'
# Speed calculation
if vehicle['t1'] and vehicle['t2'] and not vehicle['logged']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (meters_travel_pixel / delta_time) * 3.6
direction = vehicle['direction'] or 'unknown'
print(f"ID {matched_id} SPEED: {speed_kmh:.2f} km/h DIR: {direction}")
if speed_kmh >= 10:
cursor.execute(
"INSERT INTO vehicle_log (vehicle_id, direction, timestamp, speed) VALUES (%s, %s, %s, %s)",
(matched_id, direction, datetime.now(), round(speed_kmh, 2))
)
conn.commit()
vehicle['logged'] = True
# Show ID
cv2.putText(frame, f"ID:{matched_id}", (x, y - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 255), 1)
# Show speed
if vehicle['t1'] and vehicle['t2']:
delta_time = (vehicle['t2'] - vehicle['t1']).total_seconds()
if delta_time > 0:
speed_kmh = (meters_travel_pixel / delta_time) * 3.6
cv2.putText(frame, f"{speed_kmh:.1f} km/h", (x, y + h + 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 0), 2)
# Store vehicle count in zone
if vehicle_ids_in_zone != last_vehicle_ids_in_zone:
last_vehicle_ids_in_zone = vehicle_ids_in_zone.copy()
count = len(vehicle_ids_in_zone)
cursor.execute("INSERT INTO total_vehicle (count, timestamp) VALUES (%s, %s)", (count, datetime.now()))
conn.commit()
print(f"🚗 Vehicles in zone: {count}")
# Draw detection lines
cv2.line(frame, (0, speed_line_y1), (width2, speed_line_y1), (0, 255, 255), 2)
cv2.line(frame, (0, speed_line_y2), (width2, speed_line_y2), (255, 0, 255), 2)
cv2.imshow("Vehicle Detection", frame)
if cv2.waitKey(1) & 0xFF == 27:
break
cap.release()
cursor.close()
conn.close()
cv2.destroyAllWindows()

BIN
yolov8n.pt Normal file
View File

Binary file not shown.