Jallson Suryo
Published © Apache-2.0

Traffic Analysis using optimized YOLOv8 with AMD Ryzen AI

An object detection model deployed in the Vehicle Traffic Analysis use case which can track objects (car) location, speed and direction.

IntermediateFull instructions provided8 hours357
Traffic Analysis using optimized YOLOv8 with AMD Ryzen AI

Things used in this project

Hardware components

AMD Ryzen AI Laptop/Mini PC - eg. Minisforum UM790 Pro with Ryzen 9 7940HS
×1
Display, Keyboard and Mouse
×1
Webcam, Logitech® HD Pro
Webcam, Logitech® HD Pro
×1

Software apps and online services

Microsoft Windows 11
AMD Ryzen AI SW
PyTorch
Anaconda/Miniconda
Python >=3.9
Microsoft Visual Studio 2019
OpenCV
OpenCV
Ultralytic YOLOv8
Huggingface optimum-amd for Ryzen AI

Story

Read more

Code

carcounter

Python
python code with YOLO v8 object detection to count cars in video
import cv2
from ultralytics import YOLO

# Load YOLOv8 model
model = YOLO('yolov8n.pt')  # You can choose other versions like 'yolov8s.pt', 'yolov8m.pt', etc.

# Path to the video file
video_path = r'C:\Users\jalls\Documents\yolo-cars\YOLOv8\carvideo3.mp4'

# Initialize video capture
cap = cv2.VideoCapture(video_path)

if not cap.isOpened():
    print("Error: Could not open video file.")
    exit()

# Initialize car counts and tracking dictionary
car_out_count = 0
car_in_count = 0
tracked_cars = {}
next_car_id = 0

