互联网+

选自hackernoon

参与：路雪、刘晓坤

本文介绍了不使用复杂的深度学习算法计算道路交通的方法。该方法基于计算机视觉，仅使用 Python 和 OpenCV，在背景提取算法的帮助下，使用简单的移动侦测来完成任务。

今天我们将学习如何在没有复杂深度学习算法的前提下基于计算机视觉计算道路交通。

该教程中，我们仅使用 Python 和 OpenCV，在背景提取算法的帮助下，使用简单的移动侦测来完成任务。

代码地址：https://github.com/creotiv/object_detection_projects/tree/master/opencv_traffic_counting

这里是我们的计划：

1. 了解用于前景检测的背景提取算法的主要思想。

2. OpenCV 图像过滤器。

3. 基于轮廓的目标检测。

4. 构建处理管道，用于进一步的数据处理。

视频中展示了结果：

背景提取算法

背景提取有很多不同算法，但是它们的主要思想非常简单。

我们来假设你有一个自己房间的视频，该视频的很多帧都没有人／宠物，因此基本上是静态的，我们称之为 background_layer。那么，要想获取视频中移动的物体，我们只需：

foreground_objects = current_frame - background_layer

但是有时候，我们无法获取静态帧，因为光线的变化、某些物体被移动或一直移动等。在这些情况下，我们保存某些帧，尝试找出它们中相同的像素，这些像素就是 background_layer 的一部分。区别通常在于我们获取 background_layer 和用于使选择更加准确的额外过滤的方式。

本教程中，我们将使用 MOG 算法进行背景提取。视频经算法处理后，如下图所示：

左侧是原始帧，右侧是使用 MOG（带有阴影检测）算法提取的背景

如图所示，前景模板仍存在一些噪声，我们将尝试使用标准过滤技术移除噪声。

代码如下：

1. importos

2. importlogging

3. importlogging.handlers

4. importrandom

6. importnumpy asnp

7. importskvideo.io

8. importcv2

9. importmatplotlib.pyplot asplt

11. importutils

12. # without this some strange errors happen

13. cv2.ocl.setUseOpenCL(False)

14. random.seed(123)

16. # ============================================================================

17. IMAGE_DIR = "./out"

18. VIDEO_SOURCE = "input.mp4"

19. SHAPE = (720, 1280) # HxW

20. # ============================================================================

23. deftrain_bg_subtractor(inst, cap, num=500):

24. '''

25. BG substractor need process some amount of frames to start giving result

26. '''

27. print('Training BG Subtractor...')

28. i = 0

29. forframe incap:

30. inst.apply(frame, None, 0.001)

31. i += 1

32. ifi >= num:

33. returncap

36. defmain():

37. log = logging.getLogger("main")

39. # creting MOG bg subtractor with 500 frames in cache

40. # and shadow detction

41. bg_subtractor = cv2.createBackgroundSubtractorMOG2(

42. history=500, detectShadows=True)

44. # Set up image source

45. # You can use also CV2, for some reason it not working for me

46. cap = skvideo.io.vreader(VIDEO_SOURCE)

48. # skipping 500 frames to train bg subtractor

49. train_bg_subtractor(bg_subtractor, cap, num=500)

51. frame_number = -1

52. forframe incap:

53. ifnotframe.any():

54. log.error("Frame capture failed, stopping...")

55. break

57. frame_number += 1

59. utils.save_frame(frame, "./out/frame_%04d.png"% frame_number)

61. fg_mask = bg_subtractor.apply(frame, None, 0.001)

63. utils.save_frame(frame, "./out/fg_mask_%04d.png"% frame_number)

65. # ============================================================================

67. if__name__ == "__main__":

68. log = utils.init_logging()

70. ifnotos.path.exists(IMAGE_DIR):

71. log.debug("Creating image directory `%s`...", IMAGE_DIR)

72. os.makedirs(IMAGE_DIR)

74. main()

过滤

