1# -*- coding: utf-8 -*- 

2 

3# gms_preprocessing, spatial and spectral homogenization of satellite remote sensing data 

4# 

5# Copyright (C) 2020 Daniel Scheffler (GFZ Potsdam, daniel.scheffler@gfz-potsdam.de) 

6# 

7# This software was developed within the context of the GeoMultiSens project funded 

8# by the German Federal Ministry of Education and Research 

9# (project grant code: 01 IS 14 010 A-C). 

10# 

11# This program is free software: you can redistribute it and/or modify it under 

12# the terms of the GNU General Public License as published by the Free Software 

13# Foundation, either version 3 of the License, or (at your option) any later version. 

14# Please note the following exception: `gms_preprocessing` depends on tqdm, which 

15# is distributed under the Mozilla Public Licence (MPL) v2.0 except for the files 

16# "tqdm/_tqdm.py", "setup.py", "README.rst", "MANIFEST.in" and ".gitignore". 

17# Details can be found here: https://github.com/tqdm/tqdm/blob/master/LICENCE. 

18# 

19# This program is distributed in the hope that it will be useful, but WITHOUT 

20# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS 

21# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more 

22# details. 

23# 

24# You should have received a copy of the GNU Lesser General Public License along 

25# with this program. If not, see <http://www.gnu.org/licenses/>. 

26 

27 

28import collections 

29import re 

30 

31import numpy as np 

32from typing import TYPE_CHECKING # noqa F401 # flake8 issue 

33 

34from ..options.config import GMS_config as CFG 

35 

36if TYPE_CHECKING: 

37 from ..model.gms_object import GMS_identifier # noqa F401 # flake8 issue 

38 

39__author__ = 'Daniel Scheffler' 

40 

41dtype_lib_Python_IDL = {'bool_': 0, 'uint8': 1, 'int8': 1, 'int_': 1, 'int16': 2, 'uint16': 12, 'int32': 3, 

42 'uint32': 13, 'int64': 14, 'uint64': 15, 'float32': 4, 'float64': 5, 'complex_': 6, 

43 'complex64': 9} 

44dtype_lib_IDL_Python = {0: np.bool_, 1: np.uint8, 2: np.int16, 3: np.int32, 4: np.float32, 5: np.float64, 

45 6: np.complex64, 9: np.complex128, 12: np.uint16, 13: np.uint32, 14: np.int64, 15: np.uint64} 

46dtype_lib_GDAL_Python = {"uint8": 1, "int8": 1, "uint16": 2, "int16": 3, "uint32": 4, "int32": 5, "float32": 6, 

47 "float64": 7, "complex64": 10, "complex128": 11} 

48proc_chain = ['L1A', 'L1B', 'L1C', 'L2A', 'L2B', 'L2C'] 

49db_jobs_statistics_def = {'pending': 1, 'started': 2, None: 2, 'L1A': 3, 'L1B': 4, 'L1C': 5, 'L2A': 6, 'L2B': 7, 

50 'L2C': 8, 'FAILED': 9} # NOTE: OrderedDicts passed to L1A_map have proc_level=None 

51bandslist_all_errors = ['ac_errors', 'mask_clouds_confidence', 'spat_homo_errors', 'spec_homo_errors'] 

52 

53 

54def get_GMS_sensorcode(GMS_id): 

55 # type: (GMS_identifier) -> str 

56 

57 Satellite, Sensor, Subsystem = (GMS_id.satellite, GMS_id.sensor, GMS_id.subsystem) 

58 Sensor = Sensor[:-1] if re.match(r'SPOT', Satellite, re.I) and Sensor[-1] not in ['1', '2'] else Sensor 

59 meta_sensorcode = Satellite + '_' + Sensor + ('_' + Subsystem if Subsystem not in ["", None] else "") 

