I.MX8 Deep Learning Reference Designs - Implementing Custom Actions
i.MX8 Deep Learning Reference Designs |
---|
Getting Started |
Project Architecture |
Reference Designs |
Customizing the Project |
Contact Us |
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 the 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 Restricted Zone Detector 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 RZD application, the dictionary is:
{ { "id" : 556, "enabled" : "True", "bbox" : { "x" : 0, "y" : 0, "width" : 416, "height" : 416 }, "classes" : [ ], "predictions" : [ { "id" : 557, "enabled" : "True", "bbox" : { "x" : 322, "y" : 166, "width" : 82, "height" : 210 }, "classes" : [ { "Id" : 484, "Class" : 14, "Label" : "person", "Probability" : "12.757812", "Classes" : 20 } ], "predictions" : [ ] } ] }
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 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.rzd.action.action_common from datetime import datetime def log(media_name, 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 """ """ ip = "localhost" port = 5000 snapshot = Snapshot(ip, port) snapshot.call(inference_info.media_id)""" now = datetime.now() dt_string = now.strftime("%d/%m/%Y") now = now.strftime("%H:%M:%S") now = f"{now}" dt_string = f"{dt_string}" media_name = f"{media_name}" text = f'A person was detected at {now} hours in sector {media_name} on the day {dt_string}.' print("\n", text) class Stdout: """Class used to log the activity using stdout depending if all the policies are asserted """ def __init__(self, media_name): """ Constructor method """ self._media_name = media_name 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.rzd.action.action_common.is_triggered(policies): return with self._mutex: log(self._media_name, inference_info)
The result of that action shows the following message:
A person was detected at 17:30:44 hours in sector ZoneA on the day 30/09/2022.