Coverage for enpt/model/images/images_mapgeo.py: 81%

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

2 

3# EnPT, EnMAP Processing Tool - A Python package for pre-processing of EnMAP Level-1B data 

4# 

5# Copyright (C) 2018-2024 Karl Segl (GFZ Potsdam, segl@gfz-potsdam.de), Daniel Scheffler 

6# (GFZ Potsdam, danschef@gfz-potsdam.de), Niklas Bohn (GFZ Potsdam, nbohn@gfz-potsdam.de), 

7# Stéphane Guillaso (GFZ Potsdam, stephane.guillaso@gfz-potsdam.de) 

8# 

9# This software was developed within the context of the EnMAP project supported 

10# by the DLR Space Administration with funds of the German Federal Ministry of 

11# Economic Affairs and Energy (on the basis of a decision by the German Bundestag: 

12# 50 EE 1529) and contributions from DLR, GFZ and OHB System AG. 

13# 

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

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

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

17# version. Please note the following exception: `EnPT` depends on tqdm, which 

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

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

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

21# 

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

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

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

25# details. 

26# 

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

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

29 

30"""EnPT EnMAP objects in map geometry.""" 

31 

32import logging 

33from types import SimpleNamespace 

34from typing import Tuple, Optional # noqa: F401 

35from os import path, makedirs 

36 

37from geoarray import GeoArray, NoDataMask 

38 

39from .image_baseclasses import _EnMAP_Image 

40from .images_sensorgeo import EnMAPL1Product_SensorGeo 

41from ...utils.logging import EnPT_Logger 

42from ...model.metadata import EnMAP_Metadata_L2A_MapGeo # noqa: F401 # only used for type hint 

43from ...options.config import EnPTConfig 

44 

45__author__ = ['Daniel Scheffler', 'Stéphane Guillaso', 'André Hollstein'] 

46 

47 

48class EnMAP_Detector_MapGeo(_EnMAP_Image): 

49 """Base class representing a single detector of an EnMAP image (as map geometry). 

50 

51 NOTE: 

52 - Inherits all attributes from _EnMAP_Image class. 

53 - All functionality that VNIR and SWIR detectors (map geometry) have in common is to be implemented here. 

54 - All EnMAP image subclasses representing a specific EnMAP detector (sensor geometry) should inherit from 

55 _EnMAP_Detector_SensorGeo. 

56 

57 Attributes: 

58 - to be listed here. Check help(_EnMAP_Detector_SensorGeo) in the meanwhile! 

59 

60 """ 

61 

62 def __init__(self, detector_name: str, logger=None): 

63 """Get an instance of _EnMAP_Detector_MapGeo. 

64 

65 :param detector_name: 'VNIR' or 'SWIR' 

66 :param logger: 

67 """ 

68 self.detector_name = detector_name 

69 self.logger = logger or logging.getLogger() 

70 

71 # private attributes 

72 self._mask_nodata = None 

73 

74 # get all attributes of base class "_EnMAP_Image" 

75 super(EnMAP_Detector_MapGeo, self).__init__() 

76 

77 @property 

78 def mask_nodata(self) -> GeoArray: 

79 """Return the no data mask. 

80 

81 Bundled with all the corresponding metadata. 

82 

83 For usage instructions and a list of attributes refer to help(self.data). 

84 self.mask_nodata works in the same way. 

85 

86 :return: instance of geoarray.NoDataMask 

87 """ 

88 if self._mask_nodata is None and isinstance(self.data, GeoArray): 

89 self.logger.info('Calculating nodata mask...') 

90 self._mask_nodata = self.data.mask_nodata # calculates mask nodata if not already present 

91 

92 return self._mask_nodata 

93 

94 @mask_nodata.setter 

95 def mask_nodata(self, *geoArr_initArgs): 

96 self._mask_nodata = self._get_geoarray_with_datalike_geometry(geoArr_initArgs, 'mask_nodata', 

97 nodataVal=False, specialclass=NoDataMask) 

98 

99 @mask_nodata.deleter 

100 def mask_nodata(self): 

101 self._mask_nodata = None 

102 

103 def calc_mask_nodata(self, fromBand=None, overwrite=False) -> GeoArray: 

104 """Calculate a no data mask with (values: 0=nodata; 1=data). 

105 

106 :param fromBand: <int> index of the band to be used (if None, all bands are used) 

107 :param overwrite: <bool> whether to overwrite existing nodata mask that has already been calculated 

108 :return: 

109 """ 

110 self.logger.info('Calculating nodata mask...') 

111 

112 if self._mask_nodata is None or overwrite: 

113 self.data.calc_mask_nodata(fromBand=fromBand, overwrite=overwrite) 

114 self.mask_nodata = self.data.mask_nodata 

115 return self.mask_nodata 

116 

117 

118class EnMAPL2Product_MapGeo(_EnMAP_Image): 

