Receiving and processing MQTT messages from Frigate NVR

[sevmonster]

I set up a Frigate -> Mosquitto -> mqttwarn -> Apprise -> Ntfy -> Nginx flow (whew) and everything is working... Almost. The filter seems to not be blocking bad messages even though it should be based on my debugging. And, I can't figure out how to send attachments. My only lead is to save the attachment to a local file before using the attach query parameter in my Ntfy URI, but it would be nice to be able to send it through the body of the message instead, which I believe Apprise supports.

Depending on if any of this not working is a bug or potential feature request, this issue could turn into an actual issue and not a support request :)

mqttwarn.ini

[defaults]
functions = funcs.py
launch    = apprise-ntfy

status_publish = True


[config:apprise-ntfy]
module   = apprise_single
baseuri  = ntfys://ntfy/frigate


[frigate/events]
filter  = frigate_events_filter()
alldata = frigate_events()
targets = apprise-ntfy
title   = {title}
format  = {format}
click   = {click}

funcs.py

# -*- coding: utf-8 -*-
from datetime import datetime

try:
    import json
except ImportError:
    import simplejson as json

def frigate_events(topic, data, srv=None):
    if srv is not None:
        srv.logging.debug('events ******************************************')
    a = json.loads(data['payload'])['after']
    f = lambda x: (y.replace('_', ' ') for y in x)
    r = {
      'camera': a['camera'],
      'label':  a['sub_label'] or a['label'],
      'current_zones': ', '.join(f(a['current_zones'])),
      'entered_zones': ', '.join(f(a['entered_zones'])),
      'time':   datetime.fromtimestamp(a['frame_time'])
    }
    r.update({
      'title':  f"{r['label']} entered {r['entered_zones']}",
      'format': f"In zones {r['current_zones']} at {r['time']}",
      'click':  f"https://frigate/events?camera={r['camera']}&label={r['label']}&zone={a['entered_zones'][0]}"
    })
    return r

def frigate_events_filter(topic, message, section, srv=None):
    if srv is not None:
        srv.logging.debug('filter ******************************************')
    try:
        message = json.loads(message)
    finally:
        message = None
    if message and message.get('type', None) != 'end' and 'after' in message:
        a = message['after']
        return False not in (x in a and (x == 'current_zones' or a[x]) for x in
                                  ('false_positive', 'camera', 'label',
                                   'current_zones', 'entered_zones',
                                   'frame_time'))
    return False

Sample payload

{
  "before": {
    "id": "1680791459.255384-abcdef",
    "camera": "camera",
    "frame_time": 1680791459.255384,
    "snapshot_time": 0,
    "label": "car",
    "sub_label": null,
    "top_score": 0,
    "false_positive": true,
    "start_time": 1680791459.255384,
    "end_time": null,
    "score": 0.7,
    "box": [
      0,
      20,
      0,
      20
    ],
    "area": 400,
    "ratio": 1,
    "region": [
      0,
      0,
      320,
      320
    ],
    "stationary": false,
    "motionless_count": 0,
    "position_changes": 0,
    "current_zones": [],
    "entered_zones": [],
    "has_clip": false,
    "has_snapshot": false
  },
  "after": {
    "id": "1680791459.255384-abcdef",
    "camera": "camera",
    "frame_time": 1680791506.638857,
    "snapshot_time": 1680791506.638857,
    "label": "car",
    "sub_label": null,
    "top_score": 0.75,
    "false_positive": false,
    "start_time": 1680791459.255384,
    "end_time": null,
    "score": 0.8,
    "box": [
      1,
      21,
      1,
      21
    ],
    "area": 400,
    "ratio": 1,
    "region": [
      0,
      0,
      320,
      320
    ],
    "stationary": false,
    "motionless_count": 1,
    "position_changes": 2,
    "current_zones": [],
    "entered_zones": [
      "zone1"
    ],
    "has_clip": true,
    "has_snapshot": true
  },
  "type": "new"
}

[amotl]

Dear sev,

thanks for writing in. I am not 100% sure, but I think the value of the message item within a JSON payload submitted via MQTT will get used as the body payload when forwarding messages to the corresponding service plugins.

You may want to play around with this command line incantation to see if "title" and "message" will get propagated properly.

mqttwarn --plugin=apprise \
    --config='{"baseuri": "ntfy://user:password@ntfy.example.org/topic1/topic2"}' \
    --options='{"addrs": [], "title": "Example notification", "message": "Hello world"}'

