DSPRiverInspection/oss.py

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
from kafka import KafkaProducer, KafkaConsumer
##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'
    } 

def get_time(filename):
    #2021-10-09-11-44-51_frame-598-720_type-水生植被.jpg
    sps=filename.strip().split('_')[0]
    tsps=sps.split('-')
    return '%s-%s-%s %s:%s:%s'%(tsps[0],tsps[1],tsps[2],tsps[3],tsps[4],tsps[5])
def get_ms(time0,time1):
    str_time ='%.2f ms'%((time1-time0)*1000) 
    return str_time

def get_urls( platform_query_url,fp_log   ):
    try:
        if os.path.exists(platform_query_url):
            #print('###line49')
            with open('SendLog/platformQuery.json','r') as fp:
                res = json.load(fp)
        else:
            res = requests.get(platform_query_url,timeout=10).json()
        #print('###line54')    
        questionUrl = res['data']['questionUrl']  ###直播流时，问题图片的推送地址
        offlineUrl = res['data']['offlineUrl']  ###http离线视频时，问题图片的推送地址
    except Exception as ee:
        timestr=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        print('###### %s: file:send_transfer: error %s ,url:%s #####'%(timestr,ee,platform_query_url))
        outstr = '\n %s ###### get url platform error : update error:%s , url:%s'%(  time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) ,ee,platform_query_url)
        fp_log.write(outstr);fp_log.flush()
        questionUrl="http://47.96.182.154:9040/api/taskFile/submitUAVKHQuestion"
        offlineUrl ="http://47.96.182.154:9040/api/taskFile/submitUAVKHQuestion"
    return questionUrl,offlineUrl
def parse_filename(filename_base):
    #etc:2022-01-13-16-04-17_frame-823-1440_type-水生植被_hgYFEulc0dPIrG1S_s-off-XJRW20220113154959_AI.jpg
    uid =filename_base.split('.')[0].split('_')[3].strip()
    sourceType=filename_base.split('_')[4].split('-')[1]
    sourceId=filename_base.split('_')[4].split('-')[2]
    typename=filename_base.split('.')[0].split('_')[2].split('-')[1].strip()  
    return uid,sourceType,sourceId,typename
def b64encode_function(filename, filename_OR):
    if os.path.exists(filename):
        image_ori=cv2.imread(filename)
        image_ori_OR=cv2.imread(filename_OR)
    else:
        image_ori =  filename.copy()
        image_ori_OR = image_ori_OR.copy()
    image_pngcode = cv2.imencode('.jpg',image_ori)[-1]       
    image_pngcode_OR = cv2.imencode('.jpg',image_ori_OR)[-1]  
    image_code = str(base64.b64encode(image_pngcode))[2:-1]  
    image_code_OR = str(base64.b64encode(image_pngcode_OR))[2:-1]
    return  image_code, image_code_OR
def JsonSend(parIn):
   
    fp_log = parIn['fp_log']
    try:
        response=requests.post(parIn['api'],json=parIn['input_'],timeout=10).json()
        t3 = time.time()      
        print('\n file:%s encodetime:%.5f  request time:%.5f,send to %s ,return code:%s, size:%.2f M \n'%(parIn['filename_base'],parIn['t2']-parIn['t1'],t3-parIn['t2'],api,response['code'],parIn['sizeImage']))
        outstr = '%s file:%s encodetime:%.5f  request time:%.5f,send to %s ,return code:%s,size:%.2f M ,%s\n'%( time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),parIn['filename_base'],parIn['t2']-parIn['t1'],t3-parIn['t2'],parIn['api'],response['code'],parIn['sizeImage'],parIn['dic_str'])
        fp_log.write(outstr);fp_log.flush()
                   
    except Exception as ee:
        print('\n ######file:%s: upload error:%s,size:%.2f M'%(parIn['filename_base'],ee, parIn['sizeImage']))
        outstr = '\n%s ###### file:%s: upload error:%s , size:%.2f M'%(  time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) ,parIn['filename_base'],ee,parIn['sizeImage'])
        fp_log.write(outstr);fp_log.flush()
                   
        
def dic2str(dic):
    st=''
    for key in dic.keys():
        st='%s %s:%s,'%(st,key,dic[key])
    return st   
