Build an Application From Scratch


This section describes how to build a simple real-time app from scratch using the alwaysAI CLI, model catalog, and edgeIQ APIs.

Initializing the Application Directory


The very first step is to make an empty directory for your app:

$ mkdir app && cd app

Running the app configure command will make the empty directory into an alwaysAI app directory:

$ aai app configure

Answer y when prompted about initializing an app in the directory, and provide the hostname of the device you’d like to run your app on. Your app directory will now have an app config, target config, app.py, and Dockerfile:

$ ls
alwaysai.app.json  alwaysai.target.json  app.py  Dockerfile

Next, we’ll write the application code. Start by opening the app.py file in your favorite editor.

Building the Application

There are three main sections to a simple real-time application:

  1. Initialization

  2. Frame Processing Loop (perform inference and post-processing)

  3. Final Analysis and Clean Up

The following sections describe how to build these sections for a real-time object detector.

Initialization

In this stage, we’ll set up all of the necessary objects that will be used in the later sections. First, instantiate the Object Detection object. Then, load the detector on the DNN engine, which defaults to the GPU accelerator:

def main():
    obj_detect = edgeiq.ObjectDetection("alwaysai/mobilenet_ssd")
    obj_detect.load(engine=edgeiq.Engine.DNN)

To learn more about object detection, head to the object detection guide. To learn more about engines and accelerators, head to the engine and accelerator guide. To learn more about selecting and changing models, head to the changing the model guide.

Next, set up the webcam video stream and debug streamer as context managers:

def main():
...
    with edgeiq.WebcamVideoStream(cam=0) as video_stream, \
            edgeiq.Streamer() as streamer:

By setting cam=0 we’ve selected the camera which is at index 0 on the system. If your board or development machine only has one camera, this is likely the right choice. You can learn more about the webcam video stream in the webcam video guide. For the Streamer, the default setting works great for real-time use. You can learn more about the Streamer in the Streamer guide.

Next, set up the frames-per-second counter. Since the webcam takes a bit to warm up, it is a good idea to let it sleep for a short amount of time to make sure the FPS is accurate:

def main():
...
    with edgeiq.WebcamVideoStream(cam=0) as video_stream, \
            edgeiq.Streamer() as streamer:
        time.sleep(2.0)
        fps = edgeiq.FPS().start()

You’ll need to import the time module to your app in order to use the sleep() function. You can do so by adding this import command to the top of your file:

import time

You can learn more about tracking your app’s FPS in the performance guide.

Frame Processing Loop

Now, the main frame processing loop begins.

def main():
    ...
    with ...:
        ...
        while True:

Perform Inference

Grab a frame from the webcam, and perform an object detection inference on the frame:

def main():
    ...
    with ...:
        ...
        while True:
            frame = video_stream.read()
            results = obj_detect.detect_objects(frame, confidence_level=.5)

The results object contains the inference time and the list of predictions, each containing the bounding box, label, and confidence. Read the results object guide to learn more about the results objects.

Perform Post-processing

For this app, we’ll perform very simple post-processing. We’ll draw the bounding boxes on the frame, and send the frame and some text to the Streamer for viewing. To draw the bounding boxes, we’ll use the markup_image() function described in the manipulating images and video frames guide:

def main():
    ...
    with ...:
        ...
        while True:
            ...
            frame = edgeiq.markup_image(
                    frame, results.predictions, colors=obj_detect.colors)

Next, generate the text for the Streamer. Initialize the text with the inference time:

def main():
    ...
    with ...:
        ...
        while True:
            ...
            text = ["Inference time: {:1.3f} s".format(results.duration)]

For each object, we’ll add the class label and confidence:

def main():
    ...
    with ...:
        ...
        while True:
            ...
            text = ["Inference time: {:1.3f} s".format(results.duration)]
            text.append("Objects:")
            for prediction in results.predictions:
                    text.append("{}: {:2.2f}%".format(
                        prediction.label, prediction.confidence * 100))

Now update the Streamer with the frame and text, and then update the FPS counter now that all post-processing is complete:

def main():
    ...
    with ...:
        ...
        while True:
            ...
            streamer.send_data(frame, text)
            fps.update()

The Streamer has a “stop” button that can be used to safely stop real-time apps. To stop your app when the “stop” button is pressed, add the following conditional to your code:

def main():
    ...
    with ...:
        ...
        while True:
            ...
            if streamer.check_exit():
                break

Final Analysis and Clean Up

Once you exit the frame processing loop, you can end the FPS counter and print the gathered statistics:

def main():
    ...
    with ...:
        ...
        while True:
            ...
        fps.stop()
        print("elapsed time: {:.2f}".format(fps.get_elapsed_seconds()))
        print("approx. FPS: {:.2f}".format(fps.compute_fps()))

Full Application Source

Here is the full source code from following the above steps:

import time
import edgeiq

def main():
    obj_detect = edgeiq.ObjectDetection("alwaysai/mobilenet_ssd")
    obj_detect.load(engine=edgeiq.Engine.DNN)

    fps = edgeiq.FPS()

    with edgeiq.WebcamVideoStream(cam=0) as video_stream, \
                edgeiq.Streamer() as streamer:
        time.sleep(2.0)
        fps.start()

        while True:
            frame = video_stream.read()
            results = obj_detect.detect_objects(frame, confidence_level=.5)
            frame = edgeiq.markup_image(
                    frame, results.predictions, colors=obj_detect.colors)

            text = ["Model: {}".format(obj_detect.model_id)]
            text.append(
                    "Inference time: {:1.3f} s".format(results.duration))
            text.append("Objects:")

            for prediction in results.predictions:
                text.append("{}: {:2.2f}%".format(
                    prediction.label, prediction.confidence * 100))

            streamer.send_data(frame, text)

            fps.update()

            if streamer.check_exit():
                break

        fps.stop()
        print("elapsed time: {:.2f}".format(fps.get_elapsed_seconds()))
        print("approx. FPS: {:.2f}".format(fps.compute_fps()))

if __name__ == "__main__":
    main()

Running the Application

To install the app on your selected target, run the app deploy command using the CLI. Note that in order for the CLI to know to install alwaysai/mobilenet_ssd, you must also use the CLI to set the model as an app dependency:

$ aai app models add alwaysai/mobilenet_ssd
$ aai app deploy

Now you can start your app using the app start command:

$ aai app start

The Streamer will expose a graphical web interface while the application is running.

To stop the application, hit the red “stop” button on the Streamer.

The app you’ve just built is nearly identical to the realtime_object_detector starter app, so you can compare your app to the starter app if you are running into issues.