chore: 摄像头质量检测排查

isc_video_record/utils/isc_client.py

@@ -49,8 +49,7 @@ class HikVisionClient(object):
         url = self.url + uri
         urllib3.disable_warnings()
         response = requests.post(url, headers=self._sign(uri, body, 'POST'),
-                                 json=body,
-                                 verify=False)
+                                 json=body, verify=False, timeout=30)
         data = {}
         if response.status_code == 200:
             data['code'] = 200
@@ -187,3 +186,15 @@ class HikVisionClient(object):
         }
         res = self._request(uri, body)
         return res
+
+    def get_camera_quality(self, camera_index_code):
+        """
+        根据监控点编号查询视频质量诊断结果，
+        此接口可用作检测监控点位预览画面是否正常
+        """
+        uri = '/artemis/api/nms/v1/vqd/list'
+        body = {
+            'indexCodes': [camera_index_code]
+        }
+        res = self._request(uri, body)
+        return res

tests/test_devices.py

@@ -7,7 +7,44 @@ client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
                          config.get('HOST'), config.get('PORT'))
 
 devices = client.get_encode_device()
+
+desc_dict = {
+    'inspectDate': {
+        'cn': '巡检日期',
+        'value': {}
+    },
+    'resolution': {
+        'cn': '分辨率(720为基准)',
+        'value': {'1': '高清', '0': '标清', '-1': '未知'}
+    },
+    'picUrl': {
+        'cn': '图片url地址',
+        'value': {}
+    },
+    'scFrameResult': {
+        'cn': '视频丢帧结果',
+        'value': {'2': '诊断失败', '1': '正常', '0': '异常', '-1': '未知'}
+    },
+    'scCoverResult': {
+        'cn': '视频遮挡结果',
+        'value': {'2': '诊断失败', '1': '正常', '0': '异常', '-1': '未知'}
+    },
+    'inspectResult': {
+        'cn': '视频巡检结果',
+        'value': {
+            '6': '码流时延', '5': '解码失败', '4': '取流失败',
+            '3': '登录失败', '2': '异常', '1': '正常', '0': '未检测', '-1': '未检测'
+        }
+    },
+    'scMonoResult': {
+        'cn': '黑白图像结果',
+        'value': {'2': '诊断失败', '1': '正常', '0': '异常', '-1': '未知'}
+    }
+}
+
+
 for device in devices.get('list'):
+    print('=============')
     camera_info = client.get_cameras(expressions=[{
         'key': 'parentIndexCode',
         'operator': 0,
@@ -18,8 +55,31 @@ for device in devices.get('list'):
         continue
 
     camera_code = camera_info[0]['indexCode']
-    print('摄像头序列号{}的编码设备Index为{}'.format(
+    print('监控点名称：{}\n监控点分组：{}'.format(
+        camera_info[0]['name'], camera_info[0]['regionPathName']))
+    print('摄像头序列号: {}\n编码设备Index: {}'.format(
         device['deviceCode'], camera_code))
-    print(client.get_camera_online(camera_info[0]['indexCode']))
-    print('直播预览接口为', client.get_camera_preview_url(camera_code))
+
+    res = client.get_camera_online(camera_info[0]['indexCode'])
+    online_info = (res.get('list') or [{}])[0]
+    print('摄像头在线状态: ', '在线' if online_info['online'] == 1 else '离线')
+    if online_info['online'] != 1:
+        continue
+
+    print('直播预览接口: ', client.get_camera_preview_url(camera_code).get('url'))
+
     print('-------------')
+    res = client.get_camera_quality(camera_code)
+    camera_quality = res.get('list') or []
+    camera_quality = camera_quality[0] if len(camera_quality) > 0 else {}
+    text = ''
+    for key, value in camera_quality.items():
+        desc = desc_dict.get(key)
+        if not desc:
+            continue
+        name = desc.get('cn', key)
+        value = desc.get('value', {}).get(str(value) or '-1', value)
+        text += '{}: {}\n'.format(name, value)
+
+    print(text)
+