parameters: (self)
docstring: This method:
          1. Continiously listens to an input kafka (given by in_kafkaservers and
          in_kafkatopics)
          2. Performs multicam tracking
          3. Writes the tracked objects to another kafka (given by out_kafkaservers
          and out_kafkatopics)
docstring_summary: This method:
          1. Continiously listens to an input kafka (given by in_kafkaservers and
          in_kafkatopics)
          2. Performs multicam tracking
          3. Writes the tracked objects to another kafka (given by out_kafkaservers
          and out_kafkatopics)
docstring_tokens: This method : 1 . Continiously listens to an input kafka ( given by in_kafkaservers and in_kafkatopics ) 2 . Performs multicam tracking 3 . Writes the tracked objects to another kafka ( given by out_kafkaservers and out_kafkatopics )
function: def start_mctracker(self):
          """
          This method:
          1. Continiously listens to an input kafka (given by in_kafkaservers and
          in_kafkatopics)
          2. Performs multicam tracking
          3. Writes the tracked objects to another kafka (given by out_kafkaservers
          and out_kafkatopics)
          """
  
          iters = 0
          num_msgs_received = 0
          recs = []
          # Debugging-related objects
          start_time = tstart_time = ptime_taken = ttime_taken = None
          num_iters_to_print = int(
              constants.APPROX_TIME_PERIOD_TO_PRINT_INFO_IN_SEC /
              float(self.sleep_time_sec))
  
          while True:
              if self.time_prof_flag:
                  tstart_time = time.time()
  
              raw_messages = self.consumer.poll(
                  timeout_ms=self.sleep_time_sec*1000.0, max_records=5000)
  
              start_time = time.time()
              json_list = []
              for _, msg_list in raw_messages.items():
                  num_msgs_received += len(msg_list)
                  for msg in msg_list:
                      curr_time = int(round(time.time() * 1000))
                      kafka_ts = msg.timestamp
                      recs.append({'currTime': curr_time, 'kafkaTs': kafka_ts})
                      json_list.append(msg.value)
  
              if self.time_prof_flag:
                  pstart_time = time.time()
  
              retval = self.track_list(json_list)
              time_taken = time.time() - start_time
  
              if self.time_prof_flag:
                  ptime_taken = time.time() - pstart_time
                  ttime_taken = time.time() - tstart_time
  
                  res = {'currTime': start_time, 'count': len(json_list),
                         'timeTakenMs': time_taken * 1000.0,
                         'reidTimeTakenMs': ptime_taken*1000.0,
                         'totalTimeTakenMs': ttime_taken * 1000.0,
                         "num_unidentified_cars":
                         len(self.mctracker_obj.state.unidentified_cars),
                         "num_prev_list":
                         len(self.mctracker_obj.state.prev_list),
                         "num_carry_over_list":
                         len(self.mctracker_obj.state.carry_over_list),
                         "num_retval":
                         len(retval),
                         "num_match_stats":
                         len(self.mctracker_obj.state.match_stats),
                         "num_possible_parked_cars":
                         len(self.mctracker_obj.state.possible_parked_cars)}
                  self.reid_timings.append(res)
  
              iters += 1
              if (iters % num_iters_to_print) == 0:
                  logging.info(
                      "Mc-Tracker Stream: %s: Num msgs received = %d", str(datetime.now()), num_msgs_received)
              if retval:
                  self.write_to_kafka(retval)
  
              time_taken = time.time() - start_time
              tts = self.sleep_time_sec - time_taken
              if tts > 0:
                  time.sleep(tts)
  
          if self.time_prof_flag:
              if recs:
                  recs_pd = pd.DataFrame(recs)
                  recs_pd['kafkaTsDelayMs'] = recs_pd["currTime"] - \
                      recs_pd["kafkaTs"]
                  recs_pd.to_csv(DEFAULT_KAFKA_LOGS_FILE, index=False)
                  logging.debug("%s", str(recs_pd.describe(percentiles=[
                      0.05, 0.1, 0.25, 0.5, 0.75, 0.90, 0.95])))
              else:
                  logging.debug("No data received")
function_tokens: def start_mctracker ( self ) : iters = 0 num_msgs_received = 0 recs = [ ] # Debugging-related objects start_time = tstart_time = ptime_taken = ttime_taken = None num_iters_to_print = int ( constants . APPROX_TIME_PERIOD_TO_PRINT_INFO_IN_SEC / float ( self . sleep_time_sec ) ) while True : if self . time_prof_flag : tstart_time = time . time ( ) raw_messages = self . consumer . poll ( timeout_ms = self . sleep_time_sec * 1000.0 , max_records = 5000 ) start_time = time . time ( ) json_list = [ ] for _ , msg_list in raw_messages . items ( ) : num_msgs_received += len ( msg_list ) for msg in msg_list : curr_time = int ( round ( time . time ( ) * 1000 ) ) kafka_ts = msg . timestamp recs . append ( { 'currTime' : curr_time , 'kafkaTs' : kafka_ts } ) json_list . append ( msg . value ) if self . time_prof_flag : pstart_time = time . time ( ) retval = self . track_list ( json_list ) time_taken = time . time ( ) - start_time if self . time_prof_flag : ptime_taken = time . time ( ) - pstart_time ttime_taken = time . time ( ) - tstart_time res = { 'currTime' : start_time , 'count' : len ( json_list ) , 'timeTakenMs' : time_taken * 1000.0 , 'reidTimeTakenMs' : ptime_taken * 1000.0 , 'totalTimeTakenMs' : ttime_taken * 1000.0 , "num_unidentified_cars" : len ( self . mctracker_obj . state . unidentified_cars ) , "num_prev_list" : len ( self . mctracker_obj . state . prev_list ) , "num_carry_over_list" : len ( self . mctracker_obj . state . carry_over_list ) , "num_retval" : len ( retval ) , "num_match_stats" : len ( self . mctracker_obj . state . match_stats ) , "num_possible_parked_cars" : len ( self . mctracker_obj . state . possible_parked_cars ) } self . reid_timings . append ( res ) iters += 1 if ( iters % num_iters_to_print ) == 0 : logging . info ( "Mc-Tracker Stream: %s: Num msgs received = %d" , str ( datetime . now ( ) ) , num_msgs_received ) if retval : self . write_to_kafka ( retval ) time_taken = time . time ( ) - start_time tts = self . sleep_time_sec - time_taken if tts > 0 : time . sleep ( tts ) if self . time_prof_flag : if recs : recs_pd = pd . DataFrame ( recs ) recs_pd [ 'kafkaTsDelayMs' ] = recs_pd [ "currTime" ] - recs_pd [ "kafkaTs" ] recs_pd . to_csv ( DEFAULT_KAFKA_LOGS_FILE , index = False ) logging . debug ( "%s" , str ( recs_pd . describe ( percentiles = [ 0.05 , 0.1 , 0.25 , 0.5 , 0.75 , 0.90 , 0.95 ] ) ) ) else : logging . debug ( "No data received" )