119 """Class for EnPT Level-2 EnMAP object in map geometry. 

120 

121 Attributes: 

122 - logger: 

123 - logging.Logger instance or subclass instance 

124 - paths: 

125 - paths belonging to the EnMAP product 

126 - meta: 

127 - instance of EnMAP_Metadata_SensorGeo class 

128 """ 

129 

130 def __init__(self, config: EnPTConfig, logger=None): 

131 # protected attributes 

132 self._logger = None 

133 

134 # populate attributes 

135 self.cfg = config 

136 if logger: 

137 self.logger = logger 

138 

139 self.meta: Optional[EnMAP_Metadata_L2A_MapGeo] = None 

140 self.paths: Optional[SimpleNamespace] = None 

141 

142 super(EnMAPL2Product_MapGeo, self).__init__() 

143 

144 @property 

145 def logger(self) -> EnPT_Logger: 

146 """Get an instance of enpt.utils.logging.EnPT_Logger. 

147 

148 NOTE: 

149 - The logging level will be set according to the user inputs of EnPT. 

150 - The path of the log file is directly derived from the attributes of the _EnMAP_Image instance. 

151 

152 Usage: 

153 - get the logger: 

154 logger = self.logger 

155 - set the logger 

156 self.logger = logging.getLogger() # NOTE: only instances of logging.Logger are allowed here 

157 - delete the logger: 

158 del self.logger # or "self.logger = None" 

159 

160 :return: EnPT_Logger instance 

161 """ 

162 if self._logger and self._logger.handlers[:]: 

163 return self._logger 

164 else: 

165 basename = path.splitext(path.basename(self.cfg.path_l1b_enmap_image))[0] 

166 path_logfile = path.join(self.cfg.output_dir, basename + '.log') \ 

167 if self.cfg.create_logfile and self.cfg.output_dir else '' 

168 self._logger = EnPT_Logger('log__' + basename, fmt_suffix=None, path_logfile=path_logfile, 

169 log_level=self.cfg.log_level, append=False) 

170 

171 return self._logger 

172 

173 @logger.setter 

174 def logger(self, logger: logging.Logger): 

175 assert isinstance(logger, logging.Logger) or logger in ['not set', None], \ 

176 "%s.logger can not be set to %s." % (self.__class__.__name__, logger) 

177 

178 # save prior logs 

179 # if logger is None and self._logger is not None: 

180 # self.log += self.logger.captured_stream 

181 self._logger = logger 

182 

183 @property 

184 def log(self) -> str: 

185 """Return a string of all logged messages until now. 

186 

187 NOTE: self.log can also be set to a string. 

188 """ 

189 return self.logger.captured_stream 

190 

191 @log.setter 

192 def log(self, string: str): 

193 assert isinstance(string, str), "'log' can only be set to a string. Got %s." % type(string) 

194 self.logger.captured_stream = string 

195 

196 @classmethod 

197 def from_L1B_sensorgeo(cls, config: EnPTConfig, enmap_ImageL1: EnMAPL1Product_SensorGeo): 

198 from ...processors.orthorectification import Orthorectifier 

199 L2_obj = Orthorectifier(config=config).run_transformation(enmap_ImageL1=enmap_ImageL1) 

200 

201 return L2_obj 

202 

203 def get_paths(self, l2a_outdir: str): 

204 """Get all file paths associated with the current instance of EnMAP_Detector_SensorGeo. 

205 

206 NOTE: This information is read from the detector_meta. 

207 

208 :param l2a_outdir: output directory of EnMAP Level-2A dataset 

209 :return: paths as SimpleNamespace 

210 """ 

211 paths = SimpleNamespace() 

212 paths.root_dir = l2a_outdir 

213 paths.metaxml = path.join(l2a_outdir, self.meta.filename_metaxml) 

214 paths.data = path.join(l2a_outdir, self.meta.filename_data) 

215 paths.mask_landwater = path.join(l2a_outdir, self.meta.filename_mask_landwater) 

216 paths.mask_clouds = path.join(l2a_outdir, self.meta.filename_mask_clouds) 

217 paths.mask_cloudshadow = path.join(l2a_outdir, self.meta.filename_mask_cloudshadow) 

218 paths.mask_haze = path.join(l2a_outdir, self.meta.filename_mask_haze) 

219 paths.mask_snow = path.join(l2a_outdir, self.meta.filename_mask_snow) 

220 paths.mask_cirrus = path.join(l2a_outdir, self.meta.filename_mask_cirrus) 

221 paths.deadpixelmap_vnir = path.join(l2a_outdir, self.meta.filename_deadpixelmap_vnir) 

222 paths.deadpixelmap_swir = path.join(l2a_outdir, self.meta.filename_deadpixelmap_swir) 

223 paths.quicklook_vnir = path.join(l2a_outdir, self.meta.filename_quicklook_vnir) 

224 paths.quicklook_swir = path.join(l2a_outdir, self.meta.filename_quicklook_swir) 

