Coverage for enpt/model/metadata/metadata

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

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

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)

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/>.

30"""EnPT metadata objects for EnMAP data in map geometry."""

32from datetime import datetime

33from lxml import etree as ElementTree

34import logging

35import os

36import fnmatch

37from typing import Union, List, Tuple, Optional # noqa: F401

38from collections import OrderedDict

39import numpy as np

40from py_tools_ds.geo.vector.topology import Polygon, get_footprint_polygon # noqa: F401 # flake8 issue

41from geoarray import GeoArray

43from .metadata_sensorgeo import EnMAP_Metadata_L1B_SensorGeo

44from ...options.config import EnPTConfig

45from ..srf import SRF

46from ...version import __version__

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

51class EnMAP_Metadata_L2A_MapGeo(object):

52 def __init__(self,

53 config: EnPTConfig,

54 meta_l1b: EnMAP_Metadata_L1B_SensorGeo,

55 wvls_l2a: Union[List, np.ndarray],

56 dims_mapgeo: Tuple[int, int, int],

57 grid_res_l2a: Tuple[float, float],

58 logger=None):

59 """EnMAP Metadata class for the metadata of the complete EnMAP L2A product in map geometry incl. VNIR and SWIR.

61 :param config: EnPT configuration object

62 :param meta_l1b: metadata object of the L1B dataset in sensor geometry

63 :param wvls_l2a: list of center wavelengths included in the L2A product

64 :param dims_mapgeo: dimensions of the EnMAP raster data in map geometry, e.g., (1024, 1000, 218)

65 :param grid_res_l2a: Coordinate grid resolution of the L2A product (x, y)

66 :param logger: instance of logging.logger or subclassed

67 """

68 self.cfg = config

69 self._meta_l1b = meta_l1b

70 self.grid_res = grid_res_l2a

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

73 # defaults

74 self.band_means: Optional[np.ndarray] = None # band-wise means in unscaled values (percent for reflectance)

75 self.band_stds: Optional[np.ndarray] = None # band-wise standard deviations in unscaled values

76 self.fileinfos: list = [] # file information for each file belonging to the EnMAP L2A product

78 # input validation

79 if not len(wvls_l2a) == dims_mapgeo[2]:

80 raise ValueError("The list of center wavelength must be qual to the number of bands in the map geometry "

81 "dimensions.")

83 self.proc_level = 'L2A'

84 self.observation_datetime: datetime = meta_l1b.observation_datetime # Date and Time of observation

85 # FIXME VZA may be negative in DLR data

86 self.geom_view_zenith: float = meta_l1b.geom_view_zenith # viewing zenith angle

87 self.geom_view_azimuth: float = meta_l1b.geom_view_azimuth # viewing azimuth angle

88 self.geom_sun_zenith: float = meta_l1b.geom_sun_zenith # sun zenith angle

89 self.geom_sun_azimuth: float = meta_l1b.geom_sun_azimuth # sun azimuth angle

90 self.mu_sun: float = meta_l1b.mu_sun # needed by SICOR for TOARad > TOARef conversion

91 self.earthSunDist: float = meta_l1b.earthSunDist # earth-sun distance

93 # generate file names for L2A output

94 file_ext_l1b = os.path.splitext(meta_l1b.vnir.filename_data)[1]

95 file_ext_l2a = \

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

97 '.BSQ' if self.cfg.output_interleave == 'band' else \

98 '.BIL' if self.cfg.output_interleave == 'line' else \

99 '.BIP'

100

101 def convert_fn(fn):

102 return fn.replace('L1B-', 'L2A-').replace(file_ext_l1b, file_ext_l2a)

103

104 if not self.cfg.is_dummy_dataformat:

105 self.scene_basename = convert_fn(meta_l1b.vnir.filename_data.split('-SPECTRAL_IMAGE')[0])

106 else:

107 self.scene_basename = os.path.splitext(meta_l1b.vnir.filename_data)[0]

108 self.filename_data = convert_fn(meta_l1b.vnir.filename_data).replace('_VNIR', '')