60 sensorcode_dic = { 

61 'ALOS_AVNIR-2': 'AVNIR-2', 

62 'Landsat-4_TM': 'TM4', # call from layerstacker 

63 'Landsat-4_TM_SAM': 'TM4', # call from metadata object 

64 'Landsat-5_TM': 'TM5', 

65 'Landsat-5_TM_SAM': 'TM5', 

66 'Landsat-7_ETM+': 'TM7', 

67 'Landsat-7_ETM+_SAM': 'TM7', 

68 'Landsat-8_OLI': 'LDCM', 

69 'Landsat-8_OLI_TIRS': 'LDCM', 

70 'Landsat-8_LDCM': 'LDCM', 

71 'SPOT-1_HRV1': 'SPOT1a', # MS 

72 'SPOT-1_HRV2': 'SPOT1b', 

73 'SPOT-2_HRV1': 'SPOT2a', 

74 'SPOT-2_HRV2': 'SPOT2b', 

75 'SPOT-3_HRV1': 'SPOT3a', 

76 'SPOT-3_HRV2': 'SPOT3b', 

77 'SPOT-4_HRVIR1': 'SPOT4a', 

78 'SPOT-4_HRVIR2': 'SPOT4b', 

79 'SPOT-5_HRG1': 'SPOT5a', # PAN HRG2A 

80 'SPOT-5_HRG2': 'SPOT5b', # MS HRG2J 

81 'RapidEye-1_MSI': 'RE1', 

82 'RapidEye-2_MSI': 'RE2', 

83 'RapidEye-3_MSI': 'RE3', 

84 'RapidEye-4_MSI': 'RE4', 

85 'RapidEye-5_MSI': 'RE5', 

86 'SRTM_SRTM2': 'SRTM2', 

87 'Terra_ASTER': 'AST_full', 

88 'Terra_ASTER_VNIR1': 'AST_V1', 

89 'Terra_ASTER_VNIR2': 'AST_V2', 

90 'Terra_ASTER_SWIR': 'AST_S', 

91 'Terra_ASTER_TIR': 'AST_T', 

92 'Sentinel-2A_MSI': 'S2A_full', 

93 'Sentinel-2B_MSI': 'S2B_full', 

94 'Sentinel-2A_MSI_S2A10': 'S2A10', 

95 'Sentinel-2A_MSI_S2A20': 'S2A20', 

96 'Sentinel-2A_MSI_S2A60': 'S2A60', 

97 'Sentinel-2B_MSI_S2B10': 'S2B10', 

98 'Sentinel-2B_MSI_S2B20': 'S2B20', 

99 'Sentinel-2B_MSI_S2B60': 'S2B60' 

100 } 

101 try: 

102 return sensorcode_dic[meta_sensorcode] 

103 except KeyError: 

104 raise KeyError('Sensor %s is not included in sensorcode dictionary and can not be converted into GMS ' 

105 'sensorcode.' % meta_sensorcode) 

106 

107 

108def get_mask_classdefinition(maskname, satellite): 

109 if maskname == 'mask_nodata': 

110 return {'No data': 0, 

111 'Data': 1} 

112 elif maskname == 'mask_clouds': 

113 legends = { 

114 'FMASK': { 

115 'No Data': 0, 

116 'Clear': 1, 

117 'Cloud': 2, 

118 'Shadow': 3, 

119 'Snow': 4, 

120 'Water': 5}, 

121 # seems to be outdated: 

122 # {'Clear Land': 0, 'Clear Water': 1, 'Cloud Shadow': 2, 'Snow': 3, 'Cloud': 4, 'No data': 255} 

123 'Classical Bayesian': { 

124 'Clear': 10, 

125 'Thick Clouds': 20, 

126 'Thin Clouds': 30, 

127 'Snow': 40}, # Classical Bayesian py_tools_ah 

128 'SICOR': { 

129 'Clear': 10, 

130 'Water': 20, 

131 'Shadow': 30, 

132 'Cirrus': 40, 

133 'Cloud': 50, 

134 'Snow': 60} # SICOR 

135 } 

136 

137 return legends[CFG.cloud_masking_algorithm[satellite]] 

138 else: 

139 raise ValueError("'%s' is not a supported mask name." % maskname) 

140 

141 

142def get_mask_colormap(maskname): 

143 if maskname == 'mask_clouds': 

144 # return collections.OrderedDict(zip(['No data','Clear','Thick Clouds','Thin Clouds','Snow','Unknown Class'], 

145 # [[0,0,0] ,[0,255,0],[80,80,80], [175,175,175],[255,255,255],[255,0,0]])) 

146 return collections.OrderedDict(( 

147 ('No data', [0, 0, 0]), 

148 ('Clear', [0, 255, 0]), 

149 ('Water', [0, 0, 255]), 

150 ('Shadow', [50, 50, 50]), 

151 ('Cirrus', [175, 175, 175]), 

152 ('Cloud', [80, 80, 80]), 

153 ('Snow', [255, 255, 255]), 

154 ('Unknown Class', [255, 0, 0]),)) 