我们这种情况需要这些过滤器：Threshold（http://docs.opencv.org/3.1.0/d7/d4d/tutorial_py_thresholding.html）、Erode、Dilate、Opening 和 Closing（http://docs.opencv.org/3.1.0/d9/d61/tutorial_py_morphological_ops.html）。请打开链接并阅读，查看这些过滤器的工作方式（而不是简单的复制／粘贴）。

那么，现在我们将使用过滤器移除前景模板上的噪声。

首先，我们将使用 Closing 过滤器移除区域中的缝隙，然后使用 Opening 移除 1–2 个像素点，之后使用 Dilate 使物体更加清晰。

1. deffilter_mask(img):

3. kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2, 2))

5. # Fill any small holes

6. closing = cv2.morphologyEx(img, cv2.MORPH_CLOSE, kernel)

7. # Remove noise

8. opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)

10. # Dilate to merge adjacent blobs

11. dilation = cv2.dilate(opening, kernel, iterations=2)

13. # threshold

14. th = dilation[dilation < 240] = 0

16. returnth

处理后的前景如下图所示：

基于轮廓的目标检测

为达到目的，我们使用带有下列参数的标准 cv2.findContours 方法：

cv2.CV_RETR_EXTERNAL—get only outer contours.

cv2.CV_CHAIN_APPROX_TC89_L1 - use Teh-Chin chain approximation algorithm (faster)

1. defget_centroid(x, y, w, h):

2. x1 = int(w / 2)

3. y1 = int(h / 2)

5. cx = x + x1

6. cy = y + y1

8. return(cx, cy)

10. defdetect_vehicles(fg_mask, min_contour_width=35, min_contour_height=35):

12. matches = []

14. # finding external contours

15. im, contours, hierarchy = cv2.findContours(

16. fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_TC89_L1)

18. # filtering by with, height

19. for(i, contour) inenumerate(contours):

20. (x, y, w, h) = cv2.boundingRect(contour)

21. contour_valid = (w >= min_contour_width) and(

22. h >= min_contour_height)

24. ifnotcontour_valid:

25. continue

27. # getting center of the bounding box

28. centroid = get_centroid(x, y, w, h)

30. matches.append(((x, y, w, h), centroid))

32. returnmatches

很简单，对吧？

构建处理管道

你必须理解，在机器学习和计算机视觉领域中，没有一种魔术般的算法能够搞定一切，即使我们想象存在这样一种算法，我们仍然无法使用它，因为它在大规模应用时会无效。比如，几年前，Netflix 创办了一个比赛，最佳电影推荐算法奖励 300 万美元。有一支队伍创建了一个最佳算法，但问题是该算法无法大规模应用，因此对该公司没有用处。但是，Netflix 仍然奖励了他们 100 万。:)

那么，现在我们将构建简单的处理管道，该管道不是为了大规模使用，而是为了方便，但原理是一样的。

1. classPipelineRunner(object):

2. '''

3. Very simple pipline.

4. Just run passed processors in order with passing context from one to

5. another.

6. You can also set log level for processors.

7. '''

9. def__init__(self, pipeline=None, log_level=logging.DEBUG):

10. self.pipeline = pipeline or[]

11. self.context = {}

12. self.log = logging.getLogger(self.__class__.__name__)

13. self.log.setLevel(log_level)

14. self.log_level = log_level

15. self.set_log_level()

17. defset_context(self, data):

18. self.context = data

20. defadd(self, processor):

21. ifnotisinstance(processor, PipelineProcessor):

22. raiseException(

23. 'Processor should be an isinstance of PipelineProcessor.')

24. processor.log.setLevel(self.log_level)

25. self.pipeline.append(processor)

27. defremove(self, name):

28. fori, p inenumerate(self.pipeline):

29. ifp.__class__.__name__ == name:

30. delself.pipeline[i]

31. returnTrue

32. returnFalse

34. defset_log_level(self):

35. forp inself.pipeline:

36. p.log.setLevel(self.log_level)

38. defrun(self):

39. forp inself.pipeline:

40. self.context = p(self.context)

42. self.log.debug("Frame #%d processed.", self.context['frame_number'])

