detect_wrong.py

# python detect_wrong.py --source overpass.mp4 --weights ./my_coco.pt --data ./data/my_coco.yaml

import argparse
import os
import sys
from pathlib import Path
import time
import torch
import torch.backends.cudnn as cudnn
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
from pathlib import Path
import tensorflow as tf  # For TensorBoard
from datetime import datetime

FILE = Path(__file__).resolve()
ROOT = FILE.parents[0]
if str(ROOT) not in sys.path:
    sys.path.append(str(ROOT))
ROOT = Path(os.path.relpath(ROOT, Path.cwd()))

from models.common import DetectMultiBackend
from utils.dataloaders import IMG_FORMATS, VID_FORMATS, LoadImages, LoadStreams
from utils.general import (LOGGER, check_file, check_img_size, check_imshow, check_requirements, colorstr, cv2,
                           increment_path, non_max_suppression, print_args, scale_coords, strip_optimizer, xyxy2xywh)
from utils.plots import Annotator, colors, save_one_box
from utils.torch_utils import select_device, time_sync
from pyimagesearch.centroidtracker import CentroidTracker
ct = CentroidTracker()

# # TensorBoard setup
# log_dir = "logs/" + datetime.now().strftime("%Y%m%d-%H%M%S")  # Create a unique log directory based on current time
# summary_writer = tf.summary.create_file_writer(log_dir)  # TensorBoard writer
# frame_counter = 0  # For tracking frame count

