358 lines
16 KiB
Python
358 lines
16 KiB
Python
from PIL import Image
|
||
import numpy as np
|
||
import cv2
|
||
import base64
|
||
import io,os
|
||
import requests
|
||
import time,json
|
||
import string,random
|
||
import glob,string,sys
|
||
from multiprocessing import Process,Queue
|
||
import oss2,copy
|
||
from kafka import KafkaProducer, KafkaConsumer
|
||
from utilsK.sendUtils import *
|
||
|
||
from utilsK.masterUtils import create_logFile,wrtiteLog,writeELK_log,send_kafka
|
||
from voduploadsdk.UploadVideoRequest import UploadVideoRequest
|
||
from voduploadsdk.AliyunVodUtils import *
|
||
from voduploadsdk.AliyunVodUploader import AliyunVodUploader
|
||
import hashlib
|
||
from kafka.errors import kafka_errors
|
||
##for CeKanYuan
|
||
#10月21日，通过图像名称判断，是那个平台。方式不好。
|
||
#10月22日，改成访问固定的地址，从地址中读取，平台的名称与地址。每隔2分钟访问一次。
|
||
#3月18日，采用OSS阿里云存储桶
|
||
#platform_query_url='http://47.96.182.154:9051/api/suanfa/getPlatformInfo'
|
||
platform_query_url='SendLog/platformQuery.json'
|
||
api = 'http://121.40.249.52:9050/api/taskFile/submitUAVKHQuestion'
|
||
#api = 'http://47.98.157.120:9040/api/taskFile/submitUAVKHQuestion'
|
||
|
||
##这套名字，是联通的。
|
||
name_dic={
|
||
"排口":"入河、湖排口",
|
||
"排污口": "入河、湖排口",
|
||
"水生植被": "水生植物",
|
||
"漂浮物": "有大面积漂物",
|
||
"结束": "结束",
|
||
'其它' :'其它'
|
||
}
|
||
## for TH river
|
||
##这套代码是河长制度的。
|
||
'''
|
||
nameID_dic={
|
||
"排口":'00000',
|
||
"排污口": '8378',
|
||
"水生植被": '8380',
|
||
"漂浮物": '8368',
|
||
"结束":'9999',
|
||
"其它":'8888'
|
||
}
|
||
'''
|
||
|
||
msg_dict_off={
|
||
"request_id":"fflvgyntTsZCamqjuLArkiSYIbKXEeWx",#消息ID标识
|
||
"status":"running",#任务状态
|
||
"type":str(2),#消息类型 1：实时 2：离线
|
||
#"error":str(9999),#错误信息####
|
||
"error_code":"",#//错误编号
|
||
"error_msg":"",#//错误描述
|
||
"progress":"",
|
||
"results":[#问题结果
|
||
{
|
||
"original_url":"",#原图地址
|
||
"sign_url":"",#AI标记地址
|
||
"category_id":"",#分类标识
|
||
"description":"",#问题描述
|
||
"analyse_time":"",#时间戳
|
||
}
|
||
]
|
||
}
|
||
|
||
msg_dict_on={
|
||
"request_id":"nnlvgyntTsZCamqjuLArkiSYIbKXEeWx",#消息ID标识
|
||
"status":"running",#任务状态
|
||
"type":str(1),#消息类型 1：实时 2：离线
|
||
"error_code":"",#//错误编号
|
||
"error_msg":"",#//错误描述
|
||
"progressOn":"",
|
||
"results":[#问题结果
|
||
{
|
||
"original_url":"",#原视频地址(离线识别时为空不传，实时识别时需要上传)
|
||
"sign_url":"",#识别后视频地址
|
||
}
|
||
]
|
||
}
|
||
|
||
|
||
|
||
def mintor_offline_ending(parIn):
|
||
|
||
indir,server,topic,fp_log = parIn['indir'],parIn['server'],parIn['topic'] ,parIn['fp_log']
|
||
par_kafka={};par_kafka['server']=server;par_kafka['topic']=topic;
|
||
logger = parIn['logger'];thread='Send-tranfer-oss:mintor-offline-ending'
|
||
time_interval = parIn['timeInterval']
|
||
###轮询image_tmp的文件夹，每10s一次，一旦产生离线结束标志，则不停地发送没30秒heartbeat信号。
|
||
producer = KafkaProducer(
|
||
bootstrap_servers=par_kafka['server'],#tencent yun
|
||
value_serializer=lambda v: v.encode('utf-8'),
|
||
metadata_max_age_ms=120000,
|
||
)
|
||
outStrList={}
|
||
writeELK_log(msg='child processs starts',fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
time_ss0=time.time()
|
||
while True:
|
||
|
||
filelist_AI = sorted(glob.glob('%s/*_AI.txt'%(indir)),key=os.path.getmtime)
|
||
filelist = filelist_AI
|
||
|
||
off_msgs=[]
|
||
|
||
for filename in filelist[0:]:
|
||
filename_base = os.path.basename(filename)
|
||
##解析文件名
|
||
typename,requestId,onLineType = parse_filename_for_oss(filename_base)
|
||
if (onLineType=='off') and (typename=='结束'):
|
||
off_msgs.append(requestId)
|
||
|
||
for requestId in off_msgs:
|
||
msg_heart = copy.deepcopy(msg_dict_off)
|
||
msg_heart['status']='running'
|
||
msg_heart["request_id"]=requestId
|
||
msg_heart = json.dumps(msg_heart, ensure_ascii=False)
|
||
|
||
outStrList['success']= '----- send heartBeat in Transfer success, msg:%s '%(requestId)
|
||
outStrList['failure']='----- kafka error when sending heartBeat in Transfer '
|
||
outStrList['Refailure']='----- kafka error when Re-sending heartBeat in Transfer '
|
||
|
||
|
||
send_kafka(producer,par_kafka,msg_heart,outStrList,fp_log,line=sys._getframe().f_lineno,logger=logger,thread=thread );
|
||
|
||
time.sleep(time_interval)
|
||
time_ss1=time.time()
|
||
if time_ss1 - time_ss0>120:
|
||
outstrs = 'child process sleeping:%f s '%(time_ss1-time_ss0)
|
||
writeELK_log(msg=outstrs,fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
time_ss0=time_ss1
|
||
|
||
|
||
|
||
|
||
|
||
|
||
def test5(par):
|
||
|
||
indir,outdir,logdir,jsonDir = par['indir'],par['outdir'],par['logdir'],par['jsonDir']
|
||
hearBeatTimeMs = par['hearBeatTimeMs']
|
||
videoBakDir,ossPar,vodPar,kafkaPar = par['videoBakDir'], par['ossPar'],par['vodPar'],par['kafkaPar']
|
||
|
||
|
||
|
||
time0_0 = time.time();logname='SendPics.log';thread='Send-tranfer-oss:main'
|
||
fp_log=create_logFile(logdir=logdir,name=logname)
|
||
|
||
logger=logdir.replace('/','.')+'.'+logname
|
||
writeELK_log(msg='Send_tranfer_oss process starts',fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
nameID_dic=getNamedic(par['labelnamesFile'].strip())
|
||
|
||
parIn={};####给心跳信号发射的子进程参数
|
||
parIn['indir'],parIn['server'],parIn['topic'] ,parIn['fp_log']=indir,kafkaPar['boostServer'],kafkaPar['topic'],fp_log
|
||
parIn['timeInterval'] = hearBeatTimeMs;parIn['logger']=logger
|
||
HeartProcess=Process(target=mintor_offline_ending,name='process-sendHeartOnly',args=(parIn,))
|
||
HeartProcess.start()
|
||
|
||
|
||
ifind=0
|
||
time0_0 = time.time()
|
||
|
||
producer = KafkaProducer(
|
||
bootstrap_servers=kafkaPar['boostServer'],#tencent yun
|
||
value_serializer=lambda v: v.encode('utf-8'),
|
||
metadata_max_age_ms=120000,
|
||
)
|
||
|
||
|
||
|
||
|
||
###登陆准备存储桶
|
||
auth = oss2.Auth(ossPar['AId'], ossPar['ASt'])
|
||
# Endpoint以杭州为例，其它Region请按实际情况填写。
|
||
bucket = oss2.Bucket(auth, ossPar['Epoint'], ossPar['bucketName'])
|
||
|
||
##VOD
|
||
clt = init_vod_client(vodPar['AId'], vodPar['ASt'])
|
||
uploader = AliyunVodUploader(vodPar['AId'], vodPar['ASt'])
|
||
|
||
|
||
writeELK_log(msg='Load Parameter over',fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
par_heart={};outStrList={}
|
||
time_b0=time.time()
|
||
while True:
|
||
|
||
filelist_AI = sorted(glob.glob('%s/*_AI.txt'%(indir)),key=os.path.getmtime)
|
||
filelist=[]
|
||
for filename in filelist_AI:
|
||
filename_base = os.path.basename(filename)
|
||
typename,requestId,onLineType = parse_filename_for_oss(filename_base)
|
||
if typename in ["其它"]:
|
||
continue
|
||
filelist.append(filename)
|
||
|
||
if len(filelist)!=0:
|
||
time0 = time.time()
|
||
for filename in filelist[0:]:
|
||
filename_base = os.path.basename(filename)
|
||
##解析文件名
|
||
typename,requestId,onLineType = parse_filename_for_oss(filename_base)
|
||
|
||
##存储文件
|
||
filename_OR=filename.replace('_AI.','_OR.')
|
||
|
||
filename_AI_image = filename.replace('.txt','.jpg')
|
||
filename_OR_image = filename_OR.replace('.txt','.jpg')
|
||
|
||
taskInfos = lodaMsgInfos(jsonDir,requestId)
|
||
oss_dir = taskInfos['results_base_dir']
|
||
|
||
#if typename in ["排口","其它"]:
|
||
# continue
|
||
if typename not in ['结束','超时结束']:
|
||
|
||
time_s1 = time.time()
|
||
ObjectName_AI=os.path.join(oss_dir,os.path.basename(filename_AI_image))
|
||
ObjectName_OR=os.path.join(oss_dir,os.path.basename(filename_OR_image))
|
||
bucket.put_object_from_file(ObjectName_AI, filename_AI_image)
|
||
ret2=bucket.put_object_from_file(ObjectName_OR, filename_OR_image)
|
||
|
||
outstr=' oss bucket upload %s %s %s '%('***'*3,'Send:',filename)
|
||
#outstr=wrtiteLog(fp_log,outstr);print( outstr);
|
||
writeELK_log(msg=outstr,fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
|
||
msg = copy.deepcopy(msg_dict_off)
|
||
if onLineType!='off': msg['type']=str(1)
|
||
else: msg['type']=str(2)
|
||
msg['results'][0]['original_url']= ObjectName_OR
|
||
msg['results'][0]['sign_url']= ObjectName_AI
|
||
msg['results'][0]['category_id']= nameID_dic[typename]
|
||
msg['results'][0]['description']= typename
|
||
msg['results'][0]['analyse_time']= time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||
msg = update_json(taskInfos,msg)
|
||
time_s2 = time.time()
|
||
else:
|
||
time_s1 = time.time()
|
||
if onLineType!='off':
|
||
msg = copy.deepcopy(msg_dict_on)
|
||
msg["request_id"]=requestId ;msg['type']=str(1)
|
||
msg['results'][0]['original_url']= "yourAddess"
|
||
msg['results'][0]['sign_url']= "yourAddess"###最新的视频文件
|
||
upCnt=1;upLoaded=False
|
||
while upCnt<4:
|
||
try:
|
||
videoUrl= get_videoUurl(videoBakDir,requestId+'_AI.MP4')
|
||
assert os.path.exists(videoUrl) , '%s not exists'%(videoUrl)
|
||
uploadVideoRequest = UploadVideoRequest(videoUrl, 'offLineVideo')
|
||
videoId = uploader.uploadLocalVideo(uploadVideoRequest)
|
||
|
||
VideoId_AI=str(videoId['VideoId'])
|
||
|
||
videoUrl= get_videoUurl(videoBakDir,requestId+'_OR.MP4')
|
||
assert os.path.exists(videoUrl) , '%s not exists'%(videoUrl)
|
||
|
||
uploadVideoRequest = UploadVideoRequest(videoUrl, 'offLineVideo')
|
||
videoId = uploader.uploadLocalVideo(uploadVideoRequest)
|
||
VideoId_OR=str(videoId['VideoId'])
|
||
|
||
outstr=VideoId_OR+','+VideoId_AI
|
||
writeELK_log(msg=outstr,fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
msg['results'][0]['sign_url']=VideoId_AI
|
||
msg['results'][0]['original_url']=VideoId_OR
|
||
upCnt=4;upLoaded=True
|
||
except Exception as e:
|
||
writeELK_log(msg='video uploading error:%s, times:%d'%(e,upCnt),fp=fp_log,thread=thread,level='WARNING',line=sys._getframe().f_lineno,logger=logger);
|
||
upCnt+=1;upLoaded=False
|
||
if not upLoaded:
|
||
msg['error_msg']='video uploading failure' ; msg['error_code']='101' ;
|
||
|
||
|
||
else:
|
||
msg = copy.deepcopy(msg_dict_off)
|
||
msg['type']=str(2)
|
||
msg["request_id"]=requestId
|
||
msg['results'][0]['original_url']= taskInfos['original_url']
|
||
videoUrl= get_videoUurl(videoBakDir,requestId+'.MP4')
|
||
assert os.path.exists(videoUrl) , '%s not exists'%(videoUrl)
|
||
upCnt=1;upLoaded=False
|
||
while upCnt<4:
|
||
try:
|
||
uploadVideoRequest = UploadVideoRequest(videoUrl, 'offLineVideo')
|
||
videoId = uploader.uploadLocalVideo(uploadVideoRequest)
|
||
outstr=' oss upload video over %s '%(str(videoId['VideoId']))
|
||
writeELK_log(msg=outstr,fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
msg['results'][0]['sign_url']= str(videoId['VideoId'])###最新的视频文件
|
||
upCnt=4;upLoaded=True
|
||
except Exception as e:
|
||
writeELK_log(msg='video uploading error:%s, times:%d'%(e,upCnt),fp=fp_log,thread=thread,level='WARNING',line=sys._getframe().f_lineno,logger=logger);
|
||
upCnt+=1;upLoaded=False
|
||
if not upLoaded:
|
||
msg['error_msg']='video uploading failure' ; msg['error_code']='101' ;
|
||
if upLoaded:
|
||
if typename=='结束': msg["status"]="success"
|
||
else: msg["status"]="timeout"
|
||
else:
|
||
msg["status"]='failed'
|
||
time_s2 = time.time()
|
||
|
||
msg = json.dumps(msg, ensure_ascii=False)
|
||
future = producer.send(
|
||
kafkaPar['topic'],
|
||
msg
|
||
)
|
||
|
||
|
||
try:
|
||
record_metadata = future.get()
|
||
outstr='kafka send:%s msg:%s producer status:%s'%(onLineType,msg,producer.bootstrap_connected())
|
||
#outstr=wrtiteLog(fp_log,outstr);print( outstr);
|
||
writeELK_log(msg=outstr,fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
except Exception as e:
|
||
outstr='kafka ERROR:%s'%(str(e))
|
||
#outstr=wrtiteLog(fp_log,outstr);print( outstr);
|
||
writeELK_log(msg=outstr,fp=fp_log,thread=thread,level='WARNING',line=sys._getframe().f_lineno,logger=logger)
|
||
producer.close()
|
||
|
||
producer = KafkaProducer(
|
||
bootstrap_servers=kafkaPar['boostServer'],#tencent yun
|
||
value_serializer=lambda v: v.encode('utf-8')
|
||
)
|
||
try:
|
||
future = producer.send(kafkaPar['topic'], msg).get()
|
||
except Exception as e:
|
||
outstr='kafka resend ERROR:%s'%(str(e))
|
||
#poutstr=wrtiteLog(fp_log,outstr);print( outstr);
|
||
writeELK_log(msg=outstr,fp=fp_log,thread=thread,level='ERROR',line=sys._getframe().f_lineno,logger=logger)
|
||
|
||
|
||
time_s3 = time.time()
|
||
|
||
##上传后的图片，移走到另外一个文件夹###
|
||
cmd = 'mv \'%s\' \'%s\' '%(filename,outdir); os.system(cmd)
|
||
cmd = 'mv \'%s\' \'%s\' '%(filename_OR,outdir); os.system(cmd)
|
||
time_s4 = time.time()
|
||
print('-'*50)
|
||
|
||
else:
|
||
time.sleep(1)
|
||
time_b1=time.time()
|
||
if time_b1-time_b0>120:
|
||
writeELK_log(msg='send main process sleeping',fp=fp_log,thread=thread,level='INFO',line=sys._getframe().f_lineno,logger=logger)
|
||
time_b0=time_b1
|
||
fp_log.close()
|
||
|
||
if __name__=='__main__':
|
||
|
||
masterFile="conf/send_oss.json"
|
||
assert os.path.exists(masterFile)
|
||
with open(masterFile,'r') as fp:
|
||
par=json.load(fp)
|
||
|
||
test5(par)
|