I do understand you correctly that putting something into the message body would fit your needs, right? Otherwise, please let me know if you are intentionally looking into putting something into the message's attachment instead. This may require a different approach, but is well worth to explore.

With kind regards,
Andreas.

[sevmonster]

I do understand you correctly that putting something into the message body would fit your needs, right? Otherwise, please let me know if you are intentionally looking into putting something into the message's attachment instead. This may require a different approach, but is well worth to explore.

Sorry, I was speaking in reference to the Ntfy API. You can upload attachments via PUT request body. With Apprise, I think the only way to send an attachment is via URI, so you can disregard that. I really thought that there was another way to do it with Apprise, but I may have been mistaken.

For the remaining issue, filtering does not appear to work, so the unmodified JSON payload is displayed for non-matching messages, unless I am just using it wrong.

[amotl]

Hi again,

it should work to submit attachments via mqttwarn => Apprise => Ntfy. However, we have not tested it, so it will be sweet to explore this detail together.

The "Parameter Breakdown" section of the corresponding documentation of Apprise's Ntfy plugin tells us:

Variable	Required	Description
attach	No	Specify a web URL pointing at a remote attachment you would like the post to reference.
filename	No	This is only used if the attach was also provided. The ntfy server is smart enough to determine what the filename is automatically from this; however if you wish to provide a custom-override to this value, this is where you do it from.

-- https://github.com/caronc/apprise/wiki/Notify_ntfy#parameter-breakdown

So, maybe you are already successful with a dry-run snippet like this, using ?attach=https://httpbin.org/ip as an example?

mqttwarn --plugin=apprise \
    --config='{"baseuri": "ntfy://user:password@ntfy.example.org/topic1/topic2?attach=https://httpbin.org/ip"}' \
    --options='{"addrs": [], "title": "Example notification", "message": "Hello world"}'

If you can validate that any of those options to use the attach or filename parameters will work and fit your needs, we can explore how to propagate those parameters appropriately. I may think you want to use some field from the ingress message for that purpose?

With kind regards,
Andreas.

[amotl]

we can explore how to propagate those parameters appropriately. I may think you want to use some field from the ingress message for that purpose?

Ah, you actually don't want to forward an external resource as attachment, but use the return value of what you produce in frigate_events() instead, after serializing the dictionary into JSON again?

Or do you want to use frigate_events() as an alldata transformer, as usual, and add the full ingress payload additionally as an attachment?

[sevmonster]

The image is sent in another MQTT payload, so it first needs to be ingressed by the alldata function. I have not gotten around to that yet. And yes, I was previously under the impression Apprise could take raw data attachments and pass them to a plugin, but upon further reading I do not believe that is possible. I am not sure if passing a file:// URI will open the file and read it, or just send the data. But I can get it to work regardless of either case, and I will also have to save the image to disk locally regardless of either case.

I am confident I can get that to work, and I have tested attachments via URI before (only with external resources fetched with HTTP however) so my main concern now is why the filter function isn't working, unless you believe there is some work that could be done to streamline the attachment process that should be focused on.

[amotl]

The image is sent in another MQTT payload, so it first needs to be ingressed by the alldata function.

Oh I see. So you will have to introduce a kind of stateful processing for that? Will be fun!

Apprise could take raw data attachments and pass them to a plugin, but upon further reading I do not believe that is possible.

I also think it is not possible. Maybe @caronc has different things to say about this?

I am not sure if passing a file:// URI will open the file and read it, or just send the data. But I can get it to work regardless of either case, and I will also have to save the image to disk locally regardless of either case.

Yeah, it will probably need some trial-and-error. Please let me know if you find any blockers.

Do you believe there is some work that could be done to streamline the attachment process?

My thoughts would have been around how to properly propagate an "attachment" parameter from the inbound MQTT message payload or topic into the transformation dict and then into the outbound parameters to the Apprise/Ntfy plugin. But after learning you want to submit the whole inbound message itself as an attachment, I don't think this was actually your need, right?

My main concern now is why the filter function isn't working.

All right, let's focus on this detail now.

[sevmonster]

We do indeed have a problem with the first part that I was hoping would be easy :) #634

