用 python+grpc+yolo 进行目标检测

vlambda
2021-07-15

用 python+grpc+yolo 进行目标检测

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

gRPC有什么好处以及在什么场景下需要用gRPC既然是server/client模型，那么我们直接用restful api不是也可以满足吗，为什么还需要RPC呢？下面我们就来看看RPC到底有哪些优势 gRPC vs. Restful APIgRPC和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的官网介绍截图如下：

用 python+grpc+yolo 进行目标检测

一个基础的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 _descriptorfrom google.protobuf import message as _messagefrom google.protobuf import reflection as _reflectionfrom 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'] = _ACTIONREQUESTDESCRIPTOR.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-8from __future__ import division '''__Author__:沂水寒城功能： grpc 服务端'''  import ioimport osimport cv2import grpcimport timeimport jsonimport base64import numpy as npfrom PIL import Imagefrom concurrent import futuresimport data_pb2, data_pb2_grpcfrom 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-8from __future__ import division import osimport cv2import jsonimport timeimport grpcimport base64import numpy as npfrom PIL import Imageimport 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()