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
import logging as log
import sys
import flask
from argparse import ArgumentParser, SUPPRESS

import config
from person_detection import Camera
import roc


app = flask.Flask(__name__)


@app.route('/')
def index():
    global cameras
    """Video streaming home page."""
    log.info("{} - connected".format(flask.request.remote_addr))
    return flask.render_template('index.html', devices=config.cameras)

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
    log.debug("{} - video feed {}".format(flask.request.remote_addr, device))
    camera = Camera(device, flask.request.remote_addr, args)
    return flask.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.75,
            type = float)
    args.add_argument("--noroc",
            action = 'store_true',
            help = "No ROC")
    args.add_argument("--key",
            help = "ROC api key",
            type = str)
    args.add_argument("--mbrlow",
            help = "Low range of MBR",
            default = 7000000,
            type = int)
    args.add_argument("--mbrhigh",
            help = "High range of MBR",
            default = 10000000,
            type = int)
    args.add_argument("--devicegroup",
            help = "Camera device group",
            default = "menlo-4g-cameras",
            type = str)

    return parser


if __name__ == '__main__':
    log.basicConfig(format="[ %(levelname)s ] %(message)s", level=log.DEBUG, stream=sys.stdout)
    log.debug("Starting person detection app")
    args = build_argparser().parse_args()
    if not args.noroc:
        roc.set_mbr(args.key, args.devicegroup, args.mbrlow)
    app.run(host='0.0.0.0', debug=True)