44. returnself.context

47. classPipelineProcessor(object):

48. '''

49. Base class for processors.

50. '''

52. def__init__(self):

53. self.log = logging.getLogger(self.__class__.__name__)

由于输入构造函数（input constructor）将使用一串处理器，它们将按顺序运行，每个处理器处理一部分工作。那么，现在我们就来创建一个轮廓检测处理器。

1. classContourDetection(PipelineProcessor):

2. '''

3. Detecting moving objects.

4. Purpose of this processor is to subtrac background, get moving objects

5. and detect them with a cv2.findContours method, and then filter off-by

6. width and height.

7. bg_subtractor - background subtractor isinstance.

8. min_contour_width - min bounding rectangle width.

9. min_contour_height - min bounding rectangle height.

10. save_image - if True will save detected objects mask to file.

11. image_dir - where to save images(must exist).

12. '''

14. def__init__(self, bg_subtractor, min_contour_width=35, min_contour_height=35, save_image=False, image_dir='images'):

15. super(ContourDetection, self).__init__()

17. self.bg_subtractor = bg_subtractor

18. self.min_contour_width = min_contour_width

19. self.min_contour_height = min_contour_height

20. self.save_image = save_image

21. self.image_dir = image_dir

23. deffilter_mask(self, img, a=None):

24. '''

25. This filters are hand-picked just based on visual tests

26. '''

28. kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2, 2))

30. # Fill any small holes

31. closing = cv2.morphologyEx(img, cv2.MORPH_CLOSE, kernel)

32. # Remove noise

33. opening = cv2.morphologyEx(closing, cv2.MORPH_OPEN, kernel)

35. # Dilate to merge adjacent blobs

36. dilation = cv2.dilate(opening, kernel, iterations=2)

38. returndilation

40. defdetect_vehicles(self, fg_mask, context):

42. matches = []

44. # finding external contours

45. im2, contours, hierarchy = cv2.findContours(

46. fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_TC89_L1)

48. for(i, contour) inenumerate(contours):

49. (x, y, w, h) = cv2.boundingRect(contour)

50. contour_valid = (w >= self.min_contour_width) and(

51. h >= self.min_contour_height)

53. ifnotcontour_valid:

54. continue

56. centroid = utils.get_centroid(x, y, w, h)

58. matches.append(((x, y, w, h), centroid))

60. returnmatches

62. def__call__(self, context):

63. frame = context['frame'].copy()

64. frame_number = context['frame_number']

66. fg_mask = self.bg_subtractor.apply(frame, None, 0.001)

67. # just thresholding values

68. fg_mask[fg_mask < 240] = 0

69. fg_mask = self.filter_mask(fg_mask, frame_number)

71. ifself.save_image:

72. utils.save_frame(fg_mask, self.image_dir +

73. "/mask_%04d.png"% frame_number, flip=False)

75. context['objects'] = self.detect_vehicles(fg_mask, context)

76. context['fg_mask'] = fg_mask

78. returncontex

其实就是把背景提取、过滤和检测部分合并起来。

现在，我们来创建一个处理器，其将在不同帧上检测到的物体连接起来并创建路径，还能计算出口区的车辆数量。

1. '''

2. Counting vehicles that entered in exit zone.

3. Purpose of this class based on detected object and local cache create

4. objects pathes and count that entered in exit zone defined by exit masks.

5. exit_masks - list of the exit masks.

6. path_size - max number of points in a path.

7. max_dst - max distance between two points.

8. '''

10. def__init__(self, exit_masks=[], path_size=10, max_dst=30, x_weight=1.0, y_weight=1.0):

11. super(VehicleCounter, self).__init__()

13. self.exit_masks = exit_masks

15. self.vehicle_count = 0

16. self.path_size = path_size

17. self.pathes = []

18. self.max_dst = max_dst

19. self.x_weight = x_weight

20. self.y_weight = y_weight

22. defcheck_exit(self, point):

23. forexit_mask inself.exit_masks:

24. try:

25. ifexit_mask[point[1]][point[0]] == 255:

26. returnTrue

27. except:

28. returnTrue

29. returnFalse

31. def__call__(self, context):

32. objects = context['objects']

33. context['exit_masks'] = self.exit_masks

34. context['pathes'] = self.pathes

35. context['vehicle_count'] = self.vehicle_count

36. ifnotobjects:

37. returncontext

39. points = np.array(objects)[:, 0:2]

40. points = points.tolist()

42. # add new points if pathes is empty

43. ifnotself.pathes:

44. formatch inpoints:

45. self.pathes.append([match])

47. else:

48. # link new points with old pathes based on minimum distance between

49. # points

50. new_pathes = []

52. forpath inself.pathes:

53. _min = 999999

54. _match = None

55. forp inpoints:

56. iflen(path) == 1:

57. # distance from last point to current

58. d = utils.distance(p[0], path[-1][0])

59. else:

60. # based on 2 prev points predict next point and calculate

61. # distance from predicted next point to current

62. xn = 2* path[-1][0][0] - path[-2][0][0]

63. yn = 2* path[-1][0][1] - path[-2][0][1]

64. d = utils.distance(

65. p[0], (xn, yn),

66. x_weight=self.x_weight,

67. y_weight=self.y_weight

68. )

70. ifd < _min:

71. _min = d

72. _match = p

74. if_match and_min <= self.max_dst:

75. points.remove(_match)

76. path.append(_match)

77. new_pathes.append(path)

79. # do not drop path if current frame has no matches

80. if_match isNone:

81. new_pathes.append(path)

83. self.pathes = new_pathes

85. # add new pathes

86. iflen(points):

87. forp inpoints:

88. # do not add points that already should be counted

89. ifself.check_exit(p[1]):

90. continue

91. self.pathes.append([p])

93. # save only last N points in path

94. fori, _ inenumerate(self.pathes):

95. self.pathes[i] = self.pathes[i][self.path_size * -1:]

97. # count vehicles and drop counted pathes:

98. new_pathes = []

99. fori, path inenumerate(self.pathes):

100. d = path[-2:]

102. if(

103. # need at list two points to count

104. len(d) >= 2and

105. # prev point not in exit zone

106. notself.check_exit(d[0][1]) and

107. # current point in exit zone

108. self.check_exit(d[1][1]) and

109. # path len is bigger then min

110. self.path_size <= len(path)

111. ):

112. self.vehicle_count += 1

113. else:

114. # prevent linking with path that already in exit zone

115. add = True

116. forp inpath:

117. ifself.check_exit(p[1]):

118. add = False

119. break

120. ifadd:

121. new_pathes.append(path)

123. self.pathes = new_pathes

125. context['pathes'] = self.pathes

126. context['objects'] = objects

127. context['vehicle_count'] = self.vehicle_count

129. self.log.debug('#VEHICLES FOUND: %s'% self.vehicle_count)

131. returncontext

该教程有一点复杂，我们一部分一部分地过一遍。

下图中绿色的掩膜是出口区，我们在该区域计算车辆的数量。比如，我们将计算长度大于 3 个点（以移除噪声）的路径，其中第 4 个点就在绿色区域。

我们使用掩膜，因为它对很多操作都有效且比使用向量算法更简单。只需要使用二元和（binary and）运算检查该区域的点就可以了。下图显示了我们的设置方式：

1. EXIT_PTS = np.array([

2. [[732, 720], [732, 590], [1280, 500], [1280, 720]],

3. [[0, 400], [645, 400], [645, 0], [0, 0]]

4. ])

6. base = np.zeros(SHAPE + (3,), dtype='uint8')

7. exit_mask = cv2.fillPoly(base, EXIT_PTS, (255, 255, 255))[:, :, 0]

现在，我们连接路径中的点：

1. new_pathes = []

3. forpath inself.pathes:

4. _min = 999999

5. _match = None

6. forp inpoints:

7. iflen(path) == 1:

8. # distance from last point to current

