使用海康 ISAPI 与网络硬盘录像机数据交互
·
目录
1. 对需要集成使用ISAPI进行认证协议申请（ISAPI开发指南_工业测温_测温筒机）
一. 介绍
本文介绍了如何通过海康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 = username
self.password = password
self.ip = ip
self.port = port
@staticmethod
def _remove_namespace(tag):
"""去除命名空间前缀"""
return tag.split('}')[-1] if '}' in tag else tag
@staticmethod
def _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_list
def _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_list
def _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_data
else:
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 = username
self.password = password
self.ip = ip
self.port = port
@staticmethod
def _remove_namespace(tag):
"""去除命名空间前缀"""
return tag.split('}')[-1] if '}' in tag else tag
@staticmethod
def _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_list
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_list
def _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_list
def _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_list
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_data
else:
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)
四. 结语
谢谢大家观看完这篇文章，希望大家要铭记的两首歌，一首开始，一首结尾，最后希望这个篇文章对大家有帮助。
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
4
0- 0
已为社区贡献3条内容
所有评论(0)