feature: added notification

build/requirements.txt

@@ -36,3 +36,4 @@ Werkzeug>=0.15.3
 concurrent-log-handler>=0.9
 watchdog>=0.10
 dynaconf>=3.1
+apprise>=0.8.9

config.local.yaml

@@ -10,19 +10,32 @@ default:
   timeline:
     event_log: ./data/timeline-event-log.yaml
 
+  notifications:
+    providers: {}
+    template:
+      title: "[Ambianic.ai] ${event} recognized"
+      body: >
+        ${event} event has been recognized by the system.
+        Visit ${event_details_url} for more details.
+
   sources: {}
 
   ai_models:
     image_detection:
       model:
-        tflite: ai_models/mobilenet_ssd_v2_coco_quant_postprocess.tflite
-        edgetpu: ai_models/mobilenet_ssd_v2_coco_quant_postprocess_edgetpu.tflite
-      labels: ai_models/coco_labels.txt
+        tflite: /opt/ambianic-edge/ai_models/mobilenet_ssd_v2_coco_quant_postprocess.tflite
+        edgetpu: /opt/ambianic-edge/ai_models/mobilenet_ssd_v2_coco_quant_postprocess_edgetpu.tflite
+      labels: /opt/ambianic-edge/ai_models/coco_labels.txt
     face_detection:
       model:
-        tflite: ai_models/mobilenet_ssd_v2_face_quant_postprocess.tflite
-        edgetpu: ai_models/mobilenet_ssd_v2_face_quant_postprocess_edgetpu.tflite
-      labels: ai_models/coco_labels.txt
+        tflite: /opt/ambianic-edge/ai_models/mobilenet_ssd_v2_face_quant_postprocess.tflite
+        edgetpu: /opt/ambianic-edge/ai_models/mobilenet_ssd_v2_face_quant_postprocess_edgetpu.tflite
+      labels: /opt/ambianic-edge/ai_models/coco_labels.txt
       top_k: 2
+    fall_detection:
+      model:
+        tflite: /opt/ambianic-edge/ai_models/posenet_mobilenet_v1_100_257x257_multi_kpt_stripped.tflite
+        edgetpu: /opt/ambianic-edge/ai_models/posenet_mobilenet_v1_075_721_1281_quant_decoder_edgetpu.tflite
+      labels: /opt/ambianic-edge/ai_models/pose_labels.txt
 
   pipelines: {}

config.yaml

@@ -11,6 +11,18 @@ logging:
   file: ./data/ambianic-log.txt
   level: INFO
 
+# Store notifications provider configuration
+# see https://github.com/caronc/apprise#popular-notification-services for syntax examples
+# notifications:
+#   catch_all_email:
+#     include_attachments: true
+#     providers:
+#       - mailto://userid:pass@domain.com
+#   alert_fall:
+#     providers:
+#       - mailto://userid:pass@domain.com
+#       - json://hostname/a/path/to/post/to
+
 # Pipeline event timeline configuration
 timeline:
   event_log: ./data/timeline-event-log.yaml
@@ -77,5 +89,8 @@ pipelines:
      - save_detections: # save samples from the inference results
         positive_interval: 10
         idle_interval: 600000
+#        notify: # notify a thirdy party service
+#          providers:
+#            - alert_fall
 
 

src/ambianic/__init__.py

@@ -27,7 +27,12 @@ def __merge_secrets(config:Union[Dynaconf, DynaBox], src_config:Dynaconf = None)
         if isinstance(val, dict):
             __merge_secrets(val, src_config)
             continue
-        if isinstance(val, str) and val[0:2] == "${":
+        # NOTE value must be an exact match to avoid interfering 
+        # with other templates
+        if (
+            isinstance(val, str) and
+                (val[0:2] == "${" and val[-1] == "}")
+        ):
             ref_key = val[2:-1]
             ref_val = src_config.get(ref_key, None)
             if ref_val is not None:

src/ambianic/notification.py

