Implementing Custom Policies in DeepStream Reference Designs
| DeepStream Reference Designs |
|---|
 |
| Getting Started |
| Project Architecture |
| Reference Designs |
| Customizing the Project |
| Contact Us |
What are the policies in this project?
The policies act as filters of the information that the inference process generates. It is up to each application to recognize the values that will be returned by the inference to be processed and validated by the policy. Some applications can rely on the use of additional filters that allow the frames coming from the inference to be analyzed, after a filtering process where the received data is highly accurate. For example, we can use a "low-pass filter" for filtering the inference frames since will allow greater precision of the resulting value of the inference. This low-pass filter is in charge of analyzing the incoming information and filtering the data according to the configuration parameters selected by the user. Once the information filtered by the low pass filter has been processed, the policies are in charge of performing the business intelligence with the information resulting from the filter. Again, these types of extra components, such as "low-pass filters", are specific to the context of the application and the requirements of a particular project, so the custom implementation of these tools is up to you, but since this type of module can be useful in many applications, a brief explanation of the general structure of these type of filters is provided in this wiki.
Class Diagram
A policy is a structure where the user implements their business rules. To implement a new policy you must follow the next interface:
Operations
The interface methods that must be respected in any custom implementation are:
- evaluate: This method changes the status of asserted value depending on the conditions that you want to realize using the information from the inference info. The internal state of the is_asserted flag has to be changed to false or true, to allow an action to be performed or not later on.
- is_asserted: This method returns the current status of evaluating results.
Code Example
The following is a simple example of how a basic policy 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 information will be filtered by category and by threshold level.
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 create business rules. For instance, in the APLVR application the dictionary is:
{
"sensorId": "Exit",
"date": "2022-05-25",
"time": "22:44:25.019",
"instances": [
{
"id": "10",
"category": "lpd",
"bbox": {
"topleftx": 280.683,
"toplefty": 155.192,
"bottomrightx": 379.966,
"bottomrighty": 206.743
},
"extra": [
"#",
"P370651",
"0.979938"
]
},
{
"id": "13",
"category": "car",
"bbox": {
"topleftx": 137.766,
"toplefty": 0.0,
"bottomrightx": 494.419,
"bottomrighty": 309.661
},
"extra": []
}
]
}
Below is the code for a simple implementation of a policy. It should focus on changing the state of the is_asserted flag.
# -*- 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.
"""
class BasicPolicyError(RuntimeError):
"""Class used to represent the BasicPolicy exceptions
It extends from RuntimeError, and covers the exceptions that may arise during the filtering process
"""
class BasicPolicy:
def __init__(self, threshold, category):
""" This class receives the inference information data and filters it to execute customized business
rules according to threshold and categories values.
:param threshold: threshold value that the incoming inference data must exceed
:type threshold: float
:param category: categories that the inference data must have
:type category: string list
"""
self._threshold = threshold
self._category = category
self._is_asserted = False
def evaluate(self, inference_info):
"""This method changes the status of the asserted value if the inference_info categories are in the
category's private list and if its inference_info threshold is greater than or equal to the class.
:param inference_info: Structure with the categories and its threshold values to evaluate
:type inference_info: inference_info
"""
self._is_asserted = False
for instance in inference_info.inference_dict["instances"]:
instance_category = instance["category"]
instance_extra = instance["extra"]
# Check if incoming category is in policy category
# and if extra value has more than 2 spaces
max_extra_values = 2
if (
instance_category in self._category
and len(instance_extra) > max_extra_values
):
threshold_position = 2
instance_threshold = instance_extra[threshold_position]
try:
# Check if incoming threshold is more or equal than policy threshold
if float(instance_threshold) >= self._threshold:
if self.business_rule_1:
self._is_asserted = True
except ValueError as e:
raise BasicPolicyError(
"The instance threshold value is not a float"
) from e
def is_asserted(self):
"""This method returns the current status of the evaluate result
:return: Returns true or false depending on the method 'evaluate'
:rtype: bool
"""
return self._is_asserted
def business_rule_1(self):
"""This method is a proof of concept of the implementation of a business rule
:return: Return true depending the business rule
:rtype: boolean
"""
return True
Low pass filter
This filter is in charge of receiving the raw inference information in some format depending on the current application, so it must be previously parsed and filtered by other modules at this stage. The way this low pass filter works is that the inference information is temporarily stored in a hit table according to the object of importance, where depending on the assigned identifier and the value of the object the hit count will be increased or reset. If the object identifier has the same value, the number of hits is increased by 1, but if the object identifier has a different value (the inference detected another character or another value) the number of hits is reset. The identifier assigned to an object will depend on the object identifier (assigned by the inference tracker) and the name of the stream where the object was detected, thus being a tuple. Once the inference object has passed the filter, we proceed to delete that object according to its tuple identifier from the hit table and add certain information about the filtered object within the structure of the inference information. In addition to adding the filtered information within the inference information structure, the status of the is_asserted flag, which indicates the status of all the evaluation (filtered steps) mentioned above, is changed. This status indicates to the policy that there is filtered information to be processed by the business rules.
Flowchart
Elements used by the low pass filter
- Hit table: This is a table that keeps track of the number of times that the inference of the same object has been detected. If the number of hits exceeds the maximum number of hits, the information will be filtered to be processed by the business rules. If the expected inferences data changes, the hit count is reset.
- Max Hit: Value of the required hit that the inference data must reach in order to be filtered.