""" @Project ：microproduct @File ：DemMain.py @Function ：高程 @Author ：SHJ @Contact：k @Date ：2021/9/1 @Version ：1.0.0 """ import datetime import logging import os import shutil import sys import tarfile # import matplotlib as mpl #解决打包文件 from tool.algorithm.xml.AlgXmlHandle import CheckSource, ManageAlgXML from DemXmlInfo import CreateDict, CreateStadardXmlFile from tool.algorithm.xml.CreatMetafile import CreateMetafile from DemImageHandle import DemImageHandler from autorun import auto_run_main import geocoding as dem from logHandler import LogHandler from ConfigeHandle import Config as cf import sklearn.neighbors._typedefs import sklearn.neighbors._partition_nodes EXE_NAME = cf.get('exe_name') if cf.get('debug') == 'True': DEBUG = True else: DEBUG = False LogHandler.init_log_handler('run_log\\' + EXE_NAME) logger = logging.getLogger("mylog") env_str = os.path.split(os.path.realpath(__file__))[0] os.environ['PROJ_LIB'] = env_str class DemMain: """ 干涉形变主函数 """ def __init__(self, alg_xml_path): self.alg_xml_path = os.path.join(os.getcwd(), alg_xml_path) self.__alg_xml_handler = ManageAlgXML(alg_xml_path) self.__check_handler = CheckSource(self.__alg_xml_handler) self.imageHandler = DemImageHandler() self.__workspace_path = None self.__input_paras = {} self.__output_paras = {} self.__in_processing_paras = {} self.__out_para = None self.__tif_names_list = [] def check_source(self): """ 检查算法相关的配置文件，图像，辅助文件是否齐全 """ self.env_str = os.getcwd() logger.info("sysdir: %s", self.env_str) if self.__check_handler.check_alg_xml() is False: raise Exception('check_alg_xml() failed!') if self.__check_handler.check_run_env() is False: raise Exception('check_run_env() failed!') input_para_names = ["SARS", "DEM"] if self.__check_handler.check_input_paras(input_para_names) is False: raise Exception('check_input_paras() failed!') # 创建工作区 self.__workspace_path = self.__alg_xml_handler.get_workspace_path() self.__create_work_space() # 配置输入参数 self.__input_paras = self.__alg_xml_handler.get_input_paras() self.__in_processing_paras = self.__init_processing_paras(self.__input_paras) aux_path = os.path.join(self.__workspace_origin_path, "AuxDir") if os.path.exists(aux_path) is False: os.mkdir(aux_path) self.__in_processing_paras.update({'AuxDir': aux_path}) # 写入输出参数 self.__out_para = os.path.join(self.__workspace_path, EXE_NAME, 'Output', r"DemProduct.tar.gz") self.__alg_xml_handler.write_out_para("DemProduct", self.__out_para) logger.info('check_source success!') logger.info('progress bar: 5%') return True def __init_processing_paras(self, names): """ :param names:字典列表，每个字典为一个输入产品的配置信息 """ processing_paras = {} for name in names: para = self.__input_paras[name] if para is None: logger.error(name + "is None!") return False if name == 'SARS': if para['DataType'] == 'File': processing_paras.update({'slc': para['ParaValue']}) else: para_path_list = para['ParaValue'].split(";") if len(para_path_list) != 0: dem_path = os.path.join(self.__workspace_origin_path, para['ParaName']) if os.path.exists(dem_path) is False: os.mkdir(dem_path) for file_path in para_path_list: tif_name = os.path.basename(file_path) shutil.copy(file_path, os.path.join(dem_path, tif_name)) para_path = os.path.join(self.__workspace_origin_path, para['ParaName']) processing_paras.update({'slc': para_path}) if name == 'DEM': if para['DataType'] == 'File': processing_paras.update({'dem': para['ParaValue']}) else: para_path_list = para['ParaValue'].split(";") if len(para_path_list) != 0: dem_path = os.path.join(self.__workspace_origin_path, para['ParaName']) if os.path.exists(dem_path) is False: os.mkdir(dem_path) for file_path in para_path_list: tif_name = os.path.basename(file_path) shutil.copy(file_path, os.path.join(dem_path, tif_name)) para_path = os.path.join(self.__workspace_origin_path, para['ParaName']) processing_paras.update({'dem': para_path}) if name == 'Orbits': if para['DataType'] == 'File': processing_paras.update({'orbits': para['ParaValue']}) else: para_path_list = para['ParaValue'].split(";") if len(para_path_list) != 0: dem_path = os.path.join(self.__workspace_origin_path, para['ParaName']) if os.path.exists(dem_path) is False: os.mkdir(dem_path) for file_path in para_path_list: tif_name = os.path.basename(file_path) shutil.copy(file_path, os.path.join(dem_path, tif_name)) para_path = os.path.join(self.__workspace_origin_path, para['ParaName']) processing_paras.update({'orbits': para_path}) if name == 'MainImg': processing_paras.update({'mainimg': para['ParaValue']}) if name == 'box': datas = para['ParaValue'].split(';') if len(datas) != 4: msg = 'para: box is error!box:' + para['ParaValue'] raise Exception(msg) box = datas[0] + ' ' + datas[1] + ' ' + datas[2] + ' ' + datas[3] processing_paras.update({'box': box}) if name == 'AuxDir': if para['DataType'] == 'File': processing_paras.update({'AuxDir': para['ParaValue']}) else: para_path_list = para['ParaValue'].split(";") if len(para_path_list) != 0: para_path = os.path.dirname(para_path_list[0]) processing_paras.update({'AuxDir': para_path}) if name == 'NumConnections': processing_paras.update({'NumConnections': para['ParaValue']}) if name == 'EsdCoherenceThreshold': processing_paras.update({'EsdCoherenceThreshold': para['ParaValue']}) return processing_paras def verifyAndModifyWgsXml(self, xmlPath, demPath): import xml.dom.minidom as xmldom domobj = xmldom.parse(xmlPath) rootNode = domobj.documentElement # 获得子标签 propertyElementObj = rootNode.getElementsByTagName("property") for property in propertyElementObj: if property.hasAttribute("name"): if property.getAttribute("name") == "file_name": pathNode = property.getElementsByTagName("value")[0] pathInxml = pathNode.childNodes[0].data print('pathInxml1:', pathInxml) pathNode.childNodes[0].data = r"/".join(demPath.split("\\")) pathInxml = pathNode.childNodes[0].data print('pathInxml2:', pathInxml) with open(xmlPath, 'w') as f: # 缩进换行编码 domobj.writexml(f, addindent=' ', encoding='utf-8') def __create_work_space(self): """ 删除原有工作区文件夹,创建新工作区文件夹 """ self.__workspace_preprocessed_path = self.__workspace_path + EXE_NAME + r"\Temporary\preprocessed""\\" self.__workspace_processing_path = self.__workspace_path + EXE_NAME + r"\Temporary\processing""\\" self.__workspace_isce_path = os.path.join(self.__workspace_processing_path, 'isce_workspace') self.__workspace_dem_path = os.path.join(self.__workspace_preprocessed_path, 'dem') self.__product_dic = self.__workspace_processing_path + 'product\\' self.__workspace_origin_path = os.path.join(self.__workspace_path, EXE_NAME, "Temporary", "origin") path_list = [self.__workspace_preprocessed_path, self.__workspace_processing_path, self.__workspace_isce_path, self.__workspace_dem_path, self.__product_dic, self.__workspace_origin_path] for path in path_list: if os.path.exists(path): # if DEBUG is True: # continue self.del_floder(path) os.makedirs(path) else: os.makedirs(path) logger.info('create new workspace success!') def del_file(self, path_data): """ 只删除文件，不删除文件夹 """ if DEBUG is True: return for i in os.listdir(path_data): # os.listdir(path_data)#返回一个列表，里面是当前目录下面的所有东西的相对路径 file_data = path_data + "\\" + i # 当前文件夹的下面的所有东西的绝对路径 if os.path.isfile(file_data) is True: # os.path.isfile判断是否为文件,如果是文件,就删除.如果是文件夹.递归给del_file. os.remove(file_data) else: self.del_file(file_data) @staticmethod def del_floder(dic): """ 删除整个文件夹 """ if DEBUG is True: return if os.path.isdir(dic): shutil.rmtree(dic) @staticmethod def make_targz(output_filename, source_dir): """ 一次性打包整个根目录。空子目录会被打包。 如果只打包不压缩，将"w:gz"参数改为"w:"或"w"即可。 :param output_filename:输出压缩包的完整路径，eg:'E:\test.tar.gz' :param source_dir:需要打包的跟目录，eg: 'E:\testFfile\'打包文件夹里面的所有文件,'E:\testFfile'打包文件夹 """ dir = os.path.split(output_filename)[0] if os.path.exists(dir) is False: os.makedirs(dir) with tarfile.open(output_filename, "w:gz") as tar: tar.add(source_dir, arcname=os.path.basename(source_dir)) def del_temp_workspace(self): """ 临时工作区 """ if DEBUG is True: return path = self.__workspace_path + EXE_NAME + r"\Temporary" if os.path.exists(path): self.del_floder(path) def updateFile(self, out_file, in_file, old_str, new_str): """ 替换文件中的字符串 :param out_file:输出文件名 :param in_file:输入文件名 :param old_str:就字符串 :param new_str:新字符串 :return: """ file_data = "" with open(in_file, "r", encoding="utf-8") as f: for line in f: if old_str in line: line = line.replace(old_str, new_str) file_data += line with open(out_file, "w", encoding="utf-8") as f: f.write(file_data) def isce_stackSentinel(self, slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space, key_word, isce_exe_dir, box,EsdCoherenceThreshold): """执行 stackSentinel.exe""" os.chdir(isce_exe_dir) # cmd = "stackSentinel.exe -s {} -d {} -a {} -o {} -w {} -n {} -m {} -c {} --exeabsolute_dir {} -b {}".format(slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space,"'1 2 3'", key_word, '1', isce_exe_dir,box) # cmd = "stackSentinel.exe -s {} -d {} -a {} -o {} -w {} -n {} -m {} -W {} -c {} --exeabsolute_dir {} -b {}".format(slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space,"'1 2 3'", key_word,'interferogram','6', isce_exe_dir,box) cmd = "stackSentinel.exe -s {} -d {} -a {} -o {} -w {} -n '1 2 3' -m {} -W 'interferogram' -c '1' --exeabsolute_dir {} -e {} -b {}".format( slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space, key_word, isce_exe_dir,EsdCoherenceThreshold , box) # cmd = "stackSentinel.exe -s {} -d {} -a {} -o {} -w {} -c {} -b {} -m {} -n '1 2 3' -p 'vv' -W offset -e {} --exeabsolute_dir {}".format(slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space, cum_connections, box, main_img, EsdCoherenceThreshold, isce_exe_dir) # stackSentinel.exe -s I:/MicroProduct/daqiceshishuju_20221109/slc -d I:/MicroProduct/daqiceshishuju_20221109/DEM/demLat_N37_N40_Lon_E107_E112.dem.wgs84 -a I:/MicroProduct/daqiceshishuju_20221109/AuxDir -o I:/MicroProduct/daqiceshishuju_20221109/orbits -w I:/MicroProduct/daqiceshishuju_20221109/daqi_workspace/isce_workspace -n '1 2 3' -m 20190206 -W interferogram -c 6 --exeabsolute_dir I:/MicroProduct/microproduct/atmosphericDelay/ISCEApp -b '37.3 39.4 108.10 111.82' logger.info('stackSentinel_cmd:{}'.format(cmd)) result = os.system(cmd) logger.info('cmd_result:{}'.format(result)) logger.info('stackSentinel finish!') return result def process_handle(self,start): # 执行isce2.5生成干涉图 # 生成工作流配置文件 slc_dir = r"/".join(self.__in_processing_paras['slc'].split("\\")) + "/" dem_dir = r"/".join(self.__in_processing_paras['dem'].split("\\")) aux_dir = r"/".join(self.__in_processing_paras['AuxDir'].split("\\")) + "/" orbits_dir = r"/".join(self.__in_processing_paras['orbits'].split("\\")) isce_work_space = r"/".join(self.__workspace_isce_path.split("\\")) box = "'" + self.__in_processing_paras['box'] + "'" main_img = self.__in_processing_paras['mainimg'] isce_exe_dir = r"/".join(os.path.join(self.env_str, "ISCEApp").split("\\")) EsdCoherenceThreshold = self.__in_processing_paras['EsdCoherenceThreshold'] os.chdir(isce_exe_dir) ### 转换tif影像为 wgs84格式 dem_dir = '/cygdrive/' + dem_dir.replace(":/", "/") out_dem_dir = self.__workspace_dem_path out_dem_dir = '/cygdrive/' + out_dem_dir.replace(":\\", "/") cmd = "demhgt2wgs.exe --tif_path {} --hgt_path {} --ASTGTM2".format(dem_dir, out_dem_dir) logger.info('demhgt2wgs_cmd:{}'.format(cmd)) result = os.system(cmd) logger.info('cmd_result:{}'.format(result)) import glob in_tif_paths = list(glob.glob(os.path.join(self.__workspace_dem_path, '*.wgs84'))) if in_tif_paths == []: raise Exception('demhgt2wgs.exe run failed!') dem_path = r"/".join(in_tif_paths[0].split("\\")) logger.info('demhgt2wgs finish!') logger.info('progress bar: 5%') # todo 替换成大气的stackSentinel # cum_connections = self.__in_processing_paras['NumConnections'] # cmd = "stackSentinel.exe -s {} -d {} -a {} -o {} -w {} -c {} -b {} -m {} -n '1 2 3' -p 'vv' -W offset -e {} --exeabsolute_dir {}".format(slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space, cum_connections, box, main_img, EsdCoherenceThreshold, isce_exe_dir) result = self.isce_stackSentinel(slc_dir, dem_path, aux_dir, orbits_dir, isce_work_space, main_img, isce_exe_dir, box,EsdCoherenceThreshold) logger.info('stackSentinel_cmd:{}'.format(cmd)) #result = os.system(cmd) logger.info('cmd_result:{}'.format(result)) logger.info('stackSentinel finish!') logger.info('progress bar: 10%') cmd = ['-e', isce_exe_dir, '-o', isce_work_space] logger.info('autorun_cmd:{}'.format(cmd)) auto_run_main(cmd) # result = os.system(cmd) logger.info('cmd_result:{}'.format(result)) if result != 0: raise Exception('autorun.py run failed!') # 直接调用 # sys.argv.extend(['-e' + isce_exe_dir, '-o' + isce_work_space]) # logger.info('autorun_cmd:{}'.format(sys.argv)) # auto_run_main(sys.argv[1:]) logger.info('autorun_cmd success!') logger.info('progress bar: 90%') temp_path = self.__workspace_isce_path + "\\dem_temp" if not os.path.exists(temp_path): os.mkdir(temp_path) out_dir = os.path.join(self.__workspace_path, EXE_NAME, 'Output') # dem_product = os.path.join(self.__workspace_path, EXE_NAME, 'Output','dem_def.tiff') dem_product = os.path.join(self.__product_dic, 'dem_def.tiff') dem_proPath = dem.get_Dem(self.__workspace_isce_path, temp_path,out_dir, dem_product) # 生成tif self.imageHandler.write_quick_view(dem_product) # 生成快视图 # hh_flag, hv_flag, vh_flag, vv_flag = 0, 0, 0, 0 # polarization = self.imageHandler.get_polarization( # os.path.join(self.__workspace_isce_path, "reference", "IW1.xml")) # 极化方式 # if 'hh' in polarization or 'HH' in polarization: # hh_flag = 1 # elif 'hv' in polarization or 'HV' in polarization: # hv_flag = 1 # elif 'vh' in polarization or 'VH' in polarization: # vh_flag = 1 # elif 'vv' in polarization or 'VV' in polarization: # vv_flag = 1 # self.processinfo = [hh_flag, hv_flag, vh_flag, vv_flag] self.processinfo = [1, 1, 1, 1] xml_path = os.path.join(self.env_str, r"model_meta.xml") tem_folder = self.__workspace_path + EXE_NAME + r"\Temporary""\\" image_path = dem_proPath out_path1 = os.path.join(tem_folder, "trans_geo_projcs.tif") out_path2 = os.path.join(tem_folder, "trans_projcs_geo.tif") par_dict = CreateDict(image_path,self.processinfo, out_path1, out_path2).calu_nature(start) model_xml_path = os.path.join(tem_folder, "creat_standard.meta.xml") # 输出xml路径 CreateStadardXmlFile(xml_path, self.alg_xml_path, par_dict, model_xml_path).create_standard_xml() # 生成压缩包 meta_xml_path = self.__product_dic+ EXE_NAME + "Product.meta.xml" SrcImagePath = self.__input_paras["SARS"]['ParaValue'] paths = SrcImagePath.split(';') SrcImageName = os.path.split(paths[0])[1].split('.tar.gz')[0] if len(paths) >= 2: for i in range(1, len(paths)): SrcImageName = SrcImageName + ";" + os.path.split(paths[i])[1].split('.tar.gz')[0] self.image_meta_xml = None CreateMetafile(self.image_meta_xml, self.alg_xml_path, model_xml_path, meta_xml_path).process(SrcImageName) self.make_targz(self.__out_para, self.__product_dic) logger.info("write quick view and .tar.gz finish") logger.info('DEM production successful !') logger.info('progress bar: 100%') return True if __name__ == '__main__': start = datetime.datetime.now() try: if len(sys.argv) < 2: xml_path = r'Dem.xml' else: xml_path = sys.argv[1] Main = DemMain(xml_path) if not Main.check_source(): raise Exception('check_source() failed!') if not Main.process_handle(start): raise Exception('process_handle() failed!') logger.info('successful production of Deformation products!') except Exception: logger.exception("run-time error!") finally: Main.del_temp_workspace() # pass end = datetime.datetime.now() msg = 'running use time: %s ' % (end - start) logger.info(msg)