@@ -0,0 +1,97 @@
+"""Utilities to send notifications."""
+
+import logging
+import apprise
+import os
+import ambianic
+
+log = logging.getLogger(__name__)
+
+UI_BASEURL = "https://ui.ambianic.ai"
+
+class Notification:
+    def __init__(self, event:str = "detection", data:dict = {}, providers:list = ["all"]):
+        self.event: str = event
+        self.providers: list = providers
+        self.title: str = None
+        self.message: str = None
+        self.attach: list = []
+        self.data: dict = data
+
+    def add_attachments(self, *args):
+        self.attach.append(*args)
+
+    def to_dict(self) -> dict:
+        return dict(vars(self))
+
+
+class NotificationHandler:
+    def __init__(self, config: dict = None):
+        if config is None:
+            config = ambianic.config
+        self.apobj = apprise.Apprise(debug=True)
+        self.config = config.get("notifications", {})
+        for name, cfg in self.config.items():
+            providers = cfg.get("providers", [])
+            for provider in providers:
+                if not self.apobj.add(provider, tag=name):
+                    log.warning(
+                        "Failed to add notification provider: %s=%s" 
+                            % (name, provider)
+                    )
+
+    def send(self, notification: Notification):
+
+        templates = self.config.get("templates", {})
+
+        title = notification.title
+        if title is None:
+            title = templates.get("title", "[Ambianic.ai] New ${event} event" )
+
+        message = notification.message
+        if message is None:
+            message = templates.get(
+                "message", "New ${event} recognized"
+            )
+
+        attachments = []
+        for a in notification.attach:
+            if not os.path.exists(a) or not os.path.isfile(a):
+                log.warning("Attachment is not a valid file %s")
+                continue
+            attachments.append(a)
+
+        template_args = {
+            "event_type": notification.event,
+            "event": notification.data.get("label", notification.event),
+            "event_details_url": "%s/%s" % (UI_BASEURL, notification.data.get("id", ""))
+        }
+        template_args = {**template_args, **notification.data}
+
+        for key, value in template_args.items():
+            k = "${%s}" % (str(key))
+            v = str(value)
+            title = title.replace(k, v)
+            message = message.replace(k, v)
+
+        for provider in notification.providers:
+            cfg = self.config.get(provider, None)
+            if cfg is None:
+                log.warning("Skip unknown provider %s" % provider)
+                continue
+
+            include_attachments = cfg.get("include_attachments", False)
+            ok = self.apobj.notify(
+                message, 
+                title=title,
+                tag=provider, 
+                attach=attachments if include_attachments else [],
+            )
+            if ok:
+                log.debug(
+                    "Sent notification for %s to %s" % 
+                    (notification.event, provider)
+                )
+            else:
+                log.warning("Error sending notification for %s to %s" %
+                          (notification.event, provider))

src/ambianic/pipeline/store.py

@@ -8,6 +8,7 @@
 
 from ambianic import DEFAULT_DATA_DIR
 from ambianic.pipeline import PipeElement
+from ambianic.notification import Notification, NotificationHandler
 
 log = logging.getLogger(__name__)
 
@@ -18,6 +19,7 @@ class SaveDetectionSamples(PipeElement):
     def __init__(self,
                  positive_interval=2,
                  idle_interval=600,
+                 notify=None,
                  **kwargs):
         """Create SaveDetectionSamples element with the provided arguments.
 
@@ -33,6 +35,7 @@ def __init__(self,
 
         """
         super().__init__(**kwargs)
+
         log.info('Loading pipe element %r ', self.__class__.__name__)
         if self.context:
             self._sys_data_dir = self.context.data_dir
@@ -65,6 +68,12 @@ def __init__(self,
         ii = idle_interval
         self._idle_interval = datetime.timedelta(seconds=ii)
         self._time_latest_saved_idle = self._time_latest_saved_detection
+
+        # setup notification handler
+        self.notification = None
+        self.notification_config = notify        
+        if self.notification_config is not None and self.notification_config.get("providers"):
+            self.notification = NotificationHandler()
 
     def _save_sample(self,
                      inf_time=None,
@@ -119,6 +128,7 @@ def _save_sample(self,
         # e = PipelineEvent('Detected Objects', type='ObjectDetection')
         self.event_log.info('Detection Event', save_json)
         log.debug("Saved sample (detection event): %r ", save_json)
+        self.notify(save_json)
         return image_path, json_path
 
     def process_sample(self, **sample) -> Iterable[dict]:
@@ -172,3 +182,18 @@ def process_sample(self, **sample) -> Iterable[dict]:
                 }
                 log.debug('Passing sample on: %r ', processed_sample)
                 yield processed_sample
+
+    def notify(self, save_json: dict):
+        if self.notification is None:
+            return
+        log.debug("Sending notification(s)..")
+        # TODO extract inference data
+        for inference_result in save_json['inference_result']:
+            data = {
+                'id': save_json['id'],
+                'label': inference_result['label'],
+                'confidence': inference_result['confidence'],
+                'datetime': save_json['datetime'],
+            }
+            notification = Notification(data=data, providers=self.notification_config["providers"])
+            self.notification.send(notification)

src/ambianic/server.py

@@ -86,7 +86,7 @@ def _stop_servers(self, servers):
     def _healthcheck(self, servers):
         """Check the health of managed servers."""
         for s in servers.values():
-            latest_heartbeat, status = s.healthcheck()
+            latest_heartbeat, _ = s.healthcheck()
             now = time.monotonic()
             lapse = now - latest_heartbeat
             if lapse > 1: