Method run

src/server/tasks/sam.py:36–62

(
        self, file: UploadFile, inference_params: Dict[str, Any]
    )

    @torch.no_grad()
    async def run(
        self, file: UploadFile, inference_params: Dict[str, Any]
    ) -> Dict[str, Any]:
        logger.info("Run SAMImageEncoder.")
        image = await file.read()

        # Preprocess
        input_image = self.preprocess(image)

        # Prepare for inference.
        triton_inputs = [grpcclient.InferInput("INPUT__0", input_image.shape, "FP32")]
        triton_inputs[0].set_data_from_numpy(input_image)
        triton_outputs = [grpcclient.InferRequestedOutput("OUTPUT__0")]

        # Run the inference.
        result = await self.triton_client.infer(
            inputs=triton_inputs,
            outputs=triton_outputs,
            **inference_params,
        )
        # image embedding: numpy.ndarray [B, 256, 64, 64]
        image_embedding = result.as_numpy("OUTPUT__0")

        # Postprocess
        outputs = self.postprocess(image_embedding)

        return outputs

    def preprocess(self, image_byte, target_size=1024) -> torch.Tensor:
        """