109 self.filename_deadpixelmap_vnir = convert_fn(meta_l1b.vnir.filename_deadpixelmap)

110 self.filename_deadpixelmap_swir = convert_fn(meta_l1b.swir.filename_deadpixelmap)

111 self.filename_quicklook_vnir = convert_fn(meta_l1b.vnir.filename_quicklook)

112 self.filename_quicklook_swir = convert_fn(meta_l1b.swir.filename_quicklook)

113 self.filename_mask_landwater = convert_fn(meta_l1b.vnir.filename_mask_landwater)

114 self.filename_mask_clouds = convert_fn(meta_l1b.vnir.filename_mask_clouds)

115 self.filename_mask_cloudshadow = convert_fn(meta_l1b.vnir.filename_mask_cloudshadow)

116 self.filename_mask_haze = convert_fn(meta_l1b.vnir.filename_mask_haze)

117 self.filename_mask_snow = convert_fn(meta_l1b.vnir.filename_mask_snow)

118 self.filename_mask_cirrus = convert_fn(meta_l1b.vnir.filename_mask_cirrus)

119 self.filename_metaxml = convert_fn(meta_l1b.filename_metaxml)

120

121 # fuse band-wise metadata (sort all band-wise metadata by wavelengths but band number keeps as it is)

122 # get band index order

123 wvls_vswir = np.hstack([self._meta_l1b.vnir.wvl_center,

124 self._meta_l1b.swir.wvl_center])

125 bandidx_order = np.array([np.argmin(np.abs(wvls_vswir - cwl)) for cwl in wvls_l2a])

126

127 self.wvl_center = np.array(wvls_l2a)

128 self.fwhm = np.hstack([meta_l1b.vnir.fwhm, meta_l1b.swir.fwhm])[bandidx_order]

129 self.gains = np.full((dims_mapgeo[2],), 1. / self.cfg.scale_factor_boa_ref) # => value range 0-1

130 self.offsets = np.zeros((dims_mapgeo[2],))

131 self.srf = SRF.from_cwl_fwhm(self.wvl_center, self.fwhm)

132 self.solar_irrad = np.hstack([meta_l1b.vnir.solar_irrad, meta_l1b.swir.solar_irrad])[bandidx_order]

133

134 if not meta_l1b.vnir.nsmile_coef == meta_l1b.swir.nsmile_coef:

135 raise ValueError('Expected equal number of smile coefficients for VNIR and SWIR. Received %d/%s.'

136 % (meta_l1b.vnir.nsmile_coef, meta_l1b.swir.nsmile_coef))

137

138 self.nsmile_coef = meta_l1b.vnir.nsmile_coef

139 self.smile_coef = np.vstack([meta_l1b.vnir.smile_coef, meta_l1b.swir.smile_coef])[bandidx_order, :]

140 self.smile = np.hstack([meta_l1b.vnir.smile, meta_l1b.swir.smile])[:, bandidx_order]

141

142 if not self.cfg.is_dummy_dataformat:

143 all_rpc_coeffs = OrderedDict(list(meta_l1b.vnir.rpc_coeffs.items()) +

144 list(meta_l1b.swir.rpc_coeffs.items()))

145 self.rpc_coeffs = OrderedDict([(k, v) for i, (k, v) in enumerate(all_rpc_coeffs.items())

146 if i in bandidx_order])

147 else:

148 self.rpc_coeffs = OrderedDict()

149

150 self.nrows = dims_mapgeo[0]

151 self.ncols = dims_mapgeo[1]

152 self.nwvl = dims_mapgeo[2]

153 assert self.nwvl == len(self.wvl_center)

154

155 common_UL_UR_LL_LR = self.get_common_UL_UR_LL_LR()

156 self.lon_UL_UR_LL_LR = [lon for lon, lat in common_UL_UR_LL_LR]

157 self.lat_UL_UR_LL_LR = [lat for lon, lat in common_UL_UR_LL_LR]

158 self.ll_mapPoly = get_footprint_polygon(tuple(zip(self.lon_UL_UR_LL_LR,

159 self.lat_UL_UR_LL_LR)), fix_invalid=True)

