参考链接：

一、python3控制基于onvif的安防摄像头

使用的是tp-link安防摄像头

安装

在Python3中，是安装pip install onvif-zeep实现控制

二、实现

已经保证了摄像头具有的以下的能力，所以不再对设备是否具有能力进行判断。

实现获取设备、控制设备、获取摄像头视频内容进行分析

需要更改的地方，就是填写自己摄像头的用户名和密码

onvif的默认账户和密码是admin和9999，不写(空字符串)也能运行。。。

class PlcVisionSDK:

def __init__(self, kwargs_dict):

self.name = ' admin' # 自己设备的名称

self.password = '9999 ' # 自己设备的密码

self.mul_ip = "239.255.255.250"

self.mul_port = 3702

self.kwargs_dict = kwargs_dict

def zeep_pythonvalue(self, xmlvalue):

return xmlvalue

1、发现设备

使用udp组播，获取摄像头的IP和PORT

def discovery_dev(self):

local_ip = self.kwargs_dict['local_ip']

xml_str = """

uuid:a8098c1a-f86e-11da-bd1a-00112444be1e'

urn:schemas-xmlsoap-org:ws:2005:04:discovery

http://schemas.xmlsoap.org/ws/2005/04/discovery/Probe

dn:NetworkVideoTransmitter

"""

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)

s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 255)

s.bind((local_ip, self.mul_port))

s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP,

socket.inet_aton(self.mul_ip) + socket.inet_aton(local_ip))

s.setblocking(False)

s.sendto(xml_str.encode(), (self.mul_ip, self.mul_port))

while True:

data, address = s.recvfrom(65535)

content = data.decode()

a = re.search(re.compile(r"(.*?)", re.S), str(content))

if a:

b = a.group()

ip, port = b.split('/')[2].split(':')[0: 2]

return str(ip), port

2、绝对移动

对应的方法是 AbsoluteMove

根据获得的IP和PORT，进行绝对移动的控制。

def move(self):

# 需要的参数

plc_ip = self.kwargs_dict['plc_ip']

plc_port = int(self.kwargs_dict['plc_port'])

pan = self.kwargs_dict['pan'] if self.kwargs_dict.get('pan') else 0

tilt = self.kwargs_dict['tilt'] if self.kwargs_dict.get('tilt') else 0

zoom = self.kwargs_dict['zoom'] if self.kwargs_dict.get('zoom') else 0

pan_speed = self.kwargs_dict['pan_speed'] if self.kwargs_dict.get('pan_speed') else 1

tilt_speed = self.kwargs_dict['tilt_speed'] if self.kwargs_dict.get('tilt_speed') else 1

zoom_speed = self.kwargs_dict['zoom_speed'] if self.kwargs_dict.get('zoom_speed') else 1

mycam = ONVIFCamera(plc_ip, plc_port, self.name, self.password)

media = mycam.create_media_service()

ptz = mycam.create_ptz_service()

zeep.xsd.simple.AnySimpleType.pythonvalue = self.zeep_pythonvalue

media_profile = media.GetProfiles()[0]

request = ptz.create_type('GetConfigurationOptions')

request.ConfigurationToken = media_profile.PTZConfiguration.token

request = ptz.create_type('AbsoluteMove')

request.ProfileToken = media_profile.token

ptz.Stop({'ProfileToken': media_profile.token})

if request.Position is None:

request.Position = ptz.GetStatus({'ProfileToken': media_profile.token}).Position

if request.Speed is None:

request.Speed = ptz.GetStatus({'ProfileToken': media_profile.token}).Position

request.Position.PanTilt.x = pan

request.Speed.PanTilt.x = pan_speed

request.Position.PanTilt.y = tilt

request.Speed.PanTilt.y = tilt_speed

request.Position.Zoom = zoom

request.Speed.Zoom = zoom_speed

ptz.AbsoluteMove(request)

return 'finish'

3、相对移动

对应的方法是 RelativeMove

根据获得的IP和PORT，进行相对移动的控制。

def relative_move(self):

# 需要的参数

plc_ip = self.kwargs_dict['plc_ip']

plc_port = int(self.kwargs_dict['plc_port'])

re_pan = self.kwargs_dict['re_pan'] if self.kwargs_dict.get('re_pan') else 0

re_tilt = self.kwargs_dict['re_tilt'] if self.kwargs_dict.get('re_tilt') else 0

re_zoom = self.kwargs_dict['re_zoom'] if self.kwargs_dict.get('re_zoom') else 0

mycam = ONVIFCamera(plc_ip, plc_port, self.name, self.password)

media = mycam.create_media_service()

ptz = mycam.create_ptz_service()

zeep.xsd.simple.AnySimpleType.pythonvalue = self.zeep_pythonvalue

media_profile = media.GetProfiles()[0]

profile_token = media_profile.token

relative_move = ptz.create_type('RelativeMove')

relative_move.ProfileToken = profile_token

if relative_move.Translation is None:

relative_move.Translation = ptz.GetStatus({'ProfileToken': profile_token}).Position

if relative_move.Speed is None:

relative_move.Speed = ptz.GetStatus({'ProfileToken': profile_token}).Position

relative_move.Translation.PanTilt.x = re_pan

relative_move.Translation.PanTilt.y = re_tilt

relative_move.Translation.Zoom = re_zoom

try:

ptz.RelativeMove(relative_move)

except Exception as e:

return '已经无法再移动了'

return 'finish'

4、发现rtsp地址

对应的方法是 GetStreamUri

def get_rtsp(self):

plc_ip = self.kwargs_dict['plc_ip']

plc_port = int(self.kwargs_dict['plc_port'])

mycam = ONVIFCamera(plc_ip, plc_port, self.name, self.password)

media = mycam.create_media_service()

zeep.xsd.simple.AnySimpleType.pythonvalue = self.zeep_pythonvalue

media_profile = media.GetProfiles()

token = media_profile[0].token

# 实例化GetStreamUri对象

obj = media.create_type('GetStreamUri')

obj.StreamSetup = {'Stream': 'RTP-Unicast', 'Transport': {'Protocol': 'RTSP'}}

obj.ProfileToken = token

res_uri = media.GetStreamUri(obj)['Uri']

return res_uri

想要播放rtsp内容，进行分析的话，直接使用opencv就OK了。