class_docstr: Args:
      model_dir (str): root path of model.pdiparams, model.pdmodel and infer_cfg.yml
      device (str): Choose the device you want to run, it can be: CPU/GPU/XPU/NPU, default is CPU
      run_mode (str): mode of running(paddle/trt_fp32/trt_fp16)
      batch_size (int): size of pre batch in inference
      trt_min_shape (int): min shape for dynamic shape in trt
      trt_max_shape (int): max shape for dynamic shape in trt
      trt_opt_shape (int): opt shape for dynamic shape in trt
      trt_calib_mode (bool): If the model is produced by TRT offline quantitative
          calibration, trt_calib_mode need to set True
      cpu_threads (int): cpu threads
      enable_mkldnn (bool): whether to open MKLDNN
      output_dir (string): The path of output, default as 'output'
      threshold (float): Score threshold of the detected bbox, default as 0.5
      save_images (bool): Whether to save visualization image results, default as False
      save_mot_txts (bool): Whether to save tracking results (txt), default as False
      draw_center_traj (bool): Whether drawing the trajectory of center, default as False
      secs_interval (int): The seconds interval to count after tracking, default as 10
      skip_frame_num (int): Skip frame num to get faster MOT results, default as -1
      do_entrance_counting(bool): Whether counting the numbers of identifiers entering 
          or getting out from the entrance, default as False，only support single class
          counting in MOT.
      do_break_in_counting(bool): Whether counting the numbers of identifiers break in
          the area, default as False，only support single class counting in MOT,
          and the video should be taken by a static camera.
      region_type (str): Area type for entrance counting or break in counting, 'horizontal'
          and 'vertical' used when do entrance counting. 'custom' used when do break in counting. 
          Note that only support single-class MOT, and the video should be taken by a static camera.
      region_polygon (list): Clockwise point coords (x0,y0,x1,y1...) of polygon of area when
          do_break_in_counting. Note that only support single-class MOT and
          the video should be taken by a static camera.
