98 lines
3.8 KiB
Python
98 lines
3.8 KiB
Python
|
|
|
|
import base64
|
|
from io import BytesIO
|
|
import os
|
|
import requests
|
|
import grpc
|
|
|
|
from apps.ecm.service import save_dahua_pic
|
|
from . import BeiHangGrpc_pb2_grpc
|
|
from . import BeiHangGrpc_pb2
|
|
import uuid
|
|
from django.conf import settings
|
|
import cv2
|
|
requests.packages.urllib3.disable_warnings()
|
|
|
|
# 本地图像输入
|
|
# 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, imgUrl=pic_url)
|
|
|
|
|
|
def helmet(ip: str, pic_url: str):
|
|
"""安全帽事件(抠图)
|
|
|
|
Args:
|
|
ip (_type_): _description_
|
|
pic_url (_type_): _description_
|
|
"""
|
|
channel = grpc.insecure_channel(ip+':2000') # 监听频道
|
|
stub = BeiHangGrpc_pb2_grpc.Helmet_RegStub(channel) # 客户端使用Stub类发送请求,参数为频道,为了绑定链接
|
|
image_id = 3
|
|
img_path = save_dahua_pic(pic_url)
|
|
request = BeiHangGrpc_pb2.JinYu_Request(zzid=image_id, imgUrl=settings.BASE_URL_OUT + img_path)
|
|
response = stub.sendHelmet_Info(request)
|
|
if response.helmetinfoList:
|
|
for i in response.helmetinfoList:
|
|
if i.head_helmet == 1: # 未戴安全帽事件成立
|
|
return True, response
|
|
return False, response
|
|
|
|
|
|
def helmet2(ip: str, pic_url: str):
|
|
"""安全帽事件(全图)
|
|
|
|
Args:
|
|
ip (_type_): _description_
|
|
pic_url (_type_): _description_
|
|
"""
|
|
channel = grpc.insecure_channel(ip+':2002') # 监听频道
|
|
stub = BeiHangGrpc_pb2_grpc.Helmet_RegStub(channel) # 客户端使用Stub类发送请求,参数为频道,为了绑定链接
|
|
image_id = 3
|
|
img_path = save_dahua_pic(pic_url)
|
|
request = BeiHangGrpc_pb2.JinYu_Request(zzid=image_id, imgUrl=settings.BASE_URL_OUT + img_path)
|
|
response = stub.sendHelmet_Info(request)
|
|
os.remove(settings.BASE_DIR + img_path)
|
|
if response.helmetinfoList:
|
|
for i in response.helmetinfoList:
|
|
if i.head_helmet == 1: # 未戴安全帽事件成立
|
|
return True, response
|
|
return False, response
|
|
|
|
|
|
def fire1(ip: str, pic_url: str):
|
|
# 未配置灭火器事件
|
|
channel = grpc.insecure_channel(ip+':2001')
|
|
stub = BeiHangGrpc_pb2_grpc.Fire_RegStub(channel)
|
|
image_id = 3
|
|
img_path = save_dahua_pic(pic_url)
|
|
# img = cv2.imread(img_path)
|
|
# resized_img = cv2.resize(img, None, fx=0.25, fy=0.25)
|
|
# resized_img = cv2.imencode('.png', resized_img)[1]
|
|
# image_base64 = str(base64.b64encode(resized_img))[2:-1]
|
|
request = BeiHangGrpc_pb2.JinYu_Request(zzid=image_id, imgUrl=settings.BASE_URL_OUT + img_path)
|
|
response = stub.sendFire_Info(request)
|
|
os.remove(settings.BASE_DIR + img_path) # 删除临时图片
|
|
if response.FireinfoList:
|
|
for i in response.FireinfoList:
|
|
if i.fire == 0: # 配置了灭火器
|
|
return False, response
|
|
return True, 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)
|