Skip to content

I
ils-common-video
提交 20f8353c authored 作者: zw.wang's avatar zw.wang
浏览文件
操作

feat: 修改项目名称和文件结构

上级 502889c5
隐藏空白字符变更
内嵌 并排
正在显示 包含 395 行增加 和 168 行删除
Makefile 0 → 100644
浏览文件 @ 20f8353c
.PHONY: init flake8 clean build
init:
@pip install -U pip flake8
@pip install -e .
flake8:
@echo "------------------"
@flake8 isc_video_record
clean:
@echo 'removing...'
@find . -name 'dist' -print -exec rm -rf {} +
@find . -name '.eggs' -print -exec rm -rf {} +
@find . -name 'build' -print -exec rm -rf {} +
@find . -name '*.pyc' -print -exec rm -f {} +
@find . -name '*.pyo' -print -exec rm -f {} +
@find . -name '*.log' -print -exec rm -f {} +
@find . -name '__pycache__' -print -exec rm -rf {} +
@echo 'Done [clean]'
image_name := ilabservice-registry.cn-hangzhou.cr.aliyuncs.com/basic/isc_video_record/${NAME}
image_version = $(shell cat setup.py| grep -Po "version='\K(\d+\.\d+\.\d+-?\w*)")
docker_name = $(image_name):$(image_version)
build: clean
@sudo docker build --no-cache --build-arg WORKNAME=${NAME} -t $(docker_name) .
@sudo docker push $(docker_name)
@sudo docker rmi $(docker_name)
@echo 'Build successful! [$(docker_name)]'
README.md
浏览文件 @ 20f8353c
# HIKVISION iSC Python SDK
# ISC移动侦测回放视频流录制服务
This is the Python SDK to interact with Hikvision iSC platform using Hikvision's open API.
基于海康ISC平台回调消息合并成移动侦测事件，根据该事件再通过ISC平台获取摄像头本地sd卡上该时间段的回放视频流进行录制。
## CONFIG
```
# API平台配置
DYNACONF_COMMON__HOST: 'http://'
# 海康ISC平台配置
DYNACONF_ISC__HOST: ''
DYNACONF_ISC__PORT: ''
DYNACONF_ISC__KEY: ''
DYNACONF_ISC__SECRET: ''
DYNACONF_INFLUXDB__HOST: '127.0.0.1'
DYNACONF_INFLUXDB__PORT: 8086
DYNACONF_INFLUXDB__USERNAME: 'admin'
DYNACONF_INFLUXDB__PASSWORD: ''
DYNACONF_INFLUXDB__DATABASE: 'intelab'
# RABBITMQ配置
DYNACONF_RABBITMQ__HOST: 'localhost'
DYNACONF_RABBITMQ__PORT: 5672
DYNACONF_RABBITMQ__USER: 'guest'
DYNACONF_RABBITMQ__PASSWORD: 'guest'
# Redis配置
DYNACONF_REDIS__host: '127.0.0.1'
DYNACONF_REDIS__port: 6379
DYNACONF_REDIS__password: ''
DYNACONF_REDIS__db: 0
DYNACONF_REDIS__decode_responses: true
# 阿里云OSS配置
DYNACONF_ALI_OSS__access_key_id: ''
DYNACONF_ALI_OSS__access_key_secret: ''
DYNACONF_ALI_OSS__bucket_name: 'prod-jiandu'
DYNACONF_ALI_OSS__endpoint: 'https://oss-cn-shanghai.aliyuncs.com'
DYNACONF_ALI_OSS__region: 'shanghai'
# mysql配置
DYNACONF_MYSQL__HOST: ''
DYNACONF_MYSQL__PORT: ''
DYNACONF_MYSQL__USER: ''
DYNACONF_MYSQL__PASSWORD: ''
DYNACONF_MYSQL__DATABASE: 'common_camera'
# 系统异常钉钉机器人
DYNACONF_DINGTALK__WEBHOOK = https://oapi.dingtalk.com/robot/send?access_token=6c62b78c93dd935386260853591682612de1d53f106c1119b464f62fe6d15bb7
DYNACONF_DINGTALK__SECRET = SECd5986591b53e959d5076b8d53be127b0046eddf1d76b3836b52b14d1ccb170a0
```
## 依赖
```
'requests',
'python-dateutil',
'pytz',
'intelab-python-sdk',
'flask',
'dynaconf',
'influxdb',
'python-dateutil',
'pika==1.1.0',
'redis',
'mysql-connector',
'retrying',
'oss2'
```
## 本地开发
配置本地配置文件`settings.local.toml`
```
pip install -e .
```
# 设计文档
[基于ISC平台的视频服务平台](https://intelab.yuque.com/productsteam/projects/evwut6)
- EventMerger
告警消息合并成事件模块，任务每15分钟启动一次，合并最近的15分钟内influxdb中的告警消息为移动侦测事件，同时将移动事件的信息发生给rabbitmq。
- StreamRecorder
移动侦测事件取流录制模块，该模块支持多实例部署，用于消费上游EventMerger模块生产的移动侦测事件消息，根据事件的startTime和endTime向APIServer服务查询回放流地址，并进行录制，支持断点录制和视频合并。录制完成的视频将会上传到阿里云OSS，同时将链接写入Mysql。
## 部署
### 打包并上传镜像
```
make build
```
打包完成后有两个服务镜像：
- intelab
事件合并镜像-EventMerger
- intelab
移动侦测事件取流录制模块-StreamRecorder
### 日志
日志统一打印在`/var/log/event_vss/`, 部署时，可以将该目录持久化
# Release History
## 1.0.0(2021-05-10)
- 合并相近告警消息为移动侦测事件
- 获取指定时间的回放流地址并录制为MP4文件，支持断点续录和视频合并
- 视频文件上传云端并写入Mysql存储基本信息
- 录制服务支持多实例
docker-compose.yaml
浏览文件 @ 20f8353c
......@@ -31,20 +31,20 @@ services:
ports:
- 5001:5001
volumes:
- /var/log/event_rcv:/var/log/event_rcv
command: python hikvision_isc_client/app.py
- /var/log/event_vss:/var/log/event_vss
command: python isc_video_record/app.py
isc-recorder:
container_name: isc-recorder
build:
context: .
volumes:
- /var/log/event_rcv:/var/log/event_rcv
command: python hikvision_isc_client/recorder.py
- /var/log/event_vss:/var/log/event_vss
command: isc_video_record recorder
isc-merger:
container_name: isc-merger
build:
context: .
volumes:
- /data/videos/isc-record:/data/videos/isc-record
- /var/log/event_rcv:/var/log/event_rcv
command: python hikvision_isc_client/merger.py
- /var/log/event_vss:/var/log/event_vss
command: isc_video_record merger
hikvision_isc_client/db/__init__.py deleted 100644 → 0
浏览文件 @ 502889c5
from .influxdb import influxdb
from .redis import redis_connect
from .rabbitmq import rabbitmq_connect
hikvision_isc_client/merger.py deleted 100644 → 0
浏览文件 @ 502889c5
import pytz
import json
import time
import dynaconf
from datetime import datetime, timedelta
from intelab_python_sdk.logger import log
from hikvision_isc_client.db import rabbitmq_connect, redis_connect
from hikvision_isc_client.const import LAST_CHECK_TIME_KEY
from hikvision_isc_client.pre_event import PreEvent
from hikvision_isc_client.db.mysql import get_camera_info
tz = pytz.timezone('Asia/Shanghai')
class EventMergerJob:
def __init__(self):
self.queue_name = 'ISC_RECORD_JOB'
self.local_service_name = 'cloud-record'
@staticmethod
def clean():
pipe = redis_connect()
for key in pipe.keys(LAST_CHECK_TIME_KEY.format('*')):
pipe.delete(key)
pipe.close()
def start(self):
while True:
try:
self.run()
except Exception as e:
log.exception(e)
time.sleep(15 * 60)
def run(self):
pipe = redis_connect()
for camera in get_camera_info():
now = datetime.utcnow() - timedelta(minutes=3)
camera_code = camera['device_code']
last_check_time_key = LAST_CHECK_TIME_KEY.format(camera_code)
last_check_time = pipe.get(last_check_time_key)
if not last_check_time:
# 设备无上次事件,取最近的15分钟作为开始时间
last_check_time = now - timedelta(minutes=15)
else:
last_check_time = datetime.strptime(last_check_time, '%Y-%m-%d %H:%M:%S')
# 调整最大事件长度为1天
if now - last_check_time > timedelta(days=1):
last_check_time = now - timedelta(days=1)
res = pipe.set(last_check_time_key,
now.strftime('%Y-%m-%d %H:%M:%S'))
if not res:
continue
pre_event = PreEvent(
last_check_time.strftime('%Y-%m-%d %H:%M:%S'),
now.strftime('%Y-%m-%d %H:%M:%S')
)
alarm_list = list(pre_event.get_alarm_list(camera['point_index_code']))
log.info('time: {}-{}, alarm_list: {}'.format(last_check_time, now, len(alarm_list)))
connection = None
# if len(alarm_list) > 0:
connection = rabbitmq_connect()
channel = connection.channel()
channel.queue_declare(self.queue_name, durable=True)
if dynaconf.settings.get('EVENT_ON', True):
events = PreEvent.merge_alarm_to_event(alarm_list)
else:
events = [{
'start_time': last_check_time.astimezone(tz), 'end_time': now.astimezone(tz)
}]
for event in events:
body = {
'camera_index': camera['point_index_code'],
'start_time': event['start_time'].astimezone(pytz.utc).strftime('%Y-%m-%dT%H:%M:%S'),
'end_time': event['end_time'].astimezone(pytz.utc).strftime('%Y-%m-%dT%H:%M:%S'),
'event_type': camera['event_type'],
'camera_code': camera['device_code'],
'db_table': camera['db_table'],
'biz_type': camera['biz_type'],
'service_type': camera['service_type']
}
log.info(body)
channel.basic_publish(exchange='', routing_key=self.queue_name,
body=json.dumps(body, ensure_ascii=False))
if connection:
connection.close()
log.info('job end.')
pipe.close()
if __name__ == '__main__':
from intelab_python_sdk.logger import log_init
log_init('event_merger', False, '/var/log/event_rcv/')
em = EventMergerJob()
em.start()
hikvision_isc_client/app.py → isc_video_record/app.py
浏览文件 @ 20f8353c
......@@ -4,10 +4,10 @@ from flask import Flask, jsonify, request
from intelab_python_sdk.logger import log, log_init
from hikvision_isc_client.db import influxdb
from isc_video_record.db import influxdb
log_init('api-server', True, '/var/log/event_rcv')
log_init('api-server', True, '/var/log/event_vss')
app = Flask('eventRcv')
......
isc_video_record/db/__init__.py 0 → 100644
浏览文件 @ 20f8353c
from .influxdb import influxdb # noqa
from .redis import redis_connect # noqa
from .rabbitmq import rabbitmq_connect # noqa
isc_video_record/main.py 0 → 100644
浏览文件 @ 20f8353c
import argparse
from intelab_python_sdk.logger import log_init
def get_parser():
parsers = argparse.ArgumentParser(
description='ISC motion detection playback video stream recording service.'
)
parsers.add_argument('worker', choices=['merger', 'recorder'], type=str)
parsers.add_argument('-d', '--debug', required=False, help='Enable debug output',
dest='debug', action='store_true', default=False)
parsers.add_argument('-l', '--log-path', required=False, help='log file path',
dest='log_path', default='/var/log/event_vss', type=str)
return parsers
def command_line_runner():
parser = get_parser()
args = vars(parser.parse_args())
log_init(args['worker'], debug=args['debug'], log_path=args['log_path'], backupCount=3)
if args['worker'] == 'merger':
from isc_video_record.merger import EventMergerJob
em = EventMergerJob()
em.start()
elif args['worker'] == 'recorder':
from isc_video_record.recorder import StreamRecorder
sr = StreamRecorder()
sr.start()
else:
parser.print_help
if __name__ == '__main__':
command_line_runner()
isc_video_record/merger.py 0 → 100644
浏览文件 @ 20f8353c
import pytz
import json
import time
import dynaconf
from datetime import datetime, timedelta
from intelab_python_sdk.logger import log
from isc_video_record.db import rabbitmq_connect, redis_connect
from isc_video_record.const import LAST_CHECK_TIME_KEY
from isc_video_record.pre_event import PreEvent
from isc_video_record.db.mysql import get_camera_info
tz = pytz.timezone('Asia/Shanghai')
class EventMergerJob:
def __init__(self):
self.queue_name = 'ISC_RECORD_JOB'
self.local_service_name = 'cloud-record'
@staticmethod
def clean():
pipe = redis_connect()
for key in pipe.keys(LAST_CHECK_TIME_KEY.format('*')):
pipe.delete(key)
pipe.close()
def start(self):
while True:
try:
self.run()
except Exception as e:
log.exception(e)
next_run_at = time.time() + 15 * 16
log.info('next run at: %s', time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(next_run_at)))
time.sleep(15 * 60)
def run(self):
pipe = redis_connect()
for camera in get_camera_info():
try:
now = datetime.utcnow() - timedelta(minutes=3)
camera_code = camera['device_code']
last_check_time_key = LAST_CHECK_TIME_KEY.format(camera_code)
last_check_time = pipe.get(last_check_time_key)
if not last_check_time:
# 设备无上次事件,取最近的15分钟作为开始时间
last_check_time = now - timedelta(minutes=15)
else:
last_check_time = datetime.strptime(last_check_time, '%Y-%m-%d %H:%M:%S')
# 调整最大事件长度为1天
if now - last_check_time > timedelta(days=1):
last_check_time = now - timedelta(days=1)
res = pipe.set(last_check_time_key,
now.strftime('%Y-%m-%d %H:%M:%S'))
if not res:
continue
pre_event = PreEvent(
last_check_time.strftime('%Y-%m-%d %H:%M:%S'),
now.strftime('%Y-%m-%d %H:%M:%S')
)
alarm_list = list(pre_event.get_alarm_list(camera['point_index_code']))
log.info('camera_code: {}, time: {}-{}, alarm_list: {}'.format(
camera_code, last_check_time, now, len(alarm_list)))
connection = None
if len(alarm_list) > 0:
connection = rabbitmq_connect()
channel = connection.channel()
channel.queue_declare(self.queue_name, durable=True)
if dynaconf.settings.get('EVENT_ON', True):
events = PreEvent.merge_alarm_to_event(alarm_list)
else:
events = [{
'start_time': last_check_time.astimezone(tz), 'end_time': now.astimezone(tz)
}]
for event in events:
body = {
'camera_index': camera['point_index_code'],
'start_time': event['start_time'].astimezone(pytz.utc).strftime('%Y-%m-%dT%H:%M:%S'),
'end_time': event['end_time'].astimezone(pytz.utc).strftime('%Y-%m-%dT%H:%M:%S'),
'event_type': camera['event_type'],
'camera_code': camera['device_code'],
'db_table': camera['db_table'],
'biz_type': camera['biz_type'],
'service_type': camera['service_type']
}
log.info(body)
channel.basic_publish(exchange='', routing_key=self.queue_name,
body=json.dumps(body, ensure_ascii=False))
if connection:
connection.close()
except Exception as e:
log.exception(e)
pipe.close()
if __name__ == '__main__':
from intelab_python_sdk.logger import log_init
log_init('event_merger', False, '/var/log/event_vss/')
em = EventMergerJob()
em.start()
hikvision_isc_client/pre_event.py → isc_video_record/pre_event.py
浏览文件 @ 20f8353c
......@@ -2,7 +2,7 @@ import pytz
from datetime import timedelta
from dateutil.parser import parse as dt_parse
from hikvision_isc_client.db import influxdb
from isc_video_record.db import influxdb
tz = pytz.timezone('Asia/Shanghai')
......@@ -57,7 +57,7 @@ class PreEvent(object):
if __name__ == '__main__':
from hikvision_isc_client.recorder import recorder
from isc_video_record.recorder import recorder
pre_event = PreEvent('2021-04-26 00:00:00', '2021-04-26 16:00:00')
alarm_points = pre_event.get_alarm_list('f8a3c4d9b8ae42118b4db9fcf7895031')
......
hikvision_isc_client/recorder.py → isc_video_record/recorder.py
浏览文件 @ 20f8353c
......@@ -5,17 +5,16 @@ import time
import threading
import functools
import pytz
import dynaconf
from datetime import timedelta, datetime
from intelab_python_sdk.logger import log_init, log
from intelab_python_sdk.ffmpeg.ffmpeg_concat import concat
from hikvision_isc_client.db import rabbitmq_connect
from hikvision_isc_client.utils.api_helper import IntelabApiHelper
from hikvision_isc_client.utils import aliyun_oss
from hikvision_isc_client.utils.record_utils import record_thread, get_video_duration, time_to_seconds
from hikvision_isc_client.db.mysql import insert_video_info
from isc_video_record.db import rabbitmq_connect
from isc_video_record.utils.api_helper import IntelabApiHelper
from isc_video_record.utils import aliyun_oss
from isc_video_record.utils.record_utils import record_thread, get_video_duration, time_to_seconds
from isc_video_record.db.mysql import insert_video_info
tz = pytz.timezone('Asia/Shanghai')
......@@ -101,8 +100,9 @@ class StreamRecorder:
t2 = time.time()
video_info, error_log = get_video_duration(record_result['file_name'])
log.info('record Time: %s, duration: %s, size: %s',
round(t2 - t1, 2), video_info['duration'], video_info['size'])
log.info('Time consuming: %s, duration: %s, size: %sM',
round(t2 - t1, 2), time_to_seconds(video_info['duration']),
video_info['size'])
url = ''
file_name = record_result['file_name'].split('/')[-1]
if record_result['file_name'] and os.path.isfile(record_result['file_name']):
......@@ -116,7 +116,9 @@ class StreamRecorder:
video_url=url, video_resolution=video_info['resolution'],
biz_type=body['biz_type'], service_type=body['service_type'],
status=1 if record_result['is_completed'] else 2)
log.info('video_info: %s, url: %s, video_id: %s.%s', video_info, url, body['db_table'], video_id)
log.info('video_info: %s, url: %s, video_id: %s.%s, time: %s',
video_info, url, body['db_table'],
video_id, round(time.time() - t1, 2))
return True
@staticmethod
......@@ -209,6 +211,6 @@ class StreamRecorder:
if __name__ == '__main__':
log_init('recorder', False, '/var/log/event_rcv')
log_init('recorder', False, '/var/log/event_vss')
stream_recorder = StreamRecorder()
stream_recorder.start()
hikvision_isc_client/utils/aliyun_oss.py → isc_video_record/utils/aliyun_oss.py
浏览文件 @ 20f8353c
......@@ -73,5 +73,3 @@ if __name__ == '__main__':
# print(oss_download_file('https://test-qzwjtest.oss-cn-hangzhou.aliyuncs.com/test-2.mp4', 't.mp4'))
# print(oss_delete_file('https://test-qzwjtest.oss-cn-hangzhou.aliyuncs.com/test-2.mp4'))
oss_download_file('D00268229_2020-10-23_14-07-13.mp4', '/tmp/v3/videos/D00268229_2020-10-23_14-07-13.mp4')
hikvision_isc_client/utils/api_helper.py → isc_video_record/utils/api_helper.py
浏览文件 @ 20f8353c
import json
import requests
import dynaconf
import dateutil.parser
......@@ -28,6 +27,7 @@ class IntelabApiHelper:
response.raise_for_status()
res_json = response.json().get('data')
# TODO 数据为空的异常处理
results = []
for f in res_json.get('list') or []:
start_time = dateutil.parser.parse(f['beginTime'])
......
hikvision_isc_client/utils/record_utils.py → isc_video_record/utils/record_utils.py
浏览文件 @ 20f8353c
......@@ -73,7 +73,7 @@ def get_video_duration(file_name):
if 'SoundHandler' in log_buffer:
media_type += '+SoundHandler'
if os.path.isfile(file_name):
size = os.path.getsize(file_name) / 1024.0
size = os.path.getsize(file_name) / 1024.0 / 1024.0
video_info = {
'duration': duration[0],
......
settings.toml
浏览文件 @ 20f8353c
......@@ -11,3 +11,32 @@ PORT = 8086
USERNAME = 'admin'
PASSWORD = '123456'
DATABASE = 'intelab'
[default.COMMON]
HOST = ''
[default.RABBITMQ]
HOST = 'localhost'
PORT = 5672
USER = 'guest'
PASSWORD = 'guest'
[default.REDIS]
host = '127.0.0.1'
port = 6379
password = ''
db = 0
decode_responses = true
[default.ALI_OSS]
access_key_id = ''
access_key_secret = ''
bucket_name = ''
region = 'hangzhou'
[default.MYSQL]
HOST = '127.0.0.1'
PORT = 3306
USER = 'root'
PASSWORD = '123456'
DATABASE = 'common_camera'
setup.py
浏览文件 @ 20f8353c
......@@ -3,35 +3,40 @@ import setuptools
with open('README.md', 'r') as fh:
long_description = fh.read()
requires = [
'requests',
'python-dateutil',
'pytz',
'intelab-python-sdk',
'flask',
'dynaconf',
'influxdb',
'python-dateutil',
'pika==1.1.0',
'redis',
'mysql-connector',
'retrying',
'oss2'
]
setuptools.setup(
name='hikvision-isc-client',
version='0.0.2',
author='',
author_email='',
description='Client to interact with Hikvision iSC platform using open API',
name='isc-video-record',
version='1.0.0a1',
description='ISC motion detection playback video stream recording service.',
long_description=long_description,
long_description_content_type='text/markdown',
url='',
packages=setuptools.find_packages(),
python_requires='>=3.6',
classifiers=[
'Programming Language :: Python :: 3',
'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent',
],
install_requires=[
'requests',
'python-dateutil',
'pytz',
'intelab-python-sdk',
'flask',
'dynaconf',
'influxdb',
'python-dateutil',
'pika==1.1.0',
'redis',
'mysql-connector',
'retrying',
'oss2'
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 3.7',
'Private :: Do Not Upload'
],
python_requires='>=3.6',
install_requires=requires,
entry_points={
'console_scripts': [
'isc_video_record = isc_video_record.main:command_line_runner',
]
},
)
tests/test_devices.py
浏览文件 @ 20f8353c
from dynaconf import settings
from hikvision_isc_client.utils.isc_client import HikVisionClient
from isc_video_record.utils.isc_client import HikVisionClient
config = settings.get('ISC')
client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
......@@ -13,6 +13,10 @@ for device in devices.get('list'):
'operator': 0,
'values': [device['indexCode']]
}])
if not camera_info:
print('{}:{} 无监控点'.format(device['regionName'], device['indexCode']))
continue
camera_code = camera_info[0]['indexCode']
print('摄像头序列号{}的编码设备Index为{}'.format(
device['deviceCode'], camera_code))
......
tests/test_playback.py
浏览文件 @ 20f8353c
......@@ -4,8 +4,8 @@ from datetime import datetime
from intelab_python_sdk.logger import log_init
from dynaconf import settings
from hikvision_isc_client.utils.isc_client import HikVisionClient
from hikvision_isc_client.recorder import StreamRecorder
from isc_video_record.utils.isc_client import HikVisionClient
from isc_video_record.recorder import StreamRecorder
tz = pytz.timezone('Asia/Shanghai')
......
tests/test_preview.py
浏览文件 @ 20f8353c
from dynaconf import settings
from hikvision_isc_client.utils.isc_client import HikVisionClient
from isc_video_record.utils.isc_client import HikVisionClient
config = settings.get('ISC')
client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
......
tests/test_sub.py
浏览文件 @ 20f8353c
from dynaconf import settings
from hikvision_isc_client.utils.isc_client import HikVisionClient
from isc_video_record.utils.isc_client import HikVisionClient
config = settings.get('ISC')
client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
config.get('HOST'), config.get('PORT'))
print(client.event_subscription('http://139.196.239.60:5001/event/rcv'))
print(client.event_subscription())
# print(client.event_subscription('http://139.196.239.60:5001/event/rcv'))
print(client.event_unsubscription())
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论