EventDetector(), EventNotifier() and gotify

[gsczuka]

Short description of my project: detect/track cutest cat on earth (by the way, my cat)

Longer description: my inference script, based on hailo-rpi5-examples/basic_pipelines/detection.py, sends notification via gotify to my Android when cat or person or both is detected and additionally send it to database. What to track and notify is governed by an Android App I developed. The selection is written to labels.json and read by my custom detection.py. Then I decided to fine tune my own model for my purpose.

I used YOLO11s.pt to fine tune to cat and person and upload best.pt to DeGirum Compiler.

I wish I could say more about the Compiler, but all I can say is: it was the task I feared most and it turned out to be the easiest and fasted part of my project. Well done!

My model works as expected, but because I was using class_ids instead of labels my cat was a person and I was a bicycle despite using the labels-json-parameter. I couldn’ t convince the hailo pipeline to use my custom label file or persuade the cat to go to work instead of me, thus I decided to give Degirum PySDK a second chance (when I started with Hailo I couldn’t get a stable stream from camera with Degirum PySDK).

I’m impressed how much PySDK improved. I have a stable stream now - not to mention that I’m a person and my cat a cat. Long story short: I want to switch from Hailo/gstreamer-pipelines to DegirumPySDK and tools. Now I wonder how to best handle the apprise/gotify and database task. I found EventDetector and EventNotifier, but I’m unsure how to use them correctly. I found the docs and smart_nvr.ipynb, but failed to adapt it to my needs, which is: if detection result is in label-from-Android.json then do something, i.e. send a message to gotify. I even don’t know if I need those classes or if I should code them. I love the idea of gizmos and assume, this should go into a gizmo pipeline (or however it is called), but I’m unsure about that, too. Any help is highly appreciated.
Here is my code, pretty standard, I think:

import degirum as dg
import degirum_tools
import degirum_tools.streams as dgstreams
import apprise

==== Setup ====

video_source=“rtsp://myserver/input”

Loading a model

model = dg.load_model(
model_name = “mymodel–640x640_quant_hailort_hailo8_1”,
inference_host_address = “@local”,
zoo_url = “/MyModels/mymodel–640x640_quant_hailort_hailo8_1”,
overlay_show_probabilities = True,
overlay_line_width = 1
)

==== Setup Gotify ====

apobj = apprise.Apprise()
apobj.add(“gotify-address here”)
notification_config = {
“apprise”: apobj, # EventNotifier spricht mit Apprise
}

Gizmo rulez

source = dgstreams.VideoSourceGizmo(video_source)
#resize = dgstreams.ResizingGizmo(640, 640)
detection = dgstreams.AiSimpleGizmo(model, allow_drop=True)

