[to #42322933] add face recognition face mask model

Link: https://code.alibaba-inc.com/Ali-MaaS/MaaS-lib/codereview/11472495

data/test/images/ir_face_recognition_1.png

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:602b46c6ba1d18fd3b91fd3b47112d37ca9d8e1ed72f0c0ea93ad8d493f5182e
+size 20299

data/test/images/ir_face_recognition_2.png

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c0791f043b905f2e77ccf2f8c5b29182e1fc99cee16d9069e8bbc1704e917268
+size 20631

modelscope/metainfo.py

@@ -246,6 +246,7 @@ class Pipelines(object):
     realtime_video_object_detection = 'cspnet_realtime-video-object-detection_streamyolo'
     face_recognition = 'ir101-face-recognition-cfglint'
     face_recognition_ood = 'ir-face-recognition-ood-rts'
+    face_recognition_onnx_fm = 'manual-face-recognition-frfm'
     arc_face_recognition = 'ir50-face-recognition-arcface'
     mask_face_recognition = 'resnet-face-recognition-facemask'
     image_instance_segmentation = 'cascade-mask-rcnn-swin-image-instance-segmentation'

modelscope/pipelines/cv/__init__.py

@@ -21,6 +21,7 @@ if TYPE_CHECKING:
     from .face_recognition_ood_pipeline import FaceRecognitionOodPipeline
     from .arc_face_recognition_pipeline import ArcFaceRecognitionPipeline
     from .mask_face_recognition_pipeline import MaskFaceRecognitionPipeline
+    from .face_recognition_onnx_fm_pipeline import FaceRecognitionOnnxFmPipeline
     from .general_recognition_pipeline import GeneralRecognitionPipeline
     from .image_cartoon_pipeline import ImageCartoonPipeline
     from .image_classification_pipeline import GeneralImageClassificationPipeline
@@ -105,6 +106,7 @@ else:
         'face_recognition_ood_pipeline': ['FaceRecognitionOodPipeline'],
         'arc_face_recognition_pipeline': ['ArcFaceRecognitionPipeline'],
         'mask_face_recognition_pipeline': ['MaskFaceRecognitionPipeline'],
+        'face_recognition_onnx_fm_pipeline': ['FaceRecognitionOnnxFmPipeline'],
         'general_recognition_pipeline': ['GeneralRecognitionPipeline'],
         'image_classification_pipeline':
         ['GeneralImageClassificationPipeline', 'ImageClassificationPipeline'],

modelscope/pipelines/cv/face_recognition_onnx_fm_pipeline.py

@@ -0,0 +1,86 @@
+# Copyright (c) Alibaba, Inc. and its affiliates.
+import os.path as osp
+from typing import Any, Dict
+
+import cv2
+import numpy as np
+import onnxruntime
+import PIL
+import torch
+import torch.nn.functional as F
+
+from modelscope.metainfo import Pipelines
+from modelscope.models.cv.face_recognition.align_face import align_face
+from modelscope.models.cv.facial_landmark_confidence import \
+    FacialLandmarkConfidence
+from modelscope.outputs import OutputKeys
+from modelscope.pipelines import pipeline
+from modelscope.pipelines.base import Input, Pipeline
+from modelscope.pipelines.builder import PIPELINES
+from modelscope.preprocessors import LoadImage
+from modelscope.utils.constant import ModelFile, Tasks
+from modelscope.utils.logger import get_logger
+from . import FaceProcessingBasePipeline
+
+logger = get_logger()
+
+
+@PIPELINES.register_module(
+    Tasks.face_recognition, module_name=Pipelines.face_recognition_onnx_fm)
+class FaceRecognitionOnnxFmPipeline(FaceProcessingBasePipeline):
+
+    def __init__(self, model: str, **kwargs):
+        """
+        use `model` to create a face recognition face mask onnx pipeline for prediction
+        Args:
+            model: model id on modelscope hub.
+        Example:
+
+        ```python
+        >>> from modelscope.pipelines import pipeline
+        >>> frfm = pipeline('face-recognition-ood', 'damo/cv_manual_face-recognition_frfm')
+        >>> frfm("https://modelscope.oss-cn-beijing.aliyuncs.com/test/images/face_recognition_1.png")
+        {{'img_embedding': array([[ 0.02276129, -0.00761525, ...,0.05735306]],
+            dtype=float32)} }
+        ```
+        """
+        super().__init__(model=model, **kwargs)
+        onnx_path = osp.join(model, ModelFile.ONNX_MODEL_FILE)
+        logger.info(f'loading model from {onnx_path}')
+        self.sess, self.input_node_name, self.out_node_name = self.load_onnx_model(
+            onnx_path)
+        logger.info('load model done')
+
+    def load_onnx_model(self, onnx_path):
+        sess = onnxruntime.InferenceSession(onnx_path)
+        out_node_name = []
+        input_node_name = []
+        for node in sess.get_outputs():
+            out_node_name.append(node.name)
+
+        for node in sess.get_inputs():
+            input_node_name.append(node.name)
+
+        return sess, input_node_name, out_node_name
+
+    def preprocess(self, input: Input) -> Dict[str, Any]:
+        result = super().preprocess(input)
+        align_img = result['img']
+        face_img = align_img[:, :, ::-1]  # to rgb
+        face_img = (face_img / 255. - 0.5) / 0.5
+        face_img = np.expand_dims(face_img, 0).copy()
+        face_img = np.transpose(face_img, axes=(0, 3, 1, 2))
+        face_img = face_img.astype(np.float32)
+        result['input_tensor'] = face_img
+        return result
+
+    def forward(self, input: Dict[str, Any]) -> Dict[str, Any]:
+        input_feed = {}
+        input_feed[
+            self.input_node_name[0]] = input['input_tensor'].cpu().numpy()
+        emb = self.sess.run(self.out_node_name, input_feed=input_feed)[0]
+        emb /= np.sqrt(np.sum(emb**2, -1, keepdims=True))  # l2 norm
+        return {OutputKeys.IMG_EMBEDDING: emb}
+
+    def postprocess(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
+        return inputs

tests/pipelines/test_face_recognition_onnx_fm.py

@@ -0,0 +1,37 @@
+# Copyright (c) Alibaba, Inc. and its affiliates.
+import unittest
+
+import numpy as np
+
+from modelscope.outputs import OutputKeys
+from modelscope.pipelines import pipeline
+from modelscope.utils.constant import Tasks
+from modelscope.utils.demo_utils import DemoCompatibilityCheck
+from modelscope.utils.test_utils import test_level
+
+
+class FmFaceRecognitionTest(unittest.TestCase, DemoCompatibilityCheck):
+
+    def setUp(self) -> None:
+        self.task = Tasks.face_recognition
+        self.model_id = 'damo/cv_manual_face-recognition_frfm'
+
+    @unittest.skipUnless(test_level() >= 0, 'skip test in current test level')
+    def test_face_compare(self):
+        img1 = 'data/test/images/face_recognition_1.png'
+        img2 = 'data/test/images/face_recognition_2.png'
+
+        face_recognition = pipeline(
+            Tasks.face_recognition, model=self.model_id)
+        emb1 = face_recognition(img1)[OutputKeys.IMG_EMBEDDING]
+        emb2 = face_recognition(img2)[OutputKeys.IMG_EMBEDDING]
+        sim = np.dot(emb1[0], emb2[0])
+        print(f'Cos similarity={sim:.3f}, img1:{img1}  img2:{img2}')
+
+    @unittest.skipUnless(test_level() >= 0, 'skip test in current test level')
+    def test_demo_compatibility(self):
+        self.compatibility_check()
+
+
+if __name__ == '__main__':
+    unittest.main()