web_service.py 4.6 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#!flask/bin/python
B
barrierye 已提交
15 16
# pylint: disable=doc-string-missing

17 18 19 20
from flask import Flask, request, abort
from multiprocessing import Pool, Process
from paddle_serving_server import OpMaker, OpSeqMaker, Server
from paddle_serving_client import Client
M
MRXLT 已提交
21 22
from contextlib import closing
import socket
23

B
barrierye 已提交
24

25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
class WebService(object):
    def __init__(self, name="default_service"):
        self.name = name

    def load_model_config(self, model_config):
        self.model_config = model_config

    def _launch_rpc_service(self):
        op_maker = OpMaker()
        read_op = op_maker.create('general_reader')
        general_infer_op = op_maker.create('general_infer')
        general_response_op = op_maker.create('general_response')
        op_seq_maker = OpSeqMaker()
        op_seq_maker.add_op(read_op)
        op_seq_maker.add_op(general_infer_op)
        op_seq_maker.add_op(general_response_op)
        server = Server()
        server.set_op_sequence(op_seq_maker.get_op_sequence())
        server.set_num_threads(16)
        server.load_model_config(self.model_config)
        server.prepare_server(
M
MRXLT 已提交
46
            workdir=self.workdir, port=self.port_list[0], device=self.device)
47 48
        server.run_server()

M
MRXLT 已提交
49 50 51 52 53 54 55 56 57
    def port_is_available(self, port):
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
            sock.settimeout(2)
            result = sock.connect_ex(('0.0.0.0', port))
        if result != 0:
            return True
        else:
            return False

58 59 60 61
    def prepare_server(self, workdir="", port=9393, device="cpu"):
        self.workdir = workdir
        self.port = port
        self.device = device
M
MRXLT 已提交
62
        default_port = 12000
M
MRXLT 已提交
63 64 65 66 67
        self.port_list = []
        for i in range(1000):
            if self.port_is_available(default_port + i):
                self.port_list.append(default_port + i)
                break
68 69

    def _launch_web_service(self):
M
MRXLT 已提交
70 71 72 73
        self.client = Client()
        self.client.load_client_config("{}/serving_server_conf.prototxt".format(
            self.model_config))
        self.client.connect(["0.0.0.0:{}".format(self.port_list[0])])
B
barrierye 已提交
74

D
dongdaxiang 已提交
75
    def get_prediction(self, request):
D
dongdaxiang 已提交
76 77 78 79 80
        if not request.json:
            abort(400)
        if "fetch" not in request.json:
            abort(400)
        try:
M
MRXLT 已提交
81 82
            feed, fetch = self.preprocess(request.json["feed"],
                                          request.json["fetch"])
B
barrierye 已提交
83 84
            if isinstance(feed, dict) and "fetch" in feed:
                del feed["fetch"]
M
MRXLT 已提交
85
            fetch_map = self.client.predict(feed=feed, fetch=fetch)
B
barrierye 已提交
86
            for key in fetch_map:
M
MRXLT 已提交
87
                fetch_map[key] = fetch_map[key].tolist()
M
MRXLT 已提交
88
            fetch_map = self.postprocess(
M
MRXLT 已提交
89
                feed=feed, fetch=fetch, fetch_map=fetch_map)
M
MRXLT 已提交
90
            result = {"result": fetch_map}
D
dongdaxiang 已提交
91 92 93
        except ValueError:
            result = {"result": "Request Value Error"}
        return result
94 95 96 97 98

    def run_server(self):
        import socket
        localIP = socket.gethostbyname(socket.gethostname())
        print("web service address:")
B
barrierye 已提交
99 100
        print("http://{}:{}/{}/prediction".format(localIP, self.port,
                                                  self.name))
101 102 103
        p_rpc = Process(target=self._launch_rpc_service)
        p_rpc.start()

M
MRXLT 已提交
104 105 106 107 108 109 110 111 112 113 114 115
        app_instance = Flask(__name__)

        @app_instance.before_first_request
        def init():
            self._launch_web_service()

        service_name = "/" + self.name + "/prediction"

        @app_instance.route(service_name, methods=["POST"])
        def run():
            return self.get_prediction(request)

M
MRXLT 已提交
116 117 118
        self.app_instance = app_instance

    def run_flask(self):
M
fix bug  
MRXLT 已提交
119 120 121 122
        self.app_instance.run(host="0.0.0.0",
                              port=self.port,
                              threaded=False,
                              processes=1)
M
MRXLT 已提交
123 124 125

    def get_app_instance(self):
        return self.app_instance
M
MRXLT 已提交
126

M
MRXLT 已提交
127
    def preprocess(self, feed=[], fetch=[]):
128 129
        return feed, fetch

M
MRXLT 已提交
130
    def postprocess(self, feed=[], fetch=[], fetch_map=None):
131
        return fetch_map