@torch.no_grad()
def run(
        weights=ROOT / 'yolov5s.pt',
        source=ROOT / 'data/images',  # file/dir/URL/glob, 0 for webcam
        data=ROOT / 'data/coco128.yaml',  # dataset.yaml path
        imgsz=(640, 640),
        conf_thres=0.25,  # confidence threshold
        iou_thres=0.45,  # NMS IOU threshold
        max_det=1000,  # maximum detections per image
        device='',
        view_img=False,  # show results
        save_txt=False,  # save results to *.txt
        save_conf=False,  # save confidences in --save-txt labels
        save_crop=False,  # save cropped prediction boxes
        nosave=False,  # do not save images/videos
        classes=None,  # filter by class: --class 0, or --class 0 2 3
        agnostic_nms=False,  # class-agnostic NMS
        augment=False,  # augmented inference
        visualize=False,  # visualize features
        update=False,  # update all models
        project=ROOT / 'runs/detect',  # save results to project/name
        name='exp',  # save results to project/name
        exist_ok=False,  # existing project/name ok, do not increment
        line_thickness=3,  # bounding box thickness (pixels)
        hide_labels=False,  # hide labels
        hide_conf=False,  # hide confidences
        half=False,  # use FP16 half-precision inference
        dnn=False,  # use OpenCV DNN for ONNX inference
):
    source = str(source)
    save_img = not nosave and not source.endswith('.txt')  # save inference images
    is_file = Path(source).suffix[1:] in (IMG_FORMATS + VID_FORMATS)
    is_url = source.lower().startswith(('rtsp://', 'rtmp://', 'http://', 'https://'))
    webcam = source.isnumeric() or source.endswith('.txt') or (is_url and not is_file)
    if is_url and is_file:
        source = check_file(source)  # download

    time_starts = time.time()
    # Directories
    save_dir = increment_path(Path(project) / name, exist_ok=exist_ok)  # increment run
    (save_dir / 'labels' if save_txt else save_dir).mkdir(parents=True, exist_ok=True)  # make dir

    # Load model
    device = select_device(device)
    model = DetectMultiBackend(weights, device=device, dnn=dnn, data=data, fp16=half)
    stride, names, pt = model.stride, model.names, model.pt
    imgsz = check_img_size(imgsz, s=stride)  # check image size

    # Dataloader
    if webcam:
        view_img = check_imshow()
        cudnn.benchmark = True  # set True to speed up constant image size inference
        dataset = LoadStreams(source, img_size=imgsz, stride=stride, auto=pt)
        bs = len(dataset)  # batch_size
    else:
        dataset = LoadImages(source, img_size=imgsz, stride=stride, auto=pt)
        bs = 1  # batch_size
    vid_path, vid_writer = [None] * bs, [None] * bs

    # Run inference
    model.warmup(imgsz=(1 if pt else bs, 3, *imgsz))  # warmup
    dt, seen = [0.0, 0.0, 0.0], 0

    time_2nd = time.time()
    print(f" preprocess time: {time_2nd*1000 - time_starts*1000}")

    for path, im, im0s, vid_cap, s in dataset:
        time_pred_starts = time.time()
        t1 = time_sync()
        im = torch.from_numpy(im).to(device)

        im = im.half() if model.fp16 else im.float()  # uint8 to fp16/32
        im /= 255  # 0 - 255 to 0.0 - 1.0
        if len(im.shape) == 3:
            im = im[None]  # expand for batch dim
        t2 = time_sync()
        dt[0] += t2 - t1

        # Inference
        visualize = increment_path(save_dir / Path(path).stem, mkdir=True) if visualize else False
        pred = model(im, augment=augment, visualize=visualize)
        t3 = time_sync()
        dt[1] += t3 - t2

        # NMS
        pred = non_max_suppression(pred, conf_thres, iou_thres, classes, agnostic_nms, max_det=max_det)
        dt[2] += time_sync() - t3

        # pred = utils.general.apply_classifier(pred, classifier_model, im, im0s)
        rects = []
        # Process predictions
        for i, det in enumerate(pred):  # per image
            seen += 1
            if webcam:  # batch_size >= 1
                p, im0, frame = path[i], im0s[i].copy(), dataset.count
                s += f'{i}: '
            else:
                p, im0, frame = path, im0s.copy(), getattr(dataset, 'frame', 0)

            p = Path(p)  # to Path
            save_path = str(save_dir / p.name)  # im.jpg
            txt_path = str(save_dir / 'labels' / p.stem) + ('' if dataset.mode == 'image' else f'_{frame}')  # im.txt
            s += '%gx%g ' % im.shape[2:]  # print string
            gn = torch.tensor(im0.shape)[[1, 0, 1, 0]]  # normalization gain whwh
            imc = im0.copy() if save_crop else im0  # for save_crop
            annotator = Annotator(im0, line_width=line_thickness, example=str(names))


            FRAME_WIDTH = im0.shape[1]
            FRAME_HEIGHT = im0.shape[0]
            ROI_MIN = int(FRAME_HEIGHT*0.65)   #change the ROI dimension according to your camera angle
            ROI_MAX = int(FRAME_HEIGHT*0.90)

            if len(det):
                # Rescale boxes from img_size to im0 size
                det[:, :4] = scale_coords(im.shape[2:], det[:, :4], im0.shape).round()

                # Print results
                for c in det[:, -1].unique():
                    n = (det[:, -1] == c).sum()  # detections per class
                    s += f"{n} {names[int(c)]}{'s' * (n > 1)}, "  # add to string

                # Write results
                for *xyxy, conf, cls in reversed(det):
                    x1 = int(xyxy[0].detach().cpu().numpy())  #get the tof left and bottom right corner of the box
                    y1 = int(xyxy[1].detach().cpu().numpy())
                    x2 = int(xyxy[2].detach().cpu().numpy())
                    y2 = int(xyxy[3].detach().cpu().numpy())
                    box_dimension = (x1,y1,x2,y2)
                    box_center = (y1+y2)/2   # we will check the movement of box center


                    if ((ROI_MIN <= box_center <= ROI_MAX) and (int(cls)==1 or int(cls)==2)): #In our dataset, only class 1 and 2 are vehicles
                        rects.append(box_dimension)  #if box enters the ROI, save its box for tracking
                   
                    if save_txt:  # Write to file
                        xywh = (xyxy2xywh(torch.tensor(xyxy).view(1, 4)) / gn).view(-1).tolist()  # normalized xywh
                        line = (cls, *xywh, conf) if save_conf else (cls, *xywh)  # label format
                        with open(f'{txt_path}.txt', 'a') as f:
                            f.write(('%g ' * len(line)).rstrip() % line + '\n')

                    if save_img or save_crop or view_img:  # Add bbox to image
                        c = int(cls)  # integer class
                        label = None if hide_labels else (names[c] if hide_conf else f'{names[c]} {conf:.2f}')
                        annotator.box_label(xyxy, label, color=colors(c, True))
                    if save_crop:
                        save_one_box(xyxy, imc, file=save_dir / 'crops' / names[c] / f'{p.stem}.jpg', BGR=True)

            # Stream results
            im0 = annotator.result()
        #
        # # Write output to TensorBoard
        # with summary_writer.as_default():
        #     # Log inference time
        #     tf.summary.scalar("Inference Time", dt[1], step=seen)
        #
        #     # Log processed image with bounding boxes
        #     img_rgb = cv2.cvtColor(im0, cv2.COLOR_BGR2RGB)  # Convert to RGB for TensorBoard
        #     tf.summary.image("Detected Image", [img_rgb], step=seen)
            
        objects, CY1, CY2 = ct.update(rects)  #send the box to tracker

        for (objectID, centroid) in objects.items():
            cy1=list(CY1.values())[objectID]
            cy2=list(CY2.values())[objectID]
            # draw both the ID of the object and the centroid of the
            # object on the output frame
            if (cy2<=cy1):                  #check whether the vehicle is incoming or outgoing
                text = "{}".format('right')
                #text = "ID: {}".format(objectID)
                cv2.putText(im0, text, (centroid[0] - 10, centroid[1] - 10),cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255, 0), 2)
                cv2.circle(im0, (centroid[0], centroid[1]), 3, (0, 0, 255), -1)
            else:
                text = "{}".format('wrong')
                #text = "ID: {}".format(objectID)
                cv2.putText(im0, text, (centroid[0] - 10, centroid[1] - 10),cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,0, 255), 2)
                cv2.circle(im0, (centroid[0], centroid[1]), 3, (0, 0, 255), -1)

        cv2.line(im0, (5,ROI_MIN), (5, ROI_MAX), (0,255,0), 3)
        cv2.line(im0, (FRAME_WIDTH - 5,ROI_MIN), (FRAME_WIDTH - 5, ROI_MAX), (0,255,0), 3)
        cv2.line(im0, (0,ROI_MIN), (FRAME_WIDTH, ROI_MIN), (0,255,0), 3)
        cv2.line(im0, (0,ROI_MAX), (FRAME_WIDTH, ROI_MAX), (0,255,0), 3)

        #if view_img:
        cv2.imshow(str(p), im0)
        cv2.waitKey(1)  # 1 millisecond

                # Save results (image with detections)
        if save_img:
            if dataset.mode == 'image':
                cv2.imwrite(save_path, im0)
            else:  # 'video' or 'stream'
                if vid_path[i] != save_path:  # new video
                    vid_path[i] = save_path
                    if isinstance(vid_writer[i], cv2.VideoWriter):
                        vid_writer[i].release()  # release previous video writer
                    if vid_cap:  # video
                        fps = vid_cap.get(cv2.CAP_PROP_FPS)
                        w = int(vid_cap.get(cv2.CAP_PROP_FRAME_WIDTH))
                        h = int(vid_cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

                    else:  # stream
                        fps, w, h = 30, im0.shape[1], im0.shape[0]
                    save_path = str(Path(save_path).with_suffix('.mp4'))  # force *.mp4 suffix on results videos
                    vid_writer[i] = cv2.VideoWriter(save_path, cv2.VideoWriter_fourcc(*'mp4v'), fps, (w, h))

                vid_writer[i].write(im0)

        # Print time (inference-only)
        #LOGGER.info(f'{s}Done. ({t3 - t2:.3f}s)')
        #print(rects)
        time_done = time.time()
        #print(f"Time 2nd half: {time_done*1000 -  time_pred_starts*1000}")

    # Print results
    t = tuple(x / seen * 1E3 for x in dt)  # speeds per image
    LOGGER.info(f'Speed: %.1fms pre-process, %.1fms inference, %.1fms NMS per image at shape {(1, 3, *imgsz)}' % t)
    if save_txt or save_img:
        s = f"\n{len(list(save_dir.glob('labels/*.txt')))} labels saved to {save_dir / 'labels'}" if save_txt else ''
        LOGGER.info(f"Results saved to {colorstr('bold', save_dir)}{s}")
    if update:
        strip_optimizer(weights)  # update model (to fix SourceChangeWarning)
#
# # Initialize lists to store metrics
# times = []
# frame_numbers = []
#
# # Example function to simulate collecting metrics
# def collect_metrics(frame_number, time_taken):
#     times.append(time_taken)
#     frame_numbers.append(frame_number)
   


def parse_opt():
    parser = argparse.ArgumentParser()
    parser.add_argument('--weights', nargs='+', type=str, default=ROOT / 'yolov5s.pt', help='model path(s)')
    parser.add_argument('--source', type=str, default=ROOT / 'data/images', help='file/dir/URL/glob, 0 for webcam')
    parser.add_argument('--data', type=str, default=ROOT / 'data/coco128.yaml', help='(optional) dataset.yaml path')
    parser.add_argument('--imgsz', '--img', '--img-size', nargs='+', type=int, default=[640], help='inference size h,w')
    parser.add_argument('--conf-thres', type=float, default=0.25, help='confidence threshold')
    parser.add_argument('--iou-thres', type=float, default=0.45, help='NMS IoU threshold')
    parser.add_argument('--max-det', type=int, default=1000, help='maximum detections per image')
    parser.add_argument('--device', default='', help='cuda device, i.e. 0 or 0,1,2,3 or cpu')
    parser.add_argument('--view-img', action='store_true', help='show results')
    parser.add_argument('--save-txt', action='store_true', help='save results to *.txt')
    parser.add_argument('--save-conf', action='store_true', help='save confidences in --save-txt labels')
    parser.add_argument('--save-crop', action='store_true', help='save cropped prediction boxes')
    parser.add_argument('--nosave', action='store_true', help='do not save images/videos')
    parser.add_argument('--classes', nargs='+', type=int, help='filter by class: --classes 0, or --classes 0 2 3')
    parser.add_argument('--agnostic-nms', action='store_true', help='class-agnostic NMS')
    parser.add_argument('--augment', action='store_true', help='augmented inference')
    parser.add_argument('--visualize', action='store_true', help='visualize features')
    parser.add_argument('--update', action='store_true', help='update all models')
    parser.add_argument('--project', default=ROOT / 'runs/detect', help='save results to project/name')
    parser.add_argument('--name', default='exp', help='save results to project/name')
    parser.add_argument('--exist-ok', action='store_true', help='existing project/name ok, do not increment')
    parser.add_argument('--line-thickness', default=3, type=int, help='bounding box thickness (pixels)')
    parser.add_argument('--hide-labels', default=False, action='store_true', help='hide labels')
    parser.add_argument('--hide-conf', default=False, action='store_true', help='hide confidences')
    parser.add_argument('--half', action='store_true', help='use FP16 half-precision inference')
    parser.add_argument('--dnn', action='store_true', help='use OpenCV DNN for ONNX inference')
    opt = parser.parse_args()
    opt.imgsz *= 2 if len(opt.imgsz) == 1 else 1  # expand
    print_args(vars(opt))
    return opt

# # Assuming `detected_objects` is a list of tuples containing (x, y) coordinates of detected objects
# detected_objects = [(50, 100), (200, 300), (400, 500)]
#
# # Create a grid of points
# grid_x, grid_y = np.mgrid[0:600:100j, 0:800:100j]
#
# # Calculate the density of detected objects at each point
# heatmap_data = np.histogram2d([point[0] for point in detected_objects],
#                               [point[1] for point in detected_objects],
#                               bins=(np.arange(grid_x.max()+1), np.arange(grid_y.max()+1)),
#                               range=((0, grid_x.max()), (0, grid_y.max())))[0]
#
# # Plot the heatmap
# plt.figure(figsize=(10, 6))
# sns.heatmap(heatmap_data, cmap='viridis')
# plt.title('Object Detection Heatmap')
# plt.show()
#
# # Save the heatmap as a PNG file
# plt.savefig("heatmap.png", dpi=300)


def main(opt):
    check_requirements(exclude=('tensorboard', 'thop'))
    run(**vars(opt))


if __name__ == "__main__":
    opt = parse_opt()
    main(opt)