Merge pull request #70 from xmartlabs/Occupancy-SendNotifications

Occupancy Notifications

Author: renzodgc

config-coral.ini

@@ -9,6 +9,8 @@ DashboardURL = http://0.0.0.0:8000
 ScreenshotsDirectory = /repo/data/processor/static/screenshots
 EnableSlackNotifications = no
 SlackChannel = lanthorn-notifications
+; OccupancyAlertsMinInterval time is measured in seconds (if interval < 0 then no occupancy alerts are triggered)
+OccupancyAlertsMinInterval = 180
 
 [API]
 Host = 0.0.0.0

config-jetson.ini

@@ -12,6 +12,8 @@ DashboardURL = http://0.0.0.0:8000
 ScreenshotsDirectory = /repo/data/processor/static/screenshots
 EnableSlackNotifications = no
 SlackChannel = lanthorn-notifications
+; OccupancyAlertsMinInterval time is measured in seconds (if interval < 0 then no occupancy alerts are triggered)
+OccupancyAlertsMinInterval = 180
 
 [API]
 Host = 0.0.0.0

config-x86-openvino.ini

@@ -23,6 +23,8 @@ DashboardURL = http://0.0.0.0:8000
 ScreenshotsDirectory = /repo/data/processor/static/screenshots
 EnableSlackNotifications = no
 SlackChannel = lanthorn-notifications
+; OccupancyAlertsMinInterval time is measured in seconds (if interval < 0 then no occupancy alerts are triggered)
+OccupancyAlertsMinInterval = 180
 
 [Area_0]
 Id = area0

config-x86.ini

@@ -23,6 +23,8 @@ DashboardURL = http://0.0.0.0:8000
 ScreenshotsDirectory = /repo/data/processor/static/screenshots
 EnableSlackNotifications = no
 SlackChannel = lanthorn-notifications
+; OccupancyAlertsMinInterval time is measured in seconds (if interval < 0 then no occupancy alerts are triggered)
+OccupancyAlertsMinInterval = 180
 
 [Area_0]
 Id = area0

libs/area_reporting.py

@@ -0,0 +1,76 @@
+import os
+import time
+import logging
+import csv
+from datetime import date, datetime
+from collections import deque
+from .utils.mailing import MailService
+from .notifications.slack_notifications import SlackService
+
+logger = logging.getLogger(__name__)
+
+
+class AreaReporting:
+
+    def __init__(self, config, area):
+        self.processing_alerts = False
+        self.config = config
+        self.area = area
+
+        self.occupancy_sleep_time_interval = float(self.config.get_section_dict("App")["OccupancyAlertsMinInterval"])
+        self.log_dir = self.config.get_section_dict("Logger")["LogDirectory"]
+        self.idle_time = float(self.config.get_section_dict('Logger')['TimeInterval'])
+        self.area_id = self.area['id']
+        self.area_name = self.area['name']
+        self.occupancy_threshold = self.area['occupancy_threshold']
+        self.should_send_email_notifications = self.area['should_send_email_notifications']
+        self.should_send_slack_notifications = self.area['should_send_slack_notifications']
+        self.cameras = [camera for camera in self.config.get_video_sources() if camera['id'] in self.area['cameras']]
+        for camera in self.cameras:
+            camera['file_path'] = os.path.join(self.log_dir, camera['id'], "objects_log")
+            camera['last_processed_time'] = time.time()
+
+        self.mail_service = MailService(config)
+        self.slack_service = SlackService(config)
+
+    def process_area(self):
+        # Sleep for a while so cameras start processing
+        time.sleep(30)
+
+        self.processing_alerts = True
+        logger.info(f'Enabled processing alerts for - {self.area_id}: {self.area_name} with {len(self.cameras)} cameras')
+        while self.processing_alerts:
+            camera_file_paths = [os.path.join(camera['file_path'], str(date.today()) + ".csv") for camera in self.cameras]
+            if not all(list(map(os.path.isfile, camera_file_paths))):
+                # Wait before csv for this day are created
+                logger.info(f'Area reporting on - {self.area_id}: {self.area_name} is waiting for reports to be created')
+                time.sleep(5)
+
+            occupancy = 0
+            for camera in self.cameras:
+                with open(os.path.join(camera['file_path'], str(date.today()) + ".csv"), 'r') as log:
+                    last_log = deque(csv.DictReader(log), 1)[0]
+                    log_time = datetime.strptime(last_log['Timestamp'], "%Y-%m-%d %H:%M:%S")
+                    # TODO: If the TimeInterval of the Logger is more than 30 seconds this would have to be revised.
+                    if (datetime.now() - log_time).total_seconds() < 30:
+                        occupancy += int(last_log['DetectedObjects'])
+                    else:
+                        logger.warn(f"Logs aren't being updated for camera {camera['id']} - {camera['name']}")
+
+            if occupancy > self.occupancy_threshold:
+                # Trigger alerts
+                if self.should_send_email_notifications:
+                    self.mail_service.send_occupancy_notification(self.area, occupancy)
+                if self.should_send_slack_notifications:
+                    self.slack_service.occupancy_alert(self.area, occupancy)
+                # Sleep until the cooldown of the alert
+                time.sleep(self.occupancy_sleep_time_interval)
+            else:
+                # Sleep until new data is logged
+                time.sleep(self.idle_time)
+
+        self.stop_process_area()
+
+    def stop_process_area(self):
+        logger.info(f'Disabled processing alerts for area - {self.area_id}: {self.area_name}')
+        self.processing_alerts = False

libs/area_threading.py

