python+grpc+yolo 目标检测实践

偏执的太偏执、 2022-10-12 15:55 315阅读 0赞

我们模型开发完成后往往需要基于一些web服务模块将模型部署成可被外部访问的服务形式，用的最多的就是flask框架了，可以很方便地将模型暴露成web服务接口，现在有一个新的需求就是需要使用grpc方式来开发接口，用于集群服务内部之间的相互访问调用。

gRPC有什么好处以及在什么场景下需要用gRPC
    既然是server/client模型，那么我们直接用restful api不是也可以满足吗，为什么还需要RPC呢？下面我们就来看看RPC到底有哪些优势
    
    gRPC vs. Restful API
    gRPC和restful API都提供了一套通信机制，用于server/client模型通信，而且它们都使用http作为底层的传输协议(严格地说, gRPC使用的http2.0，而restful api则不一定)。不过gRPC还是有些特有的优势，如下：
    1、gRPC可以通过protobuf来定义接口，从而可以有更加严格的接口约束条件。关于protobuf可以参见笔者之前的小文Google Protobuf简明教程
    2、另外，通过protobuf可以将数据序列化为二进制编码，这会大幅减少需要传输的数据量，从而大幅提高性能。
    3、gRPC可以方便地支持流式通信(理论上通过http2.0就可以使用streaming模式, 但是通常web服务的restful api似乎很少这么用，通常的流式数据应用如视频流，一般都会使用专门的协议如HLS，RTMP等，这些就不是我们通常web服务了，而是有专门的服务器应用。）

在此之前我听过rpc（Remote Procedure Call），即：远程过程调用，但是对于grpc还是知之甚少，所以这里还是需要花点时间来学习了解一下的，这里就以一个实际的目标检测类的应用来进行实践学习。

grpc的官网介绍在[这里][Link 1]。截图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvZ2V0aGVyX0Na_size_16_color_FFFFFF_t_70][]

一个基础的grpc应用问题主要包含四个部分：

1、编写 .proto文件

RPC 是两个子系统之间进行的直接消息交互，它使用操作系统提供的套接字来作为消息的载体，以特定的消息格式来定义消息内容和边界。gRPC 是 Google 开源的基于 Protobuf 和 Http2.0 协议的通信框架，Google 深度学习框架 tensorflow 底层的 RPC 通信就完全依赖于 gRPC 库。gRPC通过protobuf来定义接口和数据类型。

这里我们编写的 data.proto 内容如下：

syntax = "proto3";
    //package example;
     
     
    service FormatData {   //定义服务,用在rpc传输中
      rpc DoFormat(actionrequest) returns (actionresponse){}
    }
    message actionrequest {
      string text = 1;
    }
    message actionresponse{
      string text=1;
    }

2、运行命令生成 data\_pb2\_grpc.py 和data\_pb2.py

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./data.proto

上述命令执行结束后会生成上述两个脚本文件，内容如下：

data\_pb2.py