160 self.epsg = self._meta_l1b.vnir.epsg_ortho

161

162 if meta_l1b.vnir.unit != meta_l1b.swir.unit or meta_l1b.vnir.unitcode != meta_l1b.swir.unitcode:

163 raise RuntimeError('L2A data should have the same radiometric unit for VNIR and SWIR. '

164 'Received %s in %s for VNIR and %s in %s for SWIR.'

165 % (meta_l1b.vnir.unitcode, meta_l1b.vnir.unit,

166 meta_l1b.swir.unitcode, meta_l1b.swir.unit))

167

168 self.unit = meta_l1b.vnir.unit

169 self.unitcode = meta_l1b.vnir.unitcode

170 self.preview_wvls_vnir = meta_l1b.vnir.preview_wvls

171 self.preview_wvls_swir = meta_l1b.swir.preview_wvls

172 self.preview_bands_vnir = meta_l1b.vnir.preview_bands

173 self.preview_bands_swir = np.array([np.argmin(np.abs(self.wvl_center - wvl))

174 for wvl in meta_l1b.swir.preview_wvls]) # must index from VNIR band 0

175

176 self.snr = None

177 if meta_l1b.vnir.snr is not None:

178 assert meta_l1b.swir.snr is not None

179 self.snr = np.dstack([meta_l1b.vnir.snr, meta_l1b.swir.snr])[:, :, bandidx_order]

180

181 def get_common_UL_UR_LL_LR(self):

182 vnir_ulx, vnir_urx, vnir_llx, vnir_lrx = self._meta_l1b.vnir.lon_UL_UR_LL_LR

183 vnir_uly, vnir_ury, vnir_lly, vnir_lry = self._meta_l1b.vnir.lat_UL_UR_LL_LR

184 swir_ulx, swir_urx, swir_llx, swir_lrx = self._meta_l1b.swir.lon_UL_UR_LL_LR

185 swir_uly, swir_ury, swir_lly, swir_lry = self._meta_l1b.swir.lat_UL_UR_LL_LR

186

187 # use OUTER coordinates

188 return ((min([vnir_ulx, swir_ulx]), max([vnir_uly, swir_uly])),

189 (max([vnir_urx, swir_urx]), max([vnir_ury, swir_ury])),

190 (min([vnir_llx, swir_llx]), min([vnir_lly, swir_lly])),

191 (max([vnir_lrx, swir_lrx]), min([vnir_lry, swir_lry])))

192

193 def add_band_statistics(self, datastack_vnir_swir: Union[np.ndarray, GeoArray]):

194 R, C, B = datastack_vnir_swir.shape

195 # NOTE: Multiply by gains to get reflectance in the range 0-1

196 data = datastack_vnir_swir[datastack_vnir_swir.mask_nodata[:]]

197 self.band_means = np.mean(data, axis=0) * self.gains

198 self.band_stds = np.std(data, axis=0) * self.gains

199

200 def add_product_fileinformation(self, filepaths: List[str], sizes: List[int] = None, versions: List[str] = None):

201 self.fileinfos = []

202

203 for i, fp in enumerate(filepaths):

204 ismeta = fp.endswith('METADATA.XML') or fp.endswith('_header.xml') # FIXME

205 if not os.path.exists(fp):

206 if ismeta:

207 pass # does not yet exist

208 else:

209 raise FileNotFoundError(fp)

210

211 ext = os.path.splitext(fp)[1]

212 fmt = 'binary' if ext in ['.GEOTIFF',

213 '.TIF',

214 '.TIFF',

215 '.GTIFF',

216 '.BSQ', '.bsq',

217 '.BIL', '.bil',

218 '.BIP', '.bip',

219 '.JPEG2000'] \

220 else 'xml' if ext == '.XML' \

221 else 'NA'

222 fileinfo_dict = dict(

223 name=os.path.basename(fp),

224 size=sizes[i] if sizes else int(os.path.getsize(fp) / 1024) if not ismeta else '',

225 version=versions[i] if versions else '',

226 format=fmt

227 )

228

229 self.fileinfos.append(fileinfo_dict)

230

231 def to_XML(self) -> str:

