Coverage for enpt/io/reader.py: 84%

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"""Reader module for reading all kinds of EnMAP images.""" 

31 

32import logging 

33import os 

34from fnmatch import filter 

35import numpy as np 

36 

37from ..model.images import EnMAPL1Product_SensorGeo 

38from ..options.config import EnPTConfig 

39 

40__author__ = 'Daniel Scheffler' 

41 

42 

43class L1B_Reader(object): 

44 """Reader for EnMAP Level-1B products.""" 

45 

46 def __init__(self, 

47 config: EnPTConfig, 

48 logger: logging.Logger = None, 

49 root_dir_main: str = None, 

50 root_dir_ext: str = None, 

51 n_line_ext: int = None): 

52 """Get an instance of L1B_Reader. 

53 

54 :param config: instance of EnPTConfig class 

55 :param logger: instance of logging.Logger (NOTE: This logger is only used to log messages within L1B_Reader. 

56 It is not appended to the read L1B EnMAP object). 

57 :param root_dir_main: Root directory of EnMAP Level-1B product (the main image) 

58 :param root_dir_ext: Root directory of EnMAP Level-1B product (to extend the main image) 

59 :param n_line_ext: [Optional] add number of line to be added to the main image from the extended image 

60 """ 

61 self.cfg = config 

62 self.logger = logger or logging.getLogger(__name__) 

63 

64 # read data if root_dir_main is given or not 

65 if root_dir_main is not None: 

66 self.read_inputdata(root_dir_main, root_dir_ext, n_line_ext) 

67 

68 def read_inputdata(self, 

69 root_dir_main, 

70 root_dir_ext: str = None, 

71 n_line_ext: int = None, 

72 compute_snr: bool = True) -> EnMAPL1Product_SensorGeo: 

73 """Read L1B EnMAP data. Extend the image by adding a second image (entire, partial). 

74 

75 :param root_dir_main: Root directory of the main EnMAP Level-1B product 

76 :param root_dir_ext: Root directory of the extended EnMAP Level-1B product (optional) 

77 :param n_line_ext: Number of lines to be added to the main image (if None, use the whole image) 

78 :param compute_snr: whether to compute SNR or not (default: True) 

79 :return: instance of EnMAPL1Product_SensorGeo 

80 """ 

81 self.validate_input(root_dir_main, root_dir_ext) 

82 self.logger.info("Reading Input Data") 

83 

84 # Get a new instance of the EnMAPL1Product_SensorGeo for the main image (TOA radiance) 

85 l1b_main_obj = EnMAPL1Product_SensorGeo(root_dir_main, config=self.cfg, logger=self.logger) 

86 

87 # append a second EnMAP L1B product below if given 

88 if root_dir_ext: 

89 l1b_main_obj.append_new_image(root_dir_ext, n_line_ext) 

90 

91 # compute SNR 

92 if compute_snr: 

93 l1b_main_obj.calc_snr_from_radiance() 

94 

95 # compute geolayer if not already done 

96 if l1b_main_obj.meta.vnir.lons is None or l1b_main_obj.meta.vnir.lats is None: 

97 l1b_main_obj.meta.vnir.lons, l1b_main_obj.meta.vnir.lats = \ 

98 l1b_main_obj.meta.vnir.compute_geolayer_for_cube() 

99 

100 if l1b_main_obj.meta.swir.lons is None or l1b_main_obj.meta.swir.lats is None: 

101 l1b_main_obj.meta.swir.lons, l1b_main_obj.meta.swir.lats = \ 

102 l1b_main_obj.meta.swir.compute_geolayer_for_cube() 

103 

104 # l1b_main_obj.correct_VNIR_SWIR_shift() 

105 

106 # Validate and return the l1b_main_obj 

107 self.validate_output() 

108 return l1b_main_obj 

109 

110 def validate_input(self, root_dir_main, root_dir_ext): 

111 """Validate user inputs.""" 

112 if not self.cfg.is_dummy_dataformat: 

