feat: [merger] 完成对事件回调status=1触发和status=2结束的合并逻辑

isc_video_record/merger.py

@@ -92,19 +92,6 @@ class EventMergerJob:
                 events = []
         return events
 
-    def look_untreated_events(self, camera, body):
-        """
-        查看当前摄像头是否存在未处理事件
-        """
-        untreated_events = mysql.get_untreated_events(
-            camera['db_table'], camera['device_code'], retry=True)
-        if untreated_events:
-            log.info('当前摄像头%s还存在未处理事件，优先处理', camera['device_code'])
-            body['ex'] = sum([(event['end_time'] - event['start_time']).total_seconds()
-                              for event in untreated_events]) + 10
-            self.send_mq_message(body)
-        return body.get('ex', 0)
-
     def get_alarm_events(self, pipe, camera, now):
         """
         摄像头录制计划为全天录制时
@@ -129,6 +116,12 @@ class EventMergerJob:
                 'start_time': last_check_time.astimezone(tz),
                 'end_time': now.astimezone(tz)
             }]
+
+        last_check_time_key = LAST_CHECK_TIME_KEY.format(camera['device_code'])
+        res = pipe.set(last_check_time_key,
+                       now.strftime('%Y-%m-%d %H:%M:%S'))
+        if not res:
+            events = []
         return events
 
     def process_camera(self, pipe, camera):
@@ -139,10 +132,18 @@ class EventMergerJob:
         }
         now = datetime.utcnow() - timedelta(minutes=3)
 
-        event_duration = self.look_untreated_events(camera, body)
+        untreated_events = mysql.get_untreated_events(
+            camera['db_table'], camera['device_code'], retry=True)
+        if untreated_events:
+            log.info('当前摄像头%s还存在未处理事件，优先处理', camera['device_code'])
+            body['ex'] = sum([(event['end_time'] - event['start_time']).total_seconds()
+                              for event in untreated_events]) + 10
+            self.send_mq_message(body)
+        event_duration = body.get('ex', 0)
+
         if event_duration == 0:
             call_get_events_func = self.get_alarm_events if camera['video_plan_type'] == 1 else  \
-                    self.get_cameras_playback_urls
+                    self.get_camera_local_events
 
             for event in call_get_events_func(pipe, camera, now):
                 insert_video_info(

isc_video_record/utils/pre_event.py

@@ -13,6 +13,8 @@ class PreEvent(object):
         influxdb.reconnect()
         self.start_time = start_time
         self.end_time = end_time
+        self.tz_start_time = dt_parse(self.start_time.replace(' ', 'T') + '.001000Z').astimezone(tz)
+        self.tz_end_time = dt_parse(self.end_time.replace(' ', 'T') + '.001000Z').astimezone(tz)
 
     def get_alarm_list(self, camera_index):
         sql = '''
@@ -34,25 +36,62 @@ class PreEvent(object):
 
         return points
 
-    @staticmethod
-    def merge_alarm_to_event(alarm_list):
+    def merge_alarm_to_event(self, alarm_list):
 
-        events = []
+        pre_events = []
+        pre_status = -1
         for alarm_point in alarm_list:
             # 报警时间转换成上海时区
             alarm_time = dt_parse(alarm_point['time']).astimezone(tz)
-            #  合并告警消息最小单位60s
-            # TODO 事件的开始和结束需要依赖于status，status=1触发，status=2结束
+
+            if alarm_point['status'] == 1:
+                if pre_status == 1:
+                    pre_events[-1]['end_time'] = alarm_time
+                else:
+                    pre_events.append({'start_time': alarm_time})
+            else:
+                if pre_status == 1:
+                    pre_events[-1]['end_time'] = alarm_time + timedelta(seconds=10)
+                elif pre_status == -1:
+                    pre_events.append({
+                        'start_time': self.tz_start_time,
+                        'end_time': alarm_time + timedelta(seconds=10)
+                    })
+                else:
+                    pre_events.append({
+                        'start_time': alarm_time - timedelta(seconds=30),
+                        'end_time': alarm_time + timedelta(seconds=10)
+                    })
+
+            pre_status = alarm_point['status']
+
+        if pre_status == 1:
+            pre_events[-1]['end_time'] = self.tz_end_time
+
+        events = []
+        for pre_event in pre_events:
+            # 合并时间间隔较短的事件
             if len(events) > 0 \
-               and alarm_time - events[-1]['end_time'] < timedelta(seconds=30) \
-               and events[-1]['end_time'] - events[-1]['start_time'] < timedelta(hours=0.5):
-                events[-1]['end_time'] = alarm_time + timedelta(seconds=10)
-                continue
-
-            events.append({
-                'start_time': alarm_time - timedelta(seconds=10),
-                'end_time': alarm_time + timedelta(seconds=10)
-            })
+               and pre_event['start_time'] - events[-1]['end_time'] <= timedelta(seconds=30) \
+               and events[-1]['end_time'] - events[-1]['start_time'] <= timedelta(hours=0.5):
+                events[-1]['end_time'] = pre_event['end_time']
+            else:
+                events.append({
+                    'start_time': pre_event['start_time'],
+                    'end_time': pre_event['end_time'],
+                })
+
+        #  合并告警消息最小单位60s
+        # if len(events) > 0 \
+        #    and alarm_time - events[-1]['end_time'] < timedelta(seconds=40) \
+        #    and events[-1]['end_time'] - events[-1]['start_time'] < timedelta(hours=0.5):
+        #     events[-1]['end_time'] = alarm_time + timedelta(seconds=10)
+        #     continue
+
+        # events.append({
+        #     'start_time': alarm_time - timedelta(seconds=10),
+        #     'end_time': alarm_time + timedelta(seconds=10)
+        # })
 
         return events
 

setup.py

@@ -24,7 +24,7 @@ requires = [
 
 setuptools.setup(
     name='isc-video-record',
-    version='1.0.0b2',
+    version='1.0.0b3',
     description='ISC motion detection playback video stream recording service.',
     long_description=long_description,
     long_description_content_type='text/markdown',

tests/test_clean_data.py

+from isc_video_record.db.mysql import query
+
+
+@query(cursor_dict=True)
+def get_camera_info(cursor, conn):
+
+    sql = '''
+    select *,concat('common_video_', mod(service_type, 2), '.',
+                   'video_data_motion_', mod(biz_type, 8)
+            ) as 'db_table' from `camera_info` where `is_valid` = 1;
+    '''
+    cursor.execute(sql)
+    return cursor.fetchall()
+
+
+@query()
+def update_status(cursor, conn, db_table, camera_code):
+    sql = '''
+        delete from {} where device_code = "{}" and `create_time` > '2021-06-15 08:40:00'
+    '''.format(db_table, camera_code)
+    cursor.execute(sql)
+    conn.commit()
+
+
+for camera in get_camera_info():
+    if camera['db_table']:
+        update_status(camera['db_table'], camera['device_code'])
+

tests/test_events.py

@@ -4,7 +4,7 @@ from dynaconf import settings
 from datetime import datetime
 
 from isc_video_record.utils.isc_client import HikVisionClient
-from isc_video_record.pre_event import PreEvent
+from isc_video_record.utils.pre_event import PreEvent
 
 tz = pytz.timezone('Asia/Shanghai')
 
@@ -13,18 +13,18 @@ config = settings.get('ISC')
 client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
                          config.get('HOST'), config.get('PORT'))
 
-camera_index = '8b86a98f6c5143219d69c9c0344b156a'
-
-start_time = datetime(2021, 5, 23, 7, 10, 0)
-end_time = datetime(2021, 5, 23, 7, 30, 0)
-res = client.get_cameras_playback_urls(camera_index,
-                                       client.iso_format(start_time.astimezone(tz)),
-                                       client.iso_format(end_time.astimezone(tz)))
-
-for event in res:
-    cur_start_time = max(event['start_time'], start_time.astimezone(tz))
-    cur_end_time = min(event['end_time'], end_time.astimezone(tz))
-    print(cur_start_time, cur_end_time)
+camera_index = 'edef21b07d7249ad8132bad6d8d1223b'
+
+start_time = datetime(2021, 6, 15, 9, 4, 48)
+end_time = datetime(2021, 6, 15, 10, 4, 48)
+# res = client.get_cameras_playback_urls(camera_index,
+#                                        client.iso_format(start_time.astimezone(tz)),
+#                                        client.iso_format(end_time.astimezone(tz)))
+#
+# for event in res:
+#     cur_start_time = max(event['start_time'], start_time.astimezone(tz))
+#     cur_end_time = min(event['end_time'], end_time.astimezone(tz))
+#     print(cur_start_time, cur_end_time)
 
 print('-------')
 pre_events = PreEvent(start_time.strftime('%Y-%m-%d %H:%M:%S'),