feat: 合并和录制

docker-compose.yaml

@@ -7,6 +7,8 @@ services:
             context: .
         ports:
             - 5001:5001
+        volumes:
+            - /data/videos/isc-record:/data/videos/isc-record
     influxdb:
         container_name: influxdb
         image: ilabservice-registry.cn-hangzhou.cr.aliyuncs.com/basic/influxdb:monitor

hikvision_isc_client/app.py

@@ -4,7 +4,7 @@ from flask import Flask, jsonify, request
 
 
 from intelab_python_sdk.logger import log, log_init
-from db import influxdb
+from hikvision_isc_client.db import influxdb
 
 
 log_init('api-server', True, '/var/log/event_rcv')
@@ -14,7 +14,7 @@ app = Flask('eventRcv')
 @app.route('/event/rcv', methods=['POST'])
 def event_rcv():
     body = request.get_json()
-    log.info(body)
+    log.info('POST body: %s', body)
     influxdb.reconnect()
     if body and body.get('method') == 'OnEventNotify':
         events = []

hikvision_isc_client/client.py

@@ -21,8 +21,8 @@ class HikVisionClient(object):
         self.url = '{}://{}:{}'.format('https' if https else 'http', self.host, self.port)
 
     @staticmethod
-    def iso_format(dtime):
-        return '{}+08:00'.format(dtime.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3])
+    def iso_format(t):
+        return '{}+08:00'.format(t.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3])
 
     def _sign(self, uri, body, method='POST'):
         content_md5 = hashlib.md5()

hikvision_isc_client/db.py

 from influxdb import InfluxDBClient
-from dynaconf import settings
+import dynaconf
 
 
 class InfluxDB(object):
@@ -7,12 +7,11 @@ class InfluxDB(object):
     def __init__(self):
 
         self._influxdb = None
+        self.config = dynaconf.settings.get('INFLUXDB')
+
         self.init_app()
 
     def init_app(self):
-
-        self.config = settings.get('INFLUXDB')
-
         config = {
             'host': self.config.get('HOST', 'localhost'),
             'port': self.config.get('PORT', 8086),
@@ -27,15 +26,15 @@ class InfluxDB(object):
     def reconnect(self):
         self.init_app()
 
-    def query(self, query_string):
-        return self._influxdb.query(query_string)
+    def query(self, query_string, *args, **kw):
+        return self._influxdb.query(query_string, *args, **kw)
 
-    def write_points(self, points, retention_policy=None):
+    def write_points(self, points, *args, **kw):
         if not points:
             return
 
         return self._influxdb.write_points(
-            points, retention_policy=retention_policy)
+            points, *args, **kw)
 
     def create_database(self, database=None):
         if not database:

hikvision_isc_client/pre_event.py

+import pytz
+from datetime import timedelta
+from dateutil.parser import parse as dt_parse
+
+from hikvision_isc_client.db import influxdb
+
+tz = pytz.timezone('Asia/Shanghai')
+
+
+class PreEvent(object):
+
+    def __init__(self, start_time, end_time):
+        influxdb.reconnect()
+        self.start_time = start_time
+        self.end_time = end_time
+
+    def get_alarm_list(self, camera_index):
+        sql = '''
+            select *
+            from "one_week"."event_vss"
+            where
+                time > $start_time
+                and time < $end_time
+                and camera_index = $camera_index
+        '''
+        res = influxdb.query(sql, bind_params={
+            'start_time': self.start_time,
+            'end_time': self.end_time,
+            'camera_index': camera_index
+        })
+        points = []
+        if res and not res.empty:
+            points = res['event_vss']
+
+        return points
+
+    @staticmethod
+    def merge_alarm_to_event(alarm_list):
+
+        events = []
+        for alarm_point in alarm_list:
+            # 报警时间转换成上海时区
+            alarm_time = dt_parse(alarm_point['time']).astimezone(tz)
+            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=20)
+            })
+
+        return events
+
+
+if __name__ == '__main__':
+
+    pre_event = PreEvent('2021-04-26 00:00:00', '2021-04-26 16:00:00')
+    alarm_points = pre_event.get_alarm_list('f8a3c4d9b8ae42118b4db9fcf7895031')
+    print(pre_event.merge_alarm_to_event(alarm_points))

hikvision_isc_client/recorder.py

