Server-py/sri-pysdk/xplugin/hydra/plugin.py

import subprocess


class Plugin:
    def __init__(self):
        model_path = '/home/server/resources/DockerDependencies/2022/target/AlgorithmModels/FaceRecognition-arcface/'
        model_path += 'arcface_r100_v1.onnx'
        # model_path += 'MFR_glintr100.onnx'

    @staticmethod
    def run_hydra():
        """执行命令"""
        target = 'ssh://58.34.94.176:22'
        username = 'server'
        password = 'server@2000'

        command = [
            'hydra',
            '-l', username,
            '-p', password,
            target,
        ]

        try:
            result = subprocess.run(command, capture_output=True, text=True, check=True)
            return result.stdout
        except subprocess.CalledProcessError as e:
            return e.stderr

    def get_face_features_normalization_by_image_array(self, image_array):
        """

        :param image_array:
        :return:
        """
        pass

    @staticmethod
    def normalize(embedding):
        """特征数据格式化"""
        embedding_norm = norm(embedding)
        normed_embedding = embedding / embedding_norm
        return normed_embedding

    def get_face_features_normalization_by_image_array(self, image_array):
        """获取特征"""

        # --- debug ---
        # methods.debug_log('FaceRecognitionEngine', f"m-92: size: {image_array.shape}")

        # --- check todo 连乘得到像素值，旨在过滤低于112的尺寸图片 ---
        # if np.prod(image_array.shape) < np.prod((112, 112, 3)):
        #     return None

        # --- check ---
        if image_array is None:
            return None

        # --- check ---
        if image_array.shape != (112, 112, 3):
            # methods.debug_log('FaceRecognitionEngine', f"m-96: image resize before is {image_array.shape}")
            image_array = cv2.resize(image_array, (112, 112))

        if not isinstance(image_array, list):
            image_array = [image_array]

        for i, img in enumerate(image_array):
            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
            img = np.transpose(img, (2, 0, 1))
            image_array[i] = img.astype(np.float32)

        image_array = np.stack(image_array)
        net_out = self.onnx_session.run(self.outputs, {self.onnx_session.get_inputs()[0].name: image_array})
        return self.normalize(net_out[0][0])


if __name__ == '__main__':
    # --- init ---
    plugin = PLUGIN()

    output = plugin.run_hydra()
    print(output)