The plan was to save the snapshot to file and provide a proxied URL (or file:// URI if it works) to Apprise, which Ntfy would then save to disk in its attachment cache. Like I said I was very confident that would work, but it seems I was overzealous.

[amotl]

I am just now getting a bit more into the details while I am looking into the snippets you provided. I will add them to the examples directory if you don't mind, so there will be a canonical reference to drive this forward.

I was hoping would be easy, but I am running into GH-634. The plan was to save the snapshot to file and provide a proxied URL (or file:// URI if it works) to Apprise, which Ntfy would then save to disk in its attachment cache.

That's sad. I was hoping that Frigate would provide an URL to the snapshot file within the "event" message already. But I see that Frigate itself has no capability to store those snapshot files anywhere, so it has no other chance than just submitting their payloads per MQTT, right? ¹

We had two other discussions about submitting images in the past, but I think your scenario is new.

• Pushover now supports image attachments - would be grand to add this to mqttwarn #284
• [Telegram] Add possibility to send photo/image #486

We do indeed have a problem with the first part.

Don't fret. Unless nothing serious will block this endeavour, I think we can make it work.

Footnotes

1. Does Frigate offer any other options in this regard, like uploading to an S3-compatible object store and then publishing a corresponding event message to MQTT? ↩

[amotl]

My main concern now is why the filter function isn't working.

All right, let's focus on this detail now.

While getting the code snippet to my machine, my editor's code highlighting immediately reveals why it is probably not working: Line 34 will always assign message = None. You probably want to say except there, instead of finally.

[sevmonster]

While getting the code snippet to my machine, the editor immediately reveals why it is probably not working: message will always be undefined, because line 34 will always assign None

Oops, that should be else not finally. But even in its current incarnation, should it not be stopping all messages since it always returns False? Why are messages getting through? Am I misunderstanding the filter function usage?

I did individually test the logic in the conditional with dummy data so the rest of it should be sound and do what I want, with the understanding that True is a passing filter and False fails.

I will add them to the examples directory if you don't mind, so there will be a canonical reference to drive this forward.

Feel free, but you may want to wait until everything works! Though, when the message comes in and matches all the criteria, the alldata does provide good output:

That's sad. I was hoping that Frigate would provide an URL to the snapshot file within the "event" message already. But I see that Frigate itself has no capability to store those snapshot files anywhere, so it has no other chance than just submitting them per MQTT?

I could just pull it from Frigate's files (all snapahots are stored in files, but the filenames are esoteric and I have not verified if the event message has enough data to compose the path) but I thought it might be easier to just rely fully on MQTT. I, too, was hoping it pushed events based on detected object ID and not the label, since there's a chance for a race condition with how I am approaching it now, but here we are.

Does Frigate offer any other options in this regard, like uploading to an S3-compatible object store and then publishing a corresponding event message to MQTT?

Frigate can optionally save snapshots to file, but that is a separate mechanism from the snapshot publishing to MQTT. I don't know if it offers object storage but I will lean towards no. (I wouldn't use it anyway.)

[amotl]

I will your code snippets to the examples directory if you don't mind, so there will be a canonical reference to drive this forward.

Feel free, but you may want to wait until everything works!

Sure. I will just add it to a branch first, so that you can use and test mqttwarn from there, including any improvements we will make while we go.

I thought it might be easier to just rely fully on MQTT.

Sure!

Frigate can optionally save snapshots to file, but that is a separate mechanism from the snapshot publishing to MQTT. I don't know if it offers object storage but I will lean towards no. (I wouldn't use it anyway.)

I wouldn't advise on using S3 at AWS, if this is what your concerns would be about. I was looking at MinIO here instead, which offers an S3-compatible interface. But if there is no MQTT message which will point out the URL, correlated with the event you are looking at, this is pointless.

Those are some corresponding items on the issue tracker of Frigate. It looks like all of them stalled, I am just referencing them for the sake of completeness, before moving on.

• S3 / Minio support blakeblackshear/frigate#1192
• [Feature Request] Custom storage provider for media files (S3-like or WebDAV) blakeblackshear/frigate#2964
• archive storage blakeblackshear/frigate#3673
• Central network-based database container for multiple Frigate processors blakeblackshear/frigate#3305

[sevmonster]

I wouldn't advise on using S3 at AWS, if this is what your concerns would be about.

Yep.

But if there is no MQTT message which will point out the URL, correlated with the event you are looking at, this is pointless.

I think the path can be constructed based on the payload. Frigate saves snapshots (previously clips) to clips/<cam>-<objectid>[-clean].<ext>. Since I save snapshots for all events, it should be enough to create the URI pointing to where that file will be/is. I will need to investigate if the file is written async or if it is written after the event is triggered, to see if I need to kludge a race condition or not.