Heute werden wir lernen, wie man den Straßenverkehr basierend auf Computer Vision und ohne umfangreiche Deep-Learning-Algorithmen zählt.
In diesem Tutorial verwenden wir nur Python und OpenCV mit der ziemlich einfachen Idee der Bewegungserkennung mithilfe des Hintergrund Subtraktionsalgorithmus.
Hier ist unser Plan:
- Verstehen Sie die Hauptidee von Hintergrund Subtraktionsalgorithmen, die für die Vordergrunderkennung verwendet werden.
- OpenCV-Bildfilter.
- Objekterkennung durch Konturen.
- Aufbau einer Verarbeitungspipeline zur weiteren Datenmanipulation.
| # import needed modules | |
| import os | |
| import csv | |
| import numpy as np | |
| import logging | |
| import logging.handlers | |
| import math | |
| import sys | |
| import random | |
| import numpy as np | |
| import skvideo.io | |
| import cv2 | |
| import matplotlib.pyplot as plt | |
| from IPython.display import HTML | |
| from base64 import b64encode | |
| # without this some strange errors happen | |
| cv2.ocl.setUseOpenCL(False) | |
| random.seed(123) | |
| # setup logging | |
| def init_logging(level=logging.INFO): | |
| main_logger = logging.getLogger() | |
| for hnd in main_logger.handlers: | |
| main_logger.removeHandler(hnd) | |
| formatter = logging.Formatter( | |
| fmt='%(asctime)s.%(msecs)03d %(levelname)-8s [%(name)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S') | |
| handler_stream = logging.StreamHandler(sys.stdout) | |
| handler_stream.setFormatter(formatter) | |
| main_logger.addHandler(handler_stream) | |
| main_logger.setLevel(level) | |
| return main_logger |
Hintergrund Subtraktionsalgorithmen
Es gibt viele verschiedene Algorithmen für die Hintergrundsubtraktion, aber die Hauptidee davon ist sehr einfach.
Nehmen wir an, Sie haben ein Video von Ihrem Zimmer und in einigen Bildern dieses Videos sind keine Menschen und Haustiere zu sehen. Im Grunde genommen ist es statisch. Nennen wir es background_layer. Um Objekte zu erhalten, die sich auf dem Video bewegen, müssen wir nur:
foreground_objects = current_frame - background_layer
In einigen Fällen können wir jedoch keinen statischen Rahmen erhalten, da sich die Beleuchtung ändern kann oder einige Objekte von jemandem bewegt werden oder immer eine Bewegung vorhanden ist usw. In solchen Fällen speichern wir eine bestimmte Anzahl von Rahmen und versuchen herauszufinden, welches der Pixel sind für die meisten von ihnen gleich, dann werden diese Pixel Teil von background_layer. Unterschied im Allgemeinen, wie wir diese Hintergrundschicht erhalten, und zusätzliche Filterung, die wir verwenden, um die Auswahl genauer zu machen.
In dieser Lektion verwenden wir den MOG-Algorithmus für die Hintergrundsubtraktion. Nach der Verarbeitung sieht er folgendermaßen aus:
| def train_bg_subtractor(inst, cap, num=500): | |
| ''' | |
| BG substractor need process some amount of frames to start giving result | |
| ''' | |
| print ('Training BG Subtractor…') | |
| i = 0 | |
| for frame in cap: | |
| inst.apply(frame, None, 0.001) | |
| i += 1 | |
| if i >= num: | |
| return cap | |
| VIDEO_SOURCE = "road.mp4" | |
| bg_subtractor = cv2.createBackgroundSubtractorMOG2( | |
| history=500, detectShadows=True) | |
| # Set up image source | |
| cap = skvideo.io.vreader(VIDEO_SOURCE) | |
| # skipping 500 frames to train bg subtractor | |
| train_bg_subtractor(bg_subtractor, cap, num=500) | |
| frame = next(cap) | |
| fg_mask = bg_subtractor.apply(frame, None, 0.001) | |
| plt.figure(figsize=(12,12)) | |
| plt.imshow(fg_mask) | |
| plt.show() |
Filtern
Für unseren Fall benötigen wir folgende Filter: Threshold, Erode, Dilate, Opening, Closing. Bitte gehen Sie über Links und lesen Sie über jeden von ihnen und schauen Sie, wie sie funktionieren (um nicht zu kopieren / einfügen).
Jetzt werden wir sie verwenden, um etwas Rauschen auf dem Vordergrund zu entfernen. Zuerst verwenden wir Closing, um Lücken in Bereichen zu entfernen, dann Opening, um 1–2 px Punkte zu entfernen, und danach Dilate, um das Objekt mutiger zu machen.
| def filter_mask(img): | |
| ''' | |
| This filters are hand-picked just based on visual tests | |
| ''' | |
| kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2, 2)) | |
| # Fill any small holes | |
| closing = cv2.morphologyEx(img, cv2.MORPH_CLOSE, kernel) | |
| # Remove noise | |
| opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel) | |
| # Dilate to merge adjacent blobs | |
| dilation = cv2.dilate(opening, kernel, iterations=2) | |
| return dilation | |
| bg_subtractor = cv2.createBackgroundSubtractorMOG2( | |
| history=500, detectShadows=True) | |
| # Set up image source | |
| cap = skvideo.io.vreader(VIDEO_SOURCE) | |
| # skipping 500 frames to train bg subtractor | |
| train_bg_subtractor(bg_subtractor, cap, num=500) | |
| frame = next(cap) | |
| fg_mask = bg_subtractor.apply(frame, None, 0.001) | |
| fg_mask[fg_mask < 240] = 0 | |
| fg_mask = filter_mask(fg_mask) | |
| plt.figure(figsize=(12,12)) | |
| plt.imshow(fg_mask) | |
| plt.show() |
Objekterkennung durch Konturen
Zu diesem Zweck verwenden wir die Standardmethode cv2.findContours mit folgenden Parametern:
cv2.CV_RETR_EXTERNAL — get only outer contours. cv2.CV_CHAIN_APPROX_TC89_L1 - use Teh-Chin chain approximation algorithm (faster)
| def get_centroid(x, y, w, h): | |
| x1 = int(w / 2) | |
| y1 = int(h / 2) | |
| cx = x + x1 | |
| cy = y + y1 | |
| return (cx, cy) | |
| class ContourDetection: | |
| ''' | |
| Detecting moving objects. | |
| Purpose of this processor is to subtrac background, get moving objects | |
| and detect them with a cv2.findContours method, and then filter off-by | |
| width and height. | |
| bg_subtractor – background subtractor isinstance. | |
| min_contour_width – min bounding rectangle width. | |
| min_contour_height – min bounding rectangle height. | |
| save_image – if True will save detected objects mask to file. | |
| image_dir – where to save images(must exist). | |
| ''' | |
| def __init__(self, bg_subtractor, min_contour_width=35, min_contour_height=35, save_image=False, image_dir='images'): | |
| super(ContourDetection, self).__init__() | |
| self.bg_subtractor = bg_subtractor | |
| self.min_contour_width = min_contour_width | |
| self.min_contour_height = min_contour_height | |
| self.save_image = save_image | |
| self.image_dir = image_dir | |
| def filter_mask(self, img, a=None): | |
| ''' | |
| This filters are hand-picked just based on visual tests | |
| ''' | |
| kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2, 2)) | |
| # Fill any small holes | |
| closing = cv2.morphologyEx(img, cv2.MORPH_CLOSE, kernel) | |
| # Remove noise | |
| opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel) | |
| # Dilate to merge adjacent blobs | |
| dilation = cv2.dilate(opening, kernel, iterations=2) | |
| return dilation | |
| def detect_vehicles(self, fg_mask): | |
| matches = [] | |
| # finding external contours | |
| contours, hierarchy = cv2.findContours( | |
| fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_TC89_L1) | |
| for (i, contour) in enumerate(contours): | |
| (x, y, w, h) = cv2.boundingRect(contour) | |
| # On the exit, we add some filtering by height, width and add centroid. | |
| contour_valid = (w >= self.min_contour_width) and ( | |
| h >= self.min_contour_height) | |
| if not contour_valid: | |
| continue | |
| centroid = get_centroid(x, y, w, h) | |
| matches.append(((x, y, w, h), centroid)) | |
| return matches | |
| def __call__(self, frame): | |
| frame = frame.copy() | |
| fg_mask = self.bg_subtractor.apply(frame, None, 0.001) | |
| # just thresholding values | |
| fg_mask[fg_mask < 240] = 0 | |
| fg_mask = self.filter_mask(fg_mask, 0) | |
| return self.detect_vehicles(fg_mask) | |
| cd = ContourDetection(bg_subtractor) | |
| bg_subtractor = cv2.createBackgroundSubtractorMOG2( | |
| history=500, detectShadows=True) | |
| # Set up image source | |
| cap = skvideo.io.vreader(VIDEO_SOURCE) | |
| # skipping 500 frames to train bg subtractor | |
| train_bg_subtractor(bg_subtractor, cap, num=500) | |
| frame = next(cap) | |
| objects = cd(frame) | |
| print('Getting list of [((x,y,w,h), (xc,yc)), …]') | |
| print(objects) |
Gebäudeverarbeitungspipeline
Sie müssen verstehen, dass es in ML und CV keinen magischen Algorithmus gibt, der insgesamt erstellt wird, selbst wenn wir uns vorstellen, dass ein solcher Algorithmus existiert, würden wir ihn dennoch nicht verwenden, da er im Maßstab nicht effektiv wäre. Zum Beispiel hat Netflix vor einigen Jahren einen Wettbewerb mit dem Preis von 3 Millionen Dollar für den besten Algorithmus für Filmempfehlungen ins Leben gerufen. Und eines der Teams hat ein solches Problem geschaffen, dass es einfach nicht in großem Maßstab funktionieren konnte und daher für das Unternehmen nutzlos war. Trotzdem hat Netflix 1 Million an sie gezahlt 🙂
Jetzt werden wir eine einfache Verarbeitungspipeline erstellen, die nicht nur aus praktischen Gründen skaliert werden soll, sondern auch die gleiche Idee hat.
| class PipelineRunner(object): | |
| ''' | |
| Very simple pipline. | |
| Just run passed processors in order with passing context from one to | |
| another. | |
| You can also set log level for processors. | |
| ''' | |
| def __init__(self, pipeline=None, log_level=logging.INFO): | |
| self.pipeline = pipeline or [] | |
| self.context = {} | |
| self.log = logging.getLogger(self.__class__.__name__) | |
| self.log.setLevel(log_level) | |
| self.log_level = log_level | |
| self.set_log_level() | |
| def set_context(self, data): | |
| self.context = data | |
| def add(self, processor): | |
| if not isinstance(processor, PipelineProcessor): | |
| raise Exception( | |
| 'Processor should be an isinstance of PipelineProcessor.') | |
| processor.log.setLevel(self.log_level) | |
| self.pipeline.append(processor) | |
| def remove(self, name): | |
| for i, p in enumerate(self.pipeline): | |
| if p.__class__.__name__ == name: | |
| del self.pipeline[i] | |
| return True | |
| return False | |
| def set_log_level(self): | |
| for p in self.pipeline: | |
| p.log.setLevel(self.log_level) | |
| def run(self): | |
| for p in self.pipeline: | |
| self.context = p(self.context) | |
| self.log.debug("Frame #%d processed.", self.context['frame_number']) | |
| return self.context | |
| class PipelineProcessor(object): | |
| ''' | |
| Base class for processors. | |
| ''' | |
| def __init__(self): | |
| self.log = logging.getLogger(self.__class__.__name__) |
Als Eingabekonstruktor wird eine Liste der Prozessoren verwendet, die der Reihe nach ausgeführt werden. Jeder Prozessor, der Teil des Auftrags ist.
Wir haben bereits die Countour Detection-Klasse, müssen sie nur leicht ändern, um den Kontext zu verwenden
| def save_frame(frame, file_name, flip=True): | |
| # flip BGR to RGB | |
| if flip: | |
| cv2.imwrite(file_name, np.flip(frame, 2)) | |
| else: | |
| cv2.imwrite(file_name, frame) | |
| class ContourDetection(PipelineProcessor): | |
| ''' | |
| Detecting moving objects. | |
| Purpose of this processor is to subtrac background, get moving objects | |
| and detect them with a cv2.findContours method, and then filter off-by | |
| width and height. | |
| bg_subtractor – background subtractor isinstance. | |
| min_contour_width – min bounding rectangle width. | |
| min_contour_height – min bounding rectangle height. | |
| save_image – if True will save detected objects mask to file. | |
| image_dir – where to save images(must exist). | |
| ''' | |
| def __init__(self, bg_subtractor, min_contour_width=35, min_contour_height=35, save_image=False, image_dir='images'): | |
| super(ContourDetection, self).__init__() | |
| self.bg_subtractor = bg_subtractor | |
| self.min_contour_width = min_contour_width | |
| self.min_contour_height = min_contour_height | |
| self.save_image = save_image | |
| self.image_dir = image_dir | |
| def filter_mask(self, img, a=None): | |
| ''' | |
| This filters are hand-picked just based on visual tests | |
| ''' | |
| kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2, 2)) | |
| # Fill any small holes | |
| closing = cv2.morphologyEx(img, cv2.MORPH_CLOSE, kernel) | |
| # Remove noise | |
| opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel) | |
| # Dilate to merge adjacent blobs | |
| dilation = cv2.dilate(opening, kernel, iterations=2) | |
| return dilation | |
| def detect_vehicles(self, fg_mask, context): | |
| matches = [] | |
| # finding external contours | |
| contours, hierarchy = cv2.findContours( | |
| fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_TC89_L1) | |
| for (i, contour) in enumerate(contours): | |
| (x, y, w, h) = cv2.boundingRect(contour) | |
| contour_valid = (w >= self.min_contour_width) and ( | |
| h >= self.min_contour_height) | |
| if not contour_valid: | |
| continue | |
| centroid = get_centroid(x, y, w, h) | |
| matches.append(((x, y, w, h), centroid)) | |
| return matches | |
| def __call__(self, context): | |
| frame = context['frame'].copy() | |
| frame_number = context['frame_number'] | |
| fg_mask = self.bg_subtractor.apply(frame, None, 0.001) | |
| # just thresholding values | |
| fg_mask[fg_mask < 240] = 0 | |
| fg_mask = self.filter_mask(fg_mask, frame_number) | |
| if self.save_image: | |
| save_frame(fg_mask, self.image_dir + | |
| "/mask_%04d.png" % frame_number, flip=False) | |
| context['objects'] = self.detect_vehicles(fg_mask, context) | |
| context['fg_mask'] = fg_mask | |
| return context |
Erstellen wir nun einen Prozessor, der erkannte Objekte in verschiedenen Frames verknüpft, Pfade erstellt und Fahrzeuge zählt, die die Ausgangszone erreicht haben.
| def distance(x, y, type='euclidian', x_weight=1.0, y_weight=1.0): | |
| if type == 'euclidian': | |
| return math.sqrt(float((x[0] – y[0])**2) / x_weight + float((x[1] – y[1])**2) / y_weight) | |
| class VehicleCounter(PipelineProcessor): | |
| ''' | |
| Counting vehicles that entered in exit zone. | |
| Purpose of this class based on detected object and local cache create | |
| objects pathes and count that entered in exit zone defined by exit masks. | |
| exit_masks – list of the exit masks. | |
| path_size – max number of points in a path. | |
| max_dst – max distance between two points. | |
| ''' | |
| def __init__(self, exit_masks=[], path_size=10, max_dst=30, x_weight=1.0, y_weight=1.0): | |
| super(VehicleCounter, self).__init__() | |
| self.exit_masks = exit_masks | |
| self.vehicle_count = 0 | |
| self.path_size = path_size | |
| self.pathes = [] | |
| self.max_dst = max_dst | |
| self.x_weight = x_weight | |
| self.y_weight = y_weight | |
| def check_exit(self, point): | |
| for exit_mask in self.exit_masks: | |
| try: | |
| if exit_mask[point[1]][point[0]] == 255: | |
| return True | |
| except: | |
| return True | |
| return False | |
| def __call__(self, context): | |
| objects = context['objects'] | |
| context['exit_masks'] = self.exit_masks | |
| context['pathes'] = self.pathes | |
| context['vehicle_count'] = self.vehicle_count | |
| if not objects: | |
| return context | |
| points = np.array(objects)[:, 0:2] | |
| points = points.tolist() | |
| # add new points if pathes is empty | |
| if not self.pathes: | |
| for match in points: | |
| self.pathes.append([match]) | |
| else: | |
| # link new points with old pathes based on minimum distance between | |
| # points | |
| new_pathes = [] | |
| for path in self.pathes: | |
| _min = 999999 | |
| _match = None | |
| for p in points: | |
| if len(path) == 1: | |
| # distance from last point to current | |
| d = distance(p[0], path[–1][0]) | |
| else: | |
| # based on 2 prev points predict next point and calculate | |
| # distance from predicted next point to current | |
| xn = 2 * path[–1][0][0] – path[–2][0][0] | |
| yn = 2 * path[–1][0][1] – path[–2][0][1] | |
| d = distance( | |
| p[0], (xn, yn), | |
| x_weight=self.x_weight, | |
| y_weight=self.y_weight | |
| ) | |
| if d < _min: | |
| _min = d | |
| _match = p | |
| if _match and _min <= self.max_dst: | |
| points.remove(_match) | |
| path.append(_match) | |
| new_pathes.append(path) | |
| # do not drop path if current frame has no matches | |
| if _match is None: | |
| new_pathes.append(path) | |
| self.pathes = new_pathes | |
| # add new pathes | |
| if len(points): | |
| for p in points: | |
| # do not add points that already should be counted | |
| if self.check_exit(p[1]): | |
| continue | |
| self.pathes.append([p]) | |
| # save only last N points in path | |
| for i, _ in enumerate(self.pathes): | |
| self.pathes[i] = self.pathes[i][self.path_size * –1:] | |
| # count vehicles and drop counted pathes: | |
| new_pathes = [] | |
| for i, path in enumerate(self.pathes): | |
| d = path[–2:] | |
| if ( | |
| # need at list two points to count | |
| len(d) >= 2 and | |
| # prev point not in exit zone | |
| not self.check_exit(d[0][1]) and | |
| # current point in exit zone | |
| self.check_exit(d[1][1]) and | |
| # path len is bigger then min | |
| self.path_size <= len(path) | |
| ): | |
| self.vehicle_count += 1 | |
| else: | |
| # prevent linking with path that already in exit zone | |
| add = True | |
| for p in path: | |
| if self.check_exit(p[1]): | |
| add = False | |
| break | |
| if add: | |
| new_pathes.append(path) | |
| self.pathes = new_pathes | |
| context['pathes'] = self.pathes | |
| context['objects'] = objects | |
| context['vehicle_count'] = self.vehicle_count | |
| self.log.debug('#VEHICLES FOUND: %s' % self.vehicle_count) | |
| return context |
Diese Klasse ist etwas kompliziert, also gehen wir sie nach Teilen durch.
Diese grüne Maske auf dem Bild ist die Ausgangszone, in der wir unsere Fahrzeuge zählen. Zum Beispiel zählen wir nur Pfade mit einer Länge von mehr als 3 Punkten (um etwas Rauschen zu entfernen) und den vierten in der grünen Zone. Wir verwenden Masken, da viele Operationen effektiver und einfacher sind als die Verwendung von Vektor-Algorithmen.
Verwenden Sie einfach die Operation “Binär und”, um diesen Punkt in der Umgebung zu überprüfen, und das ist alles. Und so stellen wir es ein:
| EXIT_PTS = np.array([ | |
| [[732, 720], [732, 590], [1280, 500], [1280, 720]], | |
| [[0, 400], [645, 400], [645, 0], [0, 0]] | |
| ]) | |
| SHAPE = (720,1280) | |
| base = np.zeros(SHAPE + (3,), dtype='uint8') | |
| exit_mask = cv2.fillPoly(base, EXIT_PTS, (255, 255, 255))[:, :, 0] | |
| plt.imshow(base) | |
| plt.show() |
Verknüpfen wir nun Punkte in Pfaden in Zeile 55
Im ersten Frame. Wir fügen einfach alle Punkte als neue Pfade hinzu.
Wenn len (path) == 1 ist, versuchen wir für jeden Pfad im Cache, den Punkt (Schwerpunkt) von neu erkannten Objekten zu finden, die den kleinsten euklidischen Abstand zum letzten Punkt des Pfads haben.
Wenn len(path)> 1 ist, sagen wir mit den letzten beiden Punkten im Pfad einen neuen Punkt auf derselben Linie voraus und ermitteln den minimalen Abstand zwischen ihm und dem aktuellen Punkt.
Der Punkt mit minimalem Abstand wird am Ende des aktuellen Pfads hinzugefügt und aus der Liste entfernt.
Wenn danach noch einige Punkte übrig sind, fügen wir sie als neue Pfade hinzu.
Außerdem begrenzen wir die Anzahl der Punkte im Pfad in Zeile 101
Jetzt werden wir versuchen, Fahrzeuge zu zählen, die in die Ausgangszone einfahren. Dazu nehmen wir nur 2 letzte Punkte im Pfad und überprüfen den letzten in der Ausgangszone und den vorherigen nicht. Außerdem überprüfen wir, ob len (Pfad) größer als der Grenzwert sein sollte. Der Teil nach sonst verhindert, dass neue Punkte mit den Punkten in der Ausgangszone rückverknüpft werden.
Die letzten beiden Prozessoren sind CSV-Writer zum Erstellen von CSV-Berichtsdateien und Visualisierungen zum Debuggen und für schöne Bilder / Videos.
| class CsvWriter(PipelineProcessor): | |
| def __init__(self, path, name, start_time=0, fps=15): | |
| super(CsvWriter, self).__init__() | |
| self.fp = open(os.path.join(path, name), 'w') | |
| self.writer = csv.DictWriter(self.fp, fieldnames=['time', 'vehicles']) | |
| self.writer.writeheader() | |
| self.start_time = start_time | |
| self.fps = fps | |
| self.path = path | |
| self.name = name | |
| self.prev = None | |
| def __call__(self, context): | |
| frame_number = context['frame_number'] | |
| count = _count = context['vehicle_count'] | |
| if self.prev: | |
| _count = count – self.prev | |
| time = ((self.start_time + int(frame_number / self.fps)) * 100 | |
| + int(100.0 / self.fps) * (frame_number % self.fps)) | |
| self.writer.writerow({'time': time, 'vehicles': _count}) | |
| self.prev = count | |
| return context | |
| BOUNDING_BOX_COLOUR = (255, 192, 0) | |
| CENTROID_COLOUR = (255, 192, 0) | |
| CAR_COLOURS = [(255, 192, 0)] | |
| EXIT_COLOR = (66, 183, 42) | |
| class Visualizer(PipelineProcessor): | |
| def __init__(self, save_image=True, image_dir='images'): | |
| super(Visualizer, self).__init__() | |
| self.save_image = save_image | |
| self.image_dir = image_dir | |
| def check_exit(self, point, exit_masks=[]): | |
| for exit_mask in exit_masks: | |
| if exit_mask[point[1]][point[0]] == 255: | |
| return True | |
| return False | |
| def draw_pathes(self, img, pathes): | |
| if not img.any(): | |
| return | |
| for i, path in enumerate(pathes): | |
| path = np.array(path)[:, 1].tolist() | |
| for point in path: | |
| cv2.circle(img, point, 2, CAR_COLOURS[0], –1) | |
| cv2.polylines(img, [np.int32(path)], False, CAR_COLOURS[0], 1) | |
| return img | |
| def draw_boxes(self, img, pathes, exit_masks=[]): | |
| for (i, match) in enumerate(pathes): | |
| contour, centroid = match[–1][:2] | |
| if self.check_exit(centroid, exit_masks): | |
| continue | |
| x, y, w, h = contour | |
| cv2.rectangle(img, (x, y), (x + w – 1, y + h – 1), | |
| BOUNDING_BOX_COLOUR, 1) | |
| cv2.circle(img, centroid, 2, CENTROID_COLOUR, –1) | |
| return img | |
| def draw_ui(self, img, vehicle_count, exit_masks=[]): | |
| # this just add green mask with opacity to the image | |
| for exit_mask in exit_masks: | |
| _img = np.zeros(img.shape, img.dtype) | |
| _img[:, :] = EXIT_COLOR | |
| mask = cv2.bitwise_and(_img, _img, mask=exit_mask) | |
| cv2.addWeighted(mask, 1, img, 1, 0, img) | |
| # drawing top block with counts | |
| cv2.rectangle(img, (0, 0), (img.shape[1], 50), (0, 0, 0), cv2.FILLED) | |
| cv2.putText(img, ("Vehicles passed: {total} ".format(total=vehicle_count)), (30, 30), | |
| cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1) | |
| return img | |
| def __call__(self, context): | |
| frame = context['frame'].copy() | |
| frame = np.ascontiguousarray(np.flip(frame, 2)) | |
| frame_number = context['frame_number'] | |
| pathes = context['pathes'] | |
| exit_masks = context['exit_masks'] | |
| vehicle_count = context['vehicle_count'] | |
| frame = self.draw_ui(frame, vehicle_count, exit_masks) | |
| frame = self.draw_pathes(frame, pathes) | |
| frame = self.draw_boxes(frame, pathes, exit_masks) | |
| if self.save_image: | |
| save_frame(frame, self.image_dir + | |
| "/processed_%04d.png" % frame_number) | |
| context['frame'] = frame | |
| return context |
CSV Writer speichert Daten nach Zeit, da wir sie für weitere Analysen benötigen. Also benutze ich diese Formel, um dem Unixtimestamp zusätzliches Frame-Timing hinzuzufügen:
time = ((self.start_time + int(frame_number / self.fps)) * 100 + int(100.0 / self.fps) * (frame_number % self.fps))so with start time=1 000 000 000 and fps=10 i will get results like thisframe 1 = 1 000 000 000 010frame 1 = 1 000 000 000 020…Nachdem Sie den vollständigen CSV-Bericht erhalten haben, können Sie diese Daten nach Ihren Wünschen zusammenfassen.
Conclusion
Wie Sie sehen, war es nicht so schwer, wie viele Leute denken.
Wenn Sie das Skript ausführen, werden Sie feststellen, dass diese Lösung nicht ideal ist und ein Problem mit überlappenden Vordergrundobjekten besteht. Außerdem gibt es keine Fahrzeugklassifizierung nach Typen (die Sie definitiv für echte Analysen benötigen). Bei guter Kameraposition (über der Straße) ergibt sich jedoch eine ziemlich gute Genauigkeit. Und das zeigt uns, dass selbst kleine und einfache Algorithmen, die richtig eingesetzt werden, gute Ergebnisse liefern können.
Was können wir also tun, um aktuelle Probleme zu beheben?
Eine Möglichkeit besteht darin, eine zusätzliche Filterung hinzuzufügen, um Objekte zur besseren Erkennung zu trennen. Eine andere Möglichkeit besteht darin, komplexere Algorithmen wie Deep Convolution Networks zu verwenden, die Sie in unserem Kurs https://learnml.today studieren können
