python目标检测基于opencv实现目标追踪示例

第python目标检测基于opencv实现目标追踪示例目录主要代码信息封装类更新utilspython-opencv3.0新增了一些比较有用的追踪器算法,这里根据官网示例写了一个追踪器类

程序只能运行在安装有opencv3.0以上版本和对应的contrib模块的python解释器

主要代码

#encoding=utf-8

importcv2

fromitemsimportMessageItem

importtime

importnumpyasnp

监视者模块,负责入侵检测,目标跟踪

classWatchDog(object):

#入侵检测者模块,用于入侵检测

def__init__(self,frame=None):

#运动检测器构造函数

self._background=None

ifframeisnotNone:

self._background=cv2.GaussianBlur(cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY),(21,21),0)

self.es=cv2.getStructuringElement(cv2.MORPH_ELLIPSE,(10,10))

defisWorking(self):

#运动检测器是否工作

returnself._backgroundisnotNone

defstartWorking(self,frame):

#运动检测器开始工作

ifframeisnotNone:

self._background=cv2.GaussianBlur(cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY),(21,21),0)

defstopWorking(self):

#运动检测器结束工作

self._background=None

defanalyze(self,frame):

#运动检测

ifframeisNoneorself._backgroundisNone:

return

sample_frame=cv2.GaussianBlur(cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY),(21,21),0)

diff=cv2.absdiff(self._background,sample_frame)

diff=cv2.threshold(diff,25,255,cv2.THRESH_BINARY)[1]

diff=cv2.dilate(diff,self.es,iterations=2)

image,cnts,hierarchy=cv2.findContours(diff.copy(),cv2.RETR_EXTERNAL,cv2.CHAIN_APPROX_SIMPLE)

coordinate=[]

bigC=None

bigMulti=0

forcincnts:

ifcv2.contourArea(c)1500:

continue

(x,y,w,h)=cv2.boundingRect(c)

ifw*hbigMulti:

bigMulti=w*h

bigC=((x,y),(x+w,y+h))

ifbigC:

cv2.rectangle(frame,bigC[0],bigC[1],(255,0,0),2,1)

coordinate.append(bigC)

message={"coord":coordinate}

message['msg']=None

returnMessageItem(frame,message)

classTracker(object):

追踪者模块,用于追踪指定目标

def__init__(self,tracker_type="BOOSTING",draw_coord=True):

初始化追踪器种类

#获得opencv版本

(major_ver,minor_ver,subminor_ver)=(cv2.__version__).split('.')

self.tracker_types=['BOOSTING','MIL','KCF','TLD','MEDIANFLOW','GOTURN']

self.tracker_type=tracker_type

self.isWorking=False

self.draw_coord=draw_coord

#构造追踪器

ifint(minor_ver)3:

self.tracker=cv2.Tracker_create(tracker_type)

else:

iftracker_type=='BOOSTING':

self.tracker=cv2.TrackerBoosting_create()

iftracker_type=='MIL':

self.tracker=cv2.TrackerMIL_create()

iftracker_type=='KCF':

self.tracker=cv2.TrackerKCF_create()

iftracker_type=='TLD':

self.tracker=cv2.TrackerTLD_create()

iftracker_type=='MEDIANFLOW':

self.tracker=cv2.TrackerMedianFlow_create()

iftracker_type=='GOTURN':

self.tracker=cv2.TrackerGOTURN_create()

definitWorking(self,frame,box):

追踪器工作初始化

frame:初始化追踪画面

box:追踪的区域

ifnotself.tracker:

raiseException("追踪器未初始化")

status=self.tracker.init(frame,box)

ifnotstatus:

raiseException("追踪器工作初始化失败")

self.coord=box

self.isWorking=True

deftrack(self,frame):

开启追踪

message=None

ifself.isWorking:

status,self.coord=self.tracker.update(frame)

ifstatus:

message={"coord":[((int(self.coord[0]),int(self.coord[1])),(int(self.coord[0]+self.coord[2]),int(self.coord[1]+self.coord[3])))]}

ifself.draw_coord:

p1=(int(self.coord[0]),int(self.coord[1]))

p2=(int(self.coord[0]+self.coord[2]),int(self.coord[1]+self.coord[3]))

cv2.rectangle(frame,p1,p2,(255,0,0),2,1)

message['msg']="istracking"

returnMessageItem(frame,message)

classObjectTracker(object):

def__init__(self,dataSet):

self.cascade=cv2.CascadeClassifier(dataSet)

deftrack(self,frame):

gray=cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY)

faces=self.cascade.detectMultiScale(gray,1.03,5)

for(x,y,w,h)infaces:

cv2.rectangle(frame,(x,y),(x+w,y+h),(255,0,0),2)

returnframe

if__name__=='__main__':

a=['BOOSTING','MIL','KCF','TLD','MEDIANFLOW','GOTURN']

tracker=Tracker(tracker_type="KCF")

video=cv2.VideoCapture(0)

ok,frame=video.read()

bbox=cv2.selectROI(frame,False)

tracker.initWorking(frame,bbox)

whileTrue:

_,frame=video.read();

if(_):

item=tracker.track(frame);

cv2.imshow("track",item.getFrame())

k=cv2.waitKey(1)0xff

ifk==27:

break

信息封装类

#encoding=utf-8

importjson

fromutilsimportIOUtil

信息封装类

classMessageItem(object):

#用于封装信息的类,包含图片和其他信息

def__init__(self,frame,message):

self._frame=frame

self._message=message

defgetFrame(self):

#图片信息

returnself._frame

defgetMessage(self):

#文字信息,json格式

returnself._message

defgetBase64Frame(self):

#返回base64格式的图片,将BGR图像转化为RGB图像

jepg=IOUtil.array_to_bytes(self._frame[...,::-1])

returnIOUtil.bytes_to_base64(jepg)

defgetBase64FrameByte(self):

#返回base64格式图片的bytes

returnbytes(self.getBase64Frame())

defgetJson(self):

#获得json数据格式

dicdata={"frame":self.getBase64Frame().decode(),"message":self.getMessage()}

returnjson.dumps(dicdata)

defgetBinaryFrame(self):

returnIOUtil.array_to_bytes(self._frame[...,::-1])

运行之后在第一帧图像上选择要追踪的部分,这里测试了一下使用KCF算法的追踪器

更新utils

#encoding=utf-8

importtime

importnumpy

importbase64

importos

importlogging

importsys

fromsettingsimport*

fromPILimportImage

fromioimportBytesIO

classIOUtil(object):

#流操作工具类

@staticmethod

defarray_to_bytes(pic,formatter="jpeg",quality=70):

静态方法,将numpy数组转化二进制流

:parampic:numpy数组

:paramformat:图片格式

:paramquality:压缩比,压缩比越高,产生的二进制数据越短

:return:

stream=BytesIO()

picture=Image.fromarray(pic)

picture.save(stream,format=formatter,quality=quality)

jepg=stream.getvalue()

stream.close()

returnjepg

@staticmethod

defbytes_to_base64(byte):

静态方法,bytes转base64编码

:parambyte:

:return:

returnbase64.b64encode(byte)

@staticmethod

deftransport_rgb(frame):

将bgr图像转化为rgb图像,或者将rgb图像转化为bgr图像

returnframe[...,::-1]

@staticmethod

defbyte_to_package(bytes,cmd,var=1):

将每一帧的图片流的二进制数据进行分包

:parambyte:二进制文件

:paramcmd:命令

:return:

head=[ver,len(byte),cmd]

headPack=struct.pack("!3I",*head)

senddata=headPack+byte

returnsenddata

@staticmethod

defmkdir(filePath):

创建文件夹

ifnotos.path.exists(filePath):

os.mkdir(filePath)

@staticmethod

defcountCenter(box):

计算一个矩形的中心

return(int(abs(box[0][0]-box[1][0])*0.5)+box[0][0],int(abs(box[0][1]-box[1][1])*0.5)+box[0][1])

@staticmethod

defcou