9. d = utils.distance(p[0], path[-1][0])

10. else:

11. # based on 2 prev points predict next point and calculate

12. # distance from predicted next point to current

13. xn = 2* path[-1][0][0] - path[-2][0][0]

14. yn = 2* path[-1][0][1] - path[-2][0][1]

15. d = utils.distance(

16. p[0], (xn, yn),

17. x_weight=self.x_weight,

18. y_weight=self.y_weight

19. )

21. ifd < _min:

22. _min = d

23. _match = p

25. if_match and_min <= self.max_dst:

26. points.remove(_match)

27. path.append(_match)

28. new_pathes.append(path)

30. # do not drop path if current frame has no matches

31. if_match isNone:

32. new_pathes.append(path)

34. self.pathes = new_pathes

36. # add new pathes

37. iflen(points):

38. forp inpoints:

39. # do not add points that already should be counted

40. ifself.check_exit(p[1]):

41. continue

42. self.pathes.append([p])

44. # save only last N points in path

45. fori, _ inenumerate(self.pathes):

46. self.pathes[i] = self.pathes[i][self.path_size * -1:]

在第一帧上，我们只需添加所有点作为新的路径。

接下来，如果 len(path) == 1，对于高速缓存中的每个路径，我们将尝试从新检测到的物体中找出点（质心），这些物体到路径最后一个点的欧几里得距离最短。

如果 len(path) > 1，我们将使用该路径中的最后两个点在同一条线上预测新的点，找出它和当前点之间的最小距离。

将最小距离的点添加至当前路径的末尾，然后将其从列表中移除。

如果还有剩下的点，我们将它们添加为新的路径。

我们还可以限制该路径中点的数量。

1. # count vehicles and drop counted pathes:

2. new_pathes = []

3. fori, path inenumerate(self.pathes):

4. d = path[-2:]

6. if(

7. # need at list two points to count

8. len(d) >= 2and

9. # prev point not in exit zone

10. notself.check_exit(d[0][1]) and

11. # current point in exit zone

12. self.check_exit(d[1][1]) and

13. # path len is bigger then min

14. self.path_size <= len(path)

15. ):

16. self.vehicle_count += 1

17. else:

18. # prevent linking with path that already in exit zone

19. add = True

20. forp inpath:

21. ifself.check_exit(p[1]):

22. add = False

23. break

24. ifadd:

25. new_pathes.append(path)

27. self.pathes = new_pathes

29. context['pathes'] = self.pathes

30. context['objects'] = objects

31. context['vehicle_count'] = self.vehicle_count

33. self.log.debug('#VEHICLES FOUND: %s'% self.vehicle_count)

35. returncontext

现在，我们尝试计算进入出口区的车辆的数量。我们需要观察路径中的最后两个点，并在出口区检查，是否其中靠后的一个在出口区，而靠前的不在，并确保 len(path) 比下限值要大。

之后的部分就是阻止新的点回联至出口区的点。

最后两个处理器是 CSV writer，可创建报告 CSV 文件和可视化文件，用于调试和输出更好的画面。

1. classCsvWriter(PipelineProcessor):

3. def__init__(self, path, name, start_time=0, fps=15):

4. super(CsvWriter, self).__init__()

6. self.fp = open(os.path.join(path, name), 'w')

7. self.writer = csv.DictWriter(self.fp, fieldnames=['time', 'vehicles'])

8. self.writer.writeheader()

9. self.start_time = start_time

10. self.fps = fps

11. self.path = path

12. self.name = name

13. self.prev = None

15. def__call__(self, context):

16. frame_number = context['frame_number']

17. count = _count = context['vehicle_count']

19. ifself.prev:

20. _count = count - self.prev

22. time = ((self.start_time + int(frame_number / self.fps)) * 100

23. + int(100.0/ self.fps) * (frame_number % self.fps))

24. self.writer.writerow({'time': time, 'vehicles': _count})

25. self.prev = count

27. returncontext

30. classVisualizer(PipelineProcessor):

32. def__init__(self, save_image=True, image_dir='images'):