# -*- coding: utf-8 -*-
    # Generated by the protocol buffer compiler.  DO NOT EDIT!
    # source: data.proto
    """Generated protocol buffer code."""
    from google.protobuf import descriptor as _descriptor
    from google.protobuf import message as _message
    from google.protobuf import reflection as _reflection
    from google.protobuf import symbol_database as _symbol_database
    # @@protoc_insertion_point(imports)
    
    _sym_db = _symbol_database.Default()
    
    
    
    
    DESCRIPTOR = _descriptor.FileDescriptor(
      name='data.proto',
      package='',
      syntax='proto3',
      serialized_options=None,
      create_key=_descriptor._internal_create_key,
      serialized_pb='......'
    )
    
    
    
    
    _ACTIONREQUEST = _descriptor.Descriptor(
      name='actionrequest',
      full_name='actionrequest',
      filename=None,
      file=DESCRIPTOR,
      containing_type=None,
      create_key=_descriptor._internal_create_key,
      fields=[
        _descriptor.FieldDescriptor(
          name='text', full_name='actionrequest.text', index=0,
          number=1, type=9, cpp_type=9, label=1,
          has_default_value=False, default_value=b"".decode('utf-8'),
          message_type=None, enum_type=None, containing_type=None,
          is_extension=False, extension_scope=None,
          serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
      ],
      extensions=[
      ],
      nested_types=[],
      enum_types=[
      ],
      serialized_options=None,
      is_extendable=False,
      syntax='proto3',
      extension_ranges=[],
      oneofs=[
      ],
      serialized_start=14,
      serialized_end=43,
    )
    
    
    _ACTIONRESPONSE = _descriptor.Descriptor(
      name='actionresponse',
      full_name='actionresponse',
      filename=None,
      file=DESCRIPTOR,
      containing_type=None,
      create_key=_descriptor._internal_create_key,
      fields=[
        _descriptor.FieldDescriptor(
          name='text', full_name='actionresponse.text', index=0,
          number=1, type=9, cpp_type=9, label=1,
          has_default_value=False, default_value=b"".decode('utf-8'),
          message_type=None, enum_type=None, containing_type=None,
          is_extension=False, extension_scope=None,
          serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
      ],
      extensions=[
      ],
      nested_types=[],
      enum_types=[
      ],
      serialized_options=None,
      is_extendable=False,
      syntax='proto3',
      extension_ranges=[],
      oneofs=[
      ],
      serialized_start=45,
      serialized_end=75,
    )
    
    DESCRIPTOR.message_types_by_name['actionrequest'] = _ACTIONREQUEST
    DESCRIPTOR.message_types_by_name['actionresponse'] = _ACTIONRESPONSE
    _sym_db.RegisterFileDescriptor(DESCRIPTOR)
    
    actionrequest = _reflection.GeneratedProtocolMessageType('actionrequest', (_message.Message,), {
      'DESCRIPTOR' : _ACTIONREQUEST,
      '__module__' : 'data_pb2'
      # @@protoc_insertion_point(class_scope:actionrequest)
      })
    _sym_db.RegisterMessage(actionrequest)
    
    actionresponse = _reflection.GeneratedProtocolMessageType('actionresponse', (_message.Message,), {
      'DESCRIPTOR' : _ACTIONRESPONSE,
      '__module__' : 'data_pb2'
      # @@protoc_insertion_point(class_scope:actionresponse)
      })
    _sym_db.RegisterMessage(actionresponse)
    
    
    
    _FORMATDATA = _descriptor.ServiceDescriptor(
      name='FormatData',
      full_name='FormatData',
      file=DESCRIPTOR,
      index=0,
      serialized_options=None,
      create_key=_descriptor._internal_create_key,
      serialized_start=77,
      serialized_end=136,
      methods=[
      _descriptor.MethodDescriptor(
        name='DoFormat',
        full_name='FormatData.DoFormat',
        index=0,
        containing_service=None,
        input_type=_ACTIONREQUEST,
        output_type=_ACTIONRESPONSE,
        serialized_options=None,
        create_key=_descriptor._internal_create_key,
      ),
    ])
    _sym_db.RegisterServiceDescriptor(_FORMATDATA)
    
    DESCRIPTOR.services_by_name['FormatData'] = _FORMATDATA
    
    # @@protoc_insertion_point(module_scope)

data\_pb2\_grpc.py

# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
    """Client and server classes corresponding to protobuf-defined services."""
    import grpc
    
    import data_pb2 as data__pb2
    
    
    class FormatDataStub(object):
        """定义服务,用在rpc传输中
        """
    
        def __init__(self, channel):
            """Constructor.
    
            Args:
                channel: A grpc.Channel.
            """
            self.DoFormat = channel.unary_unary(
                    '/FormatData/DoFormat',
                    request_serializer=data__pb2.actionrequest.SerializeToString,
                    response_deserializer=data__pb2.actionresponse.FromString,
                    )
    
    
    class FormatDataServicer(object):
        """定义服务,用在rpc传输中
        """
    
        def DoFormat(self, request, context):
            """Missing associated documentation comment in .proto file."""
            context.set_code(grpc.StatusCode.UNIMPLEMENTED)
            context.set_details('Method not implemented!')
            raise NotImplementedError('Method not implemented!')
    
    
    def add_FormatDataServicer_to_server(servicer, server):
        rpc_method_handlers = {
                'DoFormat': grpc.unary_unary_rpc_method_handler(
                        servicer.DoFormat,
                        request_deserializer=data__pb2.actionrequest.FromString,
                        response_serializer=data__pb2.actionresponse.SerializeToString,
                ),
        }
        generic_handler = grpc.method_handlers_generic_handler(
                'FormatData', rpc_method_handlers)
        server.add_generic_rpc_handlers((generic_handler,))
    
    
     # This class is part of an EXPERIMENTAL API.
    class FormatData(object):
        """定义服务,用在rpc传输中
        """
    
        @staticmethod
        def DoFormat(request,
                target,
                options=(),
                channel_credentials=None,
                call_credentials=None,
                insecure=False,
                compression=None,
                wait_for_ready=None,
                timeout=None,
                metadata=None):
            return grpc.experimental.unary_unary(request, target, '/FormatData/DoFormat',
                data__pb2.actionrequest.SerializeToString,
                data__pb2.actionresponse.FromString,
                options, channel_credentials,
                insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

至此，第二部分的工作就结束了。

3、编写sever端代码， server.py

这里主要是编写服务端的处理代码。

#!usr/bin/env python
    #encoding:utf-8
    from __future__ import division
    
    '''
    __Author__:沂水寒城
    功能： grpc 服务端
    '''
    
    
    import io
    import os
    import cv2
    import grpc
    import time
    import json
    import base64
    import numpy as np
    from PIL import Image
    from concurrent import futures
    import data_pb2, data_pb2_grpc
    from detect import *
    
    
    
    _ONE_DAY_IN_SECONDS = 60 * 60 * 24
    _HOST = 'localhost'
    _PORT = '8080'
    
    
    
    
    class FormatData(data_pb2_grpc.FormatDataServicer):
        '''
        重写接口函数
        '''
        def DoFormat(self, request, context):
            content = request.text
            decode_img = base64.b64decode(content)
            image = io.BytesIO(decode_img)
            img = Image.open(image)
            detect_res=detectImg('name',img)
            detect_res=str(detect_res)
            return data_pb2.actionresponse(text=detect_res)
     
     
    def serve():
        '''
        服务端处理计算
        '''
        grpcServer = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
        data_pb2_grpc.add_FormatDataServicer_to_server(FormatData(), grpcServer)
        grpcServer.add_insecure_port(_HOST + ':' + _PORT)
        grpcServer.start()
        try:
            while True:
                print('=================================start=================================')
                time.sleep(_ONE_DAY_IN_SECONDS)
        except KeyboardInterrupt:
            grpcServer.stop(0)
    
            
     
     
    if __name__ == '__main__':
        serve()

4、编写客户端代码，client.py

这里主要是实现客户端的请求代码：

#!usr/bin/env python
    #encoding:utf-8
    from __future__ import division
    
    
    '''
    __Author__:沂水寒城
    功能： grpc 客户端
    '''
    
    
    import os
    import cv2
    import json
    import time
    import grpc
    import base64
    import numpy as np
    from PIL import Image
    import data_pb2, data_pb2_grpc
    
    
    
    _HOST = 'localhost'
    _PORT = '8080'
    
    
    
     
    def run():
        connection = grpc.insecure_channel(_HOST + ':' + _PORT)
        print('connection: ', connection)
        client = data_pb2_grpc.FormatDataStub(channel=connection)
        print('client: ', client)
        string = base64.b64encode(img)
        response = client.DoFormat(data_pb2.actionrequest(text=string))
        print("response: " + response.text)
     
     
    if __name__ == '__main__':
        run()

到这里，一个基础的grpc应用已经开发完成，可以进行实际的测试使用了，首先在终端启动服务端，之后执行客户端的请求操作，查看结果输出。

服务端启动后输出如下：

![20210705154426252.png][]

客户端启动后输出如下：

![20210705154459366.png][]

我们这里是借助于yolov3实现的目标检测服务，原图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvZ2V0aGVyX0Na_size_16_color_FFFFFF_t_70 1][]

检测结果如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvZ2V0aGVyX0Na_size_16_color_FFFFFF_t_70 2][]

可以看到已经正常计算出来的结果。

[Link 1]: https://grpc.io/docs/what-is-grpc/introduction/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvZ2V0aGVyX0Na_size_16_color_FFFFFF_t_70]: /images/20221005/047d6d89bbee4b009cd54e5926896501.png
[20210705154426252.png]: /images/20221005/c8dded8b6fdb4a5cb11114a9cb29a4ff.png
[20210705154459366.png]: /images/20221005/d00ee8c28f1f4a1fb976000a1d24ad19.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvZ2V0aGVyX0Na_size_16_color_FFFFFF_t_70 1]: /images/20221005/309f9fab4f32444e9434123f7ca4fbcc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvZ2V0aGVyX0Na_size_16_color_FFFFFF_t_70 2]: /images/20221005/b1d717d788bb4f578e2ec1f7f3d3f5f4.png