#display = dgstreams.VideoDisplayGizmo(“Detection”, allow_drop=True, show_ai_overlay=True, show_fps=True)
streamout = dgstreams.VideoStreamerGizmo(“rtsp://rtsp://myserver/output”, show_ai_overlay=True)

dgstreams.Composition(source >> detection >> streamout).start()

[gsczuka]

I think I’ve got it: I either customize AiGizmoBase or I create an Analyzer. This allows me to use the code I already have in my previous version, more or less, right?

[vladk]

Hi @gsczuka ,

The following code illustrates what can be done to achieve your goal.
Here I use EventDetector gizmo with custom metrics function, which returns number of detected objects listed in labels set (you can load this set from your json). This gizmo generates event event_name. Then EventNotifier handles this event and sends notifications when event is detected.

The example streams annotated video stream to local preview window (using VideoDisplayGizmo gizmo) and to RTSP server (using VideoStreamerGizmo gizmo). It also launches RTSP/WebRTC server (with degirum_tools.MediaServer()).

If you do not need local preview, just remove all code related to VideoDisplayGizmo

You can observe that WebRTC stream if you run your browser and navigate to localhost:8888/stream/

You need to install ffmpeg and mediamtx software to do RTSP/WebRTC streaming.
Take mediamtx from here: Release v1.14.0 · bluenviron/mediamtx and unpack it into dir in your path.

Regarding notifications. This example just prints them to console. If you need to send them to gotify, please follow this guide: Notify_gotify · caronc/apprise Wiki

Put proper notification URL into notification_config instead of "json://console"

import degirum as dg
import degirum_tools
import degirum_tools.streams as dgstreams


video_source = 0  # 0 - for local camera, can be RTSP URL
labels = {"person", "cat"} # labels to detect
event_name = "ObjectDetected" # can be any variable name like string
notification_config = "json://console"

model = dg.load_model(
    model_name = "yolov8n_coco--640x640_quant_hailort_hailo8_1",
    inference_host_address = "@cloud",
    zoo_url = "degirum/hailo",
    overlay_show_probabilities = True,
    overlay_line_width = 1
)

# custom metric function: returns number of detected objects of interest
def check_for_labels(result, params):
    detected_labels = { r["label"] for r in result.results }
    return len(detected_labels & labels)

# event detector with custom metric function
event_detector = degirum_tools.EventDetector(
    f"""
    Trigger: {event_name}
    when: CustomMetric
    is greater than: 0
    during: [1, frame]
    """,
    custom_metric=check_for_labels,
    show_overlay=False,
)

# event notifier
notifier = degirum_tools.EventNotifier(
    event_name,
    event_name,
    message="{time}: object is detected",
    notification_config=notification_config,
)

degirum_tools.attach_analyzers(
    model, [event_detector, notifier]
)

source = dgstreams.VideoSourceGizmo(video_source)
detection = dgstreams.AiSimpleGizmo(model, allow_drop=True)
streamout = dgstreams.VideoStreamerGizmo(f"rtsp://localhost:8554/stream", show_ai_overlay=True)

# local display gizmo (just for debugging)
display = dgstreams.VideoDisplayGizmo(show_ai_overlay=True)

# start media server to serve RTSP streams
with degirum_tools.MediaServer():
    # connect gizmos into pipeline and start composition
    dgstreams.Composition(source >> detection >> streamout, detection >> display).start()

[gsczuka]

Hi @vladk
Thank you so much, this is working very well. Now I understand the concept. I always start with “Huh?! What the f***?!” and always end with “Oh my god, this is sooo cool”. Keep up the good work!

[shashi]

Hi @gsczuka

Happy to see you find the tools useful and cool.

[alex]

Hi @gsczuka, your message made me laugh that’s so relatable

If you feel like @vladk’s response answered your initial question, would you mind marking it as a solution? This helps others easily find the answer

[vladk]

Hi @gsczuka ,

Please be advised that EventNotifier gizmo is also capable of another cool feature: saving video clips around detected event in S3-compatible object storage or into local dir.

That smart_nvr.ipynb example show, how to set it up. You need to define the following args:

    clip_save=True,
    clip_duration=clip_duration,
    clip_pre_trigger_delay=clip_duration // 2,
    storage_config=storage_config,

Here clip_duration is desired clip duration in frames, clip_pre_trigger_delay is how many frames to save before the event (look back), and storage_config defines S3 storage parameters:

storage_config = degirum_tools.ObjectStorageConfig(
    endpoint="./temp",  # Object storage endpoint URL or local path
    access_key="",  # Access key for the storage account
    secret_key="",  # Secret key for the storage account
    bucket="nvr_clips",  # Bucket name for S3 or local directory name
)

– here it is defined to store files locally, into ./temp/nvr_clips dir.

[gsczuka]

I modified your check_for_labels function:

def check_for_labels(result, params):

allowed_labels = set(label_manager.get_labels()) //reads the json

detected_labels = {r\["label"\] for r in result.results}

if allowed_labels & detected_labels:

    result.custom_label = list(allowed_labels & detected_labels)\[0\]

    return 1

return 0

This allows me to use result.custom_label inside the message of the notifier. Gotify has event_name as title and message as body. Body now shows the label that triggered the notification.
Maybe this is usefull for someone.
When I’m not at home and Gotify shows “person”, I definitely have a problem A good use case for saving video clips, @vladk!

[dario]

Hi,

I’m very interested on your original request as I have the same doubt “how can I trigger customized actions based on eventDetector/eventNotifier like registering on database or triggering a sound’, but I don’t ‘see’ yet how to achieve it, by the moment I see you’re changing what? the ‘when’ in the eventDetector to a ‘custommetric’… But documentation says custommetric must return a numeric value…

Dont see how can I use it to calling to another function like register_on_database(message)

In my case I’m using almost the same trigger of the smart_nvr example but using ObjectCount instead ZoneCount.

I noticed that the first ‘event_name’ in the notifier, is in fact the subject of sent email:

event_name = "object_detected"

zone_detector = degirum_tools.EventDetector(
      f"""
      Trigger: {event_name}
      when: ObjectCount
      is greater than: 0
      during: [10, frames]
      for at least: [90, percent]
      """,
      show_overlay=False,
)    

notifier = degirum_tools.EventNotifier(
      "AI Detection Service - Object Detected",
      event_name,
      message="{time}: person is detected in zone",
      holdoff=holdoff_sec,
      notification_config=f"{notification_config}",
      clip_save=False,
      clip_duration=1,  # Send jpg with 0.22.3
      clip_pre_trigger_delay=10,
      clip_embed_ai_annotations=True,
      storage_config = clip_storage_config
    )

I also read somewhere that we can know if the notification was sent or if there was any error sending it…

[vladk]

@dario ,

Currently not - notifier only sends notification via Apprise - whatever Apprise supports. Since Apprise supports general webhooks, this is the way to invoke any web app.

Notifier is designed to not to slow down the main pipeline, therefore it does all the job in a worker process, asynchronously from the main process. This limits the argument types you can pass to a callback (if we would implement such functionality). Also that callback would be called in another process. May be this is not what you want.

But you can implement simple analyzer for calling your callback:

class CallbackCaller(degirum_tools.ResultAnalyzerBase):
    def __init__(self, callback: callable):
        super().__init__()
        self.callback = callback

    def analyze(self, result):
        if result.notifications:
            self.callback(result)
        return result

Then just add this analyzer directly after your notifier. It works this way: if there was any notification issued (result.notifications dict is not empty) then your callback is called.

[dario]

Sorry I’m not a developer so maybe I don’t use the right terms.

Maybe callback is not the proper word…

I was reading the apprise documentation, and maybe it can be used to trigger some sort of requests, but this also make me to ‘create’ a whole system to receibe apprise calls and process them.

When what I need should be so much simple… Imagine you want to put a message text on the video screen when more than 5 detections are made: “Warning 5 people on restricted Area” do you will be using Apprise and callbacks for this?

[vladk]

Hi @dario , in this case probably you will be good with little analyzer which just prints your message.

class MessagePrinter(degirum_tools.ResultAnalyzerBase):
    def __init__(self, message: str):
        super().__init__()
        self.message = message

    def analyze(self, result):
        if result.notifications:
            print(self.message) # or use your own printing function
        return result

If you want to print that particular message which is generated by EventNotifier, you can take it from result.notifications[event_name], here event_name is the name of the event you setup in EventDetector.

If you prefer to draw your message directly on the image overlay, you can do it in annotate() method of your MessagePrinter. Or, you can configure EventNotifier to do it for you: just enable show_overlay in constructor. You may control the position of the message by annotation_pos parameter. You may control the duration of the message by annotation_cool_down parameter.

[dario]

I was doing some more tests with the notifier, and there is something I misunderstood completely.

I though the notifier will be sending a email and if clip_save was True and having configured the local storage then a video_clip or a jpeg image was attached to the email.

But I found 2 files in the local dirctory instead with names -0000010.json and -0000010.mp4 how are this names created¿? can be changed?

Is there any way to send the attached clip in the email¿?

[vladk]

Hi @dario ,

Sending notifications and saving clips are two different functions of EventNotifier. Notification sending is handled by Apprise package, while clip saving is handled by S3 API. They happen completely asynchronously. Notification is sent soon after trigger event happens, but clip file is uploaded once it is fully captured. Let me explain. You may specify clip duration of, say, 5 seconds with pre-trigger delay of 2 seconds. When event is triggered, it will take 5-2=3 seconds more to finish capturing clip before it will be uploaded. Notification is already sent at this time.

Since .jpg image is considered as a sinlge-frame case of a clip, it is handled the same way.

But there is a workaround: you may include into the message text the URL of the associated clip: use {url} placeholder in the notification message text. If you use true S3 storage, it will be pre-signed URL to access that file. Meaning that you may click it in the browser and open it without S3 credentials. BTW, this approach minimizes e-mail traffic - file is downloaded only when e-mail recipient clicks on a link.

[dario]

Hi @vladk

I was thinking that maybe the notifier was listening to the clip saver (when configured) and sent the email when the clip saving action was ended

About the workarround (i’ve just installed minio so I will be testing it), it only works if the email is opened during a certain amout of time, isn’t? Beacuse it uses presigned ulr’s to add the attachment URL and this ones live by default 3600 seconds (one hour) after expiration.

So, any notification sent during the night that a recipient wants to open the next day, or any notification opened any time beyond one hour, won’t be able to see those images, right?

[vladk]

Hi @dario ,

About the workarround (i’ve just installed minio so I will be testing it), it only works if the email is opened during a certain amout of time, isn’t?

Right. This expiration you can control by setting ObjectStorageConfig.url_expiration_s parameter, when you pass storage_config parameter to EventNotifier constructor.

For AWS Signature Version 4 (SigV4) the maximum expiration time is 7 days (604,800 seconds).
I think it is more than enough for your overnight application.

[dario]

@vladk

This expiration you can control by setting ObjectStorageConfig.url_expiration_s parameter, when you pass storage_config parameter to EventNotifier constructor.

Nice, thanks!

One question more, I added the {url} parameter to the email body and I can download the files, but the message looks bad
I tried to separate the sentences using ‘\n’ but it simply becomes deleted in the email, is any otther way to enter a new line in the email message?

[vladk]

@dario

I tried to separate the sentences using ‘\n’ but it simply becomes deleted in the email, is any otther way to enter a new line in the email message?

I can change body_format to apprise.NotifyFormat.MARKDOWN. In this case you can use markdown message body formatting.

[vladk]

@dario , you can try right now by installing degirum_tools directly from github repo:

pip uninstall -y degirum_tools && pip install -U git+https://github.com/degirum/degirum_tools.git

[dario]

This should be in both devices? The one running the degirum service and the one running the script, isnt’?

Also if I understand the changes properly, markdown is now the default NotifyFormat, isn’t?

                    if not notifier.notify(
                        body=message, title=notification_title, tag=notification_tags, body_format=notifier.NotifyFormat.MARKDOWN
                    ):