실시간 사회적 거리두기 감지기

--- model.ckpt-400000.data-00000-of-00001
--- model.ckpt-400000.index
--- model.ckpt-400000.meta
--- model.tflite
--- pipeline.config
--- tflite_graph.pb
--- tflite_graph.pbtxt

converter = tf.compat.v1.lite.TFLiteConverter.from_frozen_graph(
    graph_def_file=model_to_be_quantized, 
    input_arrays=['normalized_input_image_tensor'],
    output_arrays=['TFLite_Detection_PostProcess','TFLite_Detection_PostProcess:1','TFLite_Detection_PostProcess:2','TFLite_Detection_PostProcess:3'],
    input_shapes={'normalized_input_image_tensor': [1, 320, 320, 3]}
)
converter.allow_custom_ops = True
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

converter = tf.compat.v1.lite.TFLiteConverter.from_frozen_graph(
    graph_def_file=model_to_be_quantized, 
    input_arrays=['normalized_input_image_tensor'],
    output_arrays=['TFLite_Detection_PostProcess','TFLite_Detection_PostProcess:1','TFLite_Detection_PostProcess:2','TFLite_Detection_PostProcess:3'],
    input_shapes={'normalized_input_image_tensor': [1, 320, 320, 3]}
)
converter.allow_custom_ops = True
converter.target_spec.supported_types = [tf.float16]
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

converter = tf.compat.v1.lite.TFLiteConverter.from_frozen_graph(
    graph_def_file=model_to_be_quantized, 
    input_arrays=['normalized_input_image_tensor'],
    output_arrays=['TFLite_Detection_PostProcess','TFLite_Detection_PostProcess:1','TFLite_Detection_PostProcess:2','TFLite_Detection_PostProcess:3'],
    input_shapes={'normalized_input_image_tensor': [1, 320, 320, 3]}
)
converter.allow_custom_ops = True
converter.representative_dataset = representative_dataset_gen
converter.inference_input_type = tf.uint8
converter.quantized_input_stats = {"normalized_input_image_tensor": (128, 128)}
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

def getmap(image):

  # image : first frame of the input video
  
  global grid_H, grid_W
  h,w=image.shape[:2]
  # 4 corner points of image are set by default
  # User needs to finalise the corner points of a road or floor by dragging and dropping the corners
  #corners=mouse_click_event.adjust_coor_quad(image,corners)
  corners=[(308, 67), (413, 91), (245, 351), (75, 270)]
  corners=np.array(corners,dtype="float32")
  src=order_points(corners)
  (tl,tr,br,bl)=src
  width1=np.sqrt(((br[0]-bl[0])**2)+((br[1]-bl[1])**2))
  width2=np.sqrt(((tr[0]-tl[0])**2)+((tr[1]-tl[1])**2))
  width=max(int(width1),int(width2))
  height1=np.sqrt(((tr[0]-br[0])**2)+((tr[1]-br[1])**2))
  height2=np.sqrt(((tl[0]-bl[0])**2)+((tl[1]-bl[1])**2))
  height=max(int(height1),int(height2))
  width=int(width)
  height=int(height)
  dest=np.array([[0,0],[width-1,0],[width-1,height-1],[0,height-1]],dtype="float32")
  M=cv2.getPerspectiveTransform(src,dest)
  corners=np.array([[0,h-1,1],[w-1,h-1,1],[w-1,0,1],[0,0,1]],dtype="int")
  warped_corners=np.dot(corners,M.T)
  warped_corners=warped_corners/warped_corners[:,2].reshape((len(corners),1))
  warped_corners=np.int64(warped_corners)
  warped_corners=warped_corners[:,:2]
  min_coor=np.min(warped_corners,axis=0)
  max_coor=np.max(warped_corners,axis=0)
  grid_W,grid_H=max_coor-min_coor
  dest=np.array([[abs(min_coor[0]),abs(min_coor[1])],[abs(min_coor[0])+grid_W-1,abs(min_coor[1])],[abs(min_coor[0])+grid_W-1,abs(min_coor[1])+grid_H-1],[abs(min_coor[0]),abs(min_coor[1])+grid_H-1]],dtype="float32")
  M=cv2.getPerspectiveTransform(src,dest)
  return M

def calibration(M,results):
  # calculate minimum distance for social distancing
  global MIN_DISTANCE
  rect=np.array([r[1] for r in results])
  h=np.median(rect[:,2]-rect[:,0])
  coor=np.array([[50,100,1],[50,100+h,1]],dtype="int")
  coor=np.dot(coor,M.T)
  coor=coor/coor[:,2].reshape((2,1))
  coor=np.int64(coor[:,:2])
  MIN_DISTANCE=int(round(dist.pdist(coor)[0]))

  #calculate centroid points of detected people location corresponding to bird's eye view black grid
  centroids=np.array([r[2] for r in results])
  centroids=np.c_[centroids,np.ones((centroids.shape[0],1),dtype="int")]
  warped_centroids=np.dot(centroids,M.T)
  warped_centroids=warped_centroids/warped_centroids[:,2].reshape((len(centroids),1))
  warped_centroids=np.int64(warped_centroids)
  return warped_centroids[:,:2]

def calc_dist(centroids):

  # centroids : updated centroids in top-down view coordinates

  if len(centroids)<2:  # no pair of people, no violation
    return list()
  # evaluate the pairwise distances between people
  condensed_dist=dist.pdist(centroids)
  D=dist.squareform(condensed_dist)
  locations=np.where(D<MIN_DISTANCE)
  violate=list(zip(locations[0],locations[1]))
  violate=np.sort(violate,axis=1)
  violate=np.unique(violate,axis=0)
  violate=np.asarray(list(filter(lambda x:x[0]!=x[1],violate)))
  return violate

def visualise_main(frame,results,violate):
  for (i,(prob,bbox,centroid)) in enumerate(results):
    (startX,startY,endX,endY)=bbox
    (cX,cY)=centroid
    colour=(0,255,0)
    if i in np.unique(violate):
      colour=(0,0,255)
    frame=cv2.rectangle(frame,(startX,startY),(endX,endY),colour,2)
    frame=cv2.circle(frame,(cX,cY),5,colour,1)

  #Drawing the connecting lines between violators
  for i,j in violate:
      frame=cv2.line(frame,results[i][2],results[j][2],(0,0,255),2)
  return frame

def visualise_grid(image,M,centroids,violate):
  warped=cv2.warpPerspective(image,M,(grid_W,grid_H),cv2.INTER_AREA, borderMode=cv2.BORDER_CONSTANT, borderValue=(0,0,0))
  grid=np.zeros(warped.shape,dtype=np.uint8)
  for i in range(len(centroids)):
    colour=(0,255,0)
    if i in np.unique(violate):
      colour=(0,0,255)
    grid=cv2.circle(grid,tuple(centroids[i,:]),5,colour,-1)
    warped=cv2.circle(warped,tuple(centroids[i,:]),5,colour,-1)
  for i,j in violate:
    grid=cv2.line(grid,tuple(centroids[i]),tuple(centroids[j]),(0,0,255),2)
  #cv2_imshow("Bird's eye view grid",cv2.resize(grid,image.shape[:2][::-1]))
  #cv2_imshow("warped",cv2.resize(warped,image.shape[:2][::-1]))
  grid=cv2.resize(grid,image.shape[:2][::-1])
  warped=cv2.resize(warped,image.shape[:2][::-1])
  return grid,warped

tflite_model = "ssd_mobiledet_cpu_coco_fp16.tflite"  
interpreter = tf.lite.Interpreter(model_path=tflite_model)
interpreter.allocate_tensors()
_, HEIGHT, WIDTH, _ = interpreter.get_input_details()[0]['shape']

def set_input_tensor(interpreter, image):
  tensor_index = interpreter.get_input_details()[0]['index']
  input_tensor = interpreter.tensor(tensor_index)()[0]
  input_tensor[:, :] = image

def get_output_tensor(interpreter, index):
  output_details = interpreter.get_output_details()[index]
  tensor = np.squeeze(interpreter.get_tensor(output_details['index']))
  return tensor

def pedestrian_detector(interpreter, image, threshold):
  """Returns a list of detection results, each as a tuple of object info."""
  H,W=HEIGHT,WIDTH
  set_input_tensor(interpreter, image)
  interpreter.invoke()
  # Get all output details
  boxes = get_output_tensor(interpreter, 0)
  class_id = get_output_tensor(interpreter, 1)
  scores = get_output_tensor(interpreter, 2)
  count = int(get_output_tensor(interpreter, 3))
  
  results = []
  for i in range(count):
    if class_id[i] == 0 and scores[i] >= threshold:
      [ymin,xmin,ymax,xmax]=boxes[i]
      (left, right, top, bottom) = (int(xmin * W), int(xmax * W), int(ymin * H), int(ymax * H))
      area=(right-left+1)*(bottom-top+1)
      if area>=1500:
        continue
      centerX=left+int((right-left)/2)
      centerY=top+int((bottom-top)/2)
      results.append((scores[i],(left,top,right,bottom),(centerX,centerY)))
  return results

def preprocess_frame(frame):
    frame = Image.fromarray(frame)
    preprocessed_image = frame.resize(
        (
            HEIGHT,
            WIDTH
        ),
        Image.ANTIALIAS)
    preprocessed_image = tf.keras.preprocessing.image.img_to_array(preprocessed_image)
    preprocessed_image = preprocessed_image.astype('float32') / 255.0
    preprocessed_image = np.expand_dims(preprocessed_image, axis=0)
    
    return preprocessed_image

def process(video):
  vs=cv2.VideoCapture(video)
  #Capture the first frame of video
  res,image=vs.read()
  if image is None:
    return
  image=cv2.resize(image,(320,320))
  # get transformation matrix
  mat=getmap(image)
  fourcc=cv2.VideoWriter_fourcc(*"XVID")
  out=cv2.VideoWriter("result.avi",fourcc,20.0,(320*3,320))
  fps = FPS().start()
  while True:
    res,image=vs.read()
    if image is None:
      break
    #pedestrian detection
    preprocessed_frame = preprocess_frame(image)
    results = pedestrian_detector(interpreter, preprocessed_frame, threshold=0.25)
    preprocessed_frame = np.squeeze(preprocessed_frame) * 255.0
    preprocessed_frame = preprocessed_frame.clip(0, 255)
    preprocessed_frame = preprocessed_frame.squeeze()
    image = np.uint8(preprocessed_frame)
    #calibration
    warped_centroids=calibration(mat, results)
    #Distance-Violation Determination
    violate=calc_dist(warped_centroids)
    #Visualise grid
    grid,warped=visualise_grid(image,mat,warped_centroids,violate)
    #Visualise main frame
    image=visualise_main(image,results,violate)
    #Creating final output frame
    output=cv2.hconcat((image,warped))
    output=cv2.hconcat((output,grid))
    out.write(output)
    fps.update()
  fps.stop()
  print("[INFO] elapsed time: {:.2f}".format(fps.elapsed()))
  print("[INFO] approx. FPS: {:.2f}".format(fps.fps()))
  # release the file pointers
  print("[INFO] cleaning up...")
  vs.release()
  out.release()

[INFO] elapsed time: 52.79
[INFO] approx. FPS: 10.06
[INFO] cleaning up...