class_docstr_tok: ["Args",":","model_dir","(","str",")",":","root","path","of","model.pdiparams",",","model.pdmodel","and","infer_cfg.yml","device","(","str",")",":","Choose","the","device","you","want","to","run",",","it","can","be",":","CPU\/GPU\/XPU\/NPU",",","default","is","CPU","run_mode","(","str",")",":","mode","of","running","(","paddle\/trt_fp32\/trt_fp16",")","batch_size","(","int",")",":","size","of","pre","batch","in","inference","trt_min_shape","(","int",")",":","min","shape","for","dynamic","shape","in","trt","trt_max_shape","(","int",")",":","max","shape","for","dynamic","shape","in","trt","trt_opt_shape","(","int",")",":","opt","shape","for","dynamic","shape","in","trt","trt_calib_mode","(","bool",")",":","If","the","model","is","produced","by","TRT","offline","quantitative","calibration",",","trt_calib_mode","need","to","set","True","cpu_threads","(","int",")",":","cpu","threads","enable_mkldnn","(","bool",")",":","whether","to","open","MKLDNN","output_dir","(","string",")",":","The","path","of","output",",","default","as","'output'","threshold","(","float",")",":","Score","threshold","of","the","detected","bbox",",","default","as","0.5","save_images","(","bool",")",":","Whether","to","save","visualization","image","results",",","default","as","False","save_mot_txts","(","bool",")",":","Whether","to","save","tracking","results","(","txt",")",",","default","as","False","draw_center_traj","(","bool",")",":","Whether","drawing","the","trajectory","of","center",",","default","as","False","secs_interval","(","int",")",":","The","seconds","interval","to","count","after","tracking",",","default","as","10","skip_frame_num","(","int",")",":","Skip","frame","num","to","get","faster","MOT","results",",","default","as","-1","do_entrance_counting","(","bool",")",":","Whether","counting","the","numbers","of","identifiers","entering","or","getting","out","from","the","entrance",",","default","as","False\uff0conly","support","single","class","counting","in","MOT",".","do_break_in_counting","(","bool",")",":","Whether","counting","the","numbers","of","identifiers","break","in","the","area",",","default","as","False\uff0conly","support","single","class","counting","in","MOT",",","and","the","video","should","be","taken","by","a","static","camera",".","region_type","(","str",")",":","Area","type","for","entrance","counting","or","break","in","counting",",","'horizontal'","and","'vertical","'","used","when","do","entrance","counting",".","'custom","'","used","when","do","break","in","counting",".","Note","that","only","support","single-class","MOT",",","and","the","video","should","be","taken","by","a","static","camera",".","region_polygon","(","list",")",":","Clockwise","point","coords","(","x0",",","y0",",","x1",",","y1","...",")","of","polygon","of","area","when","do_break_in_counting",".","Note","that","only","support","single-class","MOT","and","the","video","should","be","taken","by","a","static","camera","."]
code: def predict_video(self, video_file, camera_id):
      video_out_name = "mot_output.mp4"
      if camera_id != -1:
          capture = cv2.VideoCapture(camera_id)
      else:
          capture = cv2.VideoCapture(video_file)
          video_out_name = os.path.split(video_file)[-1]
      # Get Video info : resolution, fps, frame count
      width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH))
      height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
      fps = int(capture.get(cv2.CAP_PROP_FPS))
      frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
      print("fps: %d, frame_count: %d" % (fps, frame_count))
  
      if not os.path.exists(self.output_dir):
          os.makedirs(self.output_dir)
      out_path = os.path.join(self.output_dir, video_out_name)
      video_format = "mp4v"
      fourcc = cv2.VideoWriter_fourcc(*video_format)
      writer = cv2.VideoWriter(out_path, fourcc, fps, (width, height))
  
      frame_id = 0
      timer = MOTTimer()
      results = defaultdict(
          list
      )  # support single class and multi classes
      num_classes = self.num_classes
      data_type = "mcmot" if num_classes > 1 else "mot"
      ids2names = self.pred_config.labels
  
      center_traj = None
      entrance = None
      records = None
      if self.draw_center_traj:
          center_traj = [{} for i in range(num_classes)]
      if num_classes == 1:
          id_set = set()
          interval_id_set = set()
          in_id_list = list()
          out_id_list = list()
          prev_center = dict()
          records = list()
          if self.do_entrance_counting or self.do_break_in_counting:
              if self.region_type == "horizontal":
                  entrance = [0, height / 2.0, width, height / 2.0]
              elif self.region_type == "vertical":
                  entrance = [width / 2, 0.0, width / 2, height]
              elif self.region_type == "custom":
                  entrance = []
                  assert (
                      len(self.region_polygon) % 2 == 0
                  ), "region_polygon should be pairs of coords points when do break_in counting."
                  for i in range(0, len(self.region_polygon), 2):
                      entrance.append(
                          [
                              self.region_polygon[i],
                              self.region_polygon[i + 1],
                          ]
                      )
                  entrance.append([width, height])
              else:
                  raise ValueError(
                      "region_type:{} is not supported.".format(
                          self.region_type
                      )
                  )
  
      video_fps = fps
  
      while 1:
          ret, frame = capture.read()
          if not ret:
              break
          if frame_id % 10 == 0:
              print("Tracking frame: %d" % (frame_id))
  
          timer.tic()
          mot_skip_frame_num = self.skip_frame_num
          reuse_det_result = False
          if (
              mot_skip_frame_num > 1
              and frame_id > 0
              and frame_id % mot_skip_frame_num > 0
          ):
              reuse_det_result = True
          seq_name = video_out_name.split(".")[0]
          frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
          mot_results = self.predict_image(
              [frame_rgb],
              visual=False,
              seq_name=seq_name,
              reuse_det_result=reuse_det_result,
          )
          timer.toc()
  
          online_tlwhs, online_scores, online_ids = mot_results[0]
          for cls_id in range(num_classes):
              results[cls_id].append(
                  (
                      frame_id + 1,
                      online_tlwhs[cls_id],
                      online_scores[cls_id],
                      online_ids[cls_id],
                  )
              )
  
          # NOTE: just implement flow statistic for single class
          if num_classes == 1:
              result = (
                  frame_id + 1,
                  online_tlwhs[0],
                  online_scores[0],
                  online_ids[0],
              )
              statistic = flow_statistic(
                  result,
                  self.secs_interval,
                  self.do_entrance_counting,
                  self.do_break_in_counting,
                  self.region_type,
                  video_fps,
                  entrance,
                  id_set,
                  interval_id_set,
                  in_id_list,
                  out_id_list,
                  prev_center,
                  records,
                  data_type,
                  ids2names=self.pred_config.labels,
              )
              records = statistic["records"]
  
          fps = 1.0 / timer.duration
          im = plot_tracking_dict(
              frame,
              num_classes,
              online_tlwhs,
              online_ids,
              online_scores,
              frame_id=frame_id,
              fps=fps,
              ids2names=ids2names,
              do_entrance_counting=self.do_entrance_counting,
              entrance=entrance,
              records=records,
              center_traj=center_traj,
          )
  
          writer.write(im)
          if camera_id != -1:
              cv2.imshow("Mask Detection", im)
              if cv2.waitKey(1) & 0xFF == ord("q"):
                  break
          frame_id += 1
  
      if self.save_mot_txts:
          result_filename = os.path.join(
              self.output_dir, video_out_name.split(".")[-2] + ".txt"
          )
  
          write_mot_results(
              result_filename, results, data_type, num_classes
          )
  
          if num_classes == 1:
              result_filename = os.path.join(
                  self.output_dir,
                  video_out_name.split(".")[-2] + "_flow_statistic.txt",
              )
              f = open(result_filename, "w")
              for line in records:
                  f.write(line)
              print("Flow statistic save in {}".format(result_filename))
              f.close()
  
      writer.release()
  
imports: import os
  import time
  import yaml
  import cv2
  import numpy
  from collections import defaultdict
  import paddle
  from benchmark_utils import PaddleInferBenchmark
  from preprocess import decode_image
  from mot_utils import argsparser, Timer, get_current_memory_mb
  from det_infer import Detector, get_test_images, print_arguments, bench_log, PredictConfig
  import sys
  from mot import JDETracker
  from mot.utils import MOTTimer, write_mot_results, flow_statistic
  from mot.visualize import plot_tracking, plot_tracking_dict
label_desc: Use image node_id 6 for calling the JDE_Detector obj's underlying member method code with example usage: obj.predict_video(video_file, camera_id) without return types