from xml.dom import minidom from osgeo import gdal from osgeo import ogr from osgeo import gdalconst import h5py from PIL import Image import numpy as np import tarfile import zipfile import re import os import io import sys def exe_path(): """ [获取exe目录] Returns: [str]: [exe目录] """ if hasattr(sys, 'frozen'): # Handles PyInstaller return os.path.dirname(sys.executable) return os.path.dirname(os.path.realpath(__file__)) os.environ['PROJ_LIB'] = exe_path() + "/PROJ" def uint16to8(bands, lower_percent=0.001, higher_percent=99.999): """ 拉伸图像：图片16位转8位 :param bands: 输入栅格数据 :param lower_percent: 最低百分比 :param higher_percent: 最高百分比 :return: """ out = np.zeros_like(bands, dtype=np.uint8) n = bands.shape[0] for i in range(n): a = 0 # np.min(band) b = 255 # np.max(band) c = np.percentile(bands[i, :, :], lower_percent) d = np.percentile(bands[i, :, :], higher_percent) t = a + (bands[i, :, :] - c) * (b - a) / (d - c) t[t < a] = a t[t > b] = b out[i, :, :] = t return out def createXML(metadata, xlm_file): """ 创建xlm文件并写入字典 :param metadata: 元数据信息 :param xlm_file: xlm文件 :return: """ # 创建一个空的文档 document = minidom.Document() # 创建DOM文档对象 # 创建一个根节点对象 root = document.createElement('ProductMetaData') # 设置根节点的属性 # root.setAttribute('', '') # 将根节点添加到文档对象中 document.appendChild(root) # 字典转xml for key in metadata: # 创建父节点 node_name = document.createElement(key) # 给父节点设置文本 node_name.appendChild(document.createTextNode(str(metadata[key]))) # 将各父节点添加到根节点 root.appendChild(node_name) # 写入xlm文档 with open(xlm_file, 'w', encoding='utf-8') as f: document.writexml(f, indent='\t', newl='\n', addindent='\t', encoding='utf-8') f.close() def GetGFPMSData(in_file, xml_path, thumbnail_ath): """ 获取高分 PMS卫星元数据 :param thumbnail_ath: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) with tarfile.open(in_file, mode='r') as tar_file: extensions = ('MSS2_thumb.jpg', 'PAN2_thumb.jpg', 'MSS2.xml', 'PAN2.xml') file_list = [file for file in tar_file.getnames() if file.endswith(extensions)] # 解压多光谱缩略图 if file_list[1].endswith('MSS2_thumb.jpg'): tar_file.extract(file_list[1], thumbnail_ath) ThumbnailPath_MSS = in_path + "/" + file_list[1] ThumbnailName_MSS = file_list[1] else: return {"code": -1, "msg": "找不到指定文件..."} # 解压全色缩略图 if file_list[3].endswith("PAN2_thumb.jpg"): tar_file.extract(file_list[3], thumbnail_ath) ThumbnailPath_PAN = thumbnail_ath + "/" + file_list[3] ThumbnailName_PAN = file_list[3] else: return {"code": -1, "msg": "找不到指定文件..."} # 解压多光谱XML文件 if file_list[0].endswith('MSS2.xml'): # 解压XML文件 tar_file.extract(file_list[0], xml_path) xmlPath = xml_path + "/" + file_list[0] xmlFileName = file_list[0] # 获取文件流 meta_file = tar_file.extractfile(file_list[0]) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 # WidthInPixels = dom.getElementsByTagName('WidthInPixels')[0].firstChild.data # HeightInPixels = dom.getElementsByTagName('HeightInPixels')[0].firstChild.data ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 # 边界几何 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建多光谱字典 gf_mss_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, # "WidthInPixels": WidthInPixels, # "HeightInPixels": HeightInPixels, "ProjectedCoordinates": "", "CollectionCode": "", "ThumbnailPath": ThumbnailPath_MSS, "ThumbnailName": ThumbnailName_MSS, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} else: return {"code": -1, "msg": "找不到指定文件..."} # 解压全色XML文件 if file_list[2].endswith('PAN2.xml'): # 解压XML文件 tar_file.extract(file_list[2], xml_path) xmlPath = xml_path + "/" + file_list[2] xmlFileName = file_list[2] # 获取文件流 meta_file = tar_file.extractfile(file_list[2]) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 # WidthInPixels = dom.getElementsByTagName('WidthInPixels')[0].firstChild.data # HeightInPixels = dom.getElementsByTagName('HeightInPixels')[0].firstChild.data Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 # 边界几何 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建全色字典 gf_pan_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, # "WidthInPixels": WidthInPixels, # "HeightInPixels": HeightInPixels, "ProjectedCoordinates": "", "CollectionCode": "", "ThumbnailPath": ThumbnailPath_PAN, "ThumbnailName": ThumbnailName_PAN, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} else: return {"code": -1, "msg": "找不到指定文件..."} # 关闭压缩文件 tar_file.close() if (not gf_mss_dict) or (not gf_pan_dict): return {"code": -1, "msg": "没有满足条件的数据字典..."} return gf_mss_dict, gf_pan_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetGF3MDJData(in_file, xml_path, thumbnail_path): """ 获取高分3号MDJ（GF-3 MDJ）卫星元数据 :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) with tarfile.open(in_file, mode='r') as tar_file: extensions = ('.thumb.jpg', 'meta.xml') file_list = [file for file in tar_file.getnames() if file.endswith(extensions)] # 解压缩略图 if file_list[0].endswith('.thumb.jpg'): tar_file.extract(file_list[0], thumbnail_path) ThumbnailPath = thumbnail_path + "/" + file_list[0] ThumbnailName = file_list[0] else: return {"code": -1, "msg": "找不到指定文件..."} # 解压XML文件 if file_list[1].endswith('meta.xml'): tar_file.extract(file_list[1], xml_path) xmlPath = xml_path + "/" + file_list[1] xmlFileName = file_list[1] # 获取文件流 meta_file = tar_file.extractfile(file_list[1]) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 CollectionCode = "GF3_MDJ" ProduceTime = dom.getElementsByTagName('productGentime')[0].firstChild.data StartTime = dom.getElementsByTagName("imagingTime")[0].getElementsByTagName("start")[0].firstChild.data EndTime = dom.getElementsByTagName("imagingTime")[0].getElementsByTagName("end")[0].firstChild.data # 其他信息 ImageGSD = dom.getElementsByTagName('NominalResolution')[0].firstChild.data # EarthModel = dom.getElementsByTagName('EarthModel')[0].firstChild.data ProjectedCoordinates = dom.getElementsByTagName('ProjectModel')[0].firstChild.data Bands = "1,2" # 经纬度 TopLeftLatitude = dom.getElementsByTagName("topLeft")[0].getElementsByTagName("latitude")[ 0].firstChild.data TopLeftLongitude = dom.getElementsByTagName("topLeft")[0].getElementsByTagName("longitude")[ 0].firstChild.data TopRightLatitude = dom.getElementsByTagName("topRight")[0].getElementsByTagName("latitude")[ 0].firstChild.data TopRightLongitude = dom.getElementsByTagName("topRight")[0].getElementsByTagName("longitude")[ 0].firstChild.data BottomLeftLatitude = dom.getElementsByTagName("bottomLeft")[0].getElementsByTagName("latitude")[ 0].firstChild.data BottomLeftLongitude = dom.getElementsByTagName("bottomLeft")[0].getElementsByTagName("longitude")[ 0].firstChild.data BottomRightLatitude = dom.getElementsByTagName("bottomRight")[0].getElementsByTagName("latitude")[ 0].firstChild.data BottomRightLongitude = dom.getElementsByTagName("bottomRight")[0].getElementsByTagName("longitude")[ 0].firstChild.data # 边界几何 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建字典 gf3_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": "", "boundaryGeomStr": boundaryGeomStr, "Bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": ProjectedCoordinates, 'CollectionCode': CollectionCode, "ThumbnailPath": ThumbnailPath, "ThumbnailName": ThumbnailName, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} else: return {"code": -1, "msg": "找不到指定文件..."} # 判断字典是否为空 if not gf3_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return gf3_dict except Exception as e: return {"code": -1, "msg": str(e)} def GetGF4PMIData(in_file, xml_path, thumbnail_path): """ 获取高分4号PMI（GF-4 PMI）卫星元数据 PMS（可见光、近红外 5个波段 50m）、IRS（中波红外 1个波段 400m） :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) with tarfile.open(in_file, mode='r') as tar_file: extensions = ('_thumb.jpg', '.xml') file_list = [file for file in tar_file.getnames() if file.endswith(extensions)] # 解压PMS缩略图 if file_list[2].endswith('_thumb.jpg') and file_list[2].startswith('GF4_PMS_'): tar_file.extract(file_list[2], thumbnail_path) ThumbnailPath_PMS = thumbnail_path + "/" + file_list[2] ThumbnailName_PMS = file_list[2] else: return {"code": -1, "msg": "找不到指定文件..."} # 解压IRS缩略图 if file_list[0].endswith('_thumb.jpg') and file_list[0].startswith('GF4_IRS_'): tar_file.extract(file_list[0], thumbnail_path) ThumbnailPath_IRS = thumbnail_path + "/" + file_list[0] ThumbnailName_IRS = file_list[0] else: return {"code": -1, "msg": "找不到指定文件..."} # 解压PMS XML文件 if file_list[3].endswith('.xml') and file_list[3].startswith('GF4_PMS_'): # 解压XML文件 tar_file.extract(file_list[3], xml_path) xmlPath = xml_path + "/" + file_list[3] xmlFileName = file_list[3] # 获取文件流 meta_file = tar_file.extractfile(file_list[3]) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 CollectionCode = "GF4_PMS" ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 # ProjectedCoordinates = dom.getElementsByTagName('MapProjection')[0].firstChild.data # 投影坐标系 ProjectedCoordinates = "" # 投影坐标系 Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建可见光近红外(PMS)字典 gf4_pms_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, "boundaryGeomStr": boundaryGeomStr, "Bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": ProjectedCoordinates, 'CollectionCode': CollectionCode, "ThumbnailName": ThumbnailName_PMS, "ThumbnailPath": ThumbnailPath_PMS, "xmlFileName": xmlFileName, "xmlPath": xmlPath, "DirectoryDepth": "month"} else: return {"code": -1, "msg": "找不到指定文件..."} # 解压IRS XML文件 if file_list[1].endswith('.xml') and file_list[1].startswith('GF4_IRS_'): # 解压XML文件 tar_file.extract(file_list[1], xml_path) xmlPath = xml_path + "/" + file_list[1] xmlFileName = file_list[1] # 获取文件流 meta_file = tar_file.extractfile(file_list[1]) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 CollectionCode = "GF4_IRS" ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 # ProjectedCoordinates = dom.getElementsByTagName('MapProjection')[0].firstChild.data # 投影坐标系 ProjectedCoordinates = "" # 投影坐标系 Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建中红外(IRS)字典 gf4_irs_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, "boundaryGeomStr": boundaryGeomStr, "Bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": ProjectedCoordinates, 'CollectionCode': CollectionCode, "ThumbnailName": ThumbnailName_IRS, "ThumbnailPath": ThumbnailPath_IRS, "xmlFileName": xmlFileName, "xmlPath": xmlPath, "DirectoryDepth": "month"} else: return {"code": -1, "msg": "找不到指定文件..."} # 关闭压缩文件 tar_file.close() # 判断字典是否为空 if (not gf4_pms_dict) or (not gf4_irs_dict): return {"code": -1, "msg": "没有满足条件的数据字典..."} return gf4_pms_dict, gf4_irs_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetH08Data(in_file, xml_path, thumbnail_path): """ 获取葵花8卫星元数据 :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) ThumbnailName = os.path.splitext(basename)[0] + "_thumb.jpg" ThumbnailPath = os.path.join(thumbnail_path, ThumbnailName) # 其他信息 with h5py.File(in_file, mode='r') as f: start_time = f['start_time'][0] end_time = f['end_time'][0] band_id = f['band_id'][:] bands = ','.join(str(i) for i in band_id) ImageGSD = '1000, 500, 2000' # 生成缩略图 gdal.SetConfigOption("GDAL_FILENAME_IS_UTF8", "YES") in_datasets = gdal.Open(in_file) meta_data = in_datasets.GetMetadata() # 取出子数据集 datasets = in_datasets.GetSubDatasets() red_data = gdal.Open(datasets[7][0]).ReadAsArray() gre_data = gdal.Open(datasets[6][0]).ReadAsArray() blu_data = gdal.Open(datasets[5][0]).ReadAsArray() img_data = np.array([red_data, gre_data, blu_data]) img_data = uint16to8(img_data) # Array转Image img_data2 = np.transpose(img_data, (1, 2, 0)) img_data2 = img_data2[:, :, ::-1] img = Image.fromarray(img_data2) # 压缩图片大小 if img_data.shape[1] > img_data.shape[2]: width = 512 height = int(width / img_data.shape[1] * img_data.shape[2]) else: height = 512 width = int(height / img_data.shape[1] * img_data.shape[2]) img.thumbnail((width, height)) img.save(ThumbnailPath, "PNG") # 释放内存 del in_datasets del img_data del img_data2 del img # 生成XML文件 xmlFileName = os.path.splitext(basename)[0] + ".xml" xmlPath = os.path.join(xml_path, xmlFileName) createXML(meta_data, xmlPath) # 产品日期 date_created = meta_data['date_created'] # band_number = meta_data['band_number'] # 经纬度 upper_left_latitude = meta_data['upper_left_latitude'] upper_left_longitude = int(meta_data['upper_left_longitude']) - 180 upper_right_latitude = meta_data['upper_left_latitude'] upper_right_longitude = 200 - 180 lower_right_latitude = -60 lower_right_longitude = 200 - 180 lower_left_latitude = -60 lower_left_longitude = str(int(meta_data['upper_left_longitude']) - 180) boundaryGeomStr = f'POLYGON(({upper_left_longitude} {upper_left_latitude},' \ f'{upper_right_longitude} {upper_right_latitude},' \ f'{lower_right_longitude} {lower_right_latitude},' \ f'{lower_left_longitude} {lower_left_latitude},' \ f'{upper_left_longitude} {upper_left_latitude}))' # 构建字典 himawari8_dict = {"ProduceTime": date_created, "StartTime": "", "EndTime": "", "CloudPercent": "", "boundaryGeomStr": boundaryGeomStr, "bands": bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": "", "CollectionCode": "", "ThumbnailPath": ThumbnailPath, "ThumbnailName": ThumbnailName, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "day"} # 判断字典是否为空 if not himawari8_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return himawari8_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetJPSSData(in_file, xml_path, thumbnail_path): """ 获取联合极轨卫星系统（JPSS-1）元数据：NOAA-20(Joint Polar Satellite System spacecraft) :param xml_path: :param thumbnail_path: :param in_file: :return: 元数据字典 """ try: # 生成缩略图 in_path, basename = os.path.split(in_file) ThumbnailName = os.path.splitext(basename)[0] + "_thumb.jpg" ThumbnailPath = os.path.join(thumbnail_path, ThumbnailName) gdal.SetConfigOption("GDAL_FILENAME_IS_UTF8", "YES") in_datasets = gdal.Open(in_file) meta_data = in_datasets.GetMetadata() # 取出子数据集 datasets = in_datasets.GetSubDatasets() red_data = gdal.Open(datasets[0][0]).ReadAsArray() nir_data = gdal.Open(datasets[3][0]).ReadAsArray() swir_data = gdal.Open(datasets[9][0]).ReadAsArray() img_data = np.array([red_data, nir_data, swir_data]) img_data = uint16to8(img_data) # Array转Image img_data2 = np.transpose(img_data, (1, 2, 0)) img_data2 = img_data2[:, :, ::-1] img = Image.fromarray(img_data2) # 压缩图片大小 if img_data.shape[1] > img_data.shape[2]: width = 512 height = int(width / img_data.shape[1] * img_data.shape[2]) else: height = 512 width = int(height / img_data.shape[1] * img_data.shape[2]) img.thumbnail((width, height)) img.save(ThumbnailPath, "PNG") # 释放内存 del in_datasets del img_data del img_data2 del img # 生成XML文件 xmlFileName = os.path.splitext(basename)[0] + ".xml" xmlPath = os.path.join(xml_path, xmlFileName) createXML(meta_data, xmlPath) # 产品日期 ProductionTime = meta_data['ProductionTime'] StartTime = meta_data['StartTime'] EndTime = meta_data['EndTime'] # 其他信息 ImageGSD = str(meta_data['LongName']).split(" ")[-1] Bands = str(meta_data['title']).split(" ")[1] # 中心经纬度 productUpperLeftLat = meta_data['NorthBoundingCoordinate'] # 左上纬度 productUpperLeftLong = meta_data['WestBoundingCoordinate'] # 左上经度 productUpperRightLat = meta_data['NorthBoundingCoordinate'] # 右上纬度 productUpperRightLong = meta_data['EastBoundingCoordinate'] # 右上经度 productLowerLeftLat = meta_data['SouthBoundingCoordinate'] # 左下纬度 productLowerLeftLong = meta_data['WestBoundingCoordinate'] # 左下经度 productLowerRightLat = meta_data['SouthBoundingCoordinate'] # 右下纬度 productLowerRightLong = meta_data['EastBoundingCoordinate'] # 右下纬度 # 边界几何 boundaryGeomStr = f'POLYGON(({productUpperLeftLong} {productUpperLeftLat},' \ f'{productUpperRightLong} {productUpperRightLat},' \ f'{productLowerRightLong} {productLowerRightLat},' \ f'{productLowerLeftLong} {productLowerLeftLat},' \ f'{productUpperLeftLong} {productUpperLeftLat}))' # 构建字典 jpss_dict = {"ProduceTime": ProductionTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": "", # "TopLeftLatitude": productUpperLeftLat, # "TopLeftLongitude": productUpperLeftLong, # "TopRightLatitude": productUpperRightLat, # "TopRightLongitude": productUpperRightLong, # "BottomLeftLatitude": productLowerLeftLat, # "BottomLeftLongitude": productLowerLeftLong, # "BottomRightLatitude": productLowerRightLat, # "BottomRightLongitude": productLowerRightLong, "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": "", "CollectionCode": "", "ThumbnailPath": ThumbnailPath, "ThumbnailName": ThumbnailName, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "day"} # 判断字典是否为空 if not jpss_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} print(jpss_dict) return jpss_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetSNPPData(in_file, xml_path, thumbnail_path): """ 获取Suomi National Polar-orbiting Partnership（SNPP）元数据 :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) # 生成缩略图 ThumbnailName = os.path.splitext(basename)[0] + "_thumb.jpg" ThumbnailPath = os.path.join(thumbnail_path, ThumbnailName) gdal.SetConfigOption("GDAL_FILENAME_IS_UTF8", "YES") in_datasets = gdal.Open(in_file) meta_data = in_datasets.GetMetadata() # 取出子数据集 datasets = in_datasets.GetSubDatasets() red_data = gdal.Open(datasets[0][0]).ReadAsArray() gre_data = gdal.Open(datasets[3][0]).ReadAsArray() blu_data = gdal.Open(datasets[9][0]).ReadAsArray() img_data = np.array([red_data, gre_data, blu_data]) img_data = uint16to8(img_data) # Array转Image img_data2 = np.transpose(img_data, (1, 2, 0)) img_data2 = img_data2[:, :, ::-1] img = Image.fromarray(img_data2) # 压缩图片大小 if img_data.shape[1] > img_data.shape[2]: width = 512 height = int(width / img_data.shape[1] * img_data.shape[2]) else: height = 512 width = int(height / img_data.shape[1] * img_data.shape[2]) img.thumbnail((width, height)) img.save(ThumbnailPath, "PNG") # 释放内存 del in_datasets del img_data del img_data2 del img # 生成XML文件 xmlFileName = os.path.splitext(basename)[0] + ".xml" xmlPath = os.path.join(xml_path, xmlFileName) createXML(meta_data, xmlPath) # 产品日期 ProductionTime = meta_data['ProductionTime'] StartTime = meta_data['StartTime'] EndTime = meta_data['EndTime'] # 其他信息 ImageGSD = str(meta_data['LongName']).split(" ")[-1][:-1] Bands = str(meta_data['title']) # 中心经纬度 productUpperLeftLat = meta_data['NorthBoundingCoordinate'] # 左上纬度 productUpperLeftLong = meta_data['WestBoundingCoordinate'] # 左上经度 productUpperRightLat = meta_data['NorthBoundingCoordinate'] # 右上纬度 productUpperRightLong = meta_data['EastBoundingCoordinate'] # 右上经度 productLowerLeftLat = meta_data['SouthBoundingCoordinate'] # 左下纬度 productLowerLeftLong = meta_data['WestBoundingCoordinate'] # 左下经度 productLowerRightLat = meta_data['SouthBoundingCoordinate'] # 右下纬度 productLowerRightLong = meta_data['EastBoundingCoordinate'] # 右下纬度 boundaryGeomStr = f'POLYGON(({productUpperLeftLong} {productUpperLeftLat},' \ f'{productUpperRightLong} {productUpperRightLat},' \ f'{productLowerRightLong} {productLowerRightLat},' \ f'{productLowerLeftLong} {productLowerLeftLat},' \ f'{productUpperLeftLong} {productUpperLeftLat}))' # 构建字典 snpp_dict = {"ProductionTime": ProductionTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": "", "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": "", "CollectionCode": "", "ThumbnailPath": ThumbnailPath, "ThumbnailName": ThumbnailName, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "day"} # 判断字典是否为空 if not snpp_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return snpp_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetSentinel1Data(in_file, xml_path, thumbnail_path): """ 获取哨兵1卫星元数据 :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) with zipfile.ZipFile(in_file, mode='r') as zip_file: xmlFileName = os.path.splitext(basename)[0] + ".xml" xmlPath = os.path.join(xml_path, xmlFileName) ThumbnailName = os.path.splitext(basename)[0] + "_thumb.jpg" ThumbnailPath = os.path.join(thumbnail_path, ThumbnailName) for member in zip_file.namelist(): if re.match(r'[0-9a-zA-Z\_]+.SAFE/annotation/s1a-iw-grd-vv[0-9a-z\-]+.xml', member): # 输出xml文件 meta_data = zip_file.read(member) with open(xmlPath, "wb") as fout: fout.write(meta_data) # 产品日期 meta_content = zip_file.open(member) dom = minidom.parse(meta_content) ProduceTime = dom.getElementsByTagName('qualityInformation')[ 0].getElementsByTagName('qualityDataList')[ 0].getElementsByTagName('qualityData')[ 0].getElementsByTagName('azimuthTime')[ 0].firstChild.data StartTime = dom.getElementsByTagName('adsHeader')[0].getElementsByTagName('startTime')[ 0].firstChild.data StopTime = dom.getElementsByTagName('adsHeader')[0].getElementsByTagName('stopTime')[ 0].firstChild.data elif re.match(r'[0-9a-zA-Z\_]+.SAFE/preview/map-overlay.kml', member): # 读取其他信息 meta_content = zip_file.open(member) dom = minidom.parse(meta_content) coordinates = dom.getElementsByTagName('coordinates')[0].firstChild.data # 经纬度 lon_lat = re.split(r'\s', coordinates) TopLeftLatitude = re.split(r'\,', lon_lat[0])[1] # 左上纬度 TopLeftLongitude = re.split(r'\,', lon_lat[0])[0] # 左上经度 TopRightLatitude = re.split(r'\,', lon_lat[1])[1] # 右上纬度 TopRightLongitude = re.split(r'\,', lon_lat[1])[0] # 右上经度 BottomRightLatitude = re.split(r'\,', lon_lat[2])[1] # 右下纬度 BottomRightLongitude = re.split(r'\,', lon_lat[2])[0] # 右下经度 BottomLeftLatitude = re.split(r'\,', lon_lat[3])[1] # 左下纬度 BottomLeftLongitude = re.split(r'\,', lon_lat[3])[0] # 左下经度 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' elif re.match(r'[0-9a-zA-Z\_]+.SAFE/preview/quick-look.png', member): # 输出缩略图 thumb_data = zip_file.read(member) with open(ThumbnailPath, "wb") as fout: fout.write(thumb_data) else: continue # 生成字典 S1_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "StopTime": StopTime, "CloudPercent": "", "boundaryGeomStr": boundaryGeomStr, "bands": "Amplitude_VH,Intensity_VH,Amplitude_VV,Intensity_VV", # "NumberBands": "", "ImageGSD": "10", "ProjectedCoordinates": '', "CollectionCode": '', "ThumbnailName": ThumbnailName, "ThumbnailPath": ThumbnailPath, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} zip_file.close() if not S1_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return S1_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetSentinel2Data(in_file, xml_path, thumbnail_path): """ 获取哨兵2卫星元数据 :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) with zipfile.ZipFile(in_file, 'r', zipfile.ZIP_DEFLATED) as zip_file: extensions = ('_B02_60m.jp2', '_B03_60m.jp2', '_B04_60m.jp2', '.SAFE/MTD_MSIL2A.xml') file_list = [file for file in zip_file.namelist() if file.endswith(extensions)] file_list.sort() # 生成缩略图 ThumbnailName = os.path.splitext(basename)[0] + "_thumb.jpg" ThumbnailPath = os.path.join(thumbnail_path, ThumbnailName) # bgr_data = ['/vsizip/%s/%s' % (in_file, file) for file in file_list[:3]] # out_vrt = '/vsimem/stacked.vrt' # 波段合成输出虚拟路径 # # 将多个源文件合成为一个VRT（virtual gdal dataset）文件 # out_dataset = gdal.BuildVRT(out_vrt, bgr_data, separate=True) # # 将VRT文件转换为目标格式的图像 # gdal.Translate(ThumbnailPath, # out_dataset, # format='JPEG', # outputType=gdal.GDT_Byte, # widthPct=10, # heightPct=10, # creationOptions=["TILED=YES", "COMPRESS=LZW"]) # # 释放内存 # # gdal.GetDriverByName("VRT").Delete('/vsimem/stacked.vrt') # gdal.Unlink('/vsimem/stacked.vrt') # del out_dataset rgb_list = [] for file in file_list[:3]: sub_dataset = gdal.Open('/vsizip/%s/%s' % (in_file, file)) sub_array = sub_dataset.ReadAsArray() rgb_list.append(sub_array) img_data = np.array([rgb_list[2], rgb_list[1], rgb_list[0]]) img_data = uint16to8(img_data) # Array转Image img_data2 = np.transpose(img_data, (1, 2, 0)) img_data2 = img_data2[:, :, ::-1] img = Image.fromarray(img_data2) # 压缩图片大小 if img_data.shape[1] > img_data.shape[2]: width = 512 height = int(width / img_data.shape[1] * img_data.shape[2]) else: height = 512 width = int(height / img_data.shape[1] * img_data.shape[2]) img.thumbnail((width, height)) img.save(ThumbnailPath, "PNG") # 释放内存 del rgb_list del img_data del img_data2 del img # 解压多光谱XML文件 if file_list[3].endswith('.SAFE/MTD_MSIL2A.xml'): # 生成XML文件 xmlFileName = os.path.splitext(basename)[0] + ".xml" xmlPath = os.path.join(xml_path, xmlFileName) meta_data = zip_file.read(file_list[3]) with open(xmlPath, "wb") as fout: fout.write(meta_data) # 读取其他信息 meta_content = zip_file.open(file_list[3]) dom = minidom.parse(meta_content) cloud_percent = dom.getElementsByTagName('n1:Quality_Indicators_Info')[ 0].getElementsByTagName('Cloud_Coverage_Assessment')[0].firstChild.data ImageGSD = '10, 20, 60' ProjectedCoordinates = dom.getElementsByTagName('n1:Geometric_Info')[ 0].getElementsByTagName('Coordinate_Reference_System')[ 0].getElementsByTagName('GEO_TABLES')[0].firstChild.data # 产品日期 ProduceTime = dom.getElementsByTagName('n1:General_Info')[0].getElementsByTagName('Product_Info')[ 0].getElementsByTagName('GENERATION_TIME')[0].firstChild.data StartTime = dom.getElementsByTagName('n1:General_Info')[0].getElementsByTagName('Product_Info')[ 0].getElementsByTagName('PRODUCT_START_TIME')[0].firstChild.data StopTime = dom.getElementsByTagName('n1:General_Info')[0].getElementsByTagName('Product_Info')[ 0].getElementsByTagName('PRODUCT_STOP_TIME')[0].firstChild.data # 经纬度 lon_lat = dom.getElementsByTagName('n1:Geometric_Info')[0].getElementsByTagName('Product_Footprint')[ 0].getElementsByTagName('Product_Footprint')[0].getElementsByTagName('Global_Footprint')[ 0].getElementsByTagName('EXT_POS_LIST')[0].firstChild.data TopLeftLatitude = re.split(r'\s', lon_lat)[0] # 左上纬度 TopLeftLongitude = re.split(r'\s', lon_lat)[1] # 左上经度 TopRightLatitude = re.split(r'\s', lon_lat)[2] # 右上纬度 TopRightLongitude = re.split(r'\s', lon_lat)[3] # 右上经度 BottomRightLatitude = re.split(r'\s', lon_lat)[4] # 右下纬度 BottomRightLongitude = re.split(r'\s', lon_lat)[5] # 右下经度 BottomLeftLatitude = re.split(r'\s', lon_lat)[6] # 左下纬度 BottomLeftLongitude = re.split(r'\s', lon_lat)[7] # 左下经度 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 生成字典 S2_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "StopTime": StopTime, "CloudPercent": cloud_percent, "boundaryGeomStr": boundaryGeomStr, "bands": "1,2,3,4,5,6,7,8,9,10,11,12", # "NumberBands": '12', "ImageGSD": ImageGSD, "ProjectedCoordinates": ProjectedCoordinates, "CollectionCode": '', "ThumbnailName": ThumbnailName, "ThumbnailPath": ThumbnailPath, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} zip_file.close() if not S2_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return S2_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetSentinel3OLData(in_file, xml_path, thumbnail_path): """ 获取哨兵3 OLCI（海陆色度计）卫星元数据（有待修改） :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: in_path, basename = os.path.split(in_file) with zipfile.ZipFile(in_file, 'r', zipfile.ZIP_DEFLATED) as zip_file: extensions = ('Oa03_radiance.nc', 'Oa05_radiance.nc', 'Oa08_radiance.nc', 'xfdumanifest.xml') file_list = [file for file in zip_file.namelist() if file.endswith(extensions)] # 生成缩略图 ThumbnailName = os.path.splitext(basename)[0] + "_thumb.jpg" ThumbnailPath = os.path.join(thumbnail_path, ThumbnailName) rgb_list = [] for file in file_list[:3]: sub_dataset = gdal.Open('/vsizip/%s/%s' % (in_file, file)) sub_array = sub_dataset.ReadAsArray() rgb_list.append(sub_array) img_data = np.array([rgb_list[2], rgb_list[1], rgb_list[0]]) img_data = uint16to8(img_data) # Array转Image img_data2 = np.transpose(img_data, (1, 2, 0)) img_data2 = img_data2[:, :, ::-1] img = Image.fromarray(img_data2) # 压缩图片大小 if img_data.shape[1] > img_data.shape[2]: width = 512 height = int(width / img_data.shape[1] * img_data.shape[2]) else: height = 512 width = int(height / img_data.shape[1] * img_data.shape[2]) img.thumbnail((width, height)) img.save(ThumbnailPath, "PNG") # 释放内存 del rgb_list del img_data del img_data2 del img # 解压XML文件 if file_list[3].endswith('xfdumanifest.xml'): # 生成XML文件 xmlFileName = os.path.splitext(basename)[0] + ".xml" xmlPath = os.path.join(xml_path, xmlFileName) meta_data = zip_file.read(file_list[3]) with open(xmlPath, "wb") as fout: fout.write(meta_data) # 读取其他信息 CollectionCode = "Sentinel3_OLCI_L1" meta_content = zip_file.open(file_list[3]) dom = minidom.parse(meta_content) ProjectedCoordinates = "" CloudPercent = "" # 产品日期 ProduceTime = dom.getElementsByTagName('sentinel3:creationTime')[0].firstChild.data StartTime = dom.getElementsByTagName('sentinel3:receivingStartTime')[0].firstChild.data StopTime = dom.getElementsByTagName('sentinel3:receivingStopTime')[0].firstChild.data # 经纬度 TopLeftLatitude = dom.getElementsByTagName('sentinel-safe:y')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('sentinel-safe:x')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('sentinel-safe:y')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('sentinel-safe:x')[2].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('sentinel-safe:y')[2].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('sentinel-safe:x')[2].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('sentinel-safe:y')[2].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('sentinel-safe:x')[0].firstChild.data # 左下经度 # boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ # f'{TopRightLongitude} {TopRightLatitude},' \ # f'{BottomRightLongitude} {BottomRightLatitude},' \ # f'{BottomLeftLongitude} {BottomLeftLatitude},' \ # f'{TopLeftLongitude} {TopLeftLatitude}))' # 生成字典 S3_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "StopTime": StopTime, "CloudPercent": CloudPercent, "boundaryGeomStr": "", "bands": "Oa01,Oa02,Oa03,Oa04,Oa05,Oa06,Oa07,Oa08,Oa09,Oa10,Oa11,Oa12,Oa13,Oa14,Oa15,Oa16," "Oa17,Oa18,Oa19,Oa20,Oa21", # "NumberBands": '21', "ImageGSD": "270,294", "ProjectedCoordinates": ProjectedCoordinates, "CollectionCode": CollectionCode, "ThumbnailName": ThumbnailName, "ThumbnailPath": ThumbnailPath, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} zip_file.close() if not S3_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return S3_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetHJ1Data(in_file, xml_path, thumbnail_path): """ 获取环境1号（HJ-1）卫星元数据 :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: with tarfile.open(in_file, mode='r') as tar_file: in_path, basename = os.path.split(in_file) for member in tar_file.getnames(): if member.endswith("THUMB.JPG"): # 解压缩略图 tar_file.extract(member, thumbnail_path) ThumbnailPath = thumbnail_path + "/" + member ThumbnailName = member.split('/')[1] elif member.endswith(".XML"): # 解压XML文件 tar_file.extract(member, xml_path) xmlPath = xml_path + "/" + member xmlFileName = member.split('/')[1] # 获取文件流 meta_file = tar_file.extractfile(member) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 productDate = dom.getElementsByTagName('productDate')[0].firstChild.data imagingStartTime = dom.getElementsByTagName('imagingStartTime')[0].firstChild.data # 开始时间 imagingStopTime = dom.getElementsByTagName('imagingStopTime')[0].firstChild.data # 结束时间 # 其他信息 pixelSpacing = dom.getElementsByTagName('pixelSpacing')[0].firstChild.data # 分辨率 # earthModel = dom.getElementsByTagName('earthModel')[0].firstChild.data # 投影 mapProjection = dom.getElementsByTagName('mapProjection')[0].firstChild.data # 投影坐标系 # zone = dom.getElementsByTagName('zone')[0].firstChild.data # 带号 bands = dom.getElementsByTagName('bands')[0].firstChild.data # 波段 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('productUpperLeftLat')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('productUpperLeftLong')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('productUpperRightLat')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('productUpperRightLong')[0].firstChild.data # 右上经度 BottomLeftLatitude = dom.getElementsByTagName('productLowerLeftLat')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('productLowerLeftLong')[0].firstChild.data # 左下经度 BottomRightLatitude = dom.getElementsByTagName('productLowerRightLat')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('productLowerRightLong')[0].firstChild.data # 右下纬度 # 边界几何 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' else: continue # 构建字典 hj1_dict = {"ProductTime": productDate, "StartTime": imagingStartTime, "EndTime": imagingStopTime, "CloudPercent": "", # "TopLeftLatitude": TopLeftLatitude, # "TopLeftLongitude": TopLeftLongitude, # "TopRightLatitude": TopRightLatitude, # "TopRightLongitude": TopRightLongitude, # "BottomLeftLatitude": BottomLeftLatitude, # "BottomLeftLongitude": BottomLeftLongitude, # "BottomRightLatitude": BottomRightLatitude, # "BottomRightLongitude": BottomRightLongitude, "boundaryGeomStr": boundaryGeomStr, "bands": bands, "ImageGSD": pixelSpacing, "ProjectedCoordinates": mapProjection, "CollectionCode": "", "ThumbnailPath": ThumbnailPath, "ThumbnailName": ThumbnailName, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} # 关闭压缩文件 tar_file.close() # 判断字典是否为空 if not hj1_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return hj1_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetZY02CData(in_file, xml_path, thumbnail_path): """ 获取资源2（ZY-2）卫星元数据： :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: zy2_mux_dict, zy2_pan_dict = dict(), dict() in_path, basename = os.path.split(in_file) with tarfile.open(in_file, mode='r') as tar_file: for member in tar_file.getnames(): if member.endswith("MUX_thumb.jpg"): # 解压多光谱缩略图 tar_file.extract(member, thumbnail_path) ThumbnailPath_MUX = thumbnail_path + "/" + member ThumbnailName_MUX = member elif member.endswith("PAN_thumb.jpg"): # 解压全色缩略图 tar_file.extract(member, thumbnail_path) ThumbnailPath_PAN = thumbnail_path + "/" + member ThumbnailName_PAN = member if member.endswith('MUX.xml'): # 解压XML文件 tar_file.extract(member, xml_path) xmlPath = xml_path + "/" + member xmlFileName = member # 获取文件流 meta_file = tar_file.extractfile(member) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 MapProjection = dom.getElementsByTagName('MapProjection')[0].firstChild.data # 投影坐标系 EarthEllipsoid = dom.getElementsByTagName('EarthEllipsoid')[0].firstChild.data # 地理坐标系 ZoneNo = dom.getElementsByTagName('ZoneNo')[0].firstChild.data # 投影分带带号 Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 # 几何边界 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建多光谱字典 zy2_mux_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, # "TopLeftLatitude": TopLeftLatitude, # "TopLeftLongitude": TopLeftLongitude, # "TopRightLatitude": TopRightLatitude, # "TopRightLongitude": TopRightLongitude, # "BottomRightLatitude": BottomRightLatitude, # "BottomRightLongitude": BottomRightLongitude, # "BottomLeftLatitude": BottomLeftLatitude, # "BottomLeftLongitude": BottomLeftLongitude, "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": MapProjection, 'CollectionCode': "", "ThumbnailPath": ThumbnailPath_MUX, "ThumbnailName": ThumbnailName_MUX, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "day"} elif member.endswith('PAN.xml'): # 解压XML文件 tar_file.extract(member, xml_path) xmlPath = xml_path + "/" + member xmlFileName = member # 获取文件流 meta_file = tar_file.extractfile(member) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 MapProjection = dom.getElementsByTagName('MapProjection')[0].firstChild.data # 投影坐标系 EarthEllipsoid = dom.getElementsByTagName('EarthEllipsoid')[0].firstChild.data # 地理坐标系 ZoneNo = dom.getElementsByTagName('ZoneNo')[0].firstChild.data # 投影分带带号 Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 # 几何边界 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建全色字典 zy2_pan_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, # "TopLeftLatitude": TopLeftLatitude, # "TopLeftLongitude": TopLeftLongitude, # "TopRightLatitude": TopRightLatitude, # "TopRightLongitude": TopRightLongitude, # "BottomRightLatitude": BottomRightLatitude, # "BottomRightLongitude": BottomRightLongitude, # "BottomLeftLatitude": BottomLeftLatitude, # "BottomLeftLongitude": BottomLeftLongitude, "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": MapProjection, 'CollectionCode': "", "ThumbnailPath": ThumbnailPath_PAN, "ThumbnailName": ThumbnailName_PAN, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "day"} else: continue # 关闭压缩文件 tar_file.close() # 判断字典是否为空 if (not zy2_mux_dict) or (not zy2_pan_dict): return {"code": -1, "msg": "没有满足条件的数据字典..."} return zy2_mux_dict, zy2_pan_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} def GetZY3Data(in_file, xml_path, thumbnail_path): """ 获取资源3（ZY-3）卫星元数据： :param thumbnail_path: :param xml_path: :param in_file: :return: 元数据字典 """ try: zy3_dict = dict() with tarfile.open(in_file, mode='r') as tar_file: in_path, basename = os.path.split(in_file) for member in tar_file.getnames(): if member.endswith("thumb.jpg"): # 解压缩略图 tar_file.extract(member, thumbnail_path) ThumbnailPath = thumbnail_path + "/" + member ThumbnailName = member if not member.endswith('.xml'): continue elif member.endswith('order.xml'): continue else: # 解压XML文件 tar_file.extract(member, xml_path) xmlPath = xml_path + "/" + member xmlFileName = member # 获取文件流 meta_file = tar_file.extractfile(member) meta_content = meta_file.read() dom = minidom.parse(io.StringIO(meta_content.decode("utf-8"))) # 产品日期 ProduceTime = dom.getElementsByTagName('ProduceTime')[0].firstChild.data StartTime = dom.getElementsByTagName('StartTime')[0].firstChild.data EndTime = dom.getElementsByTagName('EndTime')[0].firstChild.data # 其他信息 ImageGSD = dom.getElementsByTagName('ImageGSD')[0].firstChild.data # 分辨率 MapProjection = dom.getElementsByTagName('MapProjection')[0].firstChild.data # 投影坐标系 EarthEllipsoid = dom.getElementsByTagName('EarthEllipsoid')[0].firstChild.data # 地理坐标系 ZoneNo = dom.getElementsByTagName('ZoneNo')[0].firstChild.data # 投影分带带号 Bands = dom.getElementsByTagName('Bands')[0].firstChild.data # 波段 CloudPercent = dom.getElementsByTagName('CloudPercent')[0].firstChild.data # 云覆盖 # 中心经纬度 TopLeftLatitude = dom.getElementsByTagName('TopLeftLatitude')[0].firstChild.data # 左上纬度 TopLeftLongitude = dom.getElementsByTagName('TopLeftLongitude')[0].firstChild.data # 左上经度 TopRightLatitude = dom.getElementsByTagName('TopRightLatitude')[0].firstChild.data # 右上纬度 TopRightLongitude = dom.getElementsByTagName('TopRightLongitude')[0].firstChild.data # 右上经度 BottomRightLatitude = dom.getElementsByTagName('BottomRightLatitude')[0].firstChild.data # 右下纬度 BottomRightLongitude = dom.getElementsByTagName('BottomRightLongitude')[0].firstChild.data # 右下经度 BottomLeftLatitude = dom.getElementsByTagName('BottomLeftLatitude')[0].firstChild.data # 左下纬度 BottomLeftLongitude = dom.getElementsByTagName('BottomLeftLongitude')[0].firstChild.data # 左下经度 # 边界几何 boundaryGeomStr = f'POLYGON(({TopLeftLongitude} {TopLeftLatitude},' \ f'{TopRightLongitude} {TopRightLatitude},' \ f'{BottomRightLongitude} {BottomRightLatitude},' \ f'{BottomLeftLongitude} {BottomLeftLatitude},' \ f'{TopLeftLongitude} {TopLeftLatitude}))' # 构建字典 zy3_dict = {"ProduceTime": ProduceTime, "StartTime": StartTime, "EndTime": EndTime, "CloudPercent": CloudPercent, # "TopLeftLatitude": TopLeftLatitude, # "TopLeftLongitude": TopLeftLongitude, # "TopRightLatitude": TopRightLatitude, # "TopRightLongitude": TopRightLongitude, # "BottomRightLatitude": BottomRightLatitude, # "BottomRightLongitude": BottomRightLongitude, # "BottomLeftLatitude": BottomLeftLatitude, # "BottomLeftLongitude": BottomLeftLongitude, "boundaryGeomStr": boundaryGeomStr, "bands": Bands, "ImageGSD": ImageGSD, "ProjectedCoordinates": MapProjection, "CollectionCode": "", "ThumbnailPath": ThumbnailPath, "ThumbnailName": ThumbnailName, "xmlPath": xmlPath, "xmlFileName": xmlFileName, "DirectoryDepth": "month"} # 关闭压缩文件 tar_file.close() # 判断是否为空 if not zy3_dict: return {"code": -1, "msg": "没有满足条件的数据字典..."} return zy3_dict except Exception as e: print(str(e)) return {"code": -1, "msg": str(e)} if __name__ == '__main__': HJ1FilePath = r"Y:\不同传感器数据\HJ-1\HJ1A-CCD2-450-80-20090501-L20000106616.tar.gz" JPSSFilePath = r"Y:\不同传感器数据\JPSS\VJ102IMG.A2021159.0542.002.2021159094907.nc" ZY2FilePath = r"Y:\不同传感器数据\ZY-2\ZY02C_PMS_E115.9_N36.2_20120422_L2C0000391981.tar.gz" ZY3FilePath = r"Y:\不同传感器数据\ZY-3\ZY3_MUX_E83.3_N43.3_20120405_L2A0000301226.tar.gz" S1FilePath = r'Y:\不同传感器数据\SENTINEL-1\S1A_IW_GRDH_1SDV_20210407T095634_20210407T095659_037343_046675_8E66.zip' S2FilePath = r'Y:\不同传感器数据\SENTINEL-2\S2B_MSIL2A_20210804T024549_N0301_R132_T50SQF_20210804T053331.zip' GF1PMSPath = r'Y:\不同传感器数据\GF-1\GF1_PMS2_E104.1_N36.6_20210308_L1A0005524847.tar.gz' H08FilePath = r"Y:\不同传感器数据\葵花8\NC_H08_20210802_2010_R21_FLDK.06001_06001.nc" SNPPFilePath = r"Y:\不同传感器数据\VIIRS\VNP02IMG.A2021182.0418.001.2021182100800.nc" GF3MDJPath = r'Y:\不同传感器数据\GF-3\GF3_MDJ_SS_024986_E120.8_N35.6_20210509_L1A_VHVV_L10005638033.tar.gz' GF4PMIPath = r'Y:\不同传感器数据\GF-4\GF4_PMI_E119.8_N35.3_20210908_L1A0000417337.tar.gz' S3OLFilePath = r'Y:\不同传感器数据\SENTINEL-3' \ r'\S3B_OL_1_EFR____20210910T022645_20210910T022945_20210911T064342_0179_056_374_2340_LN1_O_NT_002.zip' S3SLFilePath = r'Y:\不同传感器数据\SENTINEL-3' \ r'\S3A_SL_1_RBT____20210916T020956_20210916T021256_20210917T120953_0179_076_217_2340_LN2_O_NT_004.zip' # 读取 HJ-1 元数据 hj1_dic = GetHJ1Data(HJ1FilePath) print(hj1_dic) # 读取 JPSS 元数据 jpss_dic = GetJPSSData(JPSSFilePath) print(jpss_dic) # 读取 ZY2 元数据 zy2_mux_dic, zy2_pan_dic = GetZY02CData(ZY2FilePath) print(zy2_mux_dic) print(zy2_pan_dic) # 读取 ZY3 元数据 zy3_dic = GetZY3Data(ZY3FilePath) print(zy3_dic) # 读取GF-PMS元数据 pms_mss_dic, pms_pan_dic = GetGFPMSData(GF1PMSPath) print(pms_mss_dic) print(pms_pan_dic) # 读取葵花8元数据 h8_dic = GetH08Data(H08FilePath) print(h8_dic) # 读取 S2 元数据 s2_dic = GetSentinel2Data(S2FilePath) print(s2_dic) # 读取 S1 元数据 s1_dic = GetSentinel1Data(S1FilePath) print(s1_dic) # 读取 SNPP 元数据 snpp_dic = GetSNPPData(SNPPFilePath) print(snpp_dic) # 读取 GF3 元数据 gf3_dic = GetGF3MDJData(GF3MDJPath) print(gf3_dic) # 读取 GF4 元数据 gf4_pms_dic, gf4_irs_dic = GetGF4PMIData(GF4PMIPath) print(gf4_pms_dic) print(gf4_irs_dic) # 读取 S3 OL元数据 s3ol_dic = GetSentinel3OLData(S3OLFilePath) print(s3ol_dic) # # 读取 S3 SL元数据 # s3sl_dic = GetSentinel3SLData(S3SLFilePath) # print(s3sl_dic)