155 else: 

156 return None 

157 

158 

159def get_outFillZeroSaturated(dtype): 

160 """Returns the values for 'fill-', 'zero-' and 'saturated' pixels of an image 

161 to be written with regard to the target data type. 

162 

163 :param dtype: data type of the image to be written""" 

164 

165 dtype = str(np.dtype(dtype)) 

166 assert dtype in ['bool', 'int8', 'uint8', 'int16', 'uint16', 'float32'], \ 

167 "get_outFillZeroSaturated: Unknown dType: '%s'." % dtype 

168 dict_outFill = {'bool': None, 'int8': -128, 'uint8': 0, 'int16': -9999, 'uint16': 9999, 'float32': -9999.} 

169 dict_outZero = {'bool': None, 'int8': 0, 'uint8': 1, 'int16': 0, 'uint16': 0, 'float32': 0.} 

170 dict_outSaturated = {'bool': None, 'int8': 127, 'uint8': 256, 'int16': 32767, 'uint16': 65535, 'float32': 65535.} 

171 return dict_outFill[dtype], dict_outZero[dtype], dict_outSaturated[dtype] 

172 

173 

174def is_dataset_provided_as_fullScene(GMS_id): 

175 # type: (GMS_identifier) -> bool 

176 """Returns True if the dataset belonging to the given GMS_identifier is provided as full scene and returns False if 

177 it is provided as multiple tiles. 

178 

179 :param GMS_id: 

180 :return: 

181 """ 

182 

183 sensorcode = get_GMS_sensorcode(GMS_id) 

184 dict_fullScene_or_tiles = { 

185 'AVNIR-2': True, 

186 'AST_full': True, 

187 'AST_V1': True, 

188 'AST_V2': True, 

189 'AST_S': True, 

190 'AST_T': True, 

191 'TM4': True, 

192 'TM5': True, 

193 'TM7': True, 

194 'LDCM': True, 

195 'SPOT1a': True, 

196 'SPOT2a': True, 

197 'SPOT3a': True, 

198 'SPOT4a': True, 

199 'SPOT5a': True, 

200 'SPOT1b': True, 

201 'SPOT2b': True, 

202 'SPOT3b': True, 

203 'SPOT4b': True, 

204 'SPOT5b': True, 

205 'RE5': False, 

206 'S2A_full': False, 

207 'S2A10': False, 

208 'S2A20': False, 

209 'S2A60': False, 

210 'S2B_full': False, 

211 'S2B10': False, 

212 'S2B20': False, 

213 'S2B60': False, } 

214 return dict_fullScene_or_tiles[sensorcode] 

215 

216 

217def datasetid_to_sat_sen(dsid): 

218 # type: (int) -> tuple 

219 conv_dict = { 

220 8: ('Terra', 'ASTER'), # ASTER L1B 

221 104: ('Landsat-8', 'OLI_TIRS'), # pre-collection-ID 

222 108: ('Landsat-5', 'TM'), # pre-collection-ID 

223 112: ('Landsat-7', 'ETM+'), # pre-collection-ID SLC-off 

224 113: ('Landsat-7', 'ETM+'), # pre-collection-ID SLC-on 

225 189: ('Terra', 'ASTER'), # ASTER L1T 

226 249: ('Sentinel-2A', 'MSI'), # actually only Sentinel-2 

227 250: ('Landsat-8', 'OLI_TIRS'), 

228 251: ('Landsat-7', 'ETM+'), 

229 252: ('Landsat-5', 'TM'), # also includes Landsat-4 

230 } 

231 try: 

232 return conv_dict[dsid] 

233 except KeyError: 

234 raise ValueError('No satellite / sensor tuple available for dataset ID %s.' % dsid) 

235 

236 

237def sat_sen_to_datasetid(satellite, sensor): 

238 # type: (str, str) -> int 

239 conv_dict = { 

240 ('Landsat-5', 'TM'): 252, 

241 ('Landsat-7', 'ETM+'): 251, 

242 ('Landsat-8', 'OLI_TIRS'): 250, 

243 ('Sentinel-2A', 'MSI'): 249, 

244 ('Sentinel-2B', 'MSI'): 249, 

245 ('Terra', 'ASTER'): 189 # ASTER L1T 

246 } 

247 try: 

248 return conv_dict[(satellite, sensor)] 

249 except KeyError: 

250 raise ValueError('No dataset ID available for %s %s.' % (satellite, sensor))