232 """Generate an XML metadata string from the L2A metadata."""

233 # use an XML parser that creates properly indented XML files even if new SubElements have been added

234 parser = ElementTree.XMLParser(remove_blank_text=True)

235

236 # parse (use L1B metadata as template)

237 xml = ElementTree.parse(self._meta_l1b.path_xml, parser).getroot()

238

239 if self.cfg.is_dummy_dataformat:

240 self.logger.warning('No XML metadata conversion implemented for datasets different to the DLR format.'

241 'Metadata XML file will be empty.')

242 return ''

243

244 self.logger.warning('Currently, the L2A metadata XML file does not contain all relevant keys and contains '

245 'not updated values!') # FIXME

246

247 def uk(path, value):

248 xml.find(path).text = str(value)

249

250 def create_SubElement(_parent, _tag, attrib={}, _text=None, nsmap=None, **_extra):

251 result = ElementTree.SubElement(_parent, _tag, attrib, nsmap, **_extra)

252 result.text = _text

253 return result

254

255 ############

256 # metadata #

257 ############

258

259 uk("metadata/schema/processingLevel", self.proc_level)

260 uk("metadata/name", self.filename_metaxml)

261 # uk("metadata/comment").text = 'EnMAP Level 0 Product of datatake 987' # FIXME hardcoded

262

263 ##############

264 # processing #

265 ##############

266

267 uk("processing/terrainCorrection", 'Yes') # FIXME hardcoded {Yes, No}

268 uk("processing/ozoneValue", 'NA') # FIXME {[200-500], NA})

269 uk("processing/season", 'NA') # FIXME {summer, winter, NA}

270 uk("processing/productFormat", 'GeoTIFF+Metadata') # FIXME hardcoded

271 # {BSQ+Metadata, BIL+Metadata, BIP+Metadata, JPEG2000+Metadata, GeoTiff+Metadata}

272 uk("processing/mapProjection", 'UTM_Zone_of_Scene_Center') # FIXME hardcoded

273 # {UTM_Zone_of_Scene_Center, UTM_Zone_of_Scene_Center(-1), UTM_Zone_of_Scene_Center(+1),

274 # UTM_Zone_of_Datatake_Center, Geographic, European_Projection_LAEA, NA}

275 uk("processing/DEMDBVersion", 'SRTM-C_v4') # FIXME hardcoded

276 # {SRTM-C-X_vv.rr, best-of-DEM_vv.rr, DEM-derivedfrom-Tandem-X_vv.rr, ASTER-GDEM_vv.rr, NA}

277 uk("processing/correctionType", 'NA') # FIXME hardcoded {Combined, Land_Mode, Water_Mode, NA}

278 uk("processing/cirrusHazeRemoval", 'NA') # FIXME hardcoded {Yes, No}

279 uk("processing/bandInterpolation", 'NA') # FIXME hardcoded {Yes, No}

280 uk("processing/waterType", 'NA') # FIXME hardcoded {Clear, Turbid, Highly_Turbid, NA}

281

282 ########

283 # base #

284 ########

285

286 # TODO update corner coordinates? DLR just uses the same coordinates like in L1B

287 # xml.find("base/spatialCoverage" % lbl).text =

288 uk("base/format", 'ENMAP_%s' % self.proc_level)

289 uk("base/level", self.proc_level)

290 uk("base/size", 'NA') # FIXME Size of product. Attribute unit {byte, Kbyte, Mbyte, Gbyte}

291

292 ############

293 # specific #

294 ############

295

296 uk("specific/code", self.proc_level)

297

298 # delete old band characterisation (as it contains a different set of bands)

299 bChar_root = xml.find('specific/bandCharacterisation')

300 bChar_root.clear()

301

302 # recreate sub-elements for bandCharacterisation with respect to the current set of bands

303 for i in range(self.nwvl):

304 sub = ElementTree.SubElement(bChar_root, 'bandID', number=str(i + 1))

305 for k, val in zip(['wavelengthCenterOfBand', 'FWHMOfBand', 'GainOfBand', 'OffsetOfBand'],

306 [self.wvl_center[i], self.fwhm[i], self.gains[i], self.offsets[i]]):

