目录
一. 介绍
二. 开发准备
1. 对需要集成使用ISAPI进行认证协议申请（ISAPI开发指南_工业测温_测温筒机）
2. 参考ISAPI文档对所需要场景需求进行开发
二. 录像机数据请求
1. 导入所需依赖包
2. 海康硬盘录像机类初始化
3. Xml 数据格式解析方法
4. 对不同ISAPI数据进行组装
5. 请求结果示例
三 , 完整代码示例如下
四. 结语
一. 介绍
本文介绍了如何通过海康ISAPI与海康网络硬盘录像机进行数据交互，使用语言为Python。获取录像机的 下方数据信息。通过ISAPI，我们能够获取摄像头各类资源及进行摄像头布防操作，进而为平台数据监控展示、管理与调度提供操作支持。
通过 ISAPI 接口实现以下操作：
- 获取录像机通道信息：
- 通道名称：标识每个摄像头的名称。
- 通道号：摄像头在录像机上的唯一编号。
- 通道 IP：摄像头对应的网络地址。
- 查询视频流配置：
- 码流类型：如主码流、副码流。
- 帧率：视频每秒传输的帧数。
- 码率：视频流的传输速率。
- 分辨率：视频画面的清晰度。
- 获取设备状态：
- 摄像头是否在线/离线。
二. 开发准备
1. 对需要集成使用ISAPI进行认证协议申请（ISAPI开发指南_工业测温_测温筒机）
https://open.hikvision.com/osp#%E8%AE%BE%E5%A4%87%E9%9B%86%E6%88%90API
2. 参考ISAPI文档对所需要场景需求进行开发
二. 录像机数据请求
1. 导入所需依赖包
import requests
from requests.auth import HTTPDigestAuth
import xml.etree.ElementTree as ET2. 海康硬盘录像机类初始化
# 海康硬盘录像机获取通道监控点资源信息类
class VCRDeviceInfo:def __init__(self, username, password, ip, port):"""初始化录像机设备信息类:param username: 用户名:param password: 密码:param ip: 录像机 IP 地址:param port: 端口号"""self.username = usernameself.password = passwordself.ip = ipself.port = port@staticmethoddef _remove_namespace(tag):"""去除命名空间前缀"""return tag.split('}')[-1] if '}' in tag else tag@staticmethoddef _xml_to_dict(element):"""将 XML 转换为列表套字典结构"""result_list = []for child in element:result = {}if child.text is not None:result[VCRDeviceInfo._remove_namespace(child.tag)] = child.text.strip()if len(child) > 0:result[VCRDeviceInfo._remove_namespace(child.tag)] = VCRDeviceInfo._xml_to_dict(child)result_list.append(result)return result_list3. Xml 数据格式解析方法
def _parse_xml(self, response_info):"""解析录像机状态 XML 数据"""root = ET.fromstring(response_info.text.strip())xml_dict = self._xml_to_dict(root)device_status_list = []for i in xml_dict:channel_info = {'id': i['InputProxyChannelStatus'][0]['id'],'ip_address': i['InputProxyChannelStatus'][1]['sourceInputPortDescriptor'][2]['ipAddress'],'online': i['InputProxyChannelStatus'][2]['online'],'username': i['InputProxyChannelStatus'][1]['sourceInputPortDescriptor'][5]['userName'],'password_status': i['InputProxyChannelStatus'][6]['SecurityStatus'][0]['PasswordStatus']}device_status_list.append(channel_info)return device_status_listdef _xml_data(self, response_info):"""解析录像机基础信息 XML 数据"""root = ET.fromstring(response_info.text.strip())xml_dict = self._xml_to_dict(root)device_status_list = []for i in xml_dict:channel_info = {'id': i['InputProxyChannel'][0]['id'],'name': i['InputProxyChannel'][1]['name'],'ip_address': i['InputProxyChannel'][2]['sourceInputPortDescriptor'][2]['ipAddress'],'managePortNo': i['InputProxyChannel'][2]['sourceInputPortDescriptor'][3]['managePortNo'],'username': i['InputProxyChannel'][2]['sourceInputPortDescriptor'][5]['userName']}device_status_list.append(channel_info)return device_status_listdef _xml_info(self, response_info):"""解析录像机帧率信息 XML 数据"""root = ET.fromstring(response_info.text.strip())xml_dict = self._xml_to_dict(root)device_status_list = []for i in xml_dict:channel_info = {'id': i.get('StreamingChannel', [{}])[4].get('Video', [{}])[1].get('dynVideoInputChannelID', None),'channelName': i.get('StreamingChannel', [{}])[1].get('channelName', ''),'videoCodecType': i.get('StreamingChannel', [{}])[4].get('Video', [{}])[2].get('videoCodecType', None)}video_info = i.get('StreamingChannel', [{}])[4].get('Video', [{}])channel_info.update({'videoResolutionWidth': video_info[4].get('videoResolutionWidth', None),'videoResolutionHeight': video_info[5].get('videoResolutionHeight', None),'constantBitRate': video_info[7].get('constantBitRate', None),'maxFrameRate': str(30) if video_info[9].get('maxFrameRate', '0') == '0' else str(int(int(video_info[9]['maxFrameRate']) / 100))})if not channel_info['channelName'].endswith('2'):device_status_list.append(channel_info)return device_status_list4. 对不同ISAPI数据进行组装
def get_vcr_data_info(self):"""获取录像机的基础信息和状态"""try:# 录像机下方信息url_data = f'http://{self.ip}:{self.port}/ISAPI/ContentMgmt/InputProxy/channels'response_data = requests.get(url_data, auth=HTTPDigestAuth(self.username, self.password), timeout=15)# 录像机状态信息url_status = f'http://{self.ip}:{self.port}/ISAPI/ContentMgmt/InputProxy/channels/status/1'response_status = requests.get(url_status, auth=HTTPDigestAuth(self.username, self.password), timeout=5)# 录像机通道信息url_info = f'http://{self.ip}:{self.port}/ISAPI/Streaming/channels'response_info = requests.get(url_info, auth=HTTPDigestAuth(self.username, self.password), timeout=5)# 当所有请求状态为 200 时处理数据if response_data.status_code == 200 and response_status.status_code == 200 and response_info.status_code == 200:status_list = self._parse_xml(response_status)data_list = self._xml_data(response_data)info_list = self._xml_info(response_info)# 合并数据merged_data = [{**data,**next((status for status in status_list if status['id'] == data['id']), {}),**next((info for info in info_list if info['id'] == data['id']), {})}for data in data_list]return merged_dataelse:return []except Exception as e:return []5. 请求结果示例
三 , 完整代码示例如下
import requests
from requests.auth import HTTPDigestAuth
import xml.etree.ElementTree as ET# 海康硬盘录像机获取通道监控点资源信息类
class VCRDeviceInfo:def __init__(self, username, password, ip, port):"""初始化录像机设备信息类:param username: 用户名:param password: 密码:param ip: 录像机 IP 地址:param port: 端口号"""self.username = usernameself.password = passwordself.ip = ipself.port = port@staticmethoddef _remove_namespace(tag):"""去除命名空间前缀"""return tag.split('}')[-1] if '}' in tag else tag@staticmethoddef _xml_to_dict(element):"""将 XML 转换为列表套字典结构"""result_list = []for child in element:result = {}if child.text is not None:result[VCRDeviceInfo._remove_namespace(child.tag)] = child.text.strip()if len(child) > 0:result[VCRDeviceInfo._remove_namespace(child.tag)] = VCRDeviceInfo._xml_to_dict(child)result_list.append(result)return result_listdef _parse_xml(self, response_info):"""解析录像机状态 XML 数据"""root = ET.fromstring(response_info.text.strip())xml_dict = self._xml_to_dict(root)device_status_list = []for i in xml_dict:channel_info = {'id': i['InputProxyChannelStatus'][0]['id'],'ip_address': i['InputProxyChannelStatus'][1]['sourceInputPortDescriptor'][2]['ipAddress'],'online': i['InputProxyChannelStatus'][2]['online'],'username': i['InputProxyChannelStatus'][1]['sourceInputPortDescriptor'][5]['userName'],'password_status': i['InputProxyChannelStatus'][6]['SecurityStatus'][0]['PasswordStatus']}device_status_list.append(channel_info)return device_status_listdef _xml_data(self, response_info):"""解析录像机基础信息 XML 数据"""root = ET.fromstring(response_info.text.strip())xml_dict = self._xml_to_dict(root)device_status_list = []for i in xml_dict:channel_info = {'id': i['InputProxyChannel'][0]['id'],'name': i['InputProxyChannel'][1]['name'],'ip_address': i['InputProxyChannel'][2]['sourceInputPortDescriptor'][2]['ipAddress'],'managePortNo': i['InputProxyChannel'][2]['sourceInputPortDescriptor'][3]['managePortNo'],'username': i['InputProxyChannel'][2]['sourceInputPortDescriptor'][5]['userName']}device_status_list.append(channel_info)return device_status_listdef _xml_info(self, response_info):"""解析录像机帧率信息 XML 数据"""root = ET.fromstring(response_info.text.strip())xml_dict = self._xml_to_dict(root)device_status_list = []for i in xml_dict:channel_info = {'id': i.get('StreamingChannel', [{}])[4].get('Video', [{}])[1].get('dynVideoInputChannelID', None),'channelName': i.get('StreamingChannel', [{}])[1].get('channelName', ''),'videoCodecType': i.get('StreamingChannel', [{}])[4].get('Video', [{}])[2].get('videoCodecType', None)}video_info = i.get('StreamingChannel', [{}])[4].get('Video', [{}])channel_info.update({'videoResolutionWidth': video_info[4].get('videoResolutionWidth', None),'videoResolutionHeight': video_info[5].get('videoResolutionHeight', None),'constantBitRate': video_info[7].get('constantBitRate', None),'maxFrameRate': str(30) if video_info[9].get('maxFrameRate', '0') == '0' else str(int(int(video_info[9]['maxFrameRate']) / 100))})if not channel_info['channelName'].endswith('2'):device_status_list.append(channel_info)return device_status_listdef get_vcr_data_info(self):"""获取录像机的基础信息和状态"""try:# 录像机下方信息url_data = f'http://{self.ip}:{self.port}/ISAPI/ContentMgmt/InputProxy/channels'response_data = requests.get(url_data, auth=HTTPDigestAuth(self.username, self.password), timeout=15)# 录像机状态信息url_status = f'http://{self.ip}:{self.port}/ISAPI/ContentMgmt/InputProxy/channels/status/1'response_status = requests.get(url_status, auth=HTTPDigestAuth(self.username, self.password), timeout=5)# 录像机通道信息url_info = f'http://{self.ip}:{self.port}/ISAPI/Streaming/channels'response_info = requests.get(url_info, auth=HTTPDigestAuth(self.username, self.password), timeout=5)# 当所有请求状态为 200 时处理数据if response_data.status_code == 200 and response_status.status_code == 200 and response_info.status_code == 200:status_list = self._parse_xml(response_status)data_list = self._xml_data(response_data)info_list = self._xml_info(response_info)# 合并数据merged_data = [{**data,**next((status for status in status_list if status['id'] == data['id']), {}),**next((info for info in info_list if info['id'] == data['id']), {})}for data in data_list]return merged_dataelse:return []except Exception as e:return []# 使用示例
if __name__ == "__main__":pass# 实例化 配置相关参数vcr = VCRDeviceInfo(username='admin',password='abc123',ip='192.168.7.50',port=80)# 获取录像机下方通道数据信息device_info = vcr.get_vcr_data_info()print(device_info)
四. 结语
谢谢大家观看完这篇文章，希望大家要铭记的两首歌，一首开始，一首结尾，最后希望这个篇文章对大家有帮助。