@@ -0,0 +1,40 @@
+import os
+from threading import Thread
+from libs.area_reporting import AreaReporting as AreaEngine
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def run_area_processing(config, pipe, areas):
+    pid = os.getpid()
+    logger.info(f"[{pid}] taking on notifications for {len(areas)} areas")
+    threads = []
+    for area in areas:
+        engine = AreaThread(config, area)
+        engine.start()
+        threads.append(engine)
+
+    # Wait for a signal to die
+    pipe.recv()
+    logger.info(f"[{pid}] will stop area alerts and die")
+    for t in threads:
+        t.stop()
+
+    logger.info(f"[{pid}] Goodbye!")
+
+
+class AreaThread(Thread):
+    def __init__(self, config, area):
+        Thread.__init__(self)
+        self.engine = None
+        self.config = config
+        self.area = area
+
+    def run(self):
+        self.engine = AreaEngine(self.config, self.area)
+        self.engine.process_area()
+
+    def stop(self):
+        self.engine.stop_process_area()
+        self.join()

libs/config_engine.py

@@ -178,7 +178,7 @@ def get_areas(self):
                     if 'Cameras' in section and section['Cameras'].strip() != "":
                         area['cameras'] = section['Cameras'].split(',')
 
-                    if area['notify_every_minutes'] > 0 and (area['violation_threshold'] > 0 or area['occupancy_threshold'] > 0):
+                    if (area['notify_every_minutes'] > 0 and area['violation_threshold'] > 0) or area['occupancy_threshold'] > 0:
                         area['should_send_email_notifications'] = 'emails' in area
                         area['should_send_slack_notifications'] = bool(self.config['App']['SlackChannel'] and
                                                                        self.config.getboolean('App', 'EnableSlackNotifications'))

libs/distancing.py

@@ -60,8 +60,8 @@ def __init__(self, config, source, live_feed_enabled=True):
                 logger.info(f"Falling back using {self.default_dist_method}")
                 self.dist_method = self.default_dist_method
 
-        self.screenshot_period = float(
-            self.config.get_section_dict("App")["ScreenshotPeriod"]) * 60  # config.ini uses minutes as unit
+        # config.ini uses minutes as unit
+        self.screenshot_period = float(self.config.get_section_dict("App")["ScreenshotPeriod"]) * 60
         self.bucket_screenshots = config.get_section_dict("App")["ScreenshotS3Bucket"]
         self.uploader = S3Uploader(self.config)
         self.screenshot_path = os.path.join(self.config.get_section_dict("App")["ScreenshotsDirectory"], self.camera_id)

libs/notifications/slack_notifications.py

@@ -49,7 +49,14 @@ def violation_report(self, entity_info, number):
         msg = f"We found {number} violations in {entity_id}: {entity_name} ({entity_type})"
         self.post_message_to_channel(msg, self.channel)
 
-    def daily_report(self, entity_info, number,):
+    def daily_report(self, entity_info, number):
         entity_id, entity_type, entity_name = entity_info['id'], entity_info['type'], entity_info['name']
         msg = f"Yesterday we found {number} violations in {entity_id}: {entity_name} ({entity_type})."
         self.post_message_to_channel(msg, self.channel)
+
+    def occupancy_alert(self, entity_info, number):
+        entity_id, entity_type = entity_info['id'], entity_info['type']
+        entity_name, entity_threshold = entity_info['name'], entity_info['occupancy_threshold']
+        msg = f"Occupancy threshold was exceeded in {entity_type} {entity_id}: {entity_name}." \
+              f"We found {number} people out of a capacity of {entity_threshold}."
+        self.post_message_to_channel(msg, self.channel)

libs/processor_core.py

@@ -6,6 +6,7 @@
 from queue import Empty
 import schedule
 from libs.engine_threading import run_video_processing
+from libs.area_threading import run_area_processing
 from libs.utils.notifications import run_check_violations
 
 logger = logging.getLogger(__name__)
@@ -63,11 +64,12 @@ def _setup_scheduled_tasks(self):
             should_send_slack_notifications = area['should_send_slack_notifications']
             if should_send_email_notifications or should_send_slack_notifications:
                 interval = area['notify_every_minutes']
-                threshold = area['violation_threshold']
-                schedule.every(interval).minutes.do(
-                    run_check_violations, threshold, self.config, area, interval,
-                    should_send_email_notifications, should_send_slack_notifications
-                ).tag("notification-task")
+                violation_threshold = area['violation_threshold']
+                if violation_threshold > 0:
+                    schedule.every(interval).minutes.do(
+                        run_check_violations, violation_threshold, self.config, area, interval,
+                        should_send_email_notifications, should_send_slack_notifications
+                    ).tag("notification-task")
             else:
                 logger.info(f"should not send notification for camera {area['id']}")
 
@@ -131,6 +133,22 @@ def _start_processing(self):
             p = mp.Process(target=run_video_processing, args=(self.config, recv_conn, p_src))
             p.start()
             engines.append((send_conn, p))
+
+        # Set up occupancy alerts
+        areas_to_notify = [
+            area for area in self.config.get_areas() if area['occupancy_threshold'] > 0 and area['cameras'] and (
+                area['should_send_email_notifications'] or area['should_send_slack_notifications']
+            ) 
+        ]
+        if areas_to_notify and float(self.config.get_section_dict('App')['OccupancyAlertsMinInterval']) >= 0:
+            logger.info(f'Spinning up area alert threads for {len(areas_to_notify)} areas')
+            recv_conn, send_conn = mp.Pipe(False)
+            p = mp.Process(target=run_area_processing, args=(self.config, recv_conn, areas_to_notify))
+            p.start()
+            engines.append((send_conn, p))
+        else:
+            logger.info('Area occupancy alerts are disabled for all areas')
+
         self._engines = engines
 
     def _stop_processing(self):