307 ele = ElementTree.SubElement(sub, k)

308 ele.text = str(val)

309

310 ###########

311 # product #

312 ###########

313

314 if not self.fileinfos:

315 raise ValueError('Product file informations must be added before writing metadata. '

316 'Call add_product_fileinformation() before!')

317

318 # image #

319 #########

320

321 # recreate sub-elements for image (L1B XML contains sub-elements for VNIR and SWIR here, L2A is merged)

322 im_root = xml.find('product/image')

323 im_root.clear()

324

325 merge = ElementTree.SubElement(im_root, 'merge')

326 for subEleArgs in [

327 # tag, attribute dictionary, text

328 ('channels', {}, str(self.nwvl)),

329 ('name', {}, self.filename_data),

330 ('size', {'unit': "Kbyte"}, str([i for i in self.fileinfos if i['name'] == self.filename_data][0]['size'])),

331 ('version', {}, __version__), # we use the EnPT version here

332 ('format', {}, 'binary'),

333 ('dimension',),

334 ('dimensionGeographic',),

335 ('qlChannelsSWIR',),

336 ('qlChannelsVNIR',),

337 ]:

338 create_SubElement(merge, *subEleArgs)

339

340 dim_root = xml.find('product/image/merge/dimension')

341 for subEleArgs in [

342 # tag, attribute dictionary, text

343 ('columns', {}, str(self.ncols)),

344 ('rows', {}, str(self.nrows)),

345 ]:

346 create_SubElement(dim_root, *subEleArgs)

347

348 dimgeo_root = xml.find('product/image/merge/dimensionGeographic')

349 for subEleArgs in [

350 # tag, attribute dictionary, text

351 ('longitude', {'unit': 'DEG'}, 'NA'), # FIXME

352 ('latitude', {'unit': 'DEG'}, 'NA'), # FIXME 0.3314405

353 ]:

354 create_SubElement(dimgeo_root, *subEleArgs)

355

356 qlswir_root = xml.find('product/image/merge/qlChannelsSWIR')

357 for subEleArgs in [

358 # tag, attribute dictionary, text

359 ('red', {}, str(self.preview_wvls_vnir[0])),

360 ('green', {}, str(self.preview_wvls_vnir[1])),

361 ('blue', {}, str(self.preview_wvls_vnir[2])),

362 ]:

363 create_SubElement(qlswir_root, *subEleArgs)

364

365 qlvnir_root = xml.find('product/image/merge/qlChannelsVNIR')

366 for subEleArgs in [

367 # tag, attribute dictionary, text

368 ('red', {}, str(self.preview_wvls_vnir[0])),

369 ('green', {}, str(self.preview_wvls_vnir[1])),

370 ('blue', {}, str(self.preview_wvls_vnir[2])),

371 ]:

372 create_SubElement(qlvnir_root, *subEleArgs)

373

374 # quicklook #

375 #############

376

377 from . import L2A_product_props_DLR

378 for detName, detMetaL1B in zip(['VNIR', 'SWIR'], [self._meta_l1b.vnir, self._meta_l1b.swir]):

379 lbl = L2A_product_props_DLR['xml_detector_label'][detName]

380

381 fN_quicklook = self.filename_quicklook_vnir if detName == 'VNIR' else self.filename_quicklook_swir

382 size_quicklook = [F['size'] for F in self.fileinfos

383 if os.path.splitext(F['name'])[0].endswith('-QL_%s' % detName)][0]

384 uk("product/quicklook/%s/name" % lbl, fN_quicklook)

385 uk("product/quicklook/%s/size" % lbl, str(size_quicklook))

386 uk("product/quicklook/%s/dimension/rows" % lbl, str(self.nrows))

387 uk("product/quicklook/%s/dimension/columns" % lbl, str(self.ncols))

388 # uk("product/quicklook/%s/dimension/dimensionGeographic/longitude" % lbl, 'NA') # FIXME

389 # uk("product/quicklook/%s/dimension/dimensionGeographic/latitude" % lbl, 'NA') # FIXME