def createJsonInput(filename,offlineUrl,questionUrl):
    flag = True
    filename_base = os.path.basename(filename)             
    filename_OR=filename.replace('_AI.','_OR.')
    if not os.path.exists(filename_OR ):
        return False
              
    uid,sourceType, sourceId,typename = parse_filename(filename_base)            
    if (typename not in name_dic.keys())  or (typename == '排口'):
        return False
    api = questionUrl if sourceType=='live' else offlineUrl   
               
    time_str = get_time(filename_base)
    input_ ={
                    'imgName':os.path.basename(filename),
                    'imgNameOriginal':os.path.basename(filename_OR),
                    'time':time_str,
                    'fid':uid,  ###随机16位字符
                    'type':name_dic[typename],###这次先采用 ["排口","污口","水生植被","漂浮物","其它"]
                    'typeId':nameID_dic[typename]
                  
    }           
    if sourceType!='live':
        input_['code']=sourceId;###只有离线视频才需要code，
              
    dic_str = dic2str(input_) 
    t1 = time.time()
                
    image_code, image_code_OR = b64encode_function(filename, filename_OR) 
    input_['imgData']=image_code
    input_['imgDataOriginal']=image_code_OR
                 
    sizeImage = (len(image_code) + len(image_code_OR) )/1000000.0
                               
    parOut={};parOut['flag']=True;parOut['input_']=input_;
    parOut['sizeImage']=sizeImage;parOut['dic_str']=dic_str;  
    parOut['filename']=filename;parOut['filename_OR']=filename_OR;    
    parOut['api']=api  ;    parOut['t1']=t1  ;   parOut['filename_base']= filename_base           
    return  parOut  
    
def getLogFileFp(streamName):
    logname ='SendLog/'+ time.strftime("%Y-%m-%d", time.localtime())+'_%s.txt'%(streamName)
    if os.path.exists(logname):    
        fp_log = open(logname,'a+')    
    else:
        fp_log = open(logname,'w') 
    return                                                                 

def lodaMsgInfos(jsonDir,msgId):
    jsonUrl = os.path.join(jsonDir,msgId+'.json')
    with open(jsonUrl,'r') as fp:
        data=json.load(fp)
    return data   

def parse_filename_for_oss(name):
    splts=name.split('_')
    typename=splts[2].split('-')[1].strip()    
    msgId=splts[4].split('-')[3]
    onLineType=splts[4].split('-')[1]
    return typename,msgId,onLineType

msg_dict_off={
            "msg_id":"bblvgyntTsZCamqjuLArkiSYIbKXEeWx",#消息ID标识
	        "biz_id":"hehuzhang",#业务标识
	        "mod_id":"ai",#模型标识
	        "status":"running",#任务状态
            "type":str(1),#数据类型：1图片 2视频
	        "error":9999,#错误信息
	        "results":[#问题结果
		                {
			            "original_url":"",#原图地址
                        "sign_url":"",#AI标记地址
			            "category_id":"",#分类标识
			            "description":"",#问题描述
                        "time":"",#时间戳
		                }
	                  ]
            }

msg_dict_on={
            "msg_id":"bblvgyntTsZCamqjuLArkiSYIbKXEeWx",#消息ID标识
            "biz_id":"hehuzhang",#业务标识
            "mod_id":"qi",#模型标识
            "status":"running",#任务状态
            "type":str(2),#数据类型：1图片 2视频
            "error":9999,#错误信息
            "results":[#问题结果
                        {
                        "original_url":"",#原视频地址(离线识别时为空不传，实时识别时需要上传)
                        "sign_url":"",#识别后视频地址
                        }
                        ]
            }   

def update_json(jsonOri,jsonNew,offkeys=["msg_id","biz_id" ,"mod_id"  ]):
    #{'biz_id': 'hehuzhang', 'mod_id': 'ai', 'msg_id': 'bblvgyntTsZCamqjuLArkiSYIbKXEeWx', 'offering_id': 'http://vod.play.t-aaron.com/customerTrans/c49a2c620795d124f2ae4b10197b8d0e/303b7a58-17f3ef4494e-0004-f90c-f2c-7ec68.mp4', 'offering_type': 'mp4', 'results_base_dir': 'XJRW20220317153547', 'inSource': 'http://vod.play.t-aaron.com/customerTrans/c49a2c620795d124f2ae4b10197b8d0e/303b7a58-17f3ef4494e-0004-f90c-f2c-7ec68.mp4', 'outSource': 'NO'}
    for key in offkeys:
        jsonNew[key] = jsonOri[key]
    return    jsonNew     
