helios.py

import sys
import threading
import time

import cv2
import numpy as np

def build_model(is_cuda):
        net = cv2.dnn.readNet("yolov5s.onnx") # TODO sparseml
        if is_cuda:
                print("Attempty to use CUDA")
                net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)
                net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA_FP16)
        else:
                print("Running on CPU")
                net.setPreferableBackend(cv2.dnn.DNN_BACKEND_OPENCV)
                net.setPreferableTarget(cv2.dnn.DNN_TARGET_CPU)
        return net

INPUT_WIDTH = 640
INPUT_HEIGHT = 640
SCORE_THRESHOLD = 0.2
NMS_THRESHOLD = 0.4
CONFIDENCE_THRESHOLD = 0.4

def detect(image, net):
        blob = cv2.dnn.blobFromImage(image, 1/255.0, (INPUT_WIDTH, INPUT_HEIGHT), swapRB=True, crop=False)
        net.setInput(blob)
        preds = net.forward()
        return preds

def load_classes():
        class_list = []
        with open("classes.txt", "r") as f:
                class_list = [cname.strip() for cname in f.readlines()]
        return class_list

class_list = load_classes()

def wrap_detection(input_image, output_data):
        class_ids = []
        confidences = []
        boxes = []

        rows = output_data.shape[0]

        image_width, image_height, _ = input_image.shape

        x_factor = image_width / INPUT_WIDTH
        y_factor =  image_height / INPUT_HEIGHT

        for r in range(rows):
                row = output_data[r]
                confidence = row[4]
                if confidence >= 0.4:

                        classes_scores = row[5:]
                        _, _, _, max_indx = cv2.minMaxLoc(classes_scores)
                        class_id = max_indx[1]
                        if (classes_scores[class_id] > .25):

                                confidences.append(confidence)

                                class_ids.append(class_id)

                                x, y, w, h = row[0].item(), row[1].item(), row[2].item(), row[3].item()
                                left = int((x - 0.5 * w) * x_factor)
                                top = int((y - 0.5 * h) * y_factor)
                                width = int(w * x_factor)
                                height = int(h * y_factor)
                                box = np.array([left, top, width, height])
                                boxes.append(box)

        indexes = cv2.dnn.NMSBoxes(boxes, confidences, 0.25, 0.45)

        result_class_ids = []
        result_confidences = []
        result_boxes = []

        for i in indexes:
                result_confidences.append(confidences[i])
                result_class_ids.append(class_ids[i])
                result_boxes.append(boxes[i])

        return result_class_ids, result_confidences, result_boxes

def format_yolov5(frame):
        row, col, _ = frame.shape
        _max = max(col, row)
        result = np.zeros((_max, _max, 3), np.uint8)
        result[0:row, 0:col] = frame
        return result

colors = [(255, 255, 0), (0, 255, 0), (0, 255, 255), (255, 0, 0)]

is_cuda = len(sys.argv) > 1 and sys.argv[1] == "cuda"

class Helios:

        def __init__(self):
                print("Starting helios...")
                
                self.net = build_model(is_cuda)
                self.capture = cv2.VideoCapture(0)
                print('opened',self.capture.isOpened())

                self.start_time = time.time_ns()
                self.frame_count = 0
                self.total_frames = 0
                self.fps = -1

        def start(self):
                # This function runs in a thread to retrieve frames using opencv.
                # We then run inference on this frame and trigger the buzzes.
                
                #insert code to capture frame using opencv -> cvmat or numpy array √
                
                # Make YOLO inference -> bounding boxes √

                # Optionally display frame with bouding box overlay √

                # Pass bounding boxes to buzzer function ?
                while True:

                        _, frame = self.capture.read()
                        if frame is None:
                                print("End of stream")
                                break

                        inputImage = format_yolov5(frame)
                        outs = detect(inputImage, self.net)

                        class_ids, confidences, boxes = wrap_detection(inputImage, outs[0])

                        self.frame_count += 1
                        self.total_frames += 1

                        for (classid, confidence, box) in zip(class_ids, confidences, boxes):
                                 if class_list[classid] == "cup":
                                         color = colors[int(classid) % len(colors)]
                                         cv2.rectangle(frame, box, color, 2)
                                         cv2.rectangle(frame, (box[0], box[1] - 20), (box[0] + box[2], box[1]), color, -1)
                                         cv2.putText(frame, class_list[classid], (box[0], box[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, .5, (0,0,0))
                                         print((box[0]+box[2])/2)
                                         # divide into 3 regions by x(0-640) and then buzz
                                         break


                        if self.frame_count >= 30:
                                end = time.time_ns()
                                self.fps = 1000000000 * self.frame_count / (end - self.start_time)
                                self.frame_count = 0
                                self.start_time = time.time_ns()

                        if self.fps > 0:
                                fps_label = "FPS: %.2f" % self.fps
                                cv2.putText(frame, fps_label, (10, 25), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)

                        cv2.imshow("output", frame)

                        if cv2.waitKey(1) > -1:
                                print("finished by user")
                                break
        
def main(argv=sys.argv):
        app = Helios()
        # todo: Considerrunning start() in a thread.
        # self.thread = threading.Thread(target=start, argv=(1,))
        # self.thread.start()
        app.start()


if __name__ == '__main__':
        main(sys.argv)