113 self._validate_enmap_l1b_rootdir(root_dir_main) 

114 if root_dir_ext: 

115 self._validate_enmap_l1b_rootdir(root_dir_ext) 

116 

117 @staticmethod 

118 def _validate_enmap_l1b_rootdir(rootdir_l1b): 

119 """Check for valid EnMAP L1B root directory.""" 

120 if not os.path.isdir(rootdir_l1b): 

121 raise NotADirectoryError(rootdir_l1b, 'EnMAP images have to be provided either as zip-archives or as ' 

122 'a directory containing all extracted files.') 

123 

124 files = os.listdir(rootdir_l1b) 

125 

126 if not files: 

127 raise RuntimeError("The root directory of the EnMAP image %s is empty." % rootdir_l1b) 

128 

129 for pattern in [ 

130 # '*-HISTORY.XML', # only included in internal DLR test data, not in the zip archive on enmap.org 

131 # '*-LOG.XML', # only included in internal DLR test data, not in the zip archive on enmap.org 

132 '*-METADATA.XML', 

133 '*-QL_PIXELMASK_SWIR.?', 

134 '*-QL_PIXELMASK_VNIR.?', 

135 '*-QL_QUALITY_CIRRUS.?', 

136 '*-QL_QUALITY_CLASSES.?', 

137 '*-QL_QUALITY_CLOUD.TIF', 

138 '*-QL_QUALITY_CLOUDSHADOW.?', 

139 '*-QL_QUALITY_HAZE.?', 

140 '*-QL_QUALITY_SNOW.?', 

141 '*-QL_QUALITY_TESTFLAGS_SWIR.?', 

142 '*-QL_QUALITY_TESTFLAGS_VNIR.?', 

143 '*-QL_SWIR.?', 

144 '*-QL_VNIR.?', 

145 '*-SPECTRAL_IMAGE_SWIR.?', 

146 '*-SPECTRAL_IMAGE_VNIR.?', 

147 ]: 

148 matches = [] 

149 for ext in ['', '.TIF', '.GEOTIFF', '.BSQ', '.BIL', '.BIP', 'JPEG2000', '.JP2', '.jp2']: 

150 matches.extend(filter(files, pattern.replace('.?', ext))) 

151 

152 if matches: 

153 break 

154 

155 if not matches: 

156 raise FileNotFoundError('The root directory of the EnMAP image %s misses a file with the pattern %s.' 

157 % (rootdir_l1b, pattern)) 

158 

159 def validate_output(self): 

160 """Validate outputs of L1B_Reader.""" 

161 pass 

162 

163 

164def read_solar_irradiance(path_solar_irr_model: str, resol_nm: float = None, wvl_min_nm: float = None, 

165 wvl_max_nm: float = None) -> np.ndarray: 

166 """Read the given solar irradiance file and return an array of irradiances. 

167 

168 :param path_solar_irr_model: file path to solar irradiance model 

169 

170 -> must be arranged like that: 

171 col0 = Wavelength[nm]; col1 = Solar Irradiance [W/m2/µm] 

172 :param resol_nm: spectral resolution for returned irradiances [nanometers] 

173 :param wvl_min_nm: minimum wavelength of returned irradiances [nanometers] 

174 :param wvl_max_nm: maximum wavelength of returned irradiances [nanometers] 

175 :return: 

176 """ 

177 sol_irr = np.loadtxt(path_solar_irr_model, skiprows=1) 

178 if resol_nm is not None and isinstance(resol_nm, (float, int)): 

179 wvl_min = (np.min(sol_irr[:, 0]) if wvl_min_nm is None else wvl_min_nm) 

180 wvl_max = (np.max(sol_irr[:, 0]) if wvl_max_nm is None else wvl_max_nm) 

181 wvl_rsp = np.arange(wvl_min, wvl_max, resol_nm) 

182 sol_irr = np.interp(wvl_rsp, sol_irr[:, 0], sol_irr[:, 1]) 

183 return sol_irr