heimbsle69869/smarthome-presence-detect
2
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add remote video stream to people counter

Browse Source
This commit is contained in:
Lennart Heimbs 2019-08-04 22:15:55 +02:00
parent 91256c65fe
commit 98b9985b17
1 changed files with 73 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

119
camera/counter_people.py Normal file → Executable file
View File

@ -1,11 +1,11 @@
from imutils.object_detection import non_max_suppression
#!/usr/bin/env python
import numpy as np
import imutils
import cv2
import time
import argparse
import time
import base64
from video_stream import imagezmq
from imutils.object_detection import non_max_suppression
'''
Usage:
@ -16,6 +16,7 @@ python peopleCounter.py -c # Attempts to detect people using webcam
HOGCV = cv2.HOGDescriptor()
HOGCV.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())
VERBOSITY = False
def detector(image):
'''
@ -37,54 +38,63 @@ def detector(image):
return result
def buildPayload(variable, value, context):
return {variable: {"value": value, "context": context}}
def argsParser():
ap = argparse.ArgumentParser()
ap.add_argument("-i", "--image", default=None, help="path to image test file directory")
ap.add_argument("-c", "--camera", default=False, help="Set as true if you wish to use the camera")
ap.add_argument("-v", "--video", default=None, help="path to the video file")
ap.add_argument("-i", "--image", default=None,
help="path to image test file directory")
ap.add_argument("-c", "--camera", action="store_true", default=False,
help="Set as true if you wish to use the camera")
ap.add_argument("-v", "--video", default=None,
help="path to the video file")
ap.add_argument("-r", "--remote", action="store_true", default=False,
help="video comes from remote source via imagezmq")
ap.add_argument("--verbose", action="store_true", default=False,
help="increase output verbosity")
args = vars(ap.parse_args())
if args["verbose"]:
VERBOSITY = True
return args
def usage():
print("usage: counter_people.py [-h] [-i IMAGE] [-c] [-v] [-r REMOTE] [--verbose]")
print()
print("optional arguments:")
print(" -h, --help show this help message and exit")
print(" -i IMAGE, --image IMAGE")
print(" path to image test file directory")
print(" -c, --camera Set as true if you wish to use the camera")
print(" -v, --video path to the video file")
print(" -r REMOTE, --remote REMOTE")
print(" video comes from remote source via imagezmq")
print(" --verbose increase output verbosity")
def localDetect(image_path):
result = []
image = cv2.imread(image_path)
image = imutils.resize(image, width=min(400, image.shape[1]))
clone = image.copy()
if len(image) <= 0:
print("[ERROR] could not read local image")
return result
print("[INFO] Detecting people")
result = detector(image)
"""# shows the result
for (xA, yA, xB, yB) in result:
cv2.rectangle(image, (xA, yA), (xB, yB), (0, 255, 0), 2)
if VERBOSITY:
# shows the result
for (xA, yA, xB, yB) in result:
cv2.rectangle(image, (xA, yA), (xB, yB), (0, 255, 0), 2)
cv2.imshow("result", image)
cv2.waitKey(0)
cv2.destroyAllWindows()
cv2.imshow("result", image)
cv2.waitKey(0)
cv2.destroyWindow("result")
cv2.imwrite("result.png", np.hstack((clone, image)))"""
#cv2.imwrite("result.png", np.hstack((clone, image)))
return result#(result, image)
def cameraDetect(video_path="", sample_time=5):
if video_path:
cap = cv2.VideoCapture(video_path)
else:
cap = cv2.VideoCapture(0)
#init = time.time()
while(True):
def videoDetect(cap):
while True:
# Capture frame-by-frame
_, frame = cap.read()
@ -97,15 +107,16 @@ def cameraDetect(video_path="", sample_time=5):
# shows the result
for (xA, yA, xB, yB) in result:
cv2.rectangle(frame, (xA, yA), (xB, yB), (0, 255, 0), 2)
cv2.imshow('frame', frame)
cv2.waitKey(0)
if VERBOSITY:
cv2.imshow('frame', frame)
cv2.waitKey(0)
#if time.time() - init >= sample_time:
if len(result):
if result:
print("{} people detected.".format(len(result)))
#init = time.time()
if cv2.waitKey(1) & 0xFF == ord('q'):
break
@ -113,34 +124,46 @@ def cameraDetect(video_path="", sample_time=5):
cap.release()
cv2.destroyAllWindows()
def remoteDetect(image_hub):
while True:
rpi_name, image = image_hub.recv_image()
cv2.imshow(rpi_name, image) # 1 window for each RPi
cv2.waitKey(1)
image_hub.send_reply(b'OK')
def convert_to_base64(image):
image = imutils.resize(image, width=400)
img_str = cv2.imencode('.png', image)[1].tostring()
b64 = base64.b64encode(img_str)
return b64.decode('utf-8')
def detectPeople(args):
image_path = args["image"]
video_path = args["video"]
camera = True if str(args["camera"]) == 'true' else False
camera = True if args["camera"] else False
remote = True if args["remote"] else False
# Routine to read local image
if image_path != None and not camera and video_path == None:
if image_path is not None:
print("[INFO] Image path provided, attempting to read image")
(result, image) = localDetect(image_path)
print(str(len(result)) + " People detected.")
if video_path != None and not camera:
print("[INFO] reading video")
cameraDetect(video_path)
elif video_path is not None:
print("[INFO] Video path provided, reading video")
cap = cv2.VideoCapture(video_path)
videoDetect(cap)
# Routine to read images from webcam
if camera:
print("[INFO] reading camera images")
cameraDetect()
elif camera:
print("[INFO] Reading images from local camera")
cap = cv2.VideoCapture(0)
videoDetect(cap)
elif remote:
print("[INFO] Reading images from remote stream")
image_hub = imagezmq.ImageHub()
remoteDetect(image_hub)
else:
usage()
def main():