Rapid Prototyping

Rapid prototyping will accelerate your timeline for building computer vision applications. Two key aspects that we will focus on are reducing iteration time and writing applications that can easily be used for testing in the lab as well as running on-site.

Reduce Iteration Time

In order to reduce iteration time while building your app, you must be able to get quick feedback on the current progress. One workflow edgeIQ enables is saving model output for playback, in order to test changes to downstream algorithms, such as object tracking and business logic. This is accomplished by saving model output to analytics files, and then loading those in subsequent runs of the application.

Your object detection code may look like this:

result = obj_detect.detect_objects(
    frame,
    confidence_level=confidence,
    overlap_threshold=overlap_threshold
)

Capture model output

Add a flag to enable the capture feature, inference_test_capture, and add a call to publish_analytics to save the output to an analytics file. You can add a tag with additional information if desired, and even override the default output path (logs/analytics.txt). Note that when working with multiple streams in one app, you must write each stream to its own file, or split the results in post-processing using the tags.

if inference_test_capture:
    obj_detect.publish_analytics(
        result,
        tag={
            'stream_idx': stream_idx,
            'frame_idx': frame_idx
        },
        file_path=test_capture_file_path[stream_idx]
    )

Before running, make sure you enable analytics output to file by running:

aai app analytics enable-file-publish

Using this approach, you can collect model output for a video file once, and run the app over the video without taking the cost of inference for quick feedback cycles.

Run app on captured model output

Once you have your analytics file, move it to a safe place so it doesn’t get overwritten. Next, we will load the model output file instead of using the model for inference.

Instead of edgeiq.ObjectDetection, create an alternative object detector that loads the analytics results:

annotation_results = [edgeiq.load_analytics_results(file_path) for file_path in annotations_file_paths]
self._obj_detect = edgeiq.ObjectDetectionAnalytics(annotations=annotation_results, model_id=model_id)

This new object has the same interface as edgeiq.ObjectDetection and therefore can be used in a factory that selects between the two modes. For example:

def get_inference(
    mode: str,
    model_id: str,
    annotations_file_paths: Optional[List[str]]
) -> edgeiq.ObjectDetection:
    if mode == 'inference':
        return edgeiq.ObjectDetection(model_id=model_id)
    elif mode == 'annotations':
        annotation_results = [edgeiq.load_analytics_results(file_path) for file_path in annotations_file_paths]
        return edgeiq.ObjectDetectionAnalytics(annotations=annotation_results, model_id=model_id)
    else:
        raise ValueError(f'Unsupported mode {mode}!')

Further, you can reduce duplication in your code by reading the model ID from alwaysai.app.json. For example:

# Select the last model in the app configuration models list
model_id_list = edgeiq.AppConfig().model_id_list
if len(model_id_list) == 0:
    raise RuntimeError('No models in model ID list!')

model_id = model_id_list[-1]

Run app on annotations from an annotation tool

Additionally, if you don’t yet have a model but want to get developing, you can annotate the video and export the annotations in the following formats:

  • COCO: Load these annotations using edgeiq.parse_coco_annotations(). This has identical output to edgeiq.load_analytics_results() and can be used interchangeably.

  • MOT: Load these annotations using edgeiq.parse_mot_annotations(). This is useful for tracking annotations, and the output is a list of edgeiq.TrackingResults.

  • CVAT: Load these annotations using edgeiq.parse_cvat_annotations(). Note that this provides additional information that CVAT includes, such as tracked frames (if tracking was used to annotate) and occluded frames.

Configurable Inputs & Outputs

Much like the factory for inference to switch between using the model and using the annotations, it is very straightforward to enable different video inputs based on a configuration. For example:

def get_video_stream(mode: str, arg: str | int) -> edgeiq.BaseVideoStream:
    if mode == 'file':
        return edgeiq.FileVideoStream(arg)
    elif mode == 'usb':
        return edgeiq.WebcamVideoStream(arg)
    elif mode == 'ip':
        return edgeiq.IPVideoStream(arg)
    else:
        raise ValueError(f'Unsupported mode {mode}!')

This same approach can be used for application outputs as well. For example, it can be useful to save video output when running in test mode:

class NoVideoWriter(edgeiq.VideoWriter):
    def __init__(self):
        pass

    def write_frame(self, frame: np.ndarray):
        pass

    def close(self):
        pass


def get_video_writer(enable: bool, *args, **kwargs) -> edgeiq.VideoWriter:
    if enable:
        return edgeiq.VideoWriter(*args, **kwargs)
    else:
        return NoVideoWriter()

Now, you can parameterize your application using a configuration to select the different modules to use. To enable configuration using alwaysAI Remote Deployment, add the configs to alwaysai.app.json. You can modify that file on the cloud console so that you don’t need to redeploy to change a configuration.

For example:

{
  "scripts": {
    "start": "python app.py"
  },
  "models": {
    "alwaysai/yolo_x": 5
  },
  "app_configurations": {
    "video_stream": {
      "mode": "usb",
      "arg": 0
    },
    "inference": {
      "mode": "inference",
      "enable_test_capture": false,
      "confidence": 0.5,
      "overlap_threshold": 0.3,
    },
    "video_writer": {
      "enable": true,
      "output_path": "output.avi",
      "fps": 30,
      "codec": "MJPG",
      "chunk_duration_s": 30
    }
  },
  "analytics": {
    "enable_file_publish": true
  }
}

In your app, you can access the contents of alwaysai.app.json in the following way:

app_file = edgeiq.AppConfig().app_file
if app_file is None:
    raise RuntimeError('alwaysai.app.json not found!')
cfg = app_file['app_configurations']