class_docstr: Predictor in single camera
  
  The pipeline for image input: 
  
      1. Detection
      2. Detection -> Attribute
  
  The pipeline for video input: 
  
      1. Tracking
      2. Tracking -> Attribute
      3. Tracking -> KeyPoint -> SkeletonAction Recognition
      4. VideoAction Recognition
  
  Args:
      args (argparse.Namespace): arguments in pipeline, which contains environment and runtime settings
      cfg (dict): config of models in pipeline
      is_video (bool): whether the input is video, default as False
      multi_camera (bool): whether to use multi camera in pipeline, 
          default as False
class_docstr_tok: ["Predictor","in","single","camera","The","pipeline","for","image","input",":","1",".","Detection","2",".","Detection","-",">","Attribute","The","pipeline","for","video","input",":","1",".","Tracking","2",".","Tracking","-",">","Attribute","3",".","Tracking","-",">","KeyPoint","-",">","SkeletonAction","Recognition","4",".","VideoAction","Recognition","Args",":","args","(","argparse.Namespace",")",":","arguments","in","pipeline",",","which","contains","environment","and","runtime","settings","cfg","(","dict",")",":","config","of","models","in","pipeline","is_video","(","bool",")",":","whether","the","input","is","video",",","default","as","False","multi_camera","(","bool",")",":","whether","to","use","multi","camera","in","pipeline",",","default","as","False"]
code: def predict_video(self, video_file, thread_idx=0):
      # mot
      # mot -> attr
      # mot -> pose -> action
      capture = cv2.VideoCapture(video_file)
  
      # 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("video fps: %d, frame_count: %d" % (fps, frame_count))
  
      if len(self.pushurl) > 0:
          video_out_name = (
              "output" if self.file_name is None else self.file_name
          )
          pushurl = os.path.join(self.pushurl, video_out_name)
          print("the result will push stream to url:{}".format(pushurl))
          pushstream = PushStream(pushurl)
          pushstream.initcmd(fps, width, height)
      elif self.cfg["visual"]:
          video_out_name = (
              "output"
              if (self.file_name is None or type(self.file_name) == int)
              else self.file_name
          )
          if type(video_file) == str and "rtsp" in video_file:
              video_out_name = (
                  video_out_name
                  + "_t"
                  + str(thread_idx).zfill(2)
                  + "_rtsp"
              )
          if not os.path.exists(self.output_dir):
              os.makedirs(self.output_dir)
          out_path = os.path.join(
              self.output_dir, video_out_name + ".mp4"
          )
          fourcc = cv2.VideoWriter_fourcc(*"mp4v")
          writer = cv2.VideoWriter(
              out_path, fourcc, fps, (width, height)
          )
  
      frame_id = 0
  
      entrance, records, center_traj = None, None, None
      if self.draw_center_traj:
          center_traj = [{}]
      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
          or self.illegal_parking_time != -1
      ):
          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."
              assert (
                  len(self.region_polygon) > 6
              ), "region_type is custom, region_polygon should be at least 3 pairs of point coords."
  
              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:{} unsupported.".format(self.region_type)
              )
  
      video_fps = fps
  
      video_action_imgs = []
  
      if self.with_video_action:
          short_size = self.cfg["VIDEO_ACTION"]["short_size"]
          scale = ShortSizeScale(short_size)
  
      object_in_region_info = (
          {}
      )  # store info for vehicle parking in region
      illegal_parking_dict = None
      cars_count = 0
      retrograde_traj_len = 0
      framequeue = queue.Queue(10)
  
      thread = threading.Thread(
          target=self.capturevideo, args=(capture, framequeue)
      )
      thread.start()
      time.sleep(1)
  
      while not framequeue.empty():
          if frame_id % 10 == 0:
              print(
                  "Thread: {}; frame id: {}".format(
                      thread_idx, frame_id
                  )
              )
  
          frame_rgb = framequeue.get()
          if frame_id > self.warmup_frame:
              self.pipe_timer.total_time.start()
  
          if self.modebase["idbased"] or self.modebase["skeletonbased"]:
              if frame_id > self.warmup_frame:
                  self.pipe_timer.module_time["mot"].start()
  
              mot_skip_frame_num = self.mot_predictor.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
              res = self.mot_predictor.predict_image(
                  [copy.deepcopy(frame_rgb)],
                  visual=False,
                  reuse_det_result=reuse_det_result,
                  frame_count=frame_id,
              )
  
              # mot output format: id, class, score, xmin, ymin, xmax, ymax
              mot_res = parse_mot_res(res)
              if frame_id > self.warmup_frame:
                  self.pipe_timer.module_time["mot"].end()
                  self.pipe_timer.track_num += len(mot_res["boxes"])
  
              if frame_id % 10 == 0:
                  print(
                      "Thread: {}; trackid number: {}".format(
                          thread_idx, len(mot_res["boxes"])
                      )
                  )
  
              # flow_statistic only support single class MOT
              boxes, scores, ids = res[0]  # batch size = 1 in MOT
              mot_result = (
                  frame_id + 1,
                  boxes[0],
                  scores[0],
                  ids[0],
              )  # single class
              statistic = flow_statistic(
                  mot_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,
                  ids2names=self.mot_predictor.pred_config.labels,
              )
              records = statistic["records"]
  
              if self.illegal_parking_time != -1:
                  (
                      object_in_region_info,
                      illegal_parking_dict,
                  ) = update_object_info(
                      object_in_region_info,
                      mot_result,
                      self.region_type,
                      entrance,
                      video_fps,
                      self.illegal_parking_time,
                  )
                  if len(illegal_parking_dict) != 0:
                      # build relationship between id and plate
                      for key, value in illegal_parking_dict.items():
                          plate = self.collector.get_carlp(key)
                          illegal_parking_dict[key]["plate"] = plate
  
              # nothing detected
              if len(mot_res["boxes"]) == 0:
                  frame_id += 1
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.img_num += 1
                      self.pipe_timer.total_time.end()
                  if self.cfg["visual"]:
                      _, _, fps = self.pipe_timer.get_total_time()
                      im = self.visualize_video(
                          frame_rgb,
                          mot_res,
                          self.collector,
                          frame_id,
                          fps,
                          entrance,
                          records,
                          center_traj,
                      )  # visualize
                      if len(self.pushurl) > 0:
                          pushstream.pipe.stdin.write(im.tobytes())
                      else:
                          writer.write(im)
                          if self.file_name is None:  # use camera_id
                              cv2.imshow("Paddle-Pipeline", im)
                              if cv2.waitKey(1) & 0xFF == ord("q"):
                                  break
                  continue
  
              self.pipeline_res.update(mot_res, "mot")
              crop_input, new_bboxes, ori_bboxes = crop_image_with_mot(
                  frame_rgb, mot_res
              )
  
              if self.with_vehicleplate and frame_id % 10 == 0:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time[
                          "vehicleplate"
                      ].start()
                  plate_input, _, _ = crop_image_with_mot(
                      frame_rgb, mot_res, expand=False
                  )
                  platelicense = (
                      self.vehicleplate_detector.get_platelicense(
                          plate_input
                      )
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["vehicleplate"].end()
                  self.pipeline_res.update(platelicense, "vehicleplate")
              else:
                  self.pipeline_res.clear("vehicleplate")
  
              if self.with_human_attr:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["attr"].start()
                  attr_res = self.attr_predictor.predict_image(
                      crop_input, visual=False
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["attr"].end()
                  self.pipeline_res.update(attr_res, "attr")
  
              if self.with_vehicle_attr:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time[
                          "vehicle_attr"
                      ].start()
                  attr_res = self.vehicle_attr_predictor.predict_image(
                      crop_input, visual=False
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["vehicle_attr"].end()
                  self.pipeline_res.update(attr_res, "vehicle_attr")
  
              if (
                  self.with_vehicle_press
                  or self.with_vehicle_retrograde
              ):
                  if (
                      frame_id == 0
                      or cars_count == 0
                      or cars_count > len(mot_res["boxes"])
                  ):
                      if frame_id > self.warmup_frame:
                          self.pipe_timer.module_time["lanes"].start()
                      lanes, directions = self.laneseg_predictor.run(
                          [copy.deepcopy(frame_rgb)]
                      )
                      lanes_res = {
                          "output": lanes,
                          "directions": directions,
                      }
                      if frame_id > self.warmup_frame:
                          self.pipe_timer.module_time["lanes"].end()
  
                      if frame_id == 0 or (
                          len(lanes) > 0 and frame_id > 0
                      ):
                          self.pipeline_res.update(lanes_res, "lanes")
  
                      cars_count = len(mot_res["boxes"])
  
              if self.with_vehicle_press:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time[
                          "vehicle_press"
                      ].start()
                  press_lane = copy.deepcopy(
                      self.pipeline_res.get("lanes")
                  )
                  if press_lane is None:
                      continue
  
                  vehicle_press_res_list = (
                      self.vehicle_press_predictor.mot_run(
                          press_lane, mot_res["boxes"]
                      )
                  )
                  vehiclepress_res = {"output": vehicle_press_res_list}
  
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["vehicle_press"].end()
  
                  self.pipeline_res.update(
                      vehiclepress_res, "vehicle_press"
                  )
  
              if self.with_idbased_detaction:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["det_action"].start()
                  det_action_res = self.det_action_predictor.predict(
                      crop_input, mot_res
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["det_action"].end()
                  self.pipeline_res.update(det_action_res, "det_action")
  
                  if self.cfg["visual"]:
                      self.det_action_visual_helper.update(
                          det_action_res
                      )
  
              if self.with_idbased_clsaction:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["cls_action"].start()
                  cls_action_res = (
                      self.cls_action_predictor.predict_with_mot(
                          crop_input, mot_res
                      )
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["cls_action"].end()
                  self.pipeline_res.update(cls_action_res, "cls_action")
  
                  if self.cfg["visual"]:
                      self.cls_action_visual_helper.update(
                          cls_action_res
                      )
  
              if self.with_skeleton_action:
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["kpt"].start()
                  kpt_pred = self.kpt_predictor.predict_image(
                      crop_input, visual=False
                  )
                  (
                      keypoint_vector,
                      score_vector,
                  ) = translate_to_ori_images(
                      kpt_pred, np.array(new_bboxes)
                  )
                  kpt_res = {}
                  kpt_res["keypoint"] = (
                      [keypoint_vector.tolist(), score_vector.tolist()]
                      if len(keypoint_vector) > 0
                      else [[], []]
                  )
                  kpt_res["bbox"] = ori_bboxes
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["kpt"].end()
  
                  self.pipeline_res.update(kpt_res, "kpt")
  
                  self.kpt_buff.update(
                      kpt_res, mot_res
                  )  # collect kpt output
                  state = (
                      self.kpt_buff.get_state()
                  )  # whether frame num is enough or lost tracker
  
                  skeleton_action_res = {}
                  if state:
                      if frame_id > self.warmup_frame:
                          self.pipe_timer.module_time[
                              "skeleton_action"
                          ].start()
                      collected_keypoint = (
                          self.kpt_buff.get_collected_keypoint()
                      )  # reoragnize kpt output with ID
                      skeleton_action_input = parse_mot_keypoint(
                          collected_keypoint, self.coord_size
                      )
                      skeleton_action_res = self.skeleton_action_predictor.predict_skeleton_with_mot(
                          skeleton_action_input
                      )
                      if frame_id > self.warmup_frame:
                          self.pipe_timer.module_time[
                              "skeleton_action"
                          ].end()
                      self.pipeline_res.update(
                          skeleton_action_res, "skeleton_action"
                      )
  
                  if self.cfg["visual"]:
                      self.skeleton_action_visual_helper.update(
                          skeleton_action_res
                      )
  
              if self.with_mtmct and frame_id % 10 == 0:
                  (
                      crop_input,
                      img_qualities,
                      rects,
                  ) = self.reid_predictor.crop_image_with_mot(
                      frame_rgb, mot_res
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["reid"].start()
                  reid_res = self.reid_predictor.predict_batch(
                      crop_input
                  )
  
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["reid"].end()
  
                  reid_res_dict = {
                      "features": reid_res,
                      "qualities": img_qualities,
                      "rects": rects,
                  }
                  self.pipeline_res.update(reid_res_dict, "reid")
              else:
                  self.pipeline_res.clear("reid")
  
          if self.with_video_action:
              # get the params
              frame_len = self.cfg["VIDEO_ACTION"]["frame_len"]
              sample_freq = self.cfg["VIDEO_ACTION"]["sample_freq"]
  
              if (
                  sample_freq * frame_len > frame_count
              ):  # video is too short
                  sample_freq = int(frame_count / frame_len)
  
              # filter the warmup frames
              if frame_id > self.warmup_frame:
                  self.pipe_timer.module_time["video_action"].start()
  
              # collect frames
              if frame_id % sample_freq == 0:
                  # Scale image
                  scaled_img = scale(frame_rgb)
                  video_action_imgs.append(scaled_img)
  
              # the number of collected frames is enough to predict video action
              if len(video_action_imgs) == frame_len:
                  classes, scores = self.video_action_predictor.predict(
                      video_action_imgs
                  )
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time["video_action"].end()
  
                  video_action_res = {
                      "class": classes[0],
                      "score": scores[0],
                  }
                  self.pipeline_res.update(
                      video_action_res, "video_action"
                  )
  
                  print("video_action_res:", video_action_res)
  
                  video_action_imgs.clear()  # next clip
  
          if self.with_vehicle_retrograde:
              # get the params
              frame_len = self.cfg["VEHICLE_RETROGRADE"]["frame_len"]
              sample_freq = self.cfg["VEHICLE_RETROGRADE"][
                  "sample_freq"
              ]
  
              if (
                  sample_freq * frame_len > frame_count
              ):  # video is too short
                  sample_freq = int(frame_count / frame_len)
  
              # filter the warmup frames
              if frame_id > self.warmup_frame:
                  self.pipe_timer.module_time[
                      "vehicle_retrograde"
                  ].start()
  
              if frame_id % sample_freq == 0:
                  frame_mot_res = copy.deepcopy(
                      self.pipeline_res.get("mot")
                  )
                  self.vehicle_retrograde_predictor.update_center_traj(
                      frame_mot_res, max_len=frame_len
                  )
                  retrograde_traj_len = retrograde_traj_len + 1
  
              # the number of collected frames is enough to predict
              if retrograde_traj_len == frame_len:
                  retrograde_mot_res = copy.deepcopy(
                      self.pipeline_res.get("mot")
                  )
                  retrograde_lanes = copy.deepcopy(
                      self.pipeline_res.get("lanes")
                  )
                  frame_shape = frame_rgb.shape
  
                  if retrograde_lanes is None:
                      continue
                  (
                      retrograde_res,
                      fence_line,
                  ) = self.vehicle_retrograde_predictor.mot_run(
                      lanes_res=retrograde_lanes,
                      det_res=retrograde_mot_res,
                      frame_shape=frame_shape,
                  )
  
                  retrograde_res_update = self.pipeline_res.get(
                      "vehicle_retrograde"
                  )
  
                  if retrograde_res_update is not None:
                      retrograde_res_update = retrograde_res_update[
                          "output"
                      ]
                      if retrograde_res is not None:
                          for retrograde_res_id in retrograde_res:
                              if (
                                  retrograde_res_id
                                  not in retrograde_res_update
                              ):
                                  retrograde_res_update.append(
                                      retrograde_res_id
                                  )
                  else:
                      retrograde_res_update = []
  
                  retrograde_res_dict = {
                      "output": retrograde_res_update,
                      "fence_line": fence_line,
                  }
  
                  if (
                      retrograde_res is not None
                      and len(retrograde_res) > 0
                  ):
                      print("retrograde res:", retrograde_res)
  
                  self.pipeline_res.update(
                      retrograde_res_dict, "vehicle_retrograde"
                  )
  
                  if frame_id > self.warmup_frame:
                      self.pipe_timer.module_time[
                          "vehicle_retrograde"
                      ].end()
  
                  retrograde_traj_len = 0
  
          self.collector.append(frame_id, self.pipeline_res)
  
          if frame_id > self.warmup_frame:
              self.pipe_timer.img_num += 1
              self.pipe_timer.total_time.end()
          frame_id += 1
  
          if self.cfg["visual"]:
              _, _, fps = self.pipe_timer.get_total_time()
  
              im = self.visualize_video(
                  frame_rgb,
                  self.pipeline_res,
                  self.collector,
                  frame_id,
                  fps,
                  entrance,
                  records,
                  center_traj,
                  self.illegal_parking_time != -1,
                  illegal_parking_dict,
              )  # visualize
              if len(self.pushurl) > 0:
                  pushstream.pipe.stdin.write(im.tobytes())
              else:
                  writer.write(im)
                  if self.file_name is None:  # use camera_id
                      cv2.imshow("Paddle-Pipeline", im)
                      if cv2.waitKey(1) & 0xFF == ord("q"):
                          break
  
      if self.cfg["visual"] and len(self.pushurl) == 0:
          writer.release()
          print("save result to {}".format(out_path))
  
imports: import os
  import yaml
  import glob
  import cv2
  import numpy
  import math
  import paddle
  import sys
  import copy
  import threading
  import queue
  import time
  from collections import defaultdict
  from datacollector import DataCollector, Result
  from cfg_utils import argsparser, print_arguments, merge_cfg
  from pipe_utils import PipeTimer
  from pipe_utils import get_test_images, crop_image_with_det, crop_image_with_mot, parse_mot_res, parse_mot_keypoint
  from pipe_utils import PushStream
  from python.infer import Detector, DetectorPicoDet
  from python.keypoint_infer import KeyPointDetector
  from python.keypoint_postprocess import translate_to_ori_images
  from python.preprocess import decode_image, ShortSizeScale
  from python.visualize import visualize_box_mask, visualize_attr, visualize_pose, visualize_action, visualize_vehicleplate, visualize_vehiclepress, visualize_lane, visualize_vehicle_retrograde
  from pptracking.python.mot_sde_infer import SDE_Detector
  from pptracking.python.mot.visualize import plot_tracking_dict
  from pptracking.python.mot.utils import flow_statistic, update_object_info
  from pphuman.attr_infer import AttrDetector
  from pphuman.video_action_infer import VideoActionRecognizer
  from pphuman.action_infer import SkeletonActionRecognizer, DetActionRecognizer, ClsActionRecognizer
  from pphuman.action_utils import KeyPointBuff, ActionVisualHelper
  from pphuman.reid import ReID
  from pphuman.mtmct import mtmct_process
  from ppvehicle.vehicle_plate import PlateRecognizer
  from ppvehicle.vehicle_attr import VehicleAttr
  from ppvehicle.vehicle_pressing import VehiclePressingRecognizer
  from ppvehicle.vehicle_retrograde import VehicleRetrogradeRecognizer
  from ppvehicle.lane_seg_infer import LaneSegPredictor
  from download import auto_download_model
label_desc: Use image node_id 7 for calling the PipePredictor obj's underlying member method code with example usage: obj.predict_video(video_file, thread_idx) without return types