import base64 from io import BytesIO import requests import grpc # import BeiHangGrpc_pb2_grpc # import BeiHangGrpc_pb2 from . import BeiHangGrpc_pb2_grpc from . import BeiHangGrpc_pb2 requests.packages.urllib3.disable_warnings() # from apps.utils.tools import convert_to_base64 def analyse(ip: str, pic_url: str): """安全帽事件 Args: ip (_type_): _description_ pic_url (_type_): _description_ """ # print(pic_url) channel = grpc.insecure_channel(ip) # 监听频道 stub = BeiHangGrpc_pb2_grpc.Helmet_RegStub(channel) # 客户端使用Stub类发送请求,参数为频道,为了绑定链接 image_id = 3 # 本地图像输入 # image_path = "./onnx/image/single3.jpg" # with open(image_path, 'rb') as f: # image_base64 = str(base64.b64encode(f.read()), encoding='utf-8') # request = BeiHangGrpc_pb2.JinYu_Request(zzid=image_id,imgsbase64=image_base64) base64img = str(base64.b64encode(BytesIO(requests.get(url=pic_url, verify=False).content).read()), 'utf-8') # with open('face_test.txt', 'w') as f: # f.write(base64img) request = BeiHangGrpc_pb2.JinYu_Request(zzid=image_id, imgsbase64=base64img) # request = BeiHangGrpc_pb2.JinYu_Request(zzid=image_id, imgUrl=pic_url) response = stub.sendHelmet_Info(request) if response.helmetinfoList: for i in response.helmetinfoList: if i.head_helmet == 1: # 未戴安全帽事件成立 return True, response return False, response if __name__ == "__main__": path = 'http://10.99.5.24/evo-apigw/evo-oss/6ad010cf-ce45-11ec-9715-e4246c7d1635/20220826/1/dsf_3de82501-2521-11ed-884a-e4246c7d1635_7236037_7245601.jpg?token=399773e6-71f7-4202-b234-ff5aa8d2492c' base64img = str(base64.b64encode(BytesIO(requests.get(url=path, verify=False).content).read()), 'utf-8') with open('face_test.txt', 'w') as f: f.write(base64img) is_happend, res = analyse( ip='122.224.229.24:2000', pic_url='http://10.99.5.24/evo-apigw/evo-oss/6ad010cf-ce45-11ec-9715-e4246c7d1635/20220826/1/dsf_3de82501-2521-11ed-884a-e4246c7d1635_7236037_7245601.jpg?token=399773e6-71f7-4202-b234-ff5aa8d2492c') print(is_happend, res.helmetinfoList[0].head_helmet)