Implementing Custom Actions in DeepStream Reference Designs

What are the actions in the project?

They correspond to what the application will perform after being filtered by the policies. Actions could use filtered information, incoming frame data, or any relevant information from the incoming inference result. The user is free to add whatever actions they require for its application.

Class Diagram

An action is a structure where the user implements what the application is going to do. To implement a new action you must follow the next interface:

Operations

The interface methods that must be respected in any custom implementation are:

call: Executes the specific action if all policies are asserted. When you can use inference_info for whatever you want. The status of is_triggered method has to be checked, to allow an action to be performed or not. That status depends on incoming policies.

In ActionCommon there are methods that other actions must use:

is_triggered( ): bool: This method checks if all the incoming policies are asserted.

Code Example

The following is a simple example of how an action should be implemented. For the explanation, the APLVR application was taken as a base, which is explained in greater detail in the Reference Designs section. In this case, the action is standard output when it prints a custom message using inference_info data.

First of all, the user should know which data the inference_info is going to receive in the dictionary, with that information you are going to use with any custom action. For instance, in the APLVR application the dictionary is:

{
   "sensorId": "Entrance",
   "date": "2022-06-02",
   "time": "20:56:04.143",
   "instances": [
       {
           "id": "15",
           "category": "lpd",
           "bbox": {
               "topleftx": 287.81,
               "toplefty": 270.95,
               "bottomrightx": 312.255,
               "bottomrighty": 289.242
       },
           "extra": [
               "#",
               "7EL10",
               "0.968441"
           ]
       },
       {
           "id": "3",
           "category": "lpd",
           "bbox": {
               "topleftx": 502.654,
               "toplefty": 271.34,
               "bottomrightx": 530.109,
               "bottomrighty": 291.245
           },
           "extra": [
               "#",
               "PMB325",
               "0.979745"
           ]
       }
   ],
   "filtered": [
       {
           "sensorId": "Entrance",
           "date": "2022-06-02",
           "time": "20:56:04.143",
           "license_plate": "PMB325",
           "bbox": {
               "topleftx": 502.654,
               "toplefty": 271.34,
               "bottomrightx": 530.109,
               "bottomrighty": 291.245
           },
           "status": "new"
       }
   ]
}

In the policies, the information of the inference_info is modified, in this case, it was decided to add a list of filtered values.

The code of is_triggered() is:

# -*- coding: utf-8 -*-
"""
Copyright (C) 2022 RidgeRun, LLC (http://www.ridgerun.com)
All Rights Reserved.
 
The contents of this software are proprietary and confidential to RidgeRun,
LLC.  No part of this program may be photocopied, reproduced or translated
into another programming language without prior written consent of
RidgeRun, LLC.  The user is free to modify the source code after obtaining
a software license from RidgeRun.  All source code changes must be provided
back to RidgeRun without any encumbrance.
"""

def is_triggered(policies):
   """Get asserted status of all the policies, that were evaluated, if one of them fails, return false
 
   :param policies: List of policies entity
   :type policies: Policy
   :return: If one the policies is not asserted, return false if not returns true
   :rtype: bool
   """
   for policy in policies:
       if not policy.is_asserted():
           return False
   return True

Below is the code for a simple implementation of an action.

# -*- coding: utf-8 -*-
"""
Copyright (C) 2022 RidgeRun, LLC (http://www.ridgerun.com)
All Rights Reserved.
 
The contents of this software are proprietary and confidential to RidgeRun,
LLC.  No part of this program may be photocopied, reproduced, or translated
into another programming language without the prior written consent of
RidgeRun, LLC.  The user is free to modify the source code after obtaining
a software license from RidgeRun.  All source code changes must be provided
back to RidgeRun without any encumbrance.
"""
 
import threading
import src.aplvr.action.action_common
 
 
def log(inference_info):
   """ Logs the event using stdout
 
   :param inference_info: Structure with the categories and its threshold values to evaluate
   :type inference_info: inference_info
   """
 
   for data in inference_info.inference_dict["filtered"]:
       text = f'Vehicle with license plate: {data["license_plate"]} is in sector "{data["sensorId"]}". Status: {data["status"]}. {data["date"]},  {data["time"]}'
       print("\n", text)
 
 
class Stdout:
   """Class used to log the activity using stdout depending if all the policies are asserted
   """
 
   def __init__(self):
       """ Constructor method
       """
       self._mutex = threading.Lock()
 
   def call(self, policies, inference_info):
       """ Executes the stdout action if all policies are asserted
 
       :param policies: List of policies were executed
       :type policies: Policy
       :param inference_info: Structure with the categories and its threshold values to evaluate
       :type inference_info: inference_info
       """
       if not src.aplvr.action.action_common.is_triggered(policies):
           return
 
       with self._mutex:
           log(inference_info)

The result of that action shows the following message:

A vehicle with license plate: XYZ123 is in sector "Entrance". Status: new. 2022-06-02,  20:56:04.143

Previous: Customizing the Project/Implementing Custom Policies

Index

Next: Customizing the Project/Implementing Custom Media

❯