forked from Github/frigate
Improve motion detection and region selection (#6741)
* refactor existing motion detector * implement and use cnt bgsub * pass fps to motion detector * create a simplified motion detector * lightning detection * update default motion config * lint imports * use estimated boxes for regions * use improved motion detector * update test * use a different strategy for clustering motion and object boxes * increase alpha during calibration * simplify object consolidation * add some reasonable constraints to the estimated box * adjust cluster boundary to 10% * refactor * add disabled debug code * fix variable scope
This commit is contained in:
Blake Blackshear
22
frigate/motion/__init__.py
Normal file
22
frigate/motion/__init__.py
Normal file
@@ -0,0 +1,22 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Tuple
|
||||
|
||||
from frigate.config import MotionConfig
|
||||
|
||||
|
||||
class MotionDetector(ABC):
|
||||
@abstractmethod
|
||||
def __init__(
|
||||
self,
|
||||
frame_shape: Tuple[int, int, int],
|
||||
config: MotionConfig,
|
||||
fps: int,
|
||||
improve_contrast,
|
||||
threshold,
|
||||
contour_area,
|
||||
):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def detect(self, frame):
|
||||
pass
|
||||
154
frigate/motion/frigate_motion.py
Normal file
154
frigate/motion/frigate_motion.py
Normal file
@@ -0,0 +1,154 @@
|
||||
import cv2
|
||||
import imutils
|
||||
import numpy as np
|
||||
|
||||
from frigate.config import MotionConfig
|
||||
from frigate.motion import MotionDetector
|
||||
|
||||
|
||||
class FrigateMotionDetector(MotionDetector):
|
||||
def __init__(
|
||||
self,
|
||||
frame_shape,
|
||||
config: MotionConfig,
|
||||
fps: int,
|
||||
improve_contrast,
|
||||
threshold,
|
||||
contour_area,
|
||||
):
|
||||
self.config = config
|
||||
self.frame_shape = frame_shape
|
||||
self.resize_factor = frame_shape[0] / config.frame_height
|
||||
self.motion_frame_size = (
|
||||
config.frame_height,
|
||||
config.frame_height * frame_shape[1] // frame_shape[0],
|
||||
)
|
||||
self.avg_frame = np.zeros(self.motion_frame_size, np.float32)
|
||||
self.avg_delta = np.zeros(self.motion_frame_size, np.float32)
|
||||
self.motion_frame_count = 0
|
||||
self.frame_counter = 0
|
||||
resized_mask = cv2.resize(
|
||||
config.mask,
|
||||
dsize=(self.motion_frame_size[1], self.motion_frame_size[0]),
|
||||
interpolation=cv2.INTER_LINEAR,
|
||||
)
|
||||
self.mask = np.where(resized_mask == [0])
|
||||
self.save_images = False
|
||||
self.improve_contrast = improve_contrast
|
||||
self.threshold = threshold
|
||||
self.contour_area = contour_area
|
||||
|
||||
def detect(self, frame):
|
||||
motion_boxes = []
|
||||
|
||||
gray = frame[0 : self.frame_shape[0], 0 : self.frame_shape[1]]
|
||||
|
||||
# resize frame
|
||||
resized_frame = cv2.resize(
|
||||
gray,
|
||||
dsize=(self.motion_frame_size[1], self.motion_frame_size[0]),
|
||||
interpolation=cv2.INTER_LINEAR,
|
||||
)
|
||||
|
||||
# Improve contrast
|
||||
if self.improve_contrast.value:
|
||||
minval = np.percentile(resized_frame, 4)
|
||||
maxval = np.percentile(resized_frame, 96)
|
||||
# don't adjust if the image is a single color
|
||||
if minval < maxval:
|
||||
resized_frame = np.clip(resized_frame, minval, maxval)
|
||||
resized_frame = (
|
||||
((resized_frame - minval) / (maxval - minval)) * 255
|
||||
).astype(np.uint8)
|
||||
|
||||
# mask frame
|
||||
resized_frame[self.mask] = [255]
|
||||
|
||||
# it takes ~30 frames to establish a baseline
|
||||
# dont bother looking for motion
|
||||
if self.frame_counter < 30:
|
||||
self.frame_counter += 1
|
||||
else:
|
||||
if self.save_images:
|
||||
self.frame_counter += 1
|
||||
# compare to average
|
||||
frameDelta = cv2.absdiff(resized_frame, cv2.convertScaleAbs(self.avg_frame))
|
||||
|
||||
# compute the average delta over the past few frames
|
||||
# higher values mean the current frame impacts the delta a lot, and a single raindrop may
|
||||
# register as motion, too low and a fast moving person wont be detected as motion
|
||||
cv2.accumulateWeighted(frameDelta, self.avg_delta, self.config.delta_alpha)
|
||||
|
||||
# compute the threshold image for the current frame
|
||||
current_thresh = cv2.threshold(
|
||||
frameDelta, self.threshold.value, 255, cv2.THRESH_BINARY
|
||||
)[1]
|
||||
|
||||
# black out everything in the avg_delta where there isnt motion in the current frame
|
||||
avg_delta_image = cv2.convertScaleAbs(self.avg_delta)
|
||||
avg_delta_image = cv2.bitwise_and(avg_delta_image, current_thresh)
|
||||
|
||||
# then look for deltas above the threshold, but only in areas where there is a delta
|
||||
# in the current frame. this prevents deltas from previous frames from being included
|
||||
thresh = cv2.threshold(
|
||||
avg_delta_image, self.threshold.value, 255, cv2.THRESH_BINARY
|
||||
)[1]
|
||||
|
||||
# dilate the thresholded image to fill in holes, then find contours
|
||||
# on thresholded image
|
||||
thresh_dilated = cv2.dilate(thresh, None, iterations=2)
|
||||
cnts = cv2.findContours(
|
||||
thresh_dilated, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
|
||||
)
|
||||
cnts = imutils.grab_contours(cnts)
|
||||
|
||||
# loop over the contours
|
||||
for c in cnts:
|
||||
# if the contour is big enough, count it as motion
|
||||
contour_area = cv2.contourArea(c)
|
||||
if contour_area > self.contour_area.value:
|
||||
x, y, w, h = cv2.boundingRect(c)
|
||||
motion_boxes.append(
|
||||
(
|
||||
int(x * self.resize_factor),
|
||||
int(y * self.resize_factor),
|
||||
int((x + w) * self.resize_factor),
|
||||
int((y + h) * self.resize_factor),
|
||||
)
|
||||
)
|
||||
|
||||
if self.save_images:
|
||||
thresh_dilated = cv2.cvtColor(thresh_dilated, cv2.COLOR_GRAY2BGR)
|
||||
# print("--------")
|
||||
# print(self.frame_counter)
|
||||
for c in cnts:
|
||||
contour_area = cv2.contourArea(c)
|
||||
if contour_area > self.contour_area.value:
|
||||
x, y, w, h = cv2.boundingRect(c)
|
||||
cv2.rectangle(
|
||||
thresh_dilated,
|
||||
(x, y),
|
||||
(x + w, y + h),
|
||||
(0, 0, 255),
|
||||
2,
|
||||
)
|
||||
|
||||
cv2.imwrite(
|
||||
f"debug/frames/frigate-{self.frame_counter}.jpg", thresh_dilated
|
||||
)
|
||||
|
||||
if len(motion_boxes) > 0:
|
||||
self.motion_frame_count += 1
|
||||
if self.motion_frame_count >= 10:
|
||||
# only average in the current frame if the difference persists for a bit
|
||||
cv2.accumulateWeighted(
|
||||
resized_frame, self.avg_frame, self.config.frame_alpha
|
||||
)
|
||||
else:
|
||||
# when no motion, just keep averaging the frames together
|
||||
cv2.accumulateWeighted(
|
||||
resized_frame, self.avg_frame, self.config.frame_alpha
|
||||
)
|
||||
self.motion_frame_count = 0
|
||||
|
||||
return motion_boxes
|
||||
143
frigate/motion/improved_motion.py
Normal file
143
frigate/motion/improved_motion.py
Normal file
@@ -0,0 +1,143 @@
|
||||
import cv2
|
||||
import imutils
|
||||
import numpy as np
|
||||
|
||||
from frigate.config import MotionConfig
|
||||
from frigate.motion import MotionDetector
|
||||
|
||||
|
||||
class ImprovedMotionDetector(MotionDetector):
|
||||
def __init__(
|
||||
self,
|
||||
frame_shape,
|
||||
config: MotionConfig,
|
||||
fps: int,
|
||||
improve_contrast,
|
||||
threshold,
|
||||
contour_area,
|
||||
):
|
||||
self.config = config
|
||||
self.frame_shape = frame_shape
|
||||
self.resize_factor = frame_shape[0] / config.frame_height
|
||||
self.motion_frame_size = (
|
||||
config.frame_height,
|
||||
config.frame_height * frame_shape[1] // frame_shape[0],
|
||||
)
|
||||
self.avg_frame = np.zeros(self.motion_frame_size, np.float32)
|
||||
self.avg_delta = np.zeros(self.motion_frame_size, np.float32)
|
||||
self.motion_frame_count = 0
|
||||
self.frame_counter = 0
|
||||
resized_mask = cv2.resize(
|
||||
config.mask,
|
||||
dsize=(self.motion_frame_size[1], self.motion_frame_size[0]),
|
||||
interpolation=cv2.INTER_LINEAR,
|
||||
)
|
||||
self.mask = np.where(resized_mask == [0])
|
||||
self.save_images = False
|
||||
self.calibrating = True
|
||||
self.improve_contrast = improve_contrast
|
||||
self.threshold = threshold
|
||||
self.contour_area = contour_area
|
||||
|
||||
def detect(self, frame):
|
||||
motion_boxes = []
|
||||
|
||||
gray = frame[0 : self.frame_shape[0], 0 : self.frame_shape[1]]
|
||||
|
||||
# resize frame
|
||||
resized_frame = cv2.resize(
|
||||
gray,
|
||||
dsize=(self.motion_frame_size[1], self.motion_frame_size[0]),
|
||||
interpolation=cv2.INTER_LINEAR,
|
||||
)
|
||||
|
||||
resized_frame = cv2.GaussianBlur(resized_frame, (3, 3), cv2.BORDER_DEFAULT)
|
||||
|
||||
# Improve contrast
|
||||
if self.improve_contrast.value:
|
||||
resized_frame = cv2.equalizeHist(resized_frame)
|
||||
|
||||
# mask frame
|
||||
resized_frame[self.mask] = [255]
|
||||
|
||||
if self.save_images or self.calibrating:
|
||||
self.frame_counter += 1
|
||||
# compare to average
|
||||
frameDelta = cv2.absdiff(resized_frame, cv2.convertScaleAbs(self.avg_frame))
|
||||
|
||||
# compute the threshold image for the current frame
|
||||
thresh = cv2.threshold(
|
||||
frameDelta, self.threshold.value, 255, cv2.THRESH_BINARY
|
||||
)[1]
|
||||
|
||||
# dilate the thresholded image to fill in holes, then find contours
|
||||
# on thresholded image
|
||||
thresh_dilated = cv2.dilate(thresh, None, iterations=1)
|
||||
cnts = cv2.findContours(
|
||||
thresh_dilated, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
|
||||
)
|
||||
cnts = imutils.grab_contours(cnts)
|
||||
|
||||
# loop over the contours
|
||||
total_contour_area = 0
|
||||
for c in cnts:
|
||||
# if the contour is big enough, count it as motion
|
||||
contour_area = cv2.contourArea(c)
|
||||
total_contour_area += contour_area
|
||||
if contour_area > self.contour_area.value:
|
||||
x, y, w, h = cv2.boundingRect(c)
|
||||
motion_boxes.append(
|
||||
(
|
||||
int(x * self.resize_factor),
|
||||
int(y * self.resize_factor),
|
||||
int((x + w) * self.resize_factor),
|
||||
int((y + h) * self.resize_factor),
|
||||
)
|
||||
)
|
||||
|
||||
pct_motion = total_contour_area / (
|
||||
self.motion_frame_size[0] * self.motion_frame_size[1]
|
||||
)
|
||||
|
||||
# once the motion drops to less than 1% for the first time, assume its calibrated
|
||||
if pct_motion < 0.01:
|
||||
self.calibrating = False
|
||||
|
||||
# if calibrating or the motion contours are > 80% of the image area (lightning, ir, ptz) recalibrate
|
||||
if self.calibrating or pct_motion > self.config.lightning_threshold:
|
||||
motion_boxes = []
|
||||
self.calibrating = True
|
||||
|
||||
if self.save_images:
|
||||
thresh_dilated = cv2.cvtColor(thresh_dilated, cv2.COLOR_GRAY2BGR)
|
||||
for b in motion_boxes:
|
||||
cv2.rectangle(
|
||||
thresh_dilated,
|
||||
(int(b[0] / self.resize_factor), int(b[1] / self.resize_factor)),
|
||||
(int(b[2] / self.resize_factor), int(b[3] / self.resize_factor)),
|
||||
(0, 0, 255),
|
||||
2,
|
||||
)
|
||||
cv2.imwrite(
|
||||
f"debug/frames/improved-{self.frame_counter}.jpg", thresh_dilated
|
||||
)
|
||||
|
||||
if len(motion_boxes) > 0:
|
||||
self.motion_frame_count += 1
|
||||
if self.motion_frame_count >= 10:
|
||||
# only average in the current frame if the difference persists for a bit
|
||||
cv2.accumulateWeighted(
|
||||
resized_frame,
|
||||
self.avg_frame,
|
||||
0.2 if self.calibrating else self.config.frame_alpha,
|
||||
)
|
||||
else:
|
||||
# when no motion, just keep averaging the frames together
|
||||
cv2.accumulateWeighted(
|
||||
resized_frame,
|
||||
self.avg_frame,
|
||||
0.2 if self.calibrating else self.config.frame_alpha,
|
||||
)
|
||||
self.motion_frame_count = 0
|
||||
|
||||
return motion_boxes
|
||||
Reference in New Issue
Block a user