Reinforcement Learning-Based Adaptive Traffic Signal ControlΒΆ
Student: Shahriar Khan
Supervisor: Dr. Seena Joseph
YOLOv8 + DQN + Emergency Vehicle Priority + SUMO
Dataset DistributionΒΆ
InΒ [58]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("labels.jpg")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
YOLOv8 Training BehaviourΒΆ
InΒ [60]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("results.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
Training was performed for 30 epochs.
The model achieved:
Precision = 42.59% Recall = 33.05% mAP@50 = 34.50%
Testing ResultsΒΆ
InΒ [61]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("confusion_matrix.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [47]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("confusion_matrix.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [48]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("BoxPR_curve.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [62]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("BoxF1_curve.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
YOLOv8 Testing ResultsΒΆ
InΒ [63]:
import pandas as pd
pd.DataFrame({
"Metric":[
"Precision",
"Recall",
"mAP@50",
"mAP@50-95"
],
"Value":[
"42.59%",
"33.05%",
"34.50%",
"19.86%"
]
})
Out[63]:
| Metric | Value | |
|---|---|---|
| 0 | Precision | 42.59% |
| 1 | Recall | 33.05% |
| 2 | mAP@50 | 34.50% |
| 3 | mAP@50-95 | 19.86% |
Validation Detection ExamplesΒΆ
InΒ [64]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("val_batch0_labels.jpg")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [65]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("val_batch0_pred.jpg")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [66]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("val_batch1_labels.jpg")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [67]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("val_batch1_pred.jpg")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
Fixed-Time Traffic Control ResultsΒΆ
InΒ [68]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Figure_1_FT.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [69]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Figure_2_FT.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [70]:
ft = pd.DataFrame({
"Metric":["Waiting Time","Queue Length","Throughput"],
"Value":[14.28,6.14,1315]
})
ft
Out[70]:
| Metric | Value | |
|---|---|---|
| 0 | Waiting Time | 14.28 |
| 1 | Queue Length | 6.14 |
| 2 | Throughput | 1315.00 |
Q-Learning ResultsΒΆ
InΒ [71]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Q1.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [72]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Q2.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
DQN Controller ResultsΒΆ
InΒ [49]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Figure_1_DQN_10000.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [50]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Figure_2_DQN_10000.png")
plt.figure(figsize=(10,7))
plt.imshow(img)
plt.axis("off")
plt.title("Dataset Distribution")
plt.show()
InΒ [73]:
dqn = pd.DataFrame({
"Metric":["Waiting Time","Queue Length","Throughput"],
"Value":[2.80,5.32,1205]
})
dqn
Out[73]:
| Metric | Value | |
|---|---|---|
| 0 | Waiting Time | 2.80 |
| 1 | Queue Length | 5.32 |
| 2 | Throughput | 1205.00 |
Emergency Vehicle ResultsΒΆ
InΒ [74]:
ev = pd.DataFrame({
"Metric":[
"Travel Time",
"Delay"
],
"Value":[
21.05,
11.05
]
})
ev
Out[74]:
| Metric | Value | |
|---|---|---|
| 0 | Travel Time | 21.05 |
| 1 | Delay | 11.05 |
Final ComparisonΒΆ
InΒ [75]:
comparison = pd.DataFrame({
"Metric":[
"Waiting Time",
"Queue Length",
"Throughput"
],
"Fixed-Time":[
14.28,
6.14,
1315
],
"DQN":[
2.80,
5.32,
1205
],
"Improvement (%)":[
80.40,
13.36,
-8.37
]
})
comparison
Out[75]:
| Metric | Fixed-Time | DQN | Improvement (%) | |
|---|---|---|---|---|
| 0 | Waiting Time | 14.28 | 2.80 | 80.40 |
| 1 | Queue Length | 6.14 | 5.32 | 13.36 |
| 2 | Throughput | 1315.00 | 1205.00 | -8.37 |
ConclusionsΒΆ
YOLOv8 successfully generated traffic-state information.
The DQN controller reduced average waiting time by 80.40%.
Average queue length decreased by 13.36%.
Emergency vehicle prioritisation reduced interruptions and improved traffic management performance.
Source Code ImplementationΒΆ
Dataset Annotation ToolΒΆ
A custom annotation application was developed to generate YOLO-compatible labels for vehicle and emergency vehicle detection.
InΒ [76]:
from tkinter import *
from PIL import Image, ImageTk
import cv2
class AnnotationTool:
def load_image(self):
pass
def draw_bbox(self):
pass
def save_label(self):
pass
InΒ [77]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("annotation_tool.png")
plt.imshow(img)
plt.axis("off")
plt.show()
YOLOv8 Vehicle DetectionΒΆ
InΒ [Β ]:
from ultralytics import YOLO
model = YOLO("best.pt")
results = model.predict(
source=image_path,
conf=0.25
)
InΒ [80]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("val_batch0_pred.jpg")
plt.imshow(img)
plt.axis("off")
plt.show()
InΒ [81]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("val_batch1_pred.jpg")
plt.imshow(img)
plt.axis("off")
plt.show()
Traffic State AcquisitionΒΆ
InΒ [Β ]:
state = [
queue_eb0,
queue_eb1,
queue_eb2,
queue_sb0,
queue_sb1,
queue_sb2,
current_phase
]
The state vector contains six lane queue observations and the current signal phase.
DQN ControllerΒΆ
InΒ [Β ]:
action = np.argmax(
model.predict(state)[0]
)
InΒ [86]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("DQN final.png")
plt.imshow(img)
plt.axis("off")
plt.show()
Reward FunctionΒΆ
InΒ [Β ]:
reward = -sum(queue_lengths)
The reward function penalises queue accumulation. Lower queue lengths generate higher rewards.
Emergency Vehicle PriorityΒΆ
InΒ [Β ]:
if vehicle_type == "ambulance":
activate_priority()
elif vehicle_type == "fire":
activate_priority()
elif vehicle_type == "police":
activate_priority()
InΒ [87]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("Emergency Vehicle priority design.drawio.png")
plt.imshow(img)
plt.axis("off")
plt.show()
SUMOβTraCI IntegrationΒΆ
InΒ [Β ]:
import traci
traci.start([
"sumo-gui",
"-c",
"simulation.sumocfg"
])
traci.simulationStep()
InΒ [88]:
from PIL import Image
import matplotlib.pyplot as plt
img = Image.open("sumo.png")
plt.imshow(img)
plt.axis("off")
plt.show()
InΒ [90]:
with open("annotation_tool.py", "r", encoding="utf-8") as f:
print(f.read())
import customtkinter as ctk
from tkinter import filedialog, messagebox, Canvas
import cv2
import os
import sys
import subprocess
from PIL import Image, ImageTk
import numpy as np
import threading
# Set appearance mode and color theme
ctk.set_appearance_mode("dark")
ctk.set_default_color_theme("blue")
class YOLOAnnotator:
def __init__(self, root):
self.root = root
self.root.title("SimJam ComputerVision - YOLO Annotator")
self.root.geometry("1400x900")
# Variables
self.image_path = None
self.output_dir = None
self.original_img = None
self.display_img = None
self.rectangles = []
self.drawing = False
self.ix, self.iy = -1, -1
self.current_class_id = 3
self.scale_factor = 1.0
# Class names mapping
self.class_names = {
0: "pedestrian",
1: "people",
2: "bicycle",
3: "car",
4: "EV",
5: "truck",
6: "tricycle",
7: "awning-tricycle",
8: "bus",
9: "motor"
}
self.setup_ui()
def setup_ui(self):
# Main container
main_container = ctk.CTkFrame(self.root, fg_color="transparent")
main_container.pack(fill="both", expand=True)
# Header Frame with Logo and About/Dependencies buttons
self.create_header(main_container)
# Control Panel Frame
self.create_control_panel(main_container)
# Canvas Frame for image display
self.create_canvas_frame(main_container)
def create_header(self, parent):
"""Create header with logo and buttons"""
header_frame = ctk.CTkFrame(parent, height=60, corner_radius=0)
header_frame.pack(fill="x", padx=0, pady=(0, 10))
header_frame.pack_propagate(False)
# Left side: Logo and Title
left_frame = ctk.CTkFrame(header_frame, fg_color="transparent")
left_frame.pack(side="left", padx=15, pady=10)
# Try to load logo
logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "RoadwayVR.jpg")
if os.path.exists(logo_path):
try:
logo_image = Image.open(logo_path)
logo_image = logo_image.resize((40, 40), Image.Resampling.LANCZOS)
self.logo_photo = ctk.CTkImage(light_image=logo_image, dark_image=logo_image, size=(40, 40))
logo_label = ctk.CTkLabel(left_frame, image=self.logo_photo, text="")
logo_label.pack(side="left", padx=(0, 12))
except Exception as e:
print(f"Could not load logo: {e}")
# Title
title_label = ctk.CTkLabel(left_frame, text="SimJam ComputerVision - YOLO Annotator",
font=ctk.CTkFont(size=18, weight="bold"))
title_label.pack(side="left")
# Right side: About and Dependencies buttons
right_frame = ctk.CTkFrame(header_frame, fg_color="transparent")
right_frame.pack(side="right", padx=15, pady=10)
about_btn = ctk.CTkButton(right_frame, text="βΉ About", command=self.show_about,
width=100, height=32)
about_btn.pack(side="left", padx=(0, 10))
dependencies_btn = ctk.CTkButton(right_frame, text="π§ Dependencies",
command=self.show_dependencies, width=140, height=32)
dependencies_btn.pack(side="left")
def create_control_panel(self, parent):
"""Create control panel with all buttons"""
control_frame = ctk.CTkFrame(parent, height=80, corner_radius=0)
control_frame.pack(fill="x", padx=0, pady=(0, 10))
control_frame.pack_propagate(False)
# Container for buttons (centered)
button_container = ctk.CTkFrame(control_frame, fg_color="transparent")
button_container.pack(expand=True)
# Load Image Button
self.btn_load = ctk.CTkButton(button_container, text="π Load Image",
command=self.load_image, width=140, height=40)
self.btn_load.pack(side="left", padx=8)
# Set Output Directory Button
self.btn_output = ctk.CTkButton(button_container, text="π Set Output",
command=self.set_output_directory, width=140, height=40,
fg_color="#9333EA", hover_color="#7C3AED")
self.btn_output.pack(side="left", padx=8)
# Separator
sep1 = ctk.CTkFrame(button_container, width=2, height=40, fg_color="gray30")
sep1.pack(side="left", padx=15)
# Class Selection
class_frame = ctk.CTkFrame(button_container, fg_color="transparent")
class_frame.pack(side="left", padx=8)
ctk.CTkLabel(class_frame, text="Class:", font=ctk.CTkFont(size=12, weight="bold")).pack(side="left", padx=(0, 8))
self.class_var = ctk.StringVar(value="3 - car")
class_dropdown = ctk.CTkComboBox(class_frame, variable=self.class_var,
values=[f"{k} - {v}" for k, v in self.class_names.items()],
state="readonly", width=180, height=40,
command=self.on_class_change)
class_dropdown.pack(side="left")
# Separator
sep2 = ctk.CTkFrame(button_container, width=2, height=40, fg_color="gray30")
sep2.pack(side="left", padx=15)
# Undo Button
self.btn_undo = ctk.CTkButton(button_container, text="βΆ Undo",
command=self.undo_last, width=120, height=40,
fg_color="#F97316", hover_color="#EA580C")
self.btn_undo.pack(side="left", padx=8)
# Clear All Button
self.btn_clear = ctk.CTkButton(button_container, text="ποΈ Clear",
command=self.clear_all, width=120, height=40,
fg_color="#EF4444", hover_color="#DC2626")
self.btn_clear.pack(side="left", padx=8)
# Save Button
self.btn_save = ctk.CTkButton(button_container, text="πΎ Save Labels",
command=self.save_labels, width=140, height=40,
fg_color="#10B981", hover_color="#059669")
self.btn_save.pack(side="left", padx=8)
# Box Count Label
self.count_label = ctk.CTkLabel(button_container, text="0",
font=ctk.CTkFont(size=18, weight="bold"),
fg_color="#10B981", corner_radius=8,
width=60, height=40)
self.count_label.pack(side="left", padx=20)
def create_canvas_frame(self, parent):
"""Create canvas for image display"""
canvas_frame = ctk.CTkFrame(parent, corner_radius=0)
canvas_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10))
# Use tkinter Canvas (not CTkCanvas) for image display and mouse events
self.canvas = Canvas(canvas_frame, bg="#1a1a1a", highlightthickness=0)
self.canvas.pack(fill="both", expand=True, padx=5, pady=5)
# Bind mouse events
self.canvas.bind("<ButtonPress-1>", self.on_mouse_down)
self.canvas.bind("<B1-Motion>", self.on_mouse_move)
self.canvas.bind("<ButtonRelease-1>", self.on_mouse_up)
def on_class_change(self, choice):
"""Handle class selection change"""
self.current_class_id = int(choice.split(" - ")[0])
def load_image(self):
"""Load an image file"""
file_path = filedialog.askopenfilename(
title="Select Image",
filetypes=[("Image files", "*.jpg *.jpeg *.png *.bmp"), ("All files", "*.*")]
)
if file_path:
self.image_path = file_path
self.original_img = cv2.imread(file_path)
if self.original_img is None:
messagebox.showerror("Error", "Failed to load image!")
return
# Clear previous annotations
self.rectangles = []
self.update_count()
# Calculate scale to fit canvas
canvas_width = self.canvas.winfo_width()
canvas_height = self.canvas.winfo_height()
if canvas_width > 1 and canvas_height > 1:
img_height, img_width = self.original_img.shape[:2]
scale_w = canvas_width / img_width
scale_h = canvas_height / img_height
self.scale_factor = min(scale_w, scale_h, 1.0) * 0.95
else:
self.scale_factor = 0.8
self.display_image()
def set_output_directory(self):
"""Set output directory for labels"""
directory = filedialog.askdirectory(title="Select Output Directory")
if directory:
self.output_dir = directory
messagebox.showinfo("Success", f"Output directory set to:\n{directory}")
def display_image(self):
"""Display image on canvas"""
if self.original_img is None:
return
img_copy = self.original_img.copy()
# Draw all rectangles
for rect in self.rectangles:
xmin, ymin, xmax, ymax, class_id = rect
color = self.get_class_color(class_id)
cv2.rectangle(img_copy, (xmin, ymin), (xmax, ymax), color, 2)
# Draw label
label = f"{class_id}: {self.class_names[class_id]}"
(text_width, text_height), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
cv2.rectangle(img_copy, (xmin, ymin - text_height - 8),
(xmin + text_width + 8, ymin), color, -1)
cv2.putText(img_copy, label, (xmin + 4, ymin - 4),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# Convert and resize
img_rgb = cv2.cvtColor(img_copy, cv2.COLOR_BGR2RGB)
pil_img = Image.fromarray(img_rgb)
new_width = int(img_copy.shape[1] * self.scale_factor)
new_height = int(img_copy.shape[0] * self.scale_factor)
pil_img = pil_img.resize((new_width, new_height), Image.Resampling.LANCZOS)
self.display_img = ImageTk.PhotoImage(pil_img)
self.canvas.delete("all")
self.canvas.create_image(0, 0, anchor="nw", image=self.display_img)
def get_class_color(self, class_id):
"""Get color for each class"""
colors = [
(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0),
(255, 0, 255), (0, 255, 255), (128, 0, 128), (255, 165, 0),
(0, 128, 128), (128, 128, 0)
]
return colors[class_id % len(colors)]
def on_mouse_down(self, event):
"""Handle mouse button press"""
if self.original_img is None:
return
self.drawing = True
self.ix = int(event.x / self.scale_factor)
self.iy = int(event.y / self.scale_factor)
def on_mouse_move(self, event):
"""Handle mouse movement while drawing"""
if not self.drawing or self.original_img is None:
return
x = int(event.x / self.scale_factor)
y = int(event.y / self.scale_factor)
# Constrain to image bounds
h, w = self.original_img.shape[:2]
x = max(0, min(x, w))
y = max(0, min(y, h))
# Draw preview
img_copy = self.original_img.copy()
# Draw existing rectangles
for rect in self.rectangles:
xmin, ymin, xmax, ymax, class_id = rect
color = self.get_class_color(class_id)
cv2.rectangle(img_copy, (xmin, ymin), (xmax, ymax), color, 2)
label = f"{class_id}: {self.class_names[class_id]}"
(text_width, text_height), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
cv2.rectangle(img_copy, (xmin, ymin - text_height - 8),
(xmin + text_width + 8, ymin), color, -1)
cv2.putText(img_copy, label, (xmin + 4, ymin - 4),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# Draw current rectangle
color = self.get_class_color(self.current_class_id)
cv2.rectangle(img_copy, (self.ix, self.iy), (x, y), color, 3)
# Update display
img_rgb = cv2.cvtColor(img_copy, cv2.COLOR_BGR2RGB)
pil_img = Image.fromarray(img_rgb)
new_width = int(img_copy.shape[1] * self.scale_factor)
new_height = int(img_copy.shape[0] * self.scale_factor)
pil_img = pil_img.resize((new_width, new_height), Image.Resampling.LANCZOS)
self.display_img = ImageTk.PhotoImage(pil_img)
self.canvas.delete("all")
self.canvas.create_image(0, 0, anchor="nw", image=self.display_img)
def on_mouse_up(self, event):
"""Handle mouse button release"""
if not self.drawing or self.original_img is None:
return
self.drawing = False
x = int(event.x / self.scale_factor)
y = int(event.y / self.scale_factor)
# Constrain to image bounds
h, w = self.original_img.shape[:2]
x = max(0, min(x, w))
y = max(0, min(y, h))
# Create rectangle
xmin = min(self.ix, x)
ymin = min(self.iy, y)
xmax = max(self.ix, x)
ymax = max(self.iy, y)
# Only add if rectangle has size
if xmax - xmin > 5 and ymax - ymin > 5:
self.rectangles.append([xmin, ymin, xmax, ymax, self.current_class_id])
self.update_count()
self.display_image()
def undo_last(self):
"""Remove last annotation"""
if self.rectangles:
self.rectangles.pop()
self.update_count()
self.display_image()
def clear_all(self):
"""Clear all annotations"""
if self.rectangles:
if messagebox.askyesno("Confirm", "Clear all annotations?"):
self.rectangles = []
self.update_count()
self.display_image()
def update_count(self):
"""Update box count display"""
self.count_label.configure(text=str(len(self.rectangles)))
def save_labels(self):
"""Save annotations in YOLO format"""
if self.original_img is None:
messagebox.showwarning("Warning", "No image loaded!")
return
if not self.rectangles:
messagebox.showwarning("Warning", "No annotations to save!")
return
if not self.output_dir:
messagebox.showwarning("Warning", "Please set output directory first!")
return
# Get image dimensions
h, w = self.original_img.shape[:2]
# Get base filename
base = os.path.splitext(os.path.basename(self.image_path))[0]
# Write YOLO format labels
label_path = os.path.join(self.output_dir, f"{base}.txt")
with open(label_path, 'w') as f:
for rect in self.rectangles:
xmin, ymin, xmax, ymax, class_id = rect
# Convert to YOLO format (normalized)
x_center = ((xmin + xmax) / 2) / w
y_center = ((ymin + ymax) / 2) / h
bw = (xmax - xmin) / w
bh = (ymax - ymin) / h
f.write(f"{class_id} {x_center:.6f} {y_center:.6f} {bw:.6f} {bh:.6f}\n")
messagebox.showinfo("Success", f"Saved {len(self.rectangles)} annotations!\n\n{base}.txt")
def show_about(self):
"""Show About dialog"""
about_window = ctk.CTkToplevel(self.root)
about_window.title("About")
about_window.geometry("350x220")
about_window.resizable(False, False)
about_window.transient(self.root)
about_window.grab_set()
# Content frame
content_frame = ctk.CTkFrame(about_window, fg_color="transparent")
content_frame.pack(fill="both", expand=True, padx=25, pady=20)
# Title
ctk.CTkLabel(content_frame, text="About",
font=ctk.CTkFont(size=18, weight="bold")).pack(pady=(0, 12))
# Developed by
ctk.CTkLabel(content_frame, text="Developed by RoadwayVR",
font=ctk.CTkFont(size=13, weight="bold"),
text_color="lightblue").pack(pady=(0, 10))
# Application
ctk.CTkLabel(content_frame, text="Application:",
font=ctk.CTkFont(size=10), text_color="gray60").pack(pady=(0, 2))
ctk.CTkLabel(content_frame, text="SimJamCV Annotator",
font=ctk.CTkFont(size=12, weight="bold")).pack(pady=(0, 8))
# Version
ctk.CTkLabel(content_frame, text="Version: 1.0",
font=ctk.CTkFont(size=11)).pack(pady=(0, 8))
# Description
ctk.CTkLabel(content_frame, text="YOLO format annotation tool\nfor object detection datasets",
font=ctk.CTkFont(size=10), text_color="gray60",
justify="center").pack(pady=(0, 12))
# Close button
ctk.CTkButton(content_frame, text="Close", command=about_window.destroy,
width=100, height=28).pack()
def show_dependencies(self):
"""Show Dependencies dialog with version-specific checking"""
dep_window = ctk.CTkToplevel(self.root)
dep_window.title("Dependency Manager")
dep_window.geometry("700x500")
dep_window.resizable(False, False)
dep_window.transient(self.root)
dep_window.grab_set()
# Main frame
main_frame = ctk.CTkFrame(dep_window)
main_frame.pack(fill="both", expand=True, padx=20, pady=20)
# Header
header_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
header_frame.pack(fill="x", pady=(0, 15))
ctk.CTkLabel(header_frame, text="π¦ Package Dependencies",
font=ctk.CTkFont(size=18, weight="bold")).pack(side="left")
ctk.CTkLabel(header_frame, text="Check and install required Python packages",
text_color="gray60").pack(side="left", padx=(15, 0))
# Console
console_frame = ctk.CTkFrame(main_frame)
console_frame.pack(fill="both", expand=True, pady=(0, 15))
console = ctk.CTkTextbox(console_frame, font=ctk.CTkFont(family="Courier", size=11),
wrap="word")
console.pack(fill="both", expand=True, padx=10, pady=10)
def log_message(message):
console.configure(state="normal")
console.insert("end", message + "\n")
console.see("end")
console.configure(state="disabled")
dep_window.update()
# Buttons
button_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
button_frame.pack(fill="x")
check_btn = ctk.CTkButton(button_frame, text="π Check Status", width=150)
check_btn.pack(side="left", padx=(0, 10))
install_btn = ctk.CTkButton(button_frame, text="β¬ Install/Update", width=150,
fg_color="green", hover_color="darkgreen")
install_btn.pack(side="left", padx=(0, 10))
close_btn = ctk.CTkButton(button_frame, text="Close", width=100,
command=dep_window.destroy)
close_btn.pack(side="right")
# Required packages with EXACT versions
# Format: (import_name, pip_package_name, exact_version, description)
REQUIRED_PACKAGES = [
('customtkinter', 'customtkinter', '5.2.2', 'Modern GUI framework'),
('cv2', 'opencv-python', '4.10.0', 'Computer vision library'),
('PIL', 'pillow', '11.3.0', 'Image processing library'),
('numpy', 'numpy', '2.2.6', 'Numerical computing'),
]
def get_version(module_name):
"""Get installed version of a module"""
try:
import importlib
mod = importlib.import_module(module_name)
return getattr(mod, '__version__', 'unknown')
except:
return None
def check_dependencies():
console.configure(state="normal")
console.delete("1.0", "end")
console.configure(state="disabled")
log_message("Checking dependencies...\n")
log_message("DEPENDENCY STATUS")
log_message("=" * 60 + "\n")
installed = []
missing = []
wrong_version = []
for import_name, pip_name, exact_version, desc in REQUIRED_PACKAGES:
try:
__import__(import_name)
current_ver = get_version(import_name)
if current_ver and current_ver != exact_version and current_ver != 'unknown':
wrong_version.append((import_name, pip_name, current_ver, exact_version, desc))
log_message(f"β {pip_name:<20} v{current_ver} β v{exact_version}")
else:
installed.append((pip_name, exact_version, desc))
log_message(f"β {pip_name:<20} v{exact_version} - {desc}")
except ImportError:
missing.append((import_name, pip_name, exact_version, desc))
log_message(f"β {pip_name:<20} v{exact_version} - {desc}")
if wrong_version:
log_message("\nβ WRONG VERSION (Will be updated):")
for imp, pkg, curr, target, desc in wrong_version:
log_message(f" β’ {pkg}: v{curr} β v{target}")
if missing:
log_message("\nβ MISSING (Will be installed):")
for imp, pkg, ver, desc in missing:
log_message(f" β’ {pkg} v{ver}")
if not missing and not wrong_version:
log_message("\nπ All dependencies are correctly installed!")
install_btn.configure(state="disabled")
else:
total_issues = len(missing) + len(wrong_version)
log_message(f"\nβ‘ Click 'Install/Update' to fix {total_issues} package(s)")
install_btn.configure(state="normal")
return installed, missing, wrong_version
def install_missing():
def install_worker():
install_btn.configure(state="disabled")
check_btn.configure(state="disabled")
log_message("\n" + "=" * 60)
log_message("INSTALLING/UPDATING PACKAGES")
log_message("=" * 60 + "\n")
# Get current status
installed_list = []
to_install = []
for import_name, pip_name, exact_version, desc in REQUIRED_PACKAGES:
try:
__import__(import_name)
current_ver = get_version(import_name)
if current_ver != exact_version and current_ver != 'unknown':
to_install.append((pip_name, exact_version, "UPDATE"))
except ImportError:
to_install.append((pip_name, exact_version, "INSTALL"))
if not to_install:
log_message("β All packages already at correct versions!")
check_btn.configure(state="normal")
install_btn.configure(state="normal")
return
total = len(to_install)
for i, (package, version, action) in enumerate(to_install, 1):
log_message(f"[{action}] {i}/{total}: {package} v{version}...")
try:
result = subprocess.run(
[sys.executable, "-m", "pip", "install",
f"{package}=={version}", "--force-reinstall", "--no-deps"],
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
log_message(f" β {package} v{version} installed successfully")
else:
# Try without --no-deps
result = subprocess.run(
[sys.executable, "-m", "pip", "install",
f"{package}=={version}", "--force-reinstall"],
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
log_message(f" β {package} v{version} installed successfully")
else:
log_message(f" β Failed to install {package}")
except Exception as e:
log_message(f" β Error: {str(e)}")
log_message("\n" + "=" * 60)
log_message("Verifying installation...")
log_message("")
check_dependencies()
log_message("\nβ IMPORTANT: You may need to restart the application.")
check_btn.configure(state="normal")
install_btn.configure(state="normal")
threading.Thread(target=install_worker, daemon=True).start()
check_btn.configure(command=check_dependencies)
install_btn.configure(command=install_missing)
# Initial check
check_dependencies()
if __name__ == "__main__":
root = ctk.CTk()
app = YOLOAnnotator(root)
root.mainloop()
InΒ [91]:
with open("annotation_tool.py", "r", encoding="utf-8") as f:
print(f.read())
import customtkinter as ctk
from tkinter import filedialog, messagebox, Canvas
import cv2
import os
import sys
import subprocess
from PIL import Image, ImageTk
import numpy as np
import threading
# Set appearance mode and color theme
ctk.set_appearance_mode("dark")
ctk.set_default_color_theme("blue")
class YOLOAnnotator:
def __init__(self, root):
self.root = root
self.root.title("SimJam ComputerVision - YOLO Annotator")
self.root.geometry("1400x900")
# Variables
self.image_path = None
self.output_dir = None
self.original_img = None
self.display_img = None
self.rectangles = []
self.drawing = False
self.ix, self.iy = -1, -1
self.current_class_id = 3
self.scale_factor = 1.0
# Class names mapping
self.class_names = {
0: "pedestrian",
1: "people",
2: "bicycle",
3: "car",
4: "EV",
5: "truck",
6: "tricycle",
7: "awning-tricycle",
8: "bus",
9: "motor"
}
self.setup_ui()
def setup_ui(self):
# Main container
main_container = ctk.CTkFrame(self.root, fg_color="transparent")
main_container.pack(fill="both", expand=True)
# Header Frame with Logo and About/Dependencies buttons
self.create_header(main_container)
# Control Panel Frame
self.create_control_panel(main_container)
# Canvas Frame for image display
self.create_canvas_frame(main_container)
def create_header(self, parent):
"""Create header with logo and buttons"""
header_frame = ctk.CTkFrame(parent, height=60, corner_radius=0)
header_frame.pack(fill="x", padx=0, pady=(0, 10))
header_frame.pack_propagate(False)
# Left side: Logo and Title
left_frame = ctk.CTkFrame(header_frame, fg_color="transparent")
left_frame.pack(side="left", padx=15, pady=10)
# Try to load logo
logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "RoadwayVR.jpg")
if os.path.exists(logo_path):
try:
logo_image = Image.open(logo_path)
logo_image = logo_image.resize((40, 40), Image.Resampling.LANCZOS)
self.logo_photo = ctk.CTkImage(light_image=logo_image, dark_image=logo_image, size=(40, 40))
logo_label = ctk.CTkLabel(left_frame, image=self.logo_photo, text="")
logo_label.pack(side="left", padx=(0, 12))
except Exception as e:
print(f"Could not load logo: {e}")
# Title
title_label = ctk.CTkLabel(left_frame, text="SimJam ComputerVision - YOLO Annotator",
font=ctk.CTkFont(size=18, weight="bold"))
title_label.pack(side="left")
# Right side: About and Dependencies buttons
right_frame = ctk.CTkFrame(header_frame, fg_color="transparent")
right_frame.pack(side="right", padx=15, pady=10)
about_btn = ctk.CTkButton(right_frame, text="βΉ About", command=self.show_about,
width=100, height=32)
about_btn.pack(side="left", padx=(0, 10))
dependencies_btn = ctk.CTkButton(right_frame, text="π§ Dependencies",
command=self.show_dependencies, width=140, height=32)
dependencies_btn.pack(side="left")
def create_control_panel(self, parent):
"""Create control panel with all buttons"""
control_frame = ctk.CTkFrame(parent, height=80, corner_radius=0)
control_frame.pack(fill="x", padx=0, pady=(0, 10))
control_frame.pack_propagate(False)
# Container for buttons (centered)
button_container = ctk.CTkFrame(control_frame, fg_color="transparent")
button_container.pack(expand=True)
# Load Image Button
self.btn_load = ctk.CTkButton(button_container, text="π Load Image",
command=self.load_image, width=140, height=40)
self.btn_load.pack(side="left", padx=8)
# Set Output Directory Button
self.btn_output = ctk.CTkButton(button_container, text="π Set Output",
command=self.set_output_directory, width=140, height=40,
fg_color="#9333EA", hover_color="#7C3AED")
self.btn_output.pack(side="left", padx=8)
# Separator
sep1 = ctk.CTkFrame(button_container, width=2, height=40, fg_color="gray30")
sep1.pack(side="left", padx=15)
# Class Selection
class_frame = ctk.CTkFrame(button_container, fg_color="transparent")
class_frame.pack(side="left", padx=8)
ctk.CTkLabel(class_frame, text="Class:", font=ctk.CTkFont(size=12, weight="bold")).pack(side="left", padx=(0, 8))
self.class_var = ctk.StringVar(value="3 - car")
class_dropdown = ctk.CTkComboBox(class_frame, variable=self.class_var,
values=[f"{k} - {v}" for k, v in self.class_names.items()],
state="readonly", width=180, height=40,
command=self.on_class_change)
class_dropdown.pack(side="left")
# Separator
sep2 = ctk.CTkFrame(button_container, width=2, height=40, fg_color="gray30")
sep2.pack(side="left", padx=15)
# Undo Button
self.btn_undo = ctk.CTkButton(button_container, text="βΆ Undo",
command=self.undo_last, width=120, height=40,
fg_color="#F97316", hover_color="#EA580C")
self.btn_undo.pack(side="left", padx=8)
# Clear All Button
self.btn_clear = ctk.CTkButton(button_container, text="ποΈ Clear",
command=self.clear_all, width=120, height=40,
fg_color="#EF4444", hover_color="#DC2626")
self.btn_clear.pack(side="left", padx=8)
# Save Button
self.btn_save = ctk.CTkButton(button_container, text="πΎ Save Labels",
command=self.save_labels, width=140, height=40,
fg_color="#10B981", hover_color="#059669")
self.btn_save.pack(side="left", padx=8)
# Box Count Label
self.count_label = ctk.CTkLabel(button_container, text="0",
font=ctk.CTkFont(size=18, weight="bold"),
fg_color="#10B981", corner_radius=8,
width=60, height=40)
self.count_label.pack(side="left", padx=20)
def create_canvas_frame(self, parent):
"""Create canvas for image display"""
canvas_frame = ctk.CTkFrame(parent, corner_radius=0)
canvas_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10))
# Use tkinter Canvas (not CTkCanvas) for image display and mouse events
self.canvas = Canvas(canvas_frame, bg="#1a1a1a", highlightthickness=0)
self.canvas.pack(fill="both", expand=True, padx=5, pady=5)
# Bind mouse events
self.canvas.bind("<ButtonPress-1>", self.on_mouse_down)
self.canvas.bind("<B1-Motion>", self.on_mouse_move)
self.canvas.bind("<ButtonRelease-1>", self.on_mouse_up)
def on_class_change(self, choice):
"""Handle class selection change"""
self.current_class_id = int(choice.split(" - ")[0])
def load_image(self):
"""Load an image file"""
file_path = filedialog.askopenfilename(
title="Select Image",
filetypes=[("Image files", "*.jpg *.jpeg *.png *.bmp"), ("All files", "*.*")]
)
if file_path:
self.image_path = file_path
self.original_img = cv2.imread(file_path)
if self.original_img is None:
messagebox.showerror("Error", "Failed to load image!")
return
# Clear previous annotations
self.rectangles = []
self.update_count()
# Calculate scale to fit canvas
canvas_width = self.canvas.winfo_width()
canvas_height = self.canvas.winfo_height()
if canvas_width > 1 and canvas_height > 1:
img_height, img_width = self.original_img.shape[:2]
scale_w = canvas_width / img_width
scale_h = canvas_height / img_height
self.scale_factor = min(scale_w, scale_h, 1.0) * 0.95
else:
self.scale_factor = 0.8
self.display_image()
def set_output_directory(self):
"""Set output directory for labels"""
directory = filedialog.askdirectory(title="Select Output Directory")
if directory:
self.output_dir = directory
messagebox.showinfo("Success", f"Output directory set to:\n{directory}")
def display_image(self):
"""Display image on canvas"""
if self.original_img is None:
return
img_copy = self.original_img.copy()
# Draw all rectangles
for rect in self.rectangles:
xmin, ymin, xmax, ymax, class_id = rect
color = self.get_class_color(class_id)
cv2.rectangle(img_copy, (xmin, ymin), (xmax, ymax), color, 2)
# Draw label
label = f"{class_id}: {self.class_names[class_id]}"
(text_width, text_height), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
cv2.rectangle(img_copy, (xmin, ymin - text_height - 8),
(xmin + text_width + 8, ymin), color, -1)
cv2.putText(img_copy, label, (xmin + 4, ymin - 4),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# Convert and resize
img_rgb = cv2.cvtColor(img_copy, cv2.COLOR_BGR2RGB)
pil_img = Image.fromarray(img_rgb)
new_width = int(img_copy.shape[1] * self.scale_factor)
new_height = int(img_copy.shape[0] * self.scale_factor)
pil_img = pil_img.resize((new_width, new_height), Image.Resampling.LANCZOS)
self.display_img = ImageTk.PhotoImage(pil_img)
self.canvas.delete("all")
self.canvas.create_image(0, 0, anchor="nw", image=self.display_img)
def get_class_color(self, class_id):
"""Get color for each class"""
colors = [
(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0),
(255, 0, 255), (0, 255, 255), (128, 0, 128), (255, 165, 0),
(0, 128, 128), (128, 128, 0)
]
return colors[class_id % len(colors)]
def on_mouse_down(self, event):
"""Handle mouse button press"""
if self.original_img is None:
return
self.drawing = True
self.ix = int(event.x / self.scale_factor)
self.iy = int(event.y / self.scale_factor)
def on_mouse_move(self, event):
"""Handle mouse movement while drawing"""
if not self.drawing or self.original_img is None:
return
x = int(event.x / self.scale_factor)
y = int(event.y / self.scale_factor)
# Constrain to image bounds
h, w = self.original_img.shape[:2]
x = max(0, min(x, w))
y = max(0, min(y, h))
# Draw preview
img_copy = self.original_img.copy()
# Draw existing rectangles
for rect in self.rectangles:
xmin, ymin, xmax, ymax, class_id = rect
color = self.get_class_color(class_id)
cv2.rectangle(img_copy, (xmin, ymin), (xmax, ymax), color, 2)
label = f"{class_id}: {self.class_names[class_id]}"
(text_width, text_height), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
cv2.rectangle(img_copy, (xmin, ymin - text_height - 8),
(xmin + text_width + 8, ymin), color, -1)
cv2.putText(img_copy, label, (xmin + 4, ymin - 4),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# Draw current rectangle
color = self.get_class_color(self.current_class_id)
cv2.rectangle(img_copy, (self.ix, self.iy), (x, y), color, 3)
# Update display
img_rgb = cv2.cvtColor(img_copy, cv2.COLOR_BGR2RGB)
pil_img = Image.fromarray(img_rgb)
new_width = int(img_copy.shape[1] * self.scale_factor)
new_height = int(img_copy.shape[0] * self.scale_factor)
pil_img = pil_img.resize((new_width, new_height), Image.Resampling.LANCZOS)
self.display_img = ImageTk.PhotoImage(pil_img)
self.canvas.delete("all")
self.canvas.create_image(0, 0, anchor="nw", image=self.display_img)
def on_mouse_up(self, event):
"""Handle mouse button release"""
if not self.drawing or self.original_img is None:
return
self.drawing = False
x = int(event.x / self.scale_factor)
y = int(event.y / self.scale_factor)
# Constrain to image bounds
h, w = self.original_img.shape[:2]
x = max(0, min(x, w))
y = max(0, min(y, h))
# Create rectangle
xmin = min(self.ix, x)
ymin = min(self.iy, y)
xmax = max(self.ix, x)
ymax = max(self.iy, y)
# Only add if rectangle has size
if xmax - xmin > 5 and ymax - ymin > 5:
self.rectangles.append([xmin, ymin, xmax, ymax, self.current_class_id])
self.update_count()
self.display_image()
def undo_last(self):
"""Remove last annotation"""
if self.rectangles:
self.rectangles.pop()
self.update_count()
self.display_image()
def clear_all(self):
"""Clear all annotations"""
if self.rectangles:
if messagebox.askyesno("Confirm", "Clear all annotations?"):
self.rectangles = []
self.update_count()
self.display_image()
def update_count(self):
"""Update box count display"""
self.count_label.configure(text=str(len(self.rectangles)))
def save_labels(self):
"""Save annotations in YOLO format"""
if self.original_img is None:
messagebox.showwarning("Warning", "No image loaded!")
return
if not self.rectangles:
messagebox.showwarning("Warning", "No annotations to save!")
return
if not self.output_dir:
messagebox.showwarning("Warning", "Please set output directory first!")
return
# Get image dimensions
h, w = self.original_img.shape[:2]
# Get base filename
base = os.path.splitext(os.path.basename(self.image_path))[0]
# Write YOLO format labels
label_path = os.path.join(self.output_dir, f"{base}.txt")
with open(label_path, 'w') as f:
for rect in self.rectangles:
xmin, ymin, xmax, ymax, class_id = rect
# Convert to YOLO format (normalized)
x_center = ((xmin + xmax) / 2) / w
y_center = ((ymin + ymax) / 2) / h
bw = (xmax - xmin) / w
bh = (ymax - ymin) / h
f.write(f"{class_id} {x_center:.6f} {y_center:.6f} {bw:.6f} {bh:.6f}\n")
messagebox.showinfo("Success", f"Saved {len(self.rectangles)} annotations!\n\n{base}.txt")
def show_about(self):
"""Show About dialog"""
about_window = ctk.CTkToplevel(self.root)
about_window.title("About")
about_window.geometry("350x220")
about_window.resizable(False, False)
about_window.transient(self.root)
about_window.grab_set()
# Content frame
content_frame = ctk.CTkFrame(about_window, fg_color="transparent")
content_frame.pack(fill="both", expand=True, padx=25, pady=20)
# Title
ctk.CTkLabel(content_frame, text="About",
font=ctk.CTkFont(size=18, weight="bold")).pack(pady=(0, 12))
# Developed by
ctk.CTkLabel(content_frame, text="Developed by RoadwayVR",
font=ctk.CTkFont(size=13, weight="bold"),
text_color="lightblue").pack(pady=(0, 10))
# Application
ctk.CTkLabel(content_frame, text="Application:",
font=ctk.CTkFont(size=10), text_color="gray60").pack(pady=(0, 2))
ctk.CTkLabel(content_frame, text="SimJamCV Annotator",
font=ctk.CTkFont(size=12, weight="bold")).pack(pady=(0, 8))
# Version
ctk.CTkLabel(content_frame, text="Version: 1.0",
font=ctk.CTkFont(size=11)).pack(pady=(0, 8))
# Description
ctk.CTkLabel(content_frame, text="YOLO format annotation tool\nfor object detection datasets",
font=ctk.CTkFont(size=10), text_color="gray60",
justify="center").pack(pady=(0, 12))
# Close button
ctk.CTkButton(content_frame, text="Close", command=about_window.destroy,
width=100, height=28).pack()
def show_dependencies(self):
"""Show Dependencies dialog with version-specific checking"""
dep_window = ctk.CTkToplevel(self.root)
dep_window.title("Dependency Manager")
dep_window.geometry("700x500")
dep_window.resizable(False, False)
dep_window.transient(self.root)
dep_window.grab_set()
# Main frame
main_frame = ctk.CTkFrame(dep_window)
main_frame.pack(fill="both", expand=True, padx=20, pady=20)
# Header
header_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
header_frame.pack(fill="x", pady=(0, 15))
ctk.CTkLabel(header_frame, text="π¦ Package Dependencies",
font=ctk.CTkFont(size=18, weight="bold")).pack(side="left")
ctk.CTkLabel(header_frame, text="Check and install required Python packages",
text_color="gray60").pack(side="left", padx=(15, 0))
# Console
console_frame = ctk.CTkFrame(main_frame)
console_frame.pack(fill="both", expand=True, pady=(0, 15))
console = ctk.CTkTextbox(console_frame, font=ctk.CTkFont(family="Courier", size=11),
wrap="word")
console.pack(fill="both", expand=True, padx=10, pady=10)
def log_message(message):
console.configure(state="normal")
console.insert("end", message + "\n")
console.see("end")
console.configure(state="disabled")
dep_window.update()
# Buttons
button_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
button_frame.pack(fill="x")
check_btn = ctk.CTkButton(button_frame, text="π Check Status", width=150)
check_btn.pack(side="left", padx=(0, 10))
install_btn = ctk.CTkButton(button_frame, text="β¬ Install/Update", width=150,
fg_color="green", hover_color="darkgreen")
install_btn.pack(side="left", padx=(0, 10))
close_btn = ctk.CTkButton(button_frame, text="Close", width=100,
command=dep_window.destroy)
close_btn.pack(side="right")
# Required packages with EXACT versions
# Format: (import_name, pip_package_name, exact_version, description)
REQUIRED_PACKAGES = [
('customtkinter', 'customtkinter', '5.2.2', 'Modern GUI framework'),
('cv2', 'opencv-python', '4.10.0', 'Computer vision library'),
('PIL', 'pillow', '11.3.0', 'Image processing library'),
('numpy', 'numpy', '2.2.6', 'Numerical computing'),
]
def get_version(module_name):
"""Get installed version of a module"""
try:
import importlib
mod = importlib.import_module(module_name)
return getattr(mod, '__version__', 'unknown')
except:
return None
def check_dependencies():
console.configure(state="normal")
console.delete("1.0", "end")
console.configure(state="disabled")
log_message("Checking dependencies...\n")
log_message("DEPENDENCY STATUS")
log_message("=" * 60 + "\n")
installed = []
missing = []
wrong_version = []
for import_name, pip_name, exact_version, desc in REQUIRED_PACKAGES:
try:
__import__(import_name)
current_ver = get_version(import_name)
if current_ver and current_ver != exact_version and current_ver != 'unknown':
wrong_version.append((import_name, pip_name, current_ver, exact_version, desc))
log_message(f"β {pip_name:<20} v{current_ver} β v{exact_version}")
else:
installed.append((pip_name, exact_version, desc))
log_message(f"β {pip_name:<20} v{exact_version} - {desc}")
except ImportError:
missing.append((import_name, pip_name, exact_version, desc))
log_message(f"β {pip_name:<20} v{exact_version} - {desc}")
if wrong_version:
log_message("\nβ WRONG VERSION (Will be updated):")
for imp, pkg, curr, target, desc in wrong_version:
log_message(f" β’ {pkg}: v{curr} β v{target}")
if missing:
log_message("\nβ MISSING (Will be installed):")
for imp, pkg, ver, desc in missing:
log_message(f" β’ {pkg} v{ver}")
if not missing and not wrong_version:
log_message("\nπ All dependencies are correctly installed!")
install_btn.configure(state="disabled")
else:
total_issues = len(missing) + len(wrong_version)
log_message(f"\nβ‘ Click 'Install/Update' to fix {total_issues} package(s)")
install_btn.configure(state="normal")
return installed, missing, wrong_version
def install_missing():
def install_worker():
install_btn.configure(state="disabled")
check_btn.configure(state="disabled")
log_message("\n" + "=" * 60)
log_message("INSTALLING/UPDATING PACKAGES")
log_message("=" * 60 + "\n")
# Get current status
installed_list = []
to_install = []
for import_name, pip_name, exact_version, desc in REQUIRED_PACKAGES:
try:
__import__(import_name)
current_ver = get_version(import_name)
if current_ver != exact_version and current_ver != 'unknown':
to_install.append((pip_name, exact_version, "UPDATE"))
except ImportError:
to_install.append((pip_name, exact_version, "INSTALL"))
if not to_install:
log_message("β All packages already at correct versions!")
check_btn.configure(state="normal")
install_btn.configure(state="normal")
return
total = len(to_install)
for i, (package, version, action) in enumerate(to_install, 1):
log_message(f"[{action}] {i}/{total}: {package} v{version}...")
try:
result = subprocess.run(
[sys.executable, "-m", "pip", "install",
f"{package}=={version}", "--force-reinstall", "--no-deps"],
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
log_message(f" β {package} v{version} installed successfully")
else:
# Try without --no-deps
result = subprocess.run(
[sys.executable, "-m", "pip", "install",
f"{package}=={version}", "--force-reinstall"],
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
log_message(f" β {package} v{version} installed successfully")
else:
log_message(f" β Failed to install {package}")
except Exception as e:
log_message(f" β Error: {str(e)}")
log_message("\n" + "=" * 60)
log_message("Verifying installation...")
log_message("")
check_dependencies()
log_message("\nβ IMPORTANT: You may need to restart the application.")
check_btn.configure(state="normal")
install_btn.configure(state="normal")
threading.Thread(target=install_worker, daemon=True).start()
check_btn.configure(command=check_dependencies)
install_btn.configure(command=install_missing)
# Initial check
check_dependencies()
if __name__ == "__main__":
root = ctk.CTk()
app = YOLOAnnotator(root)
root.mainloop()
InΒ [92]:
with open("dqn_controller.py", "r", encoding="utf-8") as f:
print(f.read())
# Add modules to provide access to specific libraries and functions
import os # Module provides functions to handle file paths, directories, environment variables
import sys # Module provides access to Python-specific system parameters and functions
import random
import numpy as np
import matplotlib.pyplot as plt # Visualization
import pandas as pd
# (Additional) Imports for Deep Q-Learning
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
# Establish path to SUMO (SUMO_HOME)
if 'SUMO_HOME' in os.environ:
tools = os.path.join(os.environ['SUMO_HOME'], 'tools')
sys.path.append(tools)
else:
sys.exit("Please declare environment variable 'SUMO_HOME'")
# Add Traci module to provide access to specific libraries and functions
import traci # Static network information (such as reading and analyzing network files)
# Step 4: Define Sumo configuration
Sumo_config = [
'sumo',
'-c', 'RL.sumocfg',
'--step-length', '0.10',
'--delay', '1',
'--lateral-resolution', '0'
]
# Step 5: Open connection between SUMO and Traci
traci.start(Sumo_config)
#traci.gui.setSchema("View #0", "real world")
# -------------------------
# Step 6: Define Variables
# -------------------------
# Variables for RL State (queue lengths from detectors and current phase)
q_EB_0 = 0
q_EB_1 = 0
q_EB_2 = 0
q_SB_0 = 0
q_SB_1 = 0
q_SB_2 = 0
current_phase = 0
# ---- Reinforcement Learning Hyperparameters ----
TOTAL_STEPS = 10000 # The total number of simulation steps for continuous (online) training.
ALPHA = 0.1 # Learning rate (Ξ±) between[0, 1] #If Ξ± = 1, you fully replace the old Q-value with the newly computed estimate.
#If Ξ± = 0, you ignore the new estimate and never update the Q-value.
GAMMA = 0.9 # Discount factor (Ξ³) between[0, 1] #If Ξ³ = 0, the agent only cares about the reward at the current step (no future rewards).
#If Ξ³ = 1, the agent cares equally about current and future rewards, looking at long-term gains.
EPSILON = 0.1 # Exploration rate (Ξ΅) between[0, 1] #If Ξ΅ = 0 means very greedy, if=1 means very random
ACTIONS = [0, 1] # The discrete action space (0 = keep phase, 1 = switch phase)
# ---- Additional Stability Parameters ----
MIN_GREEN_STEPS = 100
last_switch_step = -MIN_GREEN_STEPS
# -------------------------
# Step 7: Define Functions
# -------------------------
def build_model(state_size, action_size):
"""
Build a simple feedforward neural network that approximates Q-values.
"""
model = keras.Sequential() # Feedforward neural network
model.add(layers.Input(shape=(state_size,))) # Input layer
model.add(layers.Dense(24, activation='relu')) # First hidden layer
model.add(layers.Dense(24, activation='relu')) # Second hidden layer
model.add(layers.Dense(action_size, activation='linear')) # Output layer
model.compile(
loss='mse',
optimizer=keras.optimizers.Adam(learning_rate=0.001)
)
return model
def to_array(state_tuple):
"""
Convert the state tuple into a NumPy array for neural network input.
"""
return np.array(state_tuple, dtype=np.float32).reshape((1, -1))
# Create the DQN model
state_size = 7 # (q_EB_0, q_EB_1, q_EB_2, q_SB_0, q_SB_1, q_SB_2, current_phase)
action_size = len(ACTIONS)
dqn_model = build_model(state_size, action_size)
def get_max_Q_value_of_state(s): #1. Objective Function
state_array = to_array(s)
Q_values = dqn_model.predict(state_array, verbose=0)[0] # shape: (action_size,)
return np.max(Q_values)
def get_reward(state): #2. Constraint 2
"""
Simple reward function:
Negative of total queue length to encourage shorter queues.
"""
total_queue = sum(state[:-1]) # Exclude the current_phase element
reward = -float(total_queue)
return reward
def get_state(): #3&4. Constraint 3 & 4
global q_EB_0, q_EB_1, q_EB_2, q_SB_0, q_SB_1, q_SB_2, current_phase
# Detector IDs for Node1-2-EB
detector_Node1_2_EB_0 = "Node1_2_EB_0"
detector_Node1_2_EB_1 = "Node1_2_EB_1"
detector_Node1_2_EB_2 = "Node1_2_EB_2"
# Detector IDs for Node2-7-SB
detector_Node2_7_SB_0 = "Node2_7_SB_0"
detector_Node2_7_SB_1 = "Node2_7_SB_1"
detector_Node2_7_SB_2 = "Node2_7_SB_2"
# Traffic light ID
traffic_light_id = "Node2"
# Get queue lengths from each detector
q_EB_0 = get_queue_length(detector_Node1_2_EB_0)
q_EB_1 = get_queue_length(detector_Node1_2_EB_1)
q_EB_2 = get_queue_length(detector_Node1_2_EB_2)
q_SB_0 = get_queue_length(detector_Node2_7_SB_0)
q_SB_1 = get_queue_length(detector_Node2_7_SB_1)
q_SB_2 = get_queue_length(detector_Node2_7_SB_2)
# Get current phase index
current_phase = get_current_phase(traffic_light_id)
return (q_EB_0, q_EB_1, q_EB_2, q_SB_0, q_SB_1, q_SB_2, current_phase)
def apply_action(action, tls_id="Node2"): #5. Constraint 5
"""
Executes the chosen action on the traffic light, combining:
- Min Green Time check
- Switching to the next phase if allowed
Constraint #5: Ensure at least MIN_GREEN_STEPS pass before switching again.
"""
global last_switch_step
if action == 0:
# Do nothing (keep current phase)
return
elif action == 1:
# Check if minimum green time has passed before switching
if current_simulation_step - last_switch_step >= MIN_GREEN_STEPS:
program = traci.trafficlight.getAllProgramLogics(tls_id)[0]
num_phases = len(program.phases)
next_phase = (get_current_phase(tls_id) + 1) % num_phases
traci.trafficlight.setPhase(tls_id, next_phase)
# Record when the switch happened
last_switch_step = current_simulation_step
def update_Q_table(old_state, action, reward, new_state): #6. Constraint 6
"""
In DQN, we do a single-step gradient update instead of a table update.
"""
# 1) Predict current Q-values from old_state (current state)
old_state_array = to_array(old_state)
Q_values_old = dqn_model.predict(old_state_array, verbose=0)[0]
# 2) Predict Q-values for new_state to get max future Q (new state)
new_state_array = to_array(new_state)
Q_values_new = dqn_model.predict(new_state_array, verbose=0)[0]
best_future_q = np.max(Q_values_new)
# 3) Incorporate ALPHA to partially update the Q-value
Q_values_old[action] = Q_values_old[action] + ALPHA * (reward + GAMMA * best_future_q - Q_values_old[action])
# 4) Train (fit) the DQN on this single sample
dqn_model.fit(old_state_array, np.array([Q_values_old]), verbose=0)
def get_action_from_policy(state): #7. Constraint 7
"""
Epsilon-greedy strategy using the DQN's predicted Q-values.
"""
if random.random() < EPSILON:
return random.choice(ACTIONS)
else:
state_array = to_array(state)
Q_values = dqn_model.predict(state_array, verbose=0)[0]
return int(np.argmax(Q_values))
def get_queue_length(detector_id): #8.Constraint 8
return traci.lanearea.getLastStepVehicleNumber(detector_id)
def get_current_phase(tls_id): #8.Constraint 8
return traci.trafficlight.getPhase(tls_id)
# -------------------------
# Step 8: Fully Online Continuous Learning Loop
# -------------------------
# Lists to record data for plotting
step_history = []
reward_history = []
queue_history = []
# Additional evaluation metrics
waiting_time_history = []
total_throughput = 0
emergency_start_times = {}
emergency_delays = []
cumulative_reward = 0.0
def calculate_average_waiting_time():
vehicles = traci.vehicle.getIDList()
if len(vehicles) == 0:
return 0
total_wait = 0
for veh in vehicles:
total_wait += traci.vehicle.getWaitingTime(veh)
return total_wait / len(vehicles)
def update_throughput():
global total_throughput
arrived_vehicles = (
traci.simulation.getArrivedIDList()
)
total_throughput += len(
arrived_vehicles
)
def calculate_total_queue():
return (
q_EB_0 +
q_EB_1 +
q_EB_2 +
q_SB_0 +
q_SB_1 +
q_SB_2
)
print("\n=== Starting Fully Online Continuous Learning (DQN) ===")
for step in range(TOTAL_STEPS):
current_simulation_step = step
# Current state
state = get_state()
# Select action
action = get_action_from_policy(state)
# Apply action
apply_action(action)
# Advance SUMO
traci.simulationStep()
# ---------------------------------
# Evaluation Metrics
# ---------------------------------
update_throughput()
wt = calculate_average_waiting_time()
if wt is not None:
waiting_time_history.append(wt)
current_queue = calculate_total_queue()
queue_history.append(current_queue)
# Optional debug output every 100 steps
if step % 100 == 0:
print(
"Throughput:",
total_throughput
)
print(
"Arrived this step:",
len(
traci.simulation.getArrivedIDList()
)
)
print(
"Current Queue:",
current_queue
)
# ---------------------------------
# RL update
# ---------------------------------
new_state = get_state()
reward = get_reward(new_state)
cumulative_reward += reward
update_Q_table(
state,
action,
reward,
new_state
)
# Store data for plots
step_history.append(step)
reward_history.append(
cumulative_reward
)
# Debug output
if step % 100 == 0:
updated_q_vals = dqn_model.predict(
to_array(state),
verbose=0
)[0]
print(
f"Step {step}, "
f"Reward={reward:.2f}, "
f"Cumulative={cumulative_reward:.2f}, "
f"Q={updated_q_vals}"
)
# -------------------------
# Step 9: Close connection between SUMO and Traci
# -------------------------
traci.close()
# ~~~ Print final model summary (replacing Q-table info) ~~~
print("\nOnline Training completed.")
print("DQN Model Summary:")
dqn_model.summary()
# -------------------------
# Visualization of Results
# -------------------------
if len(waiting_time_history) > 0:
average_waiting_time = (
sum(waiting_time_history)
/
len(waiting_time_history)
)
else:
average_waiting_time = 0
# Average values
average_waiting_time = (
sum(waiting_time_history)
/
len(waiting_time_history)
)
average_queue_length = (
sum(queue_history)
/
len(queue_history)
)
# Save metrics
results = pd.DataFrame({
"Average Waiting Time (s)":
[average_waiting_time],
"Average Queue Length":
[average_queue_length],
"Throughput":
[total_throughput]
})
print("\n===== FINAL METRICS =====")
print(
"Waiting Samples:",
len(waiting_time_history)
)
print(
"Queue Samples:",
len(queue_history)
)
print(
"Final Throughput:",
total_throughput
)
results.to_csv(
"traffic_results.csv",
index=False
)
print(results)
# Reward graph
plt.figure(figsize=(10,6))
plt.plot(
step_history,
reward_history,
color="blue",
linewidth=2
)
plt.xlabel("Simulation Step")
plt.ylabel("Cumulative Reward")
plt.title("DQN Training Performance")
plt.grid(True)
plt.savefig(
"reward_graph.png",
dpi=300,
bbox_inches="tight"
)
plt.close()
# Queue graph
plt.figure(figsize=(10,6))
plt.plot(
step_history,
queue_history,
color="red",
linewidth=2
)
plt.xlabel("Simulation Step")
plt.ylabel("Queue Length")
plt.title("Traffic Congestion During Training")
plt.grid(True)
plt.savefig(
"queue_graph.png",
dpi=300,
bbox_inches="tight"
)
plt.close()
print("Files created:")
print("reward_graph.png")
print("queue_graph.png")
print("traffic_results.csv")
InΒ [93]:
with open("emergency_priority.py", "r", encoding="utf-8") as f:
print(f.read())
# Step 1: Add modules to provide access to specific libraries and functions
import os # Module provides functions to handle file paths, directories, environment variables
import sys # Module provides access to Python-specific system parameters and functions
import numpy as np
import pandas as pd
# Step 2: Establish path to SUMO (SUMO_HOME)
if 'SUMO_HOME' in os.environ:
tools = os.path.join(os.environ['SUMO_HOME'], 'tools')
sys.path.append(tools)
else:
sys.exit("Please declare environment variable 'SUMO_HOME'")
# Step 3: Add Traci module to provide access to specific libraries and functions
import traci
# Step 4: Define SUMO configuration
Sumo_config = [
'sumo-gui',
'-c', 'Test1.sumocfg',
'--step-length', '0.05',
'--delay', '1000',
'--lateral-resolution', '0.1'
]
# Step 5: Open connection between SUMO and Traci
traci.start(Sumo_config)
# Step 6: Define Variables
# Dictionary that maps TLSID + direction to a desired phase index
desired_phase_mapping = {
'Node2_EW': 0, # EW represents Eastbound-Westbound
'Node2_NS': 2, # NS represents Northbound-Southbound
'Node5_EW': 0, # EW represents Eastbound-Westbound
'Node5_NS': 2, # NS represents Northbound-Southbound
# Add any other traffic lights in the route of emergency vehicles as needed
}
adjusted_tls = {} # keep track of adjusted traffic lights to avoid repeated short-cycling
step = 0
# =====================================
# Emergency Vehicle Performance Metrics
# =====================================
ev_depart_times = {}
ev_travel_times = []
IDEAL_EV_TIME = 10.0
def track_ev_departure():
current_time = traci.simulation.getTime()
for veh in traci.vehicle.getIDList():
if traci.vehicle.getTypeID(veh) == "emergency":
if veh not in ev_depart_times:
ev_depart_times[veh] = current_time
def track_ev_arrival():
current_time = traci.simulation.getTime()
remove_list = []
for veh in ev_depart_times.keys():
if veh not in traci.vehicle.getIDList():
travel_time = (
current_time
-
ev_depart_times[veh]
)
ev_travel_times.append(travel_time)
remove_list.append(veh)
for veh in remove_list:
del ev_depart_times[veh]
# Step 7: Define Functions Determine if the emergency vehicle is on a north-south edge or east-west edge
#based on the naming convention of the edges.
def get_emergency_vehicle_direction(vehicle_id):
current_edge = traci.vehicle.getRoadID(vehicle_id).lower()
print(f"Vehicle {vehicle_id} is on edge {current_edge}")
if 'nb' in current_edge or 'sb' in current_edge:
return 'NS'
elif 'eb' in current_edge or 'wb' in current_edge:
return 'EW'
else:
return None
def process_emergency_vehicles(desired_phase_mapping, adjusted_tls, step):
"""
This function identifies emergency vehicles, adjusts traffic lights
for them if needed, and resets traffic lights when no longer in use.
:param desired_phase_mapping: dict mapping "TLSID_direction" -> desired phase index
:param adjusted_tls: dict tracking which traffic lights have been adjusted
and what phase they're supposed to be in
:param step: current simulation step counter
:return: updated step counter
"""
# Identify emergency vehicles
emergency_vehicles = [
veh for veh in traci.vehicle.getIDList()
if traci.vehicle.getTypeID(veh) == "emergency"
]
# Track which TLS are actively handled this step
active_tls = set()
for veh in emergency_vehicles:
direction = get_emergency_vehicle_direction(veh)
if direction:
next_tls = traci.vehicle.getNextTLS(veh)
print(f"next_tls for {veh}: {next_tls}")
if next_tls:
tls_info = next_tls[0]
tlsID, linkIndex, distance, state = tls_info
# Construct traffic light key (e.g. "Node2_EW")
tl_key = f"{tlsID}_{direction}"
desired_phase = desired_phase_mapping.get(tl_key)
if desired_phase is not None:
current_phase = traci.trafficlight.getPhase(tlsID)
print(f"TLS {tlsID}, Current phase: {current_phase}, "
f"Desired phase: {desired_phase}")
# Mark this traffic light as active (emergency vehicle approaching)
active_tls.add(tlsID)
# Check if we have already adjusted this traffic light
# for the desired phase
if tlsID not in adjusted_tls or adjusted_tls[tlsID] != desired_phase:
adjusted_tls[tlsID] = desired_phase # Record desired phase
if current_phase == desired_phase:
# Extend the green phase if it's already in the correct one
new_duration = max(20, traci.trafficlight.getPhaseDuration(tlsID) + 10)
traci.trafficlight.setPhaseDuration(tlsID, new_duration)
print(f"Extended phase {current_phase} of {tlsID} to "
f"{new_duration} seconds")
else:
# Shorten the current phase to get to desired phase sooner
traci.trafficlight.setPhaseDuration(tlsID, 0.1)
print(f"Shortened phase {current_phase} of {tlsID} to 1 second")
# Reset traffic lights to normal operation if no emergency vehicle is approaching
for tlsID in list(adjusted_tls.keys()):
if tlsID not in active_tls:
del adjusted_tls[tlsID]
print(f"Resetting traffic light {tlsID} to normal operation.")
step += 1
return step
# Step 8: Take simulation steps until there are no more vehicles in the network
MAX_SIMULATION_TIME = 120 # seconds
while (
traci.simulation.getMinExpectedNumber() > 0
and
traci.simulation.getTime() < MAX_SIMULATION_TIME
):
traci.simulationStep()
track_ev_departure()
track_ev_arrival()
step = process_emergency_vehicles(
desired_phase_mapping,
adjusted_tls,
step
)
# ==================================
# Emergency Vehicle Statistics
# ==================================
print("EV trips recorded:", len(ev_travel_times))
print("Travel times:", ev_travel_times)
average_ev_travel_time = (
sum(ev_travel_times) / len(ev_travel_times)
if len(ev_travel_times) > 0
else 0
)
average_ev_delay = (
average_ev_travel_time
- IDEAL_EV_TIME
)
results = pd.DataFrame({
"Average EV Travel Time (s)":
[average_ev_travel_time],
"Average EV Delay (s)":
[average_ev_delay]
})
results.to_csv(
"ev_results.csv",
index=False
)
print("\n===== EV RESULTS =====")
print(results)
print(
"Average EV Travel Time:",
round(average_ev_travel_time,2)
)
print(
"Average EV Delay:",
round(average_ev_delay,2)
)
print("======================")
# Step 9: Close connection between SUMO and Traci
traci.close()
InΒ [94]:
with open("trainer.py", "r", encoding="utf-8") as f:
print(f.read())
import customtkinter as ctk
from tkinter import filedialog, messagebox
import os
import sys
import yaml
import glob
import threading
# Set appearance mode and color theme
ctk.set_appearance_mode("dark") # Options: "dark", "light", "system"
ctk.set_default_color_theme("blue") # Options: "blue", "green", "dark-blue"
class YOLOv8MiniDatasetTrainer:
def __init__(self, root):
self.root = root
self.root.title("YOLO Trainer on Mini-Datasets")
self.root.geometry("1000x750")
# Variables
self.project_path = ctk.StringVar()
self.yaml_file = ctk.StringVar()
self.model_version = ctk.StringVar(value="yolov8n.pt")
self.device = ctk.StringVar(value="0")
self.conf_threshold = ctk.DoubleVar(value=0.25)
self.train_from_scratch = ctk.BooleanVar(value=True)
self.epochs = ctk.IntVar(value=200)
self.processing = False
self.training_thread = None
self.help_visible = False
# Create main container with padding
main_frame = ctk.CTkFrame(root, fg_color="transparent")
main_frame.pack(fill="both", expand=True, padx=20, pady=20)
# Create sections
self.create_header(main_frame)
self.create_project_selection(main_frame)
self.create_validation_section(main_frame)
self.create_training_parameters(main_frame)
self.create_console_section(main_frame)
def create_header(self, parent):
"""Create header section"""
header_frame = ctk.CTkFrame(parent, fg_color="transparent")
header_frame.pack(fill="x", pady=(0, 15))
# Create a container for all header elements
header_container = ctk.CTkFrame(header_frame, fg_color="transparent")
header_container.pack(fill="x")
# Left side: Logo and Title/subtitle
left_frame = ctk.CTkFrame(header_container, fg_color="transparent")
left_frame.pack(side="left", fill="x", expand=True)
# Try to load logo
logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "RoadwayVR.jpg")
if os.path.exists(logo_path):
try:
from PIL import Image
logo_image = Image.open(logo_path)
# Resize logo to appropriate size (e.g., 40x40)
logo_image = logo_image.resize((40, 40), Image.Resampling.LANCZOS)
logo_photo = ctk.CTkImage(light_image=logo_image, dark_image=logo_image, size=(40, 40))
logo_label = ctk.CTkLabel(left_frame, image=logo_photo, text="")
logo_label.image = logo_photo # Keep a reference
logo_label.pack(side="left", padx=(0, 15))
except Exception as e:
print(f"Could not load logo: {e}")
# Title and subtitle container
text_container = ctk.CTkFrame(left_frame, fg_color="transparent")
text_container.pack(side="left", fill="x", expand=True)
title = ctk.CTkLabel(text_container, text="SimJam ComputerVision (YOLO Trainer)",
font=ctk.CTkFont(size=20, weight="bold"))
title.pack(anchor="w")
subtitle = ctk.CTkLabel(text_container,
text="Educational fine-tuning tool for small datasets (7-30 images) β’ Optimized parameters",
text_color="gray60")
subtitle.pack(anchor="w")
# Right side: Dependencies and About buttons
buttons_frame = ctk.CTkFrame(header_container, fg_color="transparent")
buttons_frame.pack(side="right", anchor="ne")
about_btn = ctk.CTkButton(buttons_frame, text="βΉ About", command=self.show_about,
width=100, height=32)
about_btn.pack(side="left", padx=(0, 10))
dependencies_btn = ctk.CTkButton(buttons_frame, text="π§ Dependencies",
command=self.show_dependencies,
width=140, height=32)
dependencies_btn.pack(side="left")
def create_project_selection(self, parent):
"""Create project path selection"""
path_frame = ctk.CTkFrame(parent)
path_frame.pack(fill="x", pady=(0, 15))
# Title
title = ctk.CTkLabel(path_frame, text="Project Configuration",
font=ctk.CTkFont(size=14, weight="bold"))
title.pack(anchor="w", padx=15, pady=(15, 10))
# Project path row
path_row = ctk.CTkFrame(path_frame, fg_color="transparent")
path_row.pack(fill="x", padx=15, pady=(0, 10))
path_label = ctk.CTkLabel(path_row, text="Project Folder:", width=120, anchor="w")
path_label.pack(side="left", padx=(0, 10))
path_entry = ctk.CTkEntry(path_row, textvariable=self.project_path, height=32)
path_entry.pack(side="left", fill="x", expand=True, padx=(0, 10))
browse_btn = ctk.CTkButton(path_row, text="Browse", command=self.browse_project, width=100)
browse_btn.pack(side="left")
# Status label
self.project_status = ctk.CTkLabel(path_frame,
text="Select project folder with images/, labels/, and .yaml file",
text_color="gray60", anchor="w")
self.project_status.pack(fill="x", padx=15, pady=(0, 10))
# Help button
self.help_btn = ctk.CTkButton(path_frame, text="? Show Expected Folder Structure",
command=self.toggle_help, width=240, height=32,
fg_color="gray30", hover_color="gray40")
self.help_btn.pack(anchor="w", padx=15, pady=(0, 10))
# Hidden info box
self.info_frame = ctk.CTkFrame(path_frame, fg_color="gray20")
info_text = """Expected folder structure:
project_folder/
βββ images/
β βββ train/ (your training images)
β βββ val/ (validation images)
β βββ test/ (test images)
βββ labels/
β βββ train/ (YOLO format labels)
β βββ val/
β βββ test/
βββ data.yaml (dataset configuration)"""
self.info_label = ctk.CTkLabel(self.info_frame, text=info_text,
text_color="gray70", anchor="w", justify="left",
font=ctk.CTkFont(family="Courier", size=11))
self.info_label.pack(anchor="w", padx=15, pady=10)
def create_validation_section(self, parent):
"""Create validation section"""
val_frame = ctk.CTkFrame(parent)
val_frame.pack(fill="x", pady=(0, 15))
# Title
title = ctk.CTkLabel(val_frame, text="Dataset Validation",
font=ctk.CTkFont(size=14, weight="bold"))
title.pack(anchor="w", padx=15, pady=(15, 10))
# Button and status row
button_status_row = ctk.CTkFrame(val_frame, fg_color="transparent")
button_status_row.pack(fill="x", padx=15, pady=(0, 15))
validate_btn = ctk.CTkButton(button_status_row, text="Validate Dataset",
command=self.validate_dataset, width=140)
validate_btn.pack(side="left", padx=(0, 20))
# Status variables
self.val_images = ctk.StringVar(value="Not checked")
self.val_labels = ctk.StringVar(value="Not checked")
self.val_yaml = ctk.StringVar(value="Not checked")
self.val_format = ctk.StringVar(value="Not checked")
# Status indicators
status_container = ctk.CTkFrame(button_status_row, fg_color="transparent")
status_container.pack(side="left", fill="x", expand=True)
# Images status
img_frame = ctk.CTkFrame(status_container, fg_color="transparent")
img_frame.pack(side="left", padx=(0, 15))
ctk.CTkLabel(img_frame, text="Images:", font=ctk.CTkFont(weight="bold")).pack(side="left", padx=(0, 5))
ctk.CTkLabel(img_frame, textvariable=self.val_images, text_color="gray60").pack(side="left")
# Labels status
lbl_frame = ctk.CTkFrame(status_container, fg_color="transparent")
lbl_frame.pack(side="left", padx=(0, 15))
ctk.CTkLabel(lbl_frame, text="Labels:", font=ctk.CTkFont(weight="bold")).pack(side="left", padx=(0, 5))
ctk.CTkLabel(lbl_frame, textvariable=self.val_labels, text_color="gray60").pack(side="left")
# YAML status
yaml_frame = ctk.CTkFrame(status_container, fg_color="transparent")
yaml_frame.pack(side="left", padx=(0, 15))
ctk.CTkLabel(yaml_frame, text="YAML:", font=ctk.CTkFont(weight="bold")).pack(side="left", padx=(0, 5))
ctk.CTkLabel(yaml_frame, textvariable=self.val_yaml, text_color="gray60").pack(side="left")
# Format status
fmt_frame = ctk.CTkFrame(status_container, fg_color="transparent")
fmt_frame.pack(side="left")
ctk.CTkLabel(fmt_frame, text="Format:", font=ctk.CTkFont(weight="bold")).pack(side="left", padx=(0, 5))
ctk.CTkLabel(fmt_frame, textvariable=self.val_format, text_color="gray60").pack(side="left")
def create_training_parameters(self, parent):
"""Create training parameters section"""
params_frame = ctk.CTkFrame(parent)
params_frame.pack(fill="x", pady=(0, 15))
# Title
title = ctk.CTkLabel(params_frame, text="Training Parameters",
font=ctk.CTkFont(size=14, weight="bold"))
title.pack(anchor="w", padx=15, pady=(15, 10))
# First row: Model Version and Device
row1 = ctk.CTkFrame(params_frame, fg_color="transparent")
row1.pack(fill="x", padx=15, pady=(0, 10))
# Model selection
model_container = ctk.CTkFrame(row1, fg_color="transparent")
model_container.pack(side="left", padx=(0, 30))
ctk.CTkLabel(model_container, text="Model Version:", width=120, anchor="w").pack(side="left", padx=(0, 10))
model_combo = ctk.CTkComboBox(model_container, variable=self.model_version, width=150,
values=["yolov8n.pt", "yolov8s.pt", "yolov8m.pt",
"yolo11n.pt", "yolo11s.pt", "yolo11m.pt"],
state="readonly")
model_combo.pack(side="left", padx=(0, 10))
self.model_info = ctk.CTkLabel(model_container, text="(Nano - fastest, smallest)",
text_color="gray60")
self.model_info.pack(side="left")
# Device selection
device_container = ctk.CTkFrame(row1, fg_color="transparent")
device_container.pack(side="left")
ctk.CTkLabel(device_container, text="Device:", width=60, anchor="w").pack(side="left", padx=(0, 10))
device_combo = ctk.CTkComboBox(device_container, variable=self.device, width=100,
values=["0", "cpu"], state="readonly")
device_combo.pack(side="left")
# Update model info
def update_model_info(*args):
model = self.model_version.get()
if 'n.pt' in model:
info = "(Nano - fastest, best for learning)"
elif 's.pt' in model:
info = "(Small - balanced)"
elif 'm.pt' in model:
info = "(Medium - higher accuracy)"
else:
info = ""
self.model_info.configure(text=info)
self.model_version.trace('w', update_model_info)
# Second row: Confidence and Epochs
row2 = ctk.CTkFrame(params_frame, fg_color="transparent")
row2.pack(fill="x", padx=15, pady=(0, 10))
# Confidence threshold
conf_container = ctk.CTkFrame(row2, fg_color="transparent")
conf_container.pack(side="left", padx=(0, 30))
ctk.CTkLabel(conf_container, text="Confidence:", width=120, anchor="w").pack(side="left", padx=(0, 10))
conf_slider = ctk.CTkSlider(conf_container, from_=0.1, to=0.9, variable=self.conf_threshold,
width=200, number_of_steps=80)
conf_slider.pack(side="left", padx=(0, 10))
self.conf_label = ctk.CTkLabel(conf_container, text=f"{self.conf_threshold.get():.2f}",
width=50, anchor="w")
self.conf_label.pack(side="left")
# Update confidence label
def update_conf_label(*args):
self.conf_label.configure(text=f"{self.conf_threshold.get():.2f}")
self.conf_threshold.trace('w', update_conf_label)
# Epochs control
epochs_container = ctk.CTkFrame(row2, fg_color="transparent")
epochs_container.pack(side="left")
ctk.CTkLabel(epochs_container, text="Epochs:", width=60, anchor="w").pack(side="left", padx=(0, 10))
epochs_entry = ctk.CTkEntry(epochs_container, textvariable=self.epochs, width=100)
epochs_entry.pack(side="left")
# Third row: Train from scratch checkbox
row3 = ctk.CTkFrame(params_frame, fg_color="transparent")
row3.pack(fill="x", padx=15, pady=(0, 10))
scratch_check = ctk.CTkCheckBox(row3, text="Train from Scratch (no pretrained weights)",
variable=self.train_from_scratch)
scratch_check.pack(side="left")
# Fourth row: Fixed parameters info
info_label = ctk.CTkLabel(params_frame,
text="Fixed optimized: LR=0.01, Batch=1, No augmentation (for mini-datasets)",
text_color="gray60")
info_label.pack(anchor="w", padx=15, pady=(0, 10))
# Training buttons
button_row = ctk.CTkFrame(params_frame, fg_color="transparent")
button_row.pack(fill="x", padx=15, pady=(0, 15))
self.train_btn = ctk.CTkButton(button_row, text="βΆ Start Fine-Tuning",
command=self.start_training, width=180, height=36,
fg_color="#2fa572", hover_color="#257a53")
self.train_btn.pack(side="left", padx=(0, 10))
self.stop_btn = ctk.CTkButton(button_row, text="β¬ Stop", command=self.stop_training,
width=100, height=36, state="disabled",
fg_color="#d63031", hover_color="#a82828")
self.stop_btn.pack(side="left")
def create_console_section(self, parent):
"""Create console and view buttons section"""
console_frame = ctk.CTkFrame(parent)
console_frame.pack(fill="both", expand=True)
# Title with view buttons
header = ctk.CTkFrame(console_frame, fg_color="transparent")
header.pack(fill="x", padx=15, pady=(15, 10))
ctk.CTkLabel(header, text="Training Console",
font=ctk.CTkFont(size=14, weight="bold")).pack(side="left")
# View buttons
btn_container = ctk.CTkFrame(header, fg_color="transparent")
btn_container.pack(side="right")
self.view_before_btn = ctk.CTkButton(btn_container, text="View Before", width=100,
command=self.view_before, state="disabled",
height=28)
self.view_before_btn.pack(side="left", padx=2)
self.view_after_btn = ctk.CTkButton(btn_container, text="View After", width=100,
command=self.view_after, state="disabled",
height=28)
self.view_after_btn.pack(side="left", padx=2)
self.view_val_btn = ctk.CTkButton(btn_container, text="View Val", width=100,
command=self.view_val, state="disabled",
height=28)
self.view_val_btn.pack(side="left", padx=2)
self.view_metrics_btn = ctk.CTkButton(btn_container, text="View Metrics", width=100,
command=self.view_metrics, state="disabled",
height=28)
self.view_metrics_btn.pack(side="left", padx=2)
# Console text area
self.console = ctk.CTkTextbox(console_frame, height=300, wrap="word",
font=ctk.CTkFont(family="Courier", size=11))
self.console.pack(fill="both", expand=True, padx=15, pady=(0, 15))
self.log("YOLO Mini-Dataset Trainer - Ready")
self.log("Select project folder and validate dataset to begin")
def toggle_help(self):
"""Toggle help information visibility"""
if self.help_visible:
self.info_frame.pack_forget()
self.help_btn.configure(text="? Show Expected Folder Structure")
self.help_visible = False
else:
self.info_frame.pack(fill="x", padx=15, pady=(0, 10))
self.help_btn.configure(text="β Hide Folder Structure")
self.help_visible = True
def browse_project(self):
"""Browse for project folder"""
folder = filedialog.askdirectory(title="Select YOLO Project Folder")
if folder:
self.project_path.set(folder)
self.project_status.configure(text=f"Selected: {folder}", text_color="gray60")
self.log(f"\nProject folder selected: {folder}")
def validate_dataset(self):
"""Validate the dataset structure"""
path = self.project_path.get()
if not path:
messagebox.showwarning("No Path", "Please select a project folder first")
return
self.log("\n" + "=" * 60)
self.log("VALIDATING DATASET...")
self.log("=" * 60)
issues = []
# Check images
img_train = os.path.join(path, "images", "train")
img_val = os.path.join(path, "images", "val")
img_test = os.path.join(path, "images", "test")
train_imgs = len(glob.glob(os.path.join(img_train, "*.*"))) if os.path.exists(img_train) else 0
val_imgs = len(glob.glob(os.path.join(img_val, "*.*"))) if os.path.exists(img_val) else 0
test_imgs = len(glob.glob(os.path.join(img_test, "*.*"))) if os.path.exists(img_test) else 0
if train_imgs > 0:
self.val_images.set(f"β {train_imgs} train, {val_imgs} val, {test_imgs} test")
self.log(f"β Images found: {train_imgs} train, {val_imgs} val, {test_imgs} test")
else:
self.val_images.set("β No images")
issues.append("No training images found")
self.log("β No training images found")
# Check labels
lbl_train = os.path.join(path, "labels", "train")
lbl_val = os.path.join(path, "labels", "val")
lbl_test = os.path.join(path, "labels", "test")
train_lbls = len(glob.glob(os.path.join(lbl_train, "*.txt"))) if os.path.exists(lbl_train) else 0
val_lbls = len(glob.glob(os.path.join(lbl_val, "*.txt"))) if os.path.exists(lbl_val) else 0
test_lbls = len(glob.glob(os.path.join(lbl_test, "*.txt"))) if os.path.exists(lbl_test) else 0
if train_lbls > 0:
self.val_labels.set(f"β {train_lbls} train, {val_lbls} val, {test_lbls} test")
self.log(f"β Labels found: {train_lbls} train, {val_lbls} val, {test_lbls} test")
else:
self.val_labels.set("β No labels")
issues.append("No training labels found")
self.log("β No training labels found")
# Check YAML
yaml_files = glob.glob(os.path.join(path, "*.yaml"))
if yaml_files:
self.yaml_file.set(yaml_files[0])
self.val_yaml.set(f"β {os.path.basename(yaml_files[0])}")
self.log(f"β YAML found: {os.path.basename(yaml_files[0])}")
try:
with open(yaml_files[0], 'r') as f:
yaml_content = yaml.safe_load(f)
if 'names' in yaml_content and 'nc' in yaml_content:
num_classes = yaml_content['nc']
class_names = yaml_content['names']
self.log(f" Classes: {num_classes} - {class_names}")
except:
pass
else:
self.val_yaml.set("β Not found")
issues.append("No .yaml file found")
self.log("β No .yaml file found")
# Check format
if train_lbls > 0:
sample_label = glob.glob(os.path.join(lbl_train, "*.txt"))[0]
try:
with open(sample_label, 'r') as f:
line = f.readline().strip()
parts = line.split()
if len(parts) == 5:
self.val_format.set("β YOLO format")
self.log("β Label format is valid YOLO")
else:
self.val_format.set("β Invalid")
issues.append("Label format invalid")
self.log("β Label format appears invalid")
except:
self.val_format.set("β Error")
issues.append("Could not read label")
self.log("β Could not read label file")
else:
self.val_format.set("Not checked")
# Summary
self.log("\n" + "=" * 60)
if not issues:
self.log("β VALIDATION PASSED - Dataset is ready!")
self.project_status.configure(text="β Dataset validated - Ready to train", text_color="#2fa572")
else:
self.log("β VALIDATION FAILED:")
for issue in issues:
self.log(f" - {issue}")
self.project_status.configure(text="β Validation failed - Check console", text_color="#d63031")
self.log("=" * 60)
def log(self, message):
"""Log message to console"""
self.console.insert("end", message + "\n")
self.console.see("end")
self.root.update_idletasks()
def view_before(self):
"""Open before training results"""
path = os.path.join(self.project_path.get(), "test_before_training")
os.startfile(path) if os.name == 'nt' else os.system(f'open "{path}"')
def view_after(self):
"""Open after training results"""
path = os.path.join(self.project_path.get(), "test_after_training")
os.startfile(path) if os.name == 'nt' else os.system(f'open "{path}"')
def view_val(self):
"""Open validation results"""
path = os.path.join(self.project_path.get(), "val_after_training")
os.startfile(path) if os.name == 'nt' else os.system(f'open "{path}"')
def view_metrics(self):
"""Open metrics/results"""
path = os.path.join(self.project_path.get(), "finetuned_model")
os.startfile(path) if os.name == 'nt' else os.system(f'open "{path}"')
def start_training(self):
"""Start the training process"""
path = self.project_path.get()
yaml_path = self.yaml_file.get()
if not path:
messagebox.showwarning("No Project", "Please select a project folder")
return
if not yaml_path:
messagebox.showwarning("No YAML", "Please validate dataset first")
return
# Disable buttons
self.train_btn.configure(state="disabled")
self.stop_btn.configure(state="normal")
self.processing = True
# Disable view buttons
self.view_before_btn.configure(state="disabled")
self.view_after_btn.configure(state="disabled")
self.view_val_btn.configure(state="disabled")
self.view_metrics_btn.configure(state="disabled")
def train_thread():
try:
from ultralytics import YOLO
import torch
self.log("\n" + "=" * 80)
self.log("YOLO FINE-TUNING STARTED (MINI-DATASET MODE)")
self.log("=" * 80)
self.log(f"\nGPU Available: {torch.cuda.is_available()}")
if torch.cuda.is_available():
self.log(f"GPU Name: {torch.cuda.get_device_name(0)}")
train_from_scratch = self.train_from_scratch.get()
self.log(f"\nConfiguration:")
self.log(f" Project: {path}")
self.log(f" Model: {self.model_version.get()}")
self.log(f" Train from Scratch: {'YES' if train_from_scratch else 'NO (Using pretrained weights)'}")
self.log(f" Epochs: {self.epochs.get()}")
self.log(f" Fixed Learning Rate: 0.01")
self.log(f" Fixed Batch Size: 1")
self.log(f" Confidence: {self.conf_threshold.get():.2f}")
self.log(f" Device: {'GPU' if self.device.get() == '0' else 'CPU'}")
# STEP 1
self.log("\n" + "=" * 80)
self.log("[STEP 1] Testing Base Model (before training)")
self.log("=" * 80)
if train_from_scratch:
self.log("Loading YOLOv8n architecture from scratch...")
base_model = YOLO("yolov8n.yaml")
else:
self.log(f"Loading pretrained model: {self.model_version.get()}")
base_model = YOLO(self.model_version.get())
test_path = os.path.join(path, "images", "test")
if os.path.exists(test_path) and len(os.listdir(test_path)) > 0:
base_model.predict(
source=test_path,
conf=self.conf_threshold.get(),
save=True,
project=path,
name="test_before_training",
exist_ok=True,
verbose=False
)
self.log("β Saved to: test_before_training/")
self.root.after(0, lambda: self.view_before_btn.configure(state="normal"))
# STEP 2
self.log("\n" + "=" * 80)
self.log("[STEP 2] Fine-tuning on Training Images")
self.log("=" * 80)
results = base_model.train(
data=yaml_path,
epochs=self.epochs.get(),
pretrained=(not train_from_scratch),
imgsz=640,
batch=1,
lr0=0.01,
lrf=0.01,
freeze=None,
hsv_h=0.0, hsv_s=0.0, hsv_v=0.0,
degrees=0.0, translate=0.0, scale=0.0,
fliplr=0.0, flipud=0.0, mosaic=0.0, mixup=0.0, copy_paste=0.0,
patience=0,
workers=0,
device=self.device.get(),
cache=False,
amp=True,
val=True,
plots=True,
project=path,
name="finetuned_model",
exist_ok=True,
verbose=True,
save=True,
save_period=-1,
)
self.log("\nβ Training complete!")
# STEP 3
self.log("\n" + "=" * 80)
self.log("[STEP 3] Testing Fine-tuned Model")
self.log("=" * 80)
finetuned_model = YOLO(os.path.join(path, "finetuned_model", "weights", "best.pt"))
if os.path.exists(test_path):
finetuned_model.predict(
source=test_path,
conf=self.conf_threshold.get(),
save=True,
project=path,
name="test_after_training",
exist_ok=True,
verbose=False
)
self.log("β Saved to: test_after_training/")
self.root.after(0, lambda: self.view_after_btn.configure(state="normal"))
val_path = os.path.join(path, "images", "val")
if os.path.exists(val_path):
finetuned_model.predict(
source=val_path,
conf=self.conf_threshold.get(),
save=True,
project=path,
name="val_after_training",
exist_ok=True,
verbose=False
)
self.root.after(0, lambda: self.view_val_btn.configure(state="normal"))
self.root.after(0, lambda: self.view_metrics_btn.configure(state="normal"))
# Results
self.log("\n" + "=" * 80)
self.log("RESULTS SUMMARY")
self.log("=" * 80)
try:
import pandas as pd
results_csv = os.path.join(path, "finetuned_model", "results.csv")
df = pd.read_csv(results_csv)
final_map50 = df['metrics/mAP50(B)'].iloc[-1]
final_precision = df['metrics/precision(B)'].iloc[-1]
final_recall = df['metrics/recall(B)'].iloc[-1]
self.log(f"\nPerformance:")
self.log(f" mAP50: {final_map50:.4f} ({final_map50*100:.1f}%)")
self.log(f" Precision: {final_precision:.4f} ({final_precision*100:.1f}%)")
self.log(f" Recall: {final_recall:.4f} ({final_recall*100:.1f}%)")
if final_map50 > 0.8:
self.log("\nβ SUCCESS! Model learned excellently!")
elif final_map50 > 0.5:
self.log("\n~ Good learning achieved")
else:
self.log("\nβ May need more data or epochs")
except Exception as e:
self.log(f"Could not read metrics: {e}")
self.log("\nβ Training complete! Use buttons above to view results")
except Exception as e:
self.log(f"\nβ ERROR: {e}")
import traceback
self.log(traceback.format_exc())
finally:
self.processing = False
self.root.after(0, lambda: self.train_btn.configure(state="normal"))
self.root.after(0, lambda: self.stop_btn.configure(state="disabled"))
self.training_thread = threading.Thread(target=train_thread)
self.training_thread.daemon = True
self.training_thread.start()
def show_about(self):
"""Show About dialog"""
about_window = ctk.CTkToplevel(self.root)
about_window.title("About")
about_window.geometry("350x220")
about_window.resizable(False, False)
# Center the window
about_window.transient(self.root)
about_window.grab_set()
# Content frame with tighter padding
content_frame = ctk.CTkFrame(about_window, fg_color="transparent")
content_frame.pack(fill="both", expand=True, padx=25, pady=20)
# Title
title = ctk.CTkLabel(content_frame, text="About",
font=ctk.CTkFont(size=18, weight="bold"))
title.pack(pady=(0, 12))
# Developed by
dev_label = ctk.CTkLabel(content_frame, text="Developed by RoadwayVR",
font=ctk.CTkFont(size=13, weight="bold"),
text_color="lightblue")
dev_label.pack(pady=(0, 10))
# Application info
app_label = ctk.CTkLabel(content_frame, text="Application:",
font=ctk.CTkFont(size=10),
text_color="gray60")
app_label.pack(pady=(0, 2))
app_name = ctk.CTkLabel(content_frame, text="SimJamCV (YOLO Trainer)",
font=ctk.CTkFont(size=12, weight="bold"))
app_name.pack(pady=(0, 8))
# Version
version_label = ctk.CTkLabel(content_frame, text="Version: 1.0",
font=ctk.CTkFont(size=11))
version_label.pack(pady=(0, 8))
# Description
desc_label = ctk.CTkLabel(content_frame,
text="For educational fine-tuning of\nYOLO on small datasets",
font=ctk.CTkFont(size=10),
text_color="gray60",
justify="center")
desc_label.pack(pady=(0, 12))
# Close button
close_btn = ctk.CTkButton(content_frame, text="Close",
command=about_window.destroy, width=100, height=28)
close_btn.pack()
def show_dependencies(self):
"""Show Dependencies dialog"""
dep_window = ctk.CTkToplevel(self.root)
dep_window.title("Dependency Manager")
dep_window.geometry("700x500")
dep_window.resizable(False, False)
# Center the window
dep_window.transient(self.root)
dep_window.grab_set()
# Main content frame
main_frame = ctk.CTkFrame(dep_window)
main_frame.pack(fill="both", expand=True, padx=20, pady=20)
# Header
header_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
header_frame.pack(fill="x", pady=(0, 15))
title = ctk.CTkLabel(header_frame, text="π¦ Package Dependencies",
font=ctk.CTkFont(size=18, weight="bold"))
title.pack(side="left")
subtitle = ctk.CTkLabel(header_frame,
text="Check and install required Python packages",
text_color="gray60")
subtitle.pack(side="left", padx=(15, 0))
# Console/Output area
console_frame = ctk.CTkFrame(main_frame)
console_frame.pack(fill="both", expand=True, pady=(0, 15))
console = ctk.CTkTextbox(console_frame, font=ctk.CTkFont(family="Courier", size=11),
wrap="word", state="normal")
console.pack(fill="both", expand=True, padx=10, pady=10)
def log_message(message):
console.configure(state="normal")
console.insert("end", message + "\n")
console.see("end")
console.configure(state="disabled")
dep_window.update()
# Progress bar
progress_bar = ctk.CTkProgressBar(main_frame)
progress_bar.pack(fill="x", pady=(0, 15))
progress_bar.set(0)
# Buttons frame
button_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
button_frame.pack(fill="x")
check_btn = ctk.CTkButton(button_frame, text="π Check Status", width=150)
check_btn.pack(side="left", padx=(0, 10))
install_btn = ctk.CTkButton(button_frame, text="β¬ Install Missing", width=150,
fg_color="green", hover_color="darkgreen")
install_btn.pack(side="left", padx=(0, 10))
close_btn = ctk.CTkButton(button_frame, text="Close", width=100,
command=dep_window.destroy)
close_btn.pack(side="right")
# Required packages
REQUIRED_PACKAGES = {
'customtkinter': 'customtkinter>=5.2.0',
'PIL': 'pillow>=10.0.0',
'cv2': 'opencv-python>=4.8.0',
'numpy': 'numpy>=1.24.0',
'pandas': 'pandas>=2.0.0',
'ultralytics': 'ultralytics>=8.0.0',
'torch': 'torch>=2.0.0',
}
def check_dependencies():
"""Check which dependencies are installed"""
console.configure(state="normal")
console.delete("1.0", "end")
console.configure(state="disabled")
log_message("Checking dependencies...\n")
log_message("DEPENDENCY STATUS")
log_message("=" * 50 + "\n")
installed = []
missing = []
for import_name, pip_name in REQUIRED_PACKAGES.items():
try:
__import__(import_name)
installed.append(pip_name.split('>=')[0])
log_message(f"β {pip_name.split('>=')[0]}")
except ImportError:
missing.append(pip_name)
if missing:
log_message("\nβ MISSING:")
for pkg in missing:
log_message(f" β’ {pkg}")
log_message(f"\nβ Need to install {len(missing)} package(s)")
install_btn.configure(state="normal")
else:
log_message("\nπ All dependencies are installed!")
install_btn.configure(state="disabled")
progress_bar.set(1.0)
return installed, missing
def install_missing():
"""Install missing packages"""
install_btn.configure(state="disabled")
check_btn.configure(state="disabled")
log_message("\n" + "=" * 50)
log_message("INSTALLING MISSING PACKAGES")
log_message("=" * 50 + "\n")
_, missing = check_dependencies_silent()
if not missing:
log_message("All packages already installed!")
check_btn.configure(state="normal")
return
total = len(missing)
for i, package in enumerate(missing, 1):
progress_bar.set(i / total)
log_message(f"Installing {i}/{total}: {package}...")
try:
import subprocess
result = subprocess.run(
[sys.executable, "-m", "pip", "install", package],
capture_output=True,
text=True,
timeout=300
)
if result.returncode == 0:
log_message(f" β {package} installed successfully")
else:
log_message(f" β Failed to install {package}")
except Exception as e:
log_message(f" β Error: {str(e)}")
log_message("\n" + "=" * 50)
log_message("Verifying installation...")
check_dependencies()
check_btn.configure(state="normal")
install_btn.configure(state="normal")
def check_dependencies_silent():
"""Silent check for install function"""
installed = []
missing = []
for import_name, pip_name in REQUIRED_PACKAGES.items():
try:
__import__(import_name)
installed.append(pip_name.split('>=')[0])
except ImportError:
missing.append(pip_name)
return installed, missing
# Connect button commands
check_btn.configure(command=check_dependencies)
install_btn.configure(command=lambda: threading.Thread(target=install_missing, daemon=True).start())
# Initial check
check_dependencies()
def stop_training(self):
"""Stop training"""
if self.processing:
self.log("\nβ Stop requested...")
self.processing = False
if __name__ == "__main__":
root = ctk.CTk()
app = YOLOv8MiniDatasetTrainer(root)
root.mainloop()
InΒ [95]:
with open("main.py", "r", encoding="utf-8") as f:
print(f.read())
"""
Traffic Analysis Application - Main GUI
Modern interface for vehicle detection and intersection performance analytics
"""
# ββ stdlib only ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
import os
import sys
import subprocess
import importlib
import threading
from tkinter import Tk, messagebox, ttk
import tkinter as tk
# ββ Bootstrap: install missing packages before anything else βββββββββββββββββ
REQUIRED_PACKAGES = {
"customtkinter": "customtkinter>=5.2.0",
"PIL": "pillow>=10.0.0",
"cv2": "opencv-python>=4.8.0",
"numpy": "numpy>=1.24.0",
"pandas": "pandas>=2.0.0",
"ultralytics": "ultralytics>=8.0.0",
"supervision": "supervision>=0.17.0",
}
def _check_missing():
missing = []
for import_name, pip_spec in REQUIRED_PACKAGES.items():
try:
importlib.import_module(import_name)
except ImportError:
missing.append(pip_spec)
return missing
def _install(pip_spec, log_fn=print):
"""Install a package, handling PEP 668 / uv-managed environments."""
log_fn(f" Installing {pip_spec} β¦")
base_cmd = [sys.executable, "-m", "pip", "install", pip_spec]
# First attempt: --break-system-packages (required for uv/PEP 668 envs)
result = subprocess.run(
base_cmd + ["--break-system-packages"],
capture_output=True, text=True, timeout=300
)
if result.returncode != 0:
# Second attempt: also add --user (for permission-restricted installs)
log_fn(f" Retrying with --user flag β¦")
result = subprocess.run(
base_cmd + ["--user", "--break-system-packages"],
capture_output=True, text=True, timeout=300
)
if result.returncode == 0:
log_fn(f" β {pip_spec}")
return True
else:
log_fn(f" β {pip_spec}: {result.stderr.strip()}")
return False
class _BootstrapWindow:
"""Minimal tkinter splash shown while packages are installing."""
def __init__(self, packages):
self.root = Tk()
self.root.title("SimJam CV Analytics β First-Run Setup")
self.root.geometry("540x360")
self.root.resizable(False, False)
self.root.configure(bg="#1a1a2e")
tk.Label(self.root, text="SimJam ComputerVision Analytics",
font=("Helvetica", 16, "bold"), fg="white", bg="#1a1a2e").pack(pady=(24, 4))
tk.Label(self.root, text="Installing required packages β please wait β¦",
font=("Helvetica", 10), fg="#aaaacc", bg="#1a1a2e").pack(pady=(0, 16))
self.log = tk.Text(self.root, height=10, bg="#0d0d1a", fg="#ccccff",
font=("Courier", 9), relief="flat", state="disabled")
self.log.pack(fill="both", expand=True, padx=20)
self.bar = ttk.Progressbar(self.root, maximum=len(packages), mode="determinate")
self.bar.pack(fill="x", padx=20, pady=12)
self.status = tk.Label(self.root, text="Starting β¦",
font=("Helvetica", 9), fg="#888899", bg="#1a1a2e")
self.status.pack(pady=(0, 12))
self.success = False
self._packages = packages
self.root.after(100, self._run)
def _append(self, text):
self.log.configure(state="normal")
self.log.insert("end", text + "\n")
self.log.see("end")
self.log.configure(state="disabled")
self.root.update_idletasks()
def _run(self):
failed = []
for i, pkg in enumerate(self._packages):
self.status.configure(text=f"Installing {i+1}/{len(self._packages)}: {pkg.split('>=')[0]}")
ok = _install(pkg, self._append)
if not ok:
failed.append(pkg)
self.bar["value"] = i + 1
self.root.update_idletasks()
if failed:
self._append(f"\nβ {len(failed)} package(s) failed β check your internet connection.")
self.status.configure(text="Some packages failed. The app may not work fully.")
messagebox.showwarning(
"Installation Warning",
f"The following packages could not be installed:\n\n" +
"\n".join(failed) +
"\n\nSome features may be unavailable.",
parent=self.root
)
else:
self._append("\nβ All packages installed successfully!")
self.status.configure(text="Done! Launching application β¦")
self.success = True
self.root.after(1200, self.root.destroy)
def run(self):
self.root.mainloop()
return self.success
def bootstrap():
"""Install any missing packages, showing a GUI splash if needed."""
missing = _check_missing()
if not missing:
print("β All dependencies already installed.")
return True
print(f"Missing {len(missing)} package(s): {', '.join(p.split('>=')[0] for p in missing)}")
print("Opening installer window β¦")
win = _BootstrapWindow(missing)
win.run()
# Final verification
still_missing = _check_missing()
if still_missing:
print(f"Still missing: {still_missing}")
return False
return True
# ββ Run bootstrap BEFORE importing anything third-party ββββββββββββββββββββββ
if __name__ == "__main__" or True: # also works when imported during testing
_bootstrap_ok = bootstrap()
# ββ Now it is safe to import third-party packages ββββββββββββββββββββββββββββ
import customtkinter as ctk # noqa: E402
from tkinter import messagebox # noqa: E402 (re-import for clarity)
from PIL import Image # noqa: E402
from dependency_manager import DependencyManager # noqa: E402
import detection_module # noqa: E402
import analytics_module # noqa: E402
# ββ Main Application ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
class TrafficAnalysisApp(ctk.CTk):
def __init__(self):
super().__init__()
self.title("SimJamComputerVisionAnalytics")
self.geometry("1200x800")
ctk.set_appearance_mode("dark")
ctk.set_default_color_theme("blue")
self.grid_columnconfigure(0, weight=1)
self.grid_rowconfigure(0, weight=1)
self.main_container = ctk.CTkFrame(self, corner_radius=0)
self.main_container.grid(row=0, column=0, sticky="nsew")
self.main_container.grid_columnconfigure(0, weight=1)
self.main_container.grid_rowconfigure(1, weight=1)
self.create_header()
self.tabview = ctk.CTkTabview(self.main_container, corner_radius=10)
self.tabview.grid(row=1, column=0, sticky="nsew", padx=20, pady=(10, 20))
self.tabview.add("Detection & Tracking")
self.tabview.add("Performance Analytics")
for tab_name in ("Detection & Tracking", "Performance Analytics"):
self.tabview.tab(tab_name).grid_columnconfigure(0, weight=1)
self.tabview.tab(tab_name).grid_rowconfigure(0, weight=1)
self.detection_frame = detection_module.DetectionTab(
self.tabview.tab("Detection & Tracking")
)
self.detection_frame.grid(row=0, column=0, sticky="nsew", padx=10, pady=10)
self.analytics_frame = analytics_module.AnalyticsTab(
self.tabview.tab("Performance Analytics")
)
self.analytics_frame.grid(row=0, column=0, sticky="nsew", padx=10, pady=10)
self.create_status_bar()
# ββ Header ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def create_header(self):
header_frame = ctk.CTkFrame(self.main_container, height=80, corner_radius=0)
header_frame.grid(row=0, column=0, sticky="ew")
header_frame.grid_columnconfigure(1, weight=1)
header_frame.grid_propagate(False)
left = ctk.CTkFrame(header_frame, fg_color="transparent")
left.grid(row=0, column=0, padx=30, pady=10, sticky="w", rowspan=2)
logo_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "RoadwayVR.jpg")
if os.path.exists(logo_path):
try:
img = Image.open(logo_path).resize((50, 50), Image.Resampling.LANCZOS)
self.logo_photo = ctk.CTkImage(light_image=img, dark_image=img, size=(50, 50))
ctk.CTkLabel(left, image=self.logo_photo, text="").pack(side="left", padx=(0, 15))
except Exception as e:
print(f"Could not load logo: {e}")
ctk.CTkLabel(left, text="SimJam ComputerVision Analytics",
font=ctk.CTkFont(size=28, weight="bold")).pack(side="left")
ctk.CTkButton(header_frame, text="βΉ About", width=100,
command=self.show_about,
fg_color="transparent", border_width=2
).grid(row=0, column=2, rowspan=2, padx=(10, 10), pady=10, sticky="e")
ctk.CTkButton(header_frame, text="π§ Dependencies", width=130,
command=self.show_dependencies,
fg_color=("gray70", "gray30"), border_width=2
).grid(row=0, column=3, rowspan=2, padx=(0, 30), pady=10, sticky="e")
# ββ Status bar ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def create_status_bar(self):
self.status_frame = ctk.CTkFrame(self.main_container, height=30, corner_radius=0)
self.status_frame.grid(row=2, column=0, sticky="ew")
self.status_frame.grid_propagate(False)
self.status_label = ctk.CTkLabel(self.status_frame, text="Ready",
font=ctk.CTkFont(size=11))
self.status_label.pack(side="left", padx=20, pady=5)
def update_status(self, message):
self.status_label.configure(text=message)
self.update_idletasks()
# ββ About dialog ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def show_about(self):
w = ctk.CTkToplevel(self)
w.title("About")
w.geometry("350x220")
w.resizable(False, False)
w.transient(self)
w.grab_set()
f = ctk.CTkFrame(w, fg_color="transparent")
f.pack(fill="both", expand=True, padx=25, pady=20)
ctk.CTkLabel(f, text="About", font=ctk.CTkFont(size=18, weight="bold")).pack(pady=(0, 12))
ctk.CTkLabel(f, text="Developed by RoadwayVR",
font=ctk.CTkFont(size=13, weight="bold"),
text_color="lightblue").pack(pady=(0, 10))
ctk.CTkLabel(f, text="Application:", font=ctk.CTkFont(size=10),
text_color="gray60").pack(pady=(0, 2))
ctk.CTkLabel(f, text="SimJamCV Analytics",
font=ctk.CTkFont(size=12, weight="bold")).pack(pady=(0, 8))
ctk.CTkLabel(f, text="Version: 1.0",
font=ctk.CTkFont(size=11)).pack(pady=(0, 8))
ctk.CTkLabel(f, text="Traffic analysis tool for\nvehicle detection and analytics",
font=ctk.CTkFont(size=10), text_color="gray60",
justify="center").pack(pady=(0, 12))
ctk.CTkButton(f, text="Close", command=w.destroy, width=100, height=28).pack()
# ββ Dependency manager dialog βββββββββββββββββββββββββββββββββββββββββββββ
def show_dependencies(self):
dw = ctk.CTkToplevel(self)
dw.title("Dependency Manager")
dw.geometry("600x500")
dw.transient(self)
dw.grab_set()
dw.update_idletasks()
x = (dw.winfo_screenwidth() // 2) - 300
y = (dw.winfo_screenheight() // 2) - 250
dw.geometry(f"+{x}+{y}")
content = ctk.CTkFrame(dw)
content.pack(fill="both", expand=True, padx=20, pady=20)
ctk.CTkLabel(content, text="π¦ Package Dependencies",
font=ctk.CTkFont(size=20, weight="bold")).pack(pady=(10, 5))
ctk.CTkLabel(content, text="Check and install required Python packages",
font=ctk.CTkFont(size=12), text_color="gray").pack(pady=(0, 15))
status_text = ctk.CTkTextbox(content, wrap="word",
font=ctk.CTkFont(family="Courier", size=11))
status_text.pack(fill="both", expand=True, padx=10, pady=10)
progress = ctk.CTkProgressBar(content)
progress.pack(fill="x", padx=10, pady=5)
progress.set(0)
btn_frame = ctk.CTkFrame(content, fg_color="transparent")
btn_frame.pack(fill="x", padx=10, pady=10)
dep_manager = DependencyManager()
def log(msg):
status_text.insert("end", f"{msg}\n")
status_text.see("end")
dw.update_idletasks()
def check_deps():
status_text.delete("1.0", "end")
log("Checking dependenciesβ¦\n")
status_text.insert("end", dep_manager.get_status_report())
installed, missing = dep_manager.check_dependencies()
if missing:
install_btn.configure(state="normal")
progress.set(len(installed) / (len(installed) + len(missing)))
else:
install_btn.configure(state="disabled")
progress.set(1.0)
def install_deps():
install_btn.configure(state="disabled", text="Installingβ¦")
check_btn.configure(state="disabled")
close_btn.configure(state="disabled")
status_text.delete("1.0", "end")
log("Starting installationβ¦\n")
def on_complete(results):
log("\n" + "=" * 50)
log("INSTALLATION COMPLETE")
log("=" * 50 + "\n")
ok = sum(1 for s, _ in results.values() if s)
fail = len(results) - ok
log(f"β Successful: {ok}")
log(f"β Failed: {fail}\n")
if fail:
log("Failed packages:")
for pkg, (s, _) in results.items():
if not s:
log(f" β’ {pkg}")
check_deps()
install_btn.configure(text="β¬ Install Missing")
check_btn.configure(state="normal")
close_btn.configure(state="normal")
if not fail:
messagebox.showinfo("Success",
"All dependencies installed successfully!\n\n"
"You can now use all features of the application.",
parent=dw)
dep_manager.install_all_missing_async(
progress_callback=log, completion_callback=on_complete)
check_btn = ctk.CTkButton(btn_frame, text="π Check Status",
command=check_deps, width=150)
check_btn.pack(side="left", padx=5)
install_btn = ctk.CTkButton(btn_frame, text="β¬ Install Missing",
command=install_deps, width=150,
fg_color=("green", "darkgreen"))
install_btn.pack(side="left", padx=5)
close_btn = ctk.CTkButton(btn_frame, text="Close",
command=dw.destroy, width=100)
close_btn.pack(side="right", padx=5)
check_deps()
# ββ Entry point βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def main():
app = TrafficAnalysisApp()
app.mainloop()
if __name__ == "__main__":
main()
InΒ [Β ]: