python opencv Achieve target tracking
Here according to the official website example wrote a tracker class
python opencv Achieve target trackingpython-opencv3.0 Some more useful tracker algorithms have been added
Here according to the official website example wrote a tracker classThe program can only be run when the installation has opencv3.0 The above version and the corresponding contrib Modular python Interpreter
#encoding=utf-8import cv2from items import MessageItemimport timeimport numpy as np''' Monitor module , Responsible for intrusion detection , Target tracking '''class WatchDog(object): # IDS module , For intrusion detection def __init__(self,frame=None): # Motion detector constructor self._background = None if frame is not None: self._background = cv2.GaussianBlur(cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY),(21,21),0) self.es = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (10, 10)) def isWorking(self): # Is the motion detector working return self._background is not None def startWorking(self,frame): # The motion detector starts to work if frame is not None: self._background = cv2.GaussianBlur(cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY), (21, 21), 0) def stopWorking(self): # The motion detector is finished self._background = None def analyze(self,frame): # Motion detection if frame is None or self._background is None: 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 for c in cnts: if cv2.contourArea(c) < 1500: continue (x,y,w,h) = cv2.boundingRect(c) if w * h > bigMulti: bigMulti = w * h bigC = ((x,y),(x+w,y+h)) if bigC: cv2.rectangle(frame, bigC[0],bigC[1], (255,0,0), 2, 1) coordinate.append(bigC) message = {"coord":coordinate} message['msg'] = None return MessageItem(frame,message)class Tracker(object): ''' Tracker module , Used to track specific targets ''' def __init__(self,tracker_type = "BOOSTING",draw_coord = True): ''' Initialize tracker type ''' # get opencv edition (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 # Build a tracker if int(minor_ver) < 3: self.tracker = cv2.Tracker_create(tracker_type) else: if tracker_type == 'BOOSTING': self.tracker = cv2.TrackerBoosting_create() if tracker_type == 'MIL': self.tracker = cv2.TrackerMIL_create() if tracker_type == 'KCF': self.tracker = cv2.TrackerKCF_create() if tracker_type == 'TLD': self.tracker = cv2.TrackerTLD_create() if tracker_type == 'MEDIANFLOW': self.tracker = cv2.TrackerMedianFlow_create() if tracker_type == 'GOTURN': self.tracker = cv2.TrackerGOTURN_create() def initWorking(self,frame,box): ''' Tracker working initialization frame: Initialize trace screen box: Tracking area ''' if not self.tracker: raise Exception(" Tracker not initialized ") status = self.tracker.init(frame,box) if not status: raise Exception(" Tracker job initialization failed ") self.coord = box self.isWorking = True def track(self,frame): ''' Turn on tracking ''' message = None if self.isWorking: status,self.coord = self.tracker.update(frame) if status: message = {"coord":[((int(self.coord[0]), int(self.coord[1])),(int(self.coord[0] + self.coord[2]), int(self.coord[1] + self.coord[3])))]} if self.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'] = "is tracking" return MessageItem(frame,message)class ObjectTracker(object): def __init__(self,dataSet): self.cascade = cv2.CascadeClassifier(dataSet) def track(self,frame): gray = cv2.cvtColor(frame,cv2.COLOR_BGR2GRAY) faces = self.cascade.detectMultiScale(gray,1.03,5) for (x,y,w,h) in faces: cv2.rectangle(frame,(x,y),(x+w,y+h),(255,0,0),2) return frameif __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) while True: _,frame = video.read(); if(_): item = tracker.track(frame); cv2.imshow("track",item.getFrame()) k = cv2.waitKey(1) & 0xff if k == 27: break#encoding=utf-8import jsonfrom utils import IOUtil''' Information encapsulation class '''class MessageItem(object): # Class used to encapsulate information , Include pictures and other information def __init__(self,frame,message): self._frame = frame self._message = message def getFrame(self): # Image information return self._frame def getMessage(self): # Text messages ,json Format return self._message def getBase64Frame(self): # return base64 Format picture , take BGR The image is transformed into RGB Images jepg = IOUtil.array_to_bytes(self._frame[...,::-1]) return IOUtil.bytes_to_base64(jepg) def getBase64FrameByte(self): # return base64 Format picture bytes return bytes(self.getBase64Frame()) def getJson(self): # get json data format dicdata = {"frame":self.getBase64Frame().decode(),"message":self.getMessage()} return json.dumps(dicdata) def getBinaryFrame(self): return IOUtil.array_to_bytes(self._frame[...,::-1])After running, select the part to be tracked on the first image , Here is a test KCF Algorithm tracker
to update ： Forget to play utils, Sorry for the trouble
#encoding=utf-8import timeimport numpyimport base64import osimport loggingimport sysfrom settings import *from PIL import Imagefrom io import BytesIO# Tool class class IOUtil(object): # Stream operation tool class @staticmethod def array_to_bytes(pic,formatter="jpeg",quality=70): ''' Static methods , take numpy Array to binary stream :param pic: numpy Array :param format: Image format :param quality: Compression ratio , Higher compression ratio , The shorter the binary data produced :return: ''' stream = BytesIO() picture = Image.fromarray(pic) picture.save(stream,format=formatter,quality=quality) jepg = stream.getvalue() stream.close() return jepg @staticmethod def bytes_to_base64(byte): ''' Static methods ,bytes turn base64 code :param byte: :return: ''' return base64.b64encode(byte) @staticmethod def transport_rgb(frame): ''' take bgr The image is transformed into rgb Images , Or will rgb The image is transformed into bgr Images ''' return frame[...,::-1] @staticmethod def byte_to_package(bytes,cmd,var=1): ''' The binary data of the picture stream of each frame is subcontracted :param byte: Binary :param cmd: command :return: ''' head = [ver,len(byte),cmd] headPack = struct.pack("!3I", *head) senddata = headPack+byte return senddata @staticmethod def mkdir(filePath): ''' Create folder ''' if not os.path.exists(filePath): os.mkdir(filePath) @staticmethod def countCenter(box): ''' Calculate the center of a rectangle ''' 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 def countBox(center): ''' Calculate from two points ,x,y,c,r ''' return (center[0][0],center[0][1],center[1][0]-center[0][0],center[1][1]-center[0][1]) @staticmethod def getImageFileName(): return time.strftime("%Y_%m_%d_%H_%M_%S", time.localtime())+'.png'# Build log logger = logging.getLogger(LOG_NAME)formatter = logging.Formatter(LOG_FORMATTER)IOUtil.mkdir(LOG_DIR);file_handler = logging.FileHandler(LOG_DIR + LOG_FILE,encoding='utf-8')file_handler.setFormatter(formatter)console_handler = logging.StreamHandler(sys.stdout)console_handler.setFormatter(formatter)logger.addHandler(file_handler)logger.addHandler(console_handler)logger.setLevel(logging.INFO)The above is personal experience , I hope I can give you a reference , I also hope you can support the software development network .
