# Face Check API response
{
  "faces": [
    {
      "confidence": 99.371,
      "face_rectangle": {
        "x": 230,
        "y": 169,
        "height": 546,
        "width": 376
      }
    },
    {
      "confidence": 99.895,
      "face_rectangle": {
        "x": 530,
        "y": 369,
        "height": 245,
        "width": 230
      }
    }
  ]
}

!pip install opencv-python
import cv2
# 读取图像
image = cv2.imread('path_to_image.jpg')
# 显示图像
cv2.imshow('Image', image)

!pip install dlib
import dlib
# 加载人脸检测器
detector = dlib.get_frontal_face_detector()
# 检测人脸
faces = detector(image)
for face in faces:
    x, y, w, h = face.left(), face.top(), face.width(), face.height()

# 加载预训练模型
def load_model(model, pretrained_path, load_to_cpu):
    print('Loading pretrained model from {}'.format(pretrained_path))
    if load_to_cpu:
        pretrained_dict = torch.load(pretrained_path, map_location=lambda storage, loc: storage)
    else:
        device = torch.cuda.current_device()
        pretrained_dict = torch.load(pretrained_path, map_location=lambda storage, loc: storage.cuda(device))
    if "state_dict" in pretrained_dict.keys():
        pretrained_dict = remove_prefix(pretrained_dict['state_dict'], 'module.')
    else:
        pretrained_dict = remove_prefix(pretrained_dict, 'module.')
    # 加载权重
    model.load_state_dict(pretrained_dict, strict=False)
    return model

def remove_prefix(state_dict, prefix):
    print('remove prefix \'{}\''.format(prefix))
    f = lambda x: x.split(prefix, 1)[-1] if x.startswith(prefix) else x
    return {f(key): value for key, value in state_dict.items()}

#预训练模型参数
trained_weigth_path = '/path/resnet50.pth' 
cpu = not torch.cuda.is_available() #当前环境没有gpu使用cpu
model = load_model(net, trained_weigth_path, cpu)
model.eval() #设置模型为评估模式

# 加载图片并预处理
def preprocess_image(image_path):
  img_raw = cv2.imread(image_path, cv2.IMREAD_COLOR)
  if img_raw is None:
    print('image cant open, check please')
    return
  img = np.float32(img_raw) #转为浮点数
  scale = torch.Tensor([img.shape[1], img.shape[0], img.shape[1], img.shape[0]]) #缩放比例 [width, height, width, height]
  img -= (104, 117, 123) #减去均值进行归一化，RGB 固定104, 117, 123
  img = img.transpose(2, 0, 1) #转置图像维度 维度从 (height, width, channels) 转换为 (channels, height, width)
  img = torch.from_numpy(img).unsqueeze(0) #添加批量维度 形状从 (channels, height, width) 变为 (1, channels, height, width)
  return img, img_raw, scale

# 执行Pytorch模型推理
model = load_model(net, trained_weigth_path, cpu)
def inference_pytorch(model, img):
  model.eval() #设置模型为评估模式
  loc, conf, landms = model(img) 
  # loc：位置预测（bounding box location predictions）
  # conf：置信度（confidence scores）
  # landms：关键点（landmark predictions）
  print("loc shape:", loc.shape)
  print("conf shape:", conf.shape)
  print("landms shape:", landms.shape)  
  return loc, conf, landms

#生成先验框
class PriorBox(object):
    def __init__(self, cfg, image_size=None, phase='train'):
        super(PriorBox, self).__init__()
        self.min_sizes = cfg['min_sizes']
        self.steps = cfg['steps']
        self.clip = cfg['clip']
        self.image_size = image_size
        self.feature_maps = [[ceil(self.image_size[0]/step), ceil(self.image_size[1]/step)] for step in self.steps]
        self.name = "s"

    def forward(self):
        anchors = []
        for k, f in enumerate(self.feature_maps):
            min_sizes = self.min_sizes[k]
            for i, j in product(range(f[0]), range(f[1])):
                for min_size in min_sizes:
                    s_kx = min_size / self.image_size[1]
                    s_ky = min_size / self.image_size[0]
                    dense_cx = [x * self.steps[k] / self.image_size[1] for x in [j + 0.5]]
                    dense_cy = [y * self.steps[k] / self.image_size[0] for y in [i + 0.5]]
                    for cy, cx in product(dense_cy, dense_cx):
                        anchors += [cx, cy, s_kx, s_ky]
        # back to torch land
        output = torch.Tensor(anchors).view(-1, 4)
        if self.clip:
            output.clamp_(max=1, min=0)
        return output