225 

226 return paths 

227 

228 def save(self, outdir: str, suffix="") -> str: 

229 """Save the product to disk using almost the same input format. 

230 

231 :param outdir: path to the output directory 

232 :param suffix: suffix to be appended to the output filename (???) 

233 :return: root path (root directory) where products were written 

234 """ 

235 # TODO optionally add more output formats 

236 product_dir = path.join(path.abspath(outdir), 

237 "{name}{suffix}".format(name=self.meta.scene_basename, suffix=suffix)) 

238 

239 self.logger.info("Write product to: %s" % product_dir) 

240 makedirs(product_dir, exist_ok=True) 

241 

242 # save raster data 

243 kwargs_save = \ 

244 dict(fmt='GTiff', 

245 creationOptions=["COMPRESS=LZW", 

246 "NUM_THREADS=%d" % self.cfg.CPUs, 

247 "INTERLEAVE=%s" % ('BAND' if self.cfg.output_interleave == 'band' else 'PIXEL')] 

248 ) if self.cfg.output_format == 'GTiff' else \ 

249 dict(fmt='ENVI', 

250 creationOptions=["INTERLEAVE=%s" % ("BSQ" if self.cfg.output_interleave == 'band' else 

251 "BIL" if self.cfg.output_interleave == 'line' else 

252 "BIP")]) 

253 outpaths = dict(metaxml=path.join(product_dir, self.meta.filename_metaxml)) 

254 

255 for attrName in ['data', 'mask_landwater', 'mask_clouds', 'mask_cloudshadow', 'mask_haze', 'mask_snow', 

256 'mask_cirrus', 'quicklook_vnir', 'quicklook_swir', 'deadpixelmap', 

257 'polymer_logchl', 'polymer_logfb', 'polymer_rgli', 'polymer_rnir', 'polymer_bitmask']: 

258 

259 if attrName == 'deadpixelmap': 

260 # TODO VNIR and SWIR must be merged 

261 self.logger.warning('Currently, L2A dead pixel masks cannot be saved yet.') 

262 continue 

263 

264 if attrName.startswith('polymer_'): 

265 ext = \ 

266 'TIF' if self.cfg.output_format == 'GTiff' else \ 

267 'BSQ' if self.cfg.output_format == 'ENVI' and self.cfg.output_interleave == 'band' else \ 

268 'BIL' if self.cfg.output_format == 'ENVI' and self.cfg.output_interleave == 'line' else \ 

269 'BIP' if self.cfg.output_format == 'ENVI' and self.cfg.output_interleave == 'pixel' else \ 

270 'NA' 

271 dict_attr_fn = dict( 

272 polymer_logchl=f'{self.meta.scene_basename}-ACOUT_POLYMER_LOGCHL.{ext}', 

273 polymer_logfb=f'{self.meta.scene_basename}-ACOUT_POLYMER_LOGFB.{ext}', 

274 polymer_rgli=f'{self.meta.scene_basename}-ACOUT_POLYMER_RGLI.{ext}', 

275 polymer_rnir=f'{self.meta.scene_basename}-ACOUT_POLYMER_RNIR.{ext}', 

276 polymer_bitmask=f'{self.meta.scene_basename}-ACOUT_POLYMER_BITMASK.{ext}', 

277 ) 

278 outpath = path.join(product_dir, dict_attr_fn[attrName]) 

279 else: 

280 outpath = path.join(product_dir, getattr(self.meta, 'filename_%s' % attrName)) 

281 

282 attr_gA = \ 

283 self.generate_quicklook(bands2use=self.meta.preview_bands_vnir) if attrName == 'quicklook_vnir' else \ 

284 self.generate_quicklook(bands2use=self.meta.preview_bands_swir) if attrName == 'quicklook_swir' else \ 

285 getattr(self, attrName) 

286 

287 if attr_gA is not None: 

288 attr_gA.save(outpath, **kwargs_save) 

289 outpaths[attrName] = outpath 

290 else: 

291 if attrName.startswith('polymer_') and \ 

292 (not self.cfg.polymer_additional_results or self.cfg.mode_ac == 'land'): 

293 # Do not show a warning if a Polymer product was intentionally not produced and cannot be saved. 

294 pass 

295 else: 

296 self.logger.warning(f"The '{attrName}' attribute cannot be saved because it does not exist in the " 

297 f"current EnMAP image.") 

298 

299 # TODO remove GDAL's *.aux.xml files? 

300 

301 # save metadata 

302 self.meta.add_product_fileinformation(filepaths=list(outpaths.values())) 

303 metadata_string = self.meta.to_XML() 

304 

305 with open(outpaths['metaxml'], 'w') as metaF: 

306 self.logger.info("Writing metadata to %s" % outpaths['metaxml']) 

307 metaF.write(metadata_string) 

308 

309 self.logger.info("L2A product successfully written!") 

310 

311 return product_dir