Custom Trigger Action

This example shows how to set custom action to be triggered when a certain event occurs. In this case, we will trigger an action when a person is detected in the frame. The action will save the exact frame to a file.

Note

Visualization in current example is done with blocking behavor. This means that the program will halt at oak.start() until the window is closed. This is done to keep the example simple. For more advanced usage, see Blocking behavior section.

Setup

Please run the install script to download all required dependencies. Please note that this script must be ran from git context, so you have to download the depthai repository first and then run the script

git clone https://github.com/luxonis/depthai.git
cd depthai/
python3 install_requirements.py

For additional information, please follow our installation guide.

Pipeline

Pipeline graph

Source Code

Python

Also available on GitHub.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from pathlib import Path
from typing import Dict

import cv2

from depthai_sdk import OakCamera, FramePacket
from depthai_sdk.trigger_action import Action, DetectionTrigger


class MyAction(Action):
    """
    Saves the latest frame from the input stream to a file.
    """

    def __init__(self, inputs, dir_path):
        super().__init__(inputs)

        self.dir_path = Path(dir_path)
        self.dir_path.mkdir(parents=True, exist_ok=True)

        self.latest_packets = None

    def activate(self):
        print('+', self.latest_packets)
        if self.latest_packets:
            for stream_name, packet in self.latest_packets.items():
                print(f'Saving {stream_name} to {self.dir_path / f"{stream_name}.jpg"}')
                cv2.imwrite(str(self.dir_path / f'{stream_name}.jpg'), packet.frame)

    def on_new_packets(self, packets: Dict[str, FramePacket]) -> None:
        self.latest_packets = packets


with OakCamera() as oak:
    color = oak.create_camera('color', '1080p')
    nn = oak.create_nn('mobilenet-ssd', color)

    oak.trigger_action(
        trigger=DetectionTrigger(input=nn, min_detections={'person': 1}, cooldown=30),
        action=MyAction(inputs=[nn], dir_path='./images/')  # `action` can be Callable as well
    )

    oak.start(blocking=True)

Got questions?

Head over to Discussion Forum for technical support or any other questions you might have.