#解码先验框
def py_cpu_nms(dets, thresh):
    """Pure Python NMS baseline."""
    x1 = dets[:, 0]
    y1 = dets[:, 1]
    x2 = dets[:, 2]
    y2 = dets[:, 3]
    scores = dets[:, 4]
    areas = (x2 - x1 + 1) * (y2 - y1 + 1)
    order = scores.argsort()[::-1]
    keep = []
    while order.size > 0:
        i = order[0]
        keep.append(i)
        xx1 = np.maximum(x1[i], x1[order[1:]])
        yy1 = np.maximum(y1[i], y1[order[1:]])
        xx2 = np.minimum(x2[i], x2[order[1:]])
        yy2 = np.minimum(y2[i], y2[order[1:]])
        w = np.maximum(0.0, xx2 - xx1 + 1)
        h = np.maximum(0.0, yy2 - yy1 + 1)
        inter = w * h
        ovr = inter / (areas[i] + areas[order[1:]] - inter)
        inds = np.where(ovr <= thresh)[0]
        order = order[inds + 1]
    return keep

#解码位置
def decode_loc(loc, priors, variances):
    boxes = torch.cat((
        priors[:, :2] + loc[:, :2] * variances[0] * priors[:, 2:],
        priors[:, 2:] * torch.exp(loc[:, 2:] * variances[1])), 1)
    boxes[:, :2] -= boxes[:, 2:] / 2
    boxes[:, 2:] += boxes[:, :2]
    return boxes
#解码关键点
def decode_landm(pre, priors, variances):
    landms = torch.cat((priors[:, :2] + pre[:, :2] * variances[0] * priors[:, 2:],
                        priors[:, :2] + pre[:, 2:4] * variances[0] * priors[:, 2:],
                        priors[:, :2] + pre[:, 4:6] * variances[0] * priors[:, 2:],
                        priors[:, :2] + pre[:, 6:8] * variances[0] * priors[:, 2:],
                        priors[:, :2] + pre[:, 8:10] * variances[0] * priors[:, 2:],
                        ), dim=1)
    return landms
    
# 解码预测结果 
def decode_results(loc, conf, landms, resize, scale, img_shape):
  im_height = img_shape[2]
  im_width = img_shape[3] 
  #### 解码预测结果 #####
  priorbox = PriorBox(cfg, image_size=(im_height, im_width)) #生成先验框（prior boxes），用于帮助模型预测目标的位置
  priors = priorbox.forward() #生成先验框
  prior_data = priors.data #将先验框数据转换为 NumPy 数组 
  #解码位置预测，将模型的输出解码为实际的边框坐标 
  boxes = decode_loc(torch.tensor(loc[0]), torch.tensor(prior_data), cfg['variance']) #使用先验框和模型的输出（loc[0]）解码出实际的边框坐标。cfg['variance'] 是解码过程中使用的方差。
  boxes = boxes * scale / resize #对解码后的边框进行缩放，使其与原图的大小一致。scale 是之前定义的缩放比例张量，resize 是缩放因子，这里为1。
  boxes = boxes.cpu().numpy() #将解码后的边框转换为 NumPy 数组
  #提取置信度 
  scores = conf[0][:, 1] #提取了每个预测框的置信度分数
  #解码关键点预测
  landms = decode_landm(torch.tensor(landms[0]), torch.tensor(prior_data), cfg['variance'])
  scale1 = torch.Tensor([img_shape[3], img_shape[2], img_shape[3], img_shape[2],
                        img_shape[3], img_shape[2], img_shape[3], img_shape[2],
                        img_shape[3], img_shape[2]])
  landms = landms * scale1 / resize
  landms = landms.cpu().numpy()
  return boxes, scores, landms

