PaddleServing图像语义分割部署实践

import os
import numpy as np
import argparse
from PIL import Image as PILImage
from paddle_serving_client import Client
from paddle_serving_app.reader import Sequential, File2Image, Resize, CenterCrop
from paddle_serving_app.reader import RGB2BGR, Transpose, Div, Normalize


def get_color_map_list(num_classes, custom_color=None):
    num_classes += 1
    color_map = num_classes * [0, 0, 0]
    for i in range(0, num_classes):
        j = 0
        lab = i
        while lab:
            color_map[i * 3] |= (((lab >> 0) & 1) << (7 - j))
            color_map[i * 3 + 1] |= (((lab >> 1) & 1) << (7 - j))
            color_map[i * 3 + 2] |= (((lab >> 2) & 1) << (7 - j))
            j += 1
            lab >>= 3
    color_map = color_map[3:]

    if custom_color:
        color_map[:len(custom_color)] = custom_color
    return color_map
    
def get_pseudo_color_map(pred, color_map=None):
    pred_mask = PILImage.fromarray(pred.astype(np.uint8), mode='P')
    if color_map is None:
        color_map = get_color_map_list(256)
    pred_mask.putpalette(color_map)
    return pred_mask
    
def parse_args():
    parser = argparse.ArgumentParser(description='')
    parser.add_argument(
        "--serving_client_path",
        help="The path of serving_client file.",
        type=str,
        required=True)
    parser.add_argument(
        "--serving_ip_port",
        help="The serving ip.",
        type=str,
        default="127.0.0.1:9292",
        required=True)
    parser.add_argument(
        "--image_path", help="The image path.", type=str, required=True)
    return parser.parse_args()
    
def run(args):
    client = Client()
    client.load_client_config(
        os.path.join(args.serving_client_path, "serving_client_conf.prototxt"))
    client.connect([args.serving_ip_port])

    seq = Sequential([
        File2Image(), RGB2BGR(), Div(255),
        Normalize([0.5, 0.5, 0.5], [0.5, 0.5, 0.5], False), Transpose((2, 0, 1))
    ])

    img = seq(args.image_path)
    fetch_map = client.predict(
        feed={"x": img}, fetch=["save_infer_model/scale_0.tmp_1"])

    result = fetch_map["save_infer_model/scale_0.tmp_1"]
    color_img = get_pseudo_color_map(result[0])
    color_img.save("./result.png")
    print("The segmentation image is saved in ./result.png")
    
if name == 'main':
    args = parse_args()
    run(args)

python deploy/python/infer.py \
    --config data/pp-matting-hrnet_w18-human_512/deploy.yaml \
    --image_path data/PPM-100/val/fg/ \
    --save_dir output/results

python -m paddle_serving_client.convert \
    --dirname ./data/pp-matting-hrnet_w18-human_512 \
    --model_filename model.pdmodel \
    --params_filename model.pdiparams

import numpy as np
import cv2
from paddle_serving_app.reader import *
import base64
from paddle_serving_server.web_service import WebService, Op

class MattingOp(Op):
    '''
    定义抠图算子
    '''
    def init_op(self):
        '''
        初始化
        '''
        self.img_preprocess = Sequential([
            #BGR2RGB(), 
            Div(255.0),
            #Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225], False),
            Resize(512), Transpose((2, 0, 1))
        ])
        self.ref_size = 512
        self.img_width = self.ref_size
        self.img_height = self.ref_size   
 
    def preprocess(self, input_dicts, data_id, log_id):
        '''
        预处理
        '''
        (_, input_dict), = input_dicts.items()
        imgs = []
        for key in input_dict.keys():
            # 解码图像
            data = base64.b64decode(input_dict[key].encode('utf8'))
            data = np.fromstring(data, np.uint8)
            im = cv2.imdecode(data, cv2.IMREAD_COLOR)
            self.im = im
            self.img_height,self.img_width,_ = im.shape
            # 短边对齐512，长边设置为32整数倍(根据算法模型要求)
            im_h, im_w, _ = im.shape  
            if im_w >= im_h:
                im_rh = self.ref_size
                im_rw = int(im_w *1.0 / im_h * self.ref_size)
            elif im_w < im_h:
                im_rw = self.ref_size
                im_rh = int(im_h *1.0 / im_w * self.ref_size)      
            im_rw = im_rw - im_rw % 32
            im_rh = im_rh - im_rh % 32
            im = cv2.resize(im,(im_rw,im_rh))

            # cv2转tensor
            im = self.img_preprocess(im)
            imgs.append({
                "img": im[np.newaxis, :],
                # "im_shape":np.array(list(im.shape[1:])).reshape(-1)[np.newaxis, :],
                # "scale_factor": np.array([1.0, 1.0]).reshape(-1)[np.newaxis, :],
            })

        # 准备输入数据
        feed_dict = {
            "img": np.concatenate(
                [x["img"] for x in imgs], axis=0),
            # "im_shape": np.concatenate(
            #     [x["im_shape"] for x in imgs], axis=0),
            # "scale_factor": np.concatenate(
            #     [x["scale_factor"] for x in imgs], axis=0)
        }
        #for key in feed_dict.keys():
        #    print(key, feed_dict[key].shape)
        return feed_dict, False, None, ""
 
    def postprocess(self, input_dicts, fetch_dict, data_id, log_id):
        '''
        后处理
        '''
        # 取出掩码图
        alpha = fetch_dict["tmp_75"]
        alpha = alpha.squeeze(0).squeeze(0)
        alpha = (alpha * 255).astype('uint8')  
        alpha = cv2.resize(
                    alpha, (self.img_width, self.img_height), interpolation=cv2.INTER_NEAREST)
        alpha = alpha[:, :, np.newaxis]
        clip = np.concatenate([self.im, alpha], axis=-1)  
        print(clip.shape)
        _, buffer_img = cv2.imencode('.png', clip)  # 在内存中编码为png格式
        img64 = base64.b64encode(buffer_img)  
        img64 = str(img64, encoding='utf-8')  # bytes转换为str类型 

        #封装成字典返回
        res_dict = {
            "alpha":img64
        }
        return res_dict, None, ""
 
class MattingService(WebService):
    '''
    定义服务
    '''
    def get_pipeline_response(self, read_op):
        matting_op = MattingOp(name="matting", input_ops=[read_op])
        return matting_op
      
# 创建服务
matting_service = MattingService(name="matting")
# 加载配置文件
matting_service.prepare_pipeline_config("config.yml")
# 启动服务
matting_service.run_service()

import numpy as np
import requests
import json
import cv2
import base64


def cv2_to_base64(image):
    return base64.b64encode(image).decode('utf8')

# 定义http接口
url = "http://127.0.0.1:27008/matting/prediction"
# 打开待预测的图像文件
with open('./data/PPM-100/val/fg/1.jpg', 'rb') as file:
    image_data1 = file.read()

# 采用base64编码图像文件
image = cv2_to_base64(image_data1)
# 按照特定格式封装成字典
data = {"key": ["image"], "value": [image]}

# 发送请求
r = requests.post(url=url, data=json.dumps(data))

# 解析返回值
r = r.json()

# 解码返回的图像数据
img = r"value"
img = bytes(img, encoding='utf-8')  # str转bytes
img = base64.b64decode(img)  # base64解码
img = np.asarray(bytearray(img), dtype="uint8")
img = cv2.imdecode(img, cv2.IMREAD_UNCHANGED)

# 保存图像到本地
if img is None:
    print('call error')
else:
	cv2.imwrite('result.png',img)
	print('完成')