person_detection/app.py

#!/usr/bin/env python
"""
SPDX-FileCopyrightText: 2020-present Open Networking Foundation <info@opennetworking.org>
SPDX-License-Identifier: LicenseRef-ONF-Member-1.01
"""
from importlib import import_module
import os
from flask import Flask, render_template, Response
from argparse import ArgumentParser, SUPPRESS

# import camera driver
if os.environ.get('CAMERA'):
    Camera = import_module('camera_' + os.environ['CAMERA']).Camera
else:
    # from camera import Camera
    from person_detection import Camera

# Raspberry Pi camera module (requires picamera package)
# from camera_pi import Camera

app = Flask(__name__)


@app.route('/')
def index():
    """Video streaming home page."""
    return render_template('index.html', devices=[0, 1, 2, 3])


def gen(camera):
    """Video streaming generator function."""
    while True:
        frame = camera.get_frame()
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')

@app.route('/video_feed/<device>')
def video_feed(device):
    """Video streaming route. Put this in the src attribute of an img tag."""
    global args
    camera = Camera(device, args)
    return Response(gen(camera),
                    mimetype='multipart/x-mixed-replace; boundary=frame')

def name_to_port(name):
    return int(name)

def build_argparser():
    parser = ArgumentParser(add_help=False)
    args = parser.add_argument_group('Options')
    args.add_argument('-h', '--help', action='help', default=SUPPRESS, help='Show this help message and exit.')
    args.add_argument("-m", "--model", help="Required. Path to an .xml file with a trained model.",
                      required=True, type=str)
    args.add_argument("-i", "--input",
                      help="Path to video file or image. 'cam' for capturing video stream from camera",
                      default = "gstreamer", type=str)
    args.add_argument("-pt", "--prob_threshold", help="Optional. Probability threshold for detections filtering",
                      default=0.0, type=float)
    args.add_argument("--idle", action='store_true', help="Idle if no clients connected")

    return parser


if __name__ == '__main__':
    args = build_argparser().parse_args()
    app.run(host='0.0.0.0', threaded=False, processes=3)