+import os
+import shutil
+import pytz
+import dynaconf
+
+from datetime import timedelta
+from intelab_python_sdk.logger import log_init, log
+from intelab_python_sdk.ffmpeg.ffmpeg_concat import concat
+
+from hikvision_isc_client.client import HikVisionClient
+from hikvision_isc_client.utils.record_utils import record_thread, get_video_duration, time_to_seconds
+
+tz = pytz.timezone('Asia/Shanghai')
+log_init(__name__, True, './log')
+video_path = '/data/videos/isc-record'
+
+os.makedirs(video_path, exist_ok=True)
+
+config = dynaconf.settings.get('ISC')
+
+client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
+                         config.get('HOST'), config.get('PORT'))
+
+
+def recorder(camera_index, start_time, end_time):
+    playback_urls = client.get_cameras_playback_urls(
+        camera_index,
+        HikVisionClient.iso_format(start_time),
+        HikVisionClient.iso_format(end_time)
+    )
+    if len(playback_urls) > 0:
+        # 可以只通过一个回放流地址取到其他时间段的流
+        playback_stream = playback_urls[0]
+    else:
+        return
+    log.info(playback_stream)
+    part_num = 1
+    part_files_set = set()
+    file_name = os.path.join(video_path, 'rtmp_{}_{}.mp4'.format(
+        start_time.astimezone(pytz.utc).strftime('%Y%m%dT%H%M%S'),
+        end_time.astimezone(pytz.utc).strftime('%Y%m%dT%H%M%S')
+    ))
+
+    while True:
+        complete_duration = (end_time - start_time).total_seconds()
+
+        file_info, _ = stream_record(playback_stream['stream_url'], start_time, end_time)
+        file_duration = time_to_seconds(file_info['duration'])
+        if not os.path.isfile(file_info['file_name']):
+            continue
+
+        if file_duration < complete_duration - 2:
+            # 视频文件时长小于完整时长
+            new_start_time = start_time + timedelta(seconds=file_duration)
+            part_file_name = 'rtmp_{}_{}_{}.mp4'.format(
+                start_time.strftime('%Y%m%dT%H%M%S'),
+                new_start_time.strftime('%Y%m%dT%H%M%S'), part_num
+            )
+            shutil.move(file_info['file_name'], part_file_name)
+            part_files_set.add(part_file_name)
+
+            start_time = new_start_time
+            part_num += 1
+        else:
+            part_files_set.add(file_info['file_name'])
+            break
+    part_files = sorted(list(part_files_set))
+    if len(part_files) > 1:
+        concat(part_files, file_name, removed=True)
+    elif len(part_files) == 1:
+        shutil.move(part_files[0], file_name)
+    log.info('The download is complete, file %s', file_name)
+
+
+def stream_record(stream, start_time, end_time):
+    start_time = start_time.strftime('%Y%m%dT%H%M%S')
+    end_time = end_time.strftime('%Y%m%dT%H%M%S')
+
+    if stream['protocol'] == 'rtsp':
+        stream_url = stream['url']
+    else:
+        stream_url = '{}?beginTime={}&endTime={}&{}'.format(stream['url'],
+                                                            start_time,
+                                                            end_time, stream['extra_args'])
+    file_name = 'rtmp_{}_{}.mp4'.format(start_time, end_time)
+    record_thread(stream_url, file_name)
+    return get_video_duration(file_name)

tests/test_playback.py

 import os
-import shutil
 import pytz
-from datetime import datetime, timedelta
-from intelab_python_sdk.logger import log_init, log
-from intelab_python_sdk.ffmpeg.ffmpeg_concat import concat
+from datetime import datetime
+from intelab_python_sdk.logger import log_init
 from dynaconf import settings
 
 from hikvision_isc_client.client import HikVisionClient
-from hikvision_isc_client.record import record_thread, get_video_duration, time_to_seconds
+from hikvision_isc_client.recorder import recorder
 
 
 tz = pytz.timezone('Asia/Shanghai')
@@ -24,70 +22,7 @@ client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
 def main():
     start_time = datetime(2021, 4, 25, 15, 19)
     end_time = datetime(2021, 4, 25, 15, 22, 20)
-    playback_urls = client.get_cameras_playback_urls(
-        'f8a3c4d9b8ae42118b4db9fcf7895031',
-        HikVisionClient.iso_format(start_time),
-        HikVisionClient.iso_format(end_time),
-    )
-    if len(playback_urls) > 0:
-        # 可以只通过一个回放流地址取到其他时间段的流
-        playback_stream = playback_urls[0]
-    else:
-        return
-
-    log.info(playback_stream)
-    part_num = 1
-    part_files_set = set()
-    file_name = 'rtmp_{}_{}.mp4'.format(
-        start_time.strftime('%Y%m%dT%H%M%S'),
-        end_time.strftime('%Y%m%dT%H%M%S')
-    )
-
-    # 断点续录
-    while True:
-        complete_duration = (end_time - start_time).total_seconds()
-
-        file_info, _ = stream_record(playback_stream['stream_url'], start_time, end_time)
-        file_duration = time_to_seconds(file_info['duration'])
-        print(complete_duration, file_duration, file_info)
-        if not os.path.isfile(file_info['file_name']):
-            continue
-
-        if file_duration < complete_duration - 1:
-            # 视频文件时长小于完整时长
-            new_start_time = start_time + timedelta(seconds=file_duration)
-            part_file_name = 'rtmp_{}_{}_{}.mp4'.format(start_time.strftime('%Y%m%dT%H%M%S'),
-                                                        new_start_time.strftime('%Y%m%dT%H%M%S'),
-                                                        part_num)
-            shutil.move(file_info['file_name'], part_file_name)
-            part_files_set.add(part_file_name)
-
-            start_time = new_start_time
-            part_num += 1
-        else:
-            part_files_set.add(file_info['file_name'])
-            break
-    part_files = sorted(list(part_files_set))
-    if len(part_files) > 1:
-        concat(part_files, file_name, removed=True)
-    elif len(part_files) == 1:
-        shutil.move(part_files[0], file_name)
-    log.info('视频文件%s录制完成！', file_name)
-
-
-def stream_record(stream, start_time, end_time):
-    start_time = start_time.strftime('%Y%m%dT%H%M%S')
-    end_time = end_time.strftime('%Y%m%dT%H%M%S')
-
-    if stream['protocol'] == 'rtsp':
-        stream_url = stream['url']
-    else:
-        stream_url = '{}?beginTime={}&endTime={}&{}'.format(stream['url'],
-                                                            start_time,
-                                                            end_time, stream['extra_args'])
-    file_name = 'rtmp_{}_{}.mp4'.format(start_time, end_time)
-    record_thread(stream_url, file_name)
-    return get_video_duration(file_name)
+    recorder('f8a3c4d9b8ae42118b4db9fcf7895031', start_time, end_time)
 
 
 if __name__ == '__main__':