def test5(indir,outdir,jsonDir,videoBakDir,ossPar,kafkaPar):    


    time0_0 = time.time()
    logname ='SendLog/'+ time.strftime("%Y-%m-%d.txt", time.localtime())
    if os.path.exists(logname):    
        fp_log = open(logname,'a+')    
    else:
        fp_log = open(logname,'w')   
    ifind=0
    time0_0 = time.time()

    producer = KafkaProducer(
                    bootstrap_servers=kafkaPar['boostServer'],#tencent yun 
                    value_serializer=lambda v: v.encode('utf-8'))
                    
        
    ###登陆准备存储桶
    auth = oss2.Auth(ossPar['AId'], ossPar['ASt'])
    # Endpoint以杭州为例，其它Region请按实际情况填写。
    bucket = oss2.Bucket(auth, ossPar['Epoint'], ossPar['bucketName'])
    

    while True:
        #filelist =  os.listdir(indir)
        filelist_AI = sorted(glob.glob('%s/*_AI.*'%(indir)),key=os.path.getmtime)
        filelist = filelist_AI
        
        if len(filelist)!=0:            
            time0 = time.time()
            for filename in filelist[0:2]: 
                filename_base = os.path.basename(filename)           
                ##解析文件名
                typename,msgId,onLineType = parse_filename_for_oss(filename_base)
                
                ##存储文件    
                filename_OR=filename.replace('_AI.','_OR.')
    
                
                
                if typename!='结束':
                    ObjectName_AI=os.path.join(ossPar['bucketName'],os.path.basename(filename))
                    ObjectName_OR=os.path.join(ossPar['bucketName'],os.path.basename(filename_OR))
                    bucket.put_object_from_file(ObjectName_AI, filename)
                    bucket.put_object_from_file(ObjectName_OR, filename_OR)
                    taskInfos = lodaMsgInfos(jsonDir,msgId)
                    #print(taskInfos)
                    ##发送返回信息
                    #if onLineType=='off':
                    msg = msg_dict_off  
                    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]['time']=  time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())                       
                    msg = update_json(taskInfos,msg)
                else:
                    msg = msg_dict_on  
                    videoList = sorted(glob.glob('%s/*'%(videoBakDir)),key=os.path.getmtime) 
                    videoName = os.path.basename(videoList[0])
                    msg["status"]="success";msg["msg_id"]=msgId
                    ObjectName_AI=os.path.join(ossPar['bucketName'],videoName)
                    bucket.put_object_from_file(ObjectName_AI, videoList[0])
                    msg['results'][0]['original_url']=  ObjectName_AI
                    msg['results'][0]['sign_url']=  ObjectName_AI###最新的视频文件   
                                        
                print('###'*3,'Send:',filename)
                msg = json.dumps(msg, ensure_ascii=False)    
                future = producer.send(
                            kafkaPar['topic'], 
                            msg
                        )
                print('***'*20,' Send transfer ',onLineType,msg)
               
                ##上传后的图片，移走到另外一个文件夹###
                cmd = 'mv \'%s\' \'%s\' '%(filename,outdir); os.system(cmd)
                cmd = 'mv \'%s\' \'%s\' '%(filename_OR,outdir); os.system(cmd)

        else:
            time.sleep(1)    
    fp_log.close()
  
if __name__=='__main__':
    indir='problems/images_tmp'
    outdir='problems/images_save'
    jsonDir = 'mintors/kafka/'
    videoBakDir='../../data/video_live_bak/1945'
    ossPar={'Epoint':'http://oss-cn-shanghai.aliyuncs.com',
            'AId':'LTAI5tSJ62TLMUb4SZuf285A',
            'ASt':'MWYynm30filZ7x0HqSHlU3pdLVNeI7',
            'bucketName':'ta-tech-image',
           }
    #kafkaPar={'boostServer':['212.129.223.66:9092'],'topic':'testReturn'}
    kafkaPar={'boostServer':['101.132.127.1:19092'],'topic':'alg-task-results'}
    test5(indir,outdir,jsonDir,videoBakDir,ossPar,kafkaPar)