390

391 # productFileInformation #

392 ##########################

393

394 # get L1B product file information

395 l1b_fileinfos = xmlSubtree2dict(xml, 'product/productFileInformation/')

396

397 # clear old L1B file information in XML

398 pFI_root = xml.find('product/productFileInformation')

399 pFI_root.clear()

400

401 # recreate sub-elements for productFileInformation according to L2A file information

402 for i, fileInfo in enumerate(self.fileinfos):

403 fn_l1b_exp = fileInfo['name'].replace('L2A', '*').replace('-SPECTRAL_IMAGE', '-SPECTRAL_IMAGE_VNIR')

404 l1b_fileInfo = [fI for fI in l1b_fileinfos.values() if fnmatch.fnmatch(fI['name'], fn_l1b_exp)]

405

406 if l1b_fileInfo:

407 # TODO update file size of METADATA.XML (has not been written yet)

408 fileInfo['size'] = fileInfo['size'] or l1b_fileInfo[0]['size']

409 fileInfo['version'] = fileInfo['version'] or l1b_fileInfo[0]['version']

410 else:

411 # FIXME if no L1B equivalent is found for the file to be written, the file version will be empty ('')

412 pass

413

414 sub = ElementTree.SubElement(pFI_root, 'file', number=str(i))

415

416 for k, kw in zip(['name', 'size', 'version', 'format'], [{}, {'unit': 'kbytes'}, {}, {}]):

417 ele = ElementTree.SubElement(sub, k, **kw)

418 ele.text = str(fileInfo[k])

419

420 # ortho #

421 #########

422

423 if self.epsg == 4326:

424 proj_str = 'Geographic'

425 elif self.epsg == 3035:

426 proj_str = 'LAEA-ETRS89'

427 elif len(str(self.epsg)) == 5 and str(self.epsg)[:3] in ['326', '327']:

428 zone = int(str(self.epsg)[-2:])

429 if not 0 <= zone <= 60:

430 raise RuntimeError('Invalid L2A UTM zone: %d.' % zone)

431 proj_str = 'UTM_Zone%d_North' % zone if str(self.epsg).startswith('326') else 'UTM_Zone%d_South' % zone

432 else:

433 proj_str = 'NA'

434

435 uk('product/ortho/projection', proj_str) # {UTM_ZoneX_North, UTM_ZoneX_South (where X in {1..60}), Geographic, LAEA-ETRS89, NA} # noqa

436 uk('product/ortho/resolution', self.grid_res[0])

437 uk('product/ortho/resampling', self.cfg.ortho_resampAlg)

438

439 # band statistics

440 #################

441

442 if self.band_means is None or self.band_stds is None:

443 raise ValueError('Band statistics have not yet been computed. Compute them first by calling '

444 'add_band_statistics()!')

445

446 # delete old L1B band statistics

447 bStats_root = xml.find('product/bandStatistics')

448 bStats_root.clear()

449

450 # recreate sub-elements for bandStatistics with respect to the current set of bands

451 for i in range(self.nwvl):

452 sub = ElementTree.SubElement(bStats_root, 'bandID', number=str(i + 1))

453 for k, val in zip(['wavelength', 'mean', 'stdDeviation'],

454 [self.wvl_center[i], self.band_means[i], self.band_stds[i]]):

455 ele = ElementTree.SubElement(sub, k)

456 ele.text = str(val)

457

458 #######################

459 # generate XML string #

460 #######################

461

462 xml_string = ElementTree.tostring(xml, encoding='unicode', pretty_print=True)

463

464 return xml_string

465

466

467def xmlSubtree2dict(xml_root, path_subtree) -> OrderedDict:

468 outDict = OrderedDict()

469 allEle = xml_root.findall(path_subtree)

470

471 for ele in allEle:

472 eleKey = '%s_%s' % (ele.tag, ele.get('number'))

473 outDict[eleKey] = dict()

474 for subele in ele:

475 outDict[eleKey][subele.tag] = subele.text

476

477 return outDict

Coverage for enpt/model/metadata/metadata_mapgeo.py: 96%

226 statements