# 过滤检测框
def filter_boxes(boxes, scores, landms, confidence_threshold, nms_threshold):
  # 将 PyTorch 张量转换为 NumPy 数组(这里有修改，和原来的代码不一样)
  scores = scores.detach().cpu().numpy()
  # 过滤低置信度检测结果
  inds = np.where(scores > confidence_threshold)[0]
  boxes = boxes[inds] #保留置信度高于阈值的边框
  landms = landms[inds] #保留置信度高于阈值的关键点
  scores = scores[inds] #保留置信度高于阈值的置信度分数
  # 按置信度排序
  order = scores.argsort()[::-1] #[::-1] 表示从高到低排序。
  boxes = boxes[order] #按排序后的索引重新排列边框
  landms = landms[order] #按排序后的索引重新排列关键点
  scores = scores[order] #按排序后的索引重新排列置信度分数
  # 将边框和置信度分数水平堆叠在一起，形成一个新的数组 dets。scores[:, np.newaxis] 将置信度分数转换为列向量。
  dets = np.hstack((boxes, scores[:, np.newaxis], landms)).astype(np.float32, copy=False)
  # 使用 NMS 算法去除重叠的检测框。nms_threshold 是 NMS 的阈值，py_cpu_nms 是一个实现 NMS 的函数。keep 是保留下来的检测框的索引。
  keep = py_cpu_nms(dets[:, :5], nms_threshold)
  dets = dets[keep, :] #保留 NMS 后的检测框
  landms = landms[keep] #保留 NMS 后的关键点
  return dets

# 在图上绘制检测结果
def draw_detection(image, dets, save_path, vis_thres=0.6):
  # 绘制检测结果（可能多张人脸）
  for i, b in enumerate(dets):
      if b[4] < vis_thres: #只展示高置信度人脸
          continue
      text = "{:.4f}".format(b[4])
      b = list(map(int, b))
      #绘制人脸边框
      cv2.rectangle(image, (b[0], b[1]), (b[2], b[3]), (0, 0, 255), 5)
      cx = b[0]
      cy = b[1] + 12
      #写上置信度
      cv2.putText(image, text, (cx, cy),
                  cv2.FONT_HERSHEY_DUPLEX, 0.5, (255, 255, 255))
      # 绘制关键点
      cv2.circle(image, (b[5], b[6]), 3, (0, 0, 255), 8) #1是半径，4是线宽
      cv2.circle(image, (b[7], b[8]), 3, (0, 255, 255), 8)
      cv2.circle(image, (b[9], b[10]), 3, (255, 0, 255), 8)
      cv2.circle(image, (b[11], b[12]), 3, (0, 255, 0), 8)
      cv2.circle(image, (b[13], b[14]), 3, (255, 0, 0), 8)
  # 保存结果图像
  cv2.imwrite(save_path, image)
  print(f"Detection result saved to {save_path}")

#人脸检测,使用Pytorch模型
def check_face_use_pytorch(model, image_path):
  # 加载图片并预处理
  img, img_raw, scale = preprocess
  # 执行Pytorch模型推理
  loc, conf, landms = inference_pytorch(model, img)
  # 推理结果解码
  boxes, scores, landms = decode_results(loc, conf, landms, resize, scale, img.shape)
  # 过滤检测框
  dets = filter_boxes(boxes, scores, landms, confidence_threshold, nms_threshold)
  # 在图像上结果标注并保存
  draw_detection(img_raw, dets, save_path, vis_thres)

# 创建dummy input，使用任意大小的图像，但批量大小固定为1
dummy_input = torch.randn(1, 3, 480, 640)  # 这个大小只是一个示例
# 导出为ONNX模型，设置动态输入形状
torch.onnx.export(
    model,                   # PyTorch 模型
    dummy_input,             # 示例输入张量
    "path/kavin_face_model.onnx",  # 导出ONNX路径
    verbose=True,            # 是否打印导出过程的详细信息
    input_names=['input'],   # 输入的名字
    output_names=['loc', 'conf', 'landms'], # 输出的名字
    opset_version=11,        # ONNX 的操作集版本
    dynamic_axes={
        'input': {0: 'batch_size', 2: 'height', 3: 'width'},  # 设置动态维度
        'loc': {0: 'batch_size'},  # 批量大小是动态的
        'conf': {0: 'batch_size'},  # 批量大小是动态的
        'landms': {0: 'batch_size'}  # 批量大小是动态的
    }
)
print("ONNX模型导出成功")