33. super(Visualizer, self).__init__()

35. self.save_image = save_image

36. self.image_dir = image_dir

38. defcheck_exit(self, point, exit_masks=[]):

39. forexit_mask inexit_masks:

40. ifexit_mask[point[1]][point[0]] == 255:

41. returnTrue

42. returnFalse

44. defdraw_pathes(self, img, pathes):

45. ifnotimg.any():

46. return

48. fori, path inenumerate(pathes):

49. path = np.array(path)[:, 1].tolist()

50. forpoint inpath:

51. cv2.circle(img, point, 2, CAR_COLOURS[0], -1)

52. cv2.polylines(img, [np.int32(path)], False, CAR_COLOURS[0], 1)

54. returnimg

56. defdraw_boxes(self, img, pathes, exit_masks=[]):

57. for(i, match) inenumerate(pathes):

59. contour, centroid = match[-1][:2]

60. ifself.check_exit(centroid, exit_masks):

61. continue

63. x, y, w, h = contour

65. cv2.rectangle(img, (x, y), (x + w - 1, y + h - 1),

66. BOUNDING_BOX_COLOUR, 1)

67. cv2.circle(img, centroid, 2, CENTROID_COLOUR, -1)

69. returnimg

71. defdraw_ui(self, img, vehicle_count, exit_masks=[]):

73. # this just add green mask with opacity to the image

74. forexit_mask inexit_masks:

75. _img = np.zeros(img.shape, img.dtype)

76. _img[:, :] = EXIT_COLOR

77. mask = cv2.bitwise_and(_img, _img, mask=exit_mask)

78. cv2.addWeighted(mask, 1, img, 1, 0, img)

80. # drawing top block with counts

81. cv2.rectangle(img, (0, 0), (img.shape[1], 50), (0, 0, 0), cv2.FILLED)

82. cv2.putText(img, ("Vehicles passed: {total} ".format(total=vehicle_count)), (30, 30),

83. cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 1)

84. returnimg

86. def__call__(self, context):

87. frame = context['frame'].copy()

88. frame_number = context['frame_number']

89. pathes = context['pathes']

90. exit_masks = context['exit_masks']

91. vehicle_count = context['vehicle_count']

93. frame = self.draw_ui(frame, vehicle_count, exit_masks)

94. frame = self.draw_pathes(frame, pathes)

95. frame = self.draw_boxes(frame, pathes, exit_masks)

97. utils.save_frame(frame, self.image_dir +

98. "/processed_%04d.png"% frame_number)

100. returncontext

CSV writer 按时间保存数据，因为我们需要用它做进一步的分析。因此我使用下列公式向 unix 时间戳添加额外的帧计时：

time = ((self.start_time + int(frame_number / self.fps)) * 100

+ int(100.0 / self.fps) * (frame_number % self.fps))

在 start time=1 000 000 000 和 fps=10 的情况下，结果如下：

frame 1 = 1 000 000 000 010

frame 1 = 1 000 000 000 020

…

获取完整的 csv 报告后，你可以随意聚合这些数据。

该项目的完整代码地址：https://github.com/creotiv/object_detection_projects/tree/master/opencv_traffic_counting

结论

所以，这并不像人们想象的那么难。

但是，如果你运行该脚本，你会发现该解决方案并不完美，它存在一个问题——背景物体重叠，而且它还无法按类型进行车辆分类（实分析时你肯定需要）。但是，该方法拥有好的摄像头位置（道路上方），能够提供相当不错的准确率。这告诉我们即使简单的小算法用好了也能取得不错的结果。

那么我们要怎么做才能解决当前的问题呢？

一种方式是添加额外的过滤，使物体分离，以进行更好的检测。另一种方式是使用更复杂的算法，比如深度卷积网络。

原文地址：https://hackernoon.com/tutorial-making-road-traffic-counting-app-based-on-computer-vision-and-opencv-166937911660

本文为机器之心编译，转载请联系本公众号获得授权。返回搜狐，查看更多

责任编辑：