while True:
    ret, frame = cap.read()
    if not ret:
        print("End of video file.")
        break

    # Perform detection
    results = model(frame)

    # Process the results
    current_frame_cars = []

    for result in results:
        # Iterate over detected objects
        for obj in result.boxes:
            # Check if the detected object is a car (class 2 in COCO dataset)
            cls_id = int(obj.cls[0])  # Ensure it's an integer
            if cls_id == 2:
                # Get bounding box coordinates
                x1, y1, x2, y2 = map(int, obj.xyxy[0])  # Extract coordinates from tensor
                bbox = (x1, y1, x2, y2)

                # Check if this car is already being tracked
                car_found = False
                for car_id, car_data in tracked_cars.items():
                    prev_bbox = car_data['bbox']
                    if abs(prev_bbox[0] - x1) < 50 and abs(prev_bbox[1] - y1) < 50:
                        tracked_cars[car_id]['bbox'] = bbox
                        car_found = True
                        
                        # Check for "car-out" crossing
                        if car_data['crossed_out'] == False and y1 < 640 and y2 >=640 and 0 <= x1 <= 600:
                            car_out_count += 1
                            tracked_cars[car_id]['crossed_out'] = True
                        
                        # Check for "car-in" crossing
                        if car_data['crossed_in'] == False and y1 < 640 and y2 >= 640 and 680 <= x1 <= 1280:
                            car_in_count += 1
                            tracked_cars[car_id]['crossed_in'] = True
                        
                        break

                if not car_found:
                    tracked_cars[next_car_id] = {'bbox': bbox, 'crossed_out': False, 'crossed_in': False}
                    next_car_id += 1

                current_frame_cars.append(bbox)

                # Draw bounding box
                cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
                # Put label
                label = f'{model.names[cls_id]} {obj.conf[0]:.2f}'
                cv2.putText(frame, label, (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)

    # Remove cars not detected in the current frame from the tracking dictionary
    tracked_cars = {car_id: car_data for car_id, car_data in tracked_cars.items() if car_data['bbox'] in current_frame_cars}

    # Display car-out count on the frame (upper left)
    car_out_label = f'Mobil Keluar: {car_out_count}'
    cv2.putText(frame, car_out_label, (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (139, 0 , 0), 2)

    # Display car-in count on the frame (upper right)
    car_in_label = f'Mobil Masuk: {car_in_count}'
    cv2.putText(frame, car_in_label, (frame.shape[1] - 300, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (139, 0 , 0), 2)

    # Draw the y-coordinate line
    cv2.line(frame, (0, 640), (frame.shape[1], 640), (0, 0, 255), 3)

    # Display the frame with detections
    cv2.imshow('YOLOv8 Object Detection', frame)

    # Break the loop if 'q' key is pressed
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

# Release the video capture and close windows
cap.release()
cv2.destroyAllWindows()

tracker

Python
additional python code library to track bounding boxes. put this file in the same folder with your main python program
import math


class Tracker:
    def __init__(self):
        # Store the center positions of the objects
        self.center_points = {}
        # Keep the count of the IDs
        # each time a new object id detected, the count will increase by one
        self.id_count = 0


    def update(self, objects_rect):
        # Objects boxes and ids
        objects_bbs_ids = []

        # Get center point of new object
        for rect in objects_rect:
            x, y, w, h = rect
            cx = (x + x + w) // 2
            cy = (y + y + h) // 2

            # Find out if that object was detected already
            same_object_detected = False
            for id, pt in self.center_points.items():
                dist = math.hypot(cx - pt[0], cy - pt[1])

                if dist < 35:
                    self.center_points[id] = (cx, cy)
#                    print(self.center_points)
                    objects_bbs_ids.append([x, y, w, h, id])
                    same_object_detected = True
                    break

            # New object is detected we assign the ID to that object
            if same_object_detected is False:
                self.center_points[self.id_count] = (cx, cy)
                objects_bbs_ids.append([x, y, w, h, self.id_count])
                self.id_count += 1

        # Clean the dictionary by center points to remove IDS not used anymore
        new_center_points = {}
        for obj_bb_id in objects_bbs_ids:
            _, _, _, _, object_id = obj_bb_id
            center = self.center_points[object_id]
            new_center_points[object_id] = center

        # Update dictionary with IDs not used removed
        self.center_points = new_center_points.copy()
        return objects_bbs_ids

speedncount

Python
Python code with YOLOv8 to track speed, count, direction of cars
import cv2
import pandas as pd
import numpy as np
from ultralytics import YOLO
from tracker import Tracker
import time
from math import dist

model = YOLO('yolov8s.pt')

def RGB(event, x, y, flags, param):
    if event == cv2.EVENT_MOUSEMOVE:
        colorsBGR = [x, y]
        print(colorsBGR)

cv2.namedWindow('RGB')
cv2.setMouseCallback('RGB', RGB)

# Video path
video_path = r'C:\Users\s4mue\Documents\yolo-cars\YOLOv8\carvideo4.mp4'

# Video capture
cap = cv2.VideoCapture(video_path)

my_file = open("coco.txt", "r")
data = my_file.read()
class_list = data.split("\n")

count = 0
tracker = Tracker()

cy1 = 300
cy2 = 360
offset = 4

vh_down = {}
counter = []

vh_up = {}
counter1 = []

vehicle_speeds = {}  # Dictionary to store calculated speeds
speed_display_timestamps = {}  # Dictionary to store timestamps for speed display

while True:
    ret, frame = cap.read()
    if not ret:
        break
    count += 1
    if count % 3 != 0:
        continue
    frame = cv2.resize(frame, (1020, 500))

    results = model.predict(frame)
    a = results[0].boxes.data
    px = pd.DataFrame(a).astype("float")
    list = []

    for index, row in px.iterrows():
        x1 = int(row[0])
        y1 = int(row[1])
        x2 = int(row[2])
        y2 = int(row[3])
        d = int(row[5])
        c = class_list[d]
        if 'car' in c:
            list.append([x1, y1, x2, y2])
    bbox_id = tracker.update(list)
    for bbox in bbox_id:
        x3, y3, x4, y4, id = bbox
        cx = int(x3 + x4) // 2
        cy = int(y3 + y4) // 2

        cv2.rectangle(frame, (x3, y3), (x4, y4), (0, 255, 0), 1)

        # Vehicle going down
        if cy2 < (cy + offset) and cy2 > (cy - offset):
            vh_down[id] = time.time()
        if id in vh_down:
            if cy1 < (cy + offset) and cy1 > (cy - offset):
                elapsed_time = time.time() - vh_down[id]
                if counter.count(id) == 0:
                    counter.append(id)
                    distance = 20  # meters
                    a_speed_ms = distance / elapsed_time
                    a_speed_kh = a_speed_ms * 3.6
                    vehicle_speeds[id] = a_speed_kh
                    speed_display_timestamps[id] = time.time()  # Store timestamp for speed display

        # Vehicle going up
        if cy1 < (cy + offset) and cy1 > (cy - offset):
            vh_up[id] = time.time()
        if id in vh_up:
            if cy2 < (cy + offset) and cy2 > (cy - offset):
                elapsed1_time = time.time() - vh_up[id]
                if counter1.count(id) == 0:
                    counter1.append(id)
                    distance1 = 15  # meters
                    a_speed_ms1 = distance1 / elapsed1_time
                    a_speed_kh1 = a_speed_ms1 * 3.6
                    vehicle_speeds[id] = a_speed_kh1
                    speed_display_timestamps[id] = time.time()  # Store timestamp for speed display

        # Display speed information with bounding box for 2 seconds
        if id in vehicle_speeds and time.time() - speed_display_timestamps[id] <= 2:
            speed_text = f"{int(vehicle_speeds[id])} kph"
            cv2.putText(frame, speed_text, (x3, y3 - 20), cv2.FONT_HERSHEY_COMPLEX, 0.7, (10, 255, 0), 2)

           

    cv2.line(frame,(274,cy1),(814,cy1),(200,200,150),1)

    cv2.putText(frame,('L1'),(277,295),cv2.FONT_HERSHEY_COMPLEX,0.5,(150,150,255),1)


    cv2.line(frame,(177,cy2),(927,cy2),(200,200,150),1)
 
    cv2.putText(frame,('L2'),(180,380),cv2.FONT_HERSHEY_COMPLEX,0.5,(150,150,255),1)
    d=(len(counter))
    u=(len(counter1))
    cv2.putText(frame,('keluar: ')+str(d),(40,60),cv2.FONT_HERSHEY_COMPLEX,0.6,(0,255,0),2)

    cv2.putText(frame,('masuk: ')+str(u),(40,100),cv2.FONT_HERSHEY_COMPLEX,0.6,(0,255,0),2)
    cv2.imshow("RGB", frame)
    if cv2.waitKey(1)&0xFF==27:
        break
cap.release()
cv2.destroyAllWindows()

speedncount_amd

Python
Modified speedncount python code using Huggingface's optimum-amd Ryzen Ai optimized models
import cv2
import pandas as pd
import numpy as np
from optimum.amd.ryzenai import pipeline  # Use Huggingface optimum pipeline
from tracker import Tracker
import time
from math import dist

model_id = "amd/yolov8m"
# Load Huggingface object detection pipeline
# model = pipeline('object-detection', model='amd/yolox-s')
detector = pipeline("object-detection", model=model_id, model_type="yolov8")

def RGB(event, x, y, flags, param):
    if event == cv2.EVENT_MOUSEMOVE:
        colorsBGR = [x, y]
        print(colorsBGR)

cv2.namedWindow('RGB')
cv2.setMouseCallback('RGB', RGB)

# Video path
video_path = r'C:\Users\s4mue\Documents\yolo-cars\YOLOv8\optimum-amd\carvideo4.mp4'

# Video capture
cap = cv2.VideoCapture(video_path)

my_file = open("coco.txt", "r")
data = my_file.read()
class_list = data.split("\n")

count = 0
tracker = Tracker()

cy1 = 300
cy2 = 360
offset = 4

vh_down = {}
counter = []

vh_up = {}
counter1 = []

vehicle_speeds = {}  # Dictionary to store calculated speeds
speed_display_timestamps = {}  # Dictionary to store timestamps for speed display

while True:
    ret, frame = cap.read()
    if not ret:
        break
    count += 1
    if count % 3 != 0:
        continue
    frame = cv2.resize(frame, (1020, 500))

    # Perform detection
    results = detector(frame)

    list = []

    for result in results:
        if result['label'] == 'car':
            x1, y1, x2, y2 = map(int, result['box'])
            list.append([x1, y1, x2, y2])
    
    bbox_id = tracker.update(list)
    for bbox in bbox_id:
        x3, y3, x4, y4, id = bbox
        cx = int(x3 + x4) // 2
        cy = int(y3 + y4) // 2

        cv2.rectangle(frame, (x3, y3), (x4, y4), (0, 255, 0), 1)

        # Vehicle going down
        if cy2 < (cy + offset) and cy2 > (cy - offset):
            vh_down[id] = time.time()
        if id in vh_down:
            if cy1 < (cy + offset) and cy1 > (cy - offset):
                elapsed_time = time.time() - vh_down[id]
                if counter.count(id) == 0:
                    counter.append(id)
                    distance = 20  # meters
                    a_speed_ms = distance / elapsed_time
                    a_speed_kh = a_speed_ms * 3.6
                    vehicle_speeds[id] = a_speed_kh
                    speed_display_timestamps[id] = time.time()  # Store timestamp for speed display

        # Vehicle going up
        if cy1 < (cy + offset) and cy1 > (cy - offset):
            vh_up[id] = time.time()
        if id in vh_up:
            if cy2 < (cy + offset) and cy2 > (cy - offset):
                elapsed1_time = time.time() - vh_up[id]
                if counter1.count(id) == 0:
                    counter1.append(id)
                    distance1 = 15  # meters
                    a_speed_ms1 = distance1 / elapsed1_time
                    a_speed_kh1 = a_speed_ms1 * 3.6
                    vehicle_speeds[id] = a_speed_kh1
                    speed_display_timestamps[id] = time.time()  # Store timestamp for speed display

        # Display speed information with bounding box for 2 seconds
        if id in vehicle_speeds and time.time() - speed_display_timestamps[id] <= 2:
            speed_text = f"{int(vehicle_speeds[id])} kph"
            cv2.putText(frame, speed_text, (x3, y3 - 20), cv2.FONT_HERSHEY_COMPLEX, 0.7, (10, 255, 0), 2)

    cv2.line(frame, (274, cy1), (814, cy1), (200, 200, 150), 1)
    cv2.putText(frame, 'L1', (277, 295), cv2.FONT_HERSHEY_COMPLEX, 0.5, (150, 150, 255), 1)

    cv2.line(frame, (177, cy2), (927, cy2), (200, 200, 150), 1)
    cv2.putText(frame, 'L2', (180, 380), cv2.FONT_HERSHEY_COMPLEX, 0.5, (150, 150, 255), 1)
    d = len(counter)
    u = len(counter1)
    cv2.putText(frame, 'keluar: ' + str(d), (40, 60), cv2.FONT_HERSHEY_COMPLEX, 0.6, (0, 255, 0), 2)
    cv2.putText(frame, 'masuk: ' + str(u), (40, 100), cv2.FONT_HERSHEY_COMPLEX, 0.6, (0, 255, 0), 2)
    
    cv2.imshow("RGB", frame)
    if cv2.waitKey(1) & 0xFF == 27:
        break

cap.release()
cv2.destroyAllWindows()

Credits

Jallson Suryo

Jallson Suryo

5 projects • 21 followers
Tech integrator for schools. Also works as a maker and his activities include disassembling, fixing, and making things.

Comments