import onnxruntime as ort
# 加载 ONNX 模型
onnx_model_path = "/path/kavin_face_model.onnx"
ort_session = ort.InferenceSession(onnx_model_path)
# 打印模型输入名称和形状
input_name = ort_session.get_inputs()[0].name
print(f"Input name: {input_name}")
print("Input shape:", ort_session.get_inputs()[0].shape)
# 打印模型输出名称和形状
output_names = [output.name for output in ort_session.get_outputs()]
for output_name in output_names:
    print(f"Output name: {output_name}")
    print("Output shape:", ort_session.get_outputs()[output_names.index(output_name)].shape)

import onnxruntime as ort
# 加载 ONNX 模型
onnx_model_path = "model_repository/kavin_face_model/1/model.onnx"
ort_session = ort.InferenceSession(onnx_model_path)

# 执行ONNX模型推理
def inference_onnx(ort_session, img):
  # 前向传播
  input_name = ort_session.get_inputs()[0].name #获取输入名称
  loc, conf, landms = ort_session.run(None, {input_name: img}) #使用 ONNX Runtime 进行模型推理，第一个参数 None 表示我们要获取模型的所有输出，第二个参数是一个字典，键是输入名称，值是输入数据。
  # loc：位置预测（bounding box location predictions）
  # conf：置信度（confidence scores）
  # landms：关键点（landmark predictions）
  print("loc shape:", loc.shape)
  print("conf shape:", conf.shape)
  print("landms shape:", landms.shape)  
  return loc, conf, landms

#人脸检测,使用ONNX模型
def check_face_use_onnx(ort_session, image_path):
  # 加载图片并预处理
  img, img_raw, scale = preprocess_image(image_path)
  # 执行ONNX模型推理
  loc, conf, landms = inference_onnx(ort_session, img)  
  # 解码结果，过滤，在图上标注并保存
  # ...同上

name: "kavin_face_model"
platform: "onnxruntime_onnx"
max_batch_size: 1
input [
  {
    name: "input"
    data_type: TYPE_FP32
    dims: [3, -1, -1]
  }
]
output [
  {
    name: "loc"
    data_type: TYPE_FP32
    dims: [-1, 4]
  },
  {
    name: "conf"
    data_type: TYPE_FP32
    dims: [-1, 2]
  },
  {
    name: "landms"
    data_type: TYPE_FP32
    dims: [-1, 10]
  }
]
instance_group [
  {
    count: 1
    kind: KIND_CPU
  }
]

version: '3.8'
services:
  kavin_face_server:
    image: nvcr.io/nvidia/tritonserver:24.07-py3
    container_name: kavin_face_server
    ports:
      - "8000:8000"
      - "8001:8001"
      - "8002:8002"
    volumes:
      - ./model_repository:/models
    command: tritonserver --model-repository=/models
    deploy:
      replicas: 3  # 设置为3个实例来创建一个集群
      resources:
        limits:
          cpus: '1'
          memory: 2G
    restart: always

import requests
def inference_api(img):
  # 构造推理请求的JSON结构
  infer_request = {
        "inputs": [{
                "name": "input",
                "shape": img.shape,
                "datatype": "FP32",
                "data": img.flatten().tolist()
            }]
  }  
  # 发送推理请求
  try:
    url = 'http://localhost:8000/v2/models/kavin_face_model/infer'
    response = requests.post(url, json=infer_request)
    result = response.json()
    if response.status_code != 200:
      raise Exception(result)
  except Exception as e:
    print("Error in inference request:", e)
    return
  # 解析推理结果,注意这里的返回值顺序
  conf = np.array(result['outputs'][0]['data']).reshape(result['outputs'][0]['shape'])
  landms = np.array(result['outputs'][1]['data']).reshape(result['outputs'][1]['shape'])
  loc = np.array(result['outputs'][2]['data']).reshape(result['outputs'][2]['shape'])
  return loc, conf, landms
#人脸检测,使用API
def check_face_use_api(image_path):
  # 加载图片并预处理
  img, img_raw, scale = preprocess_image(image_path)
  # 执行API推理
  loc, conf, landms = inference_api(img)
  # 解码结果，过滤，在图上标注并保存
  # ...同上