import os import sys import json import glob import re from collections import OrderedDict # band reference should be required to have the format # _ # instead, the sensor name should be stored somewhere else, # and band names should be STAC common names, see # https://stacspec.org/ # https://github.com/radiantearth/stac-spec/blob/master/extensions/eo/README.md#band-object # custom names must be possible class BandReferenceReaderError(Exception): pass class BandReferenceReader: """Band references reader""" def __init__(self): self._json_files = glob.glob( os.path.join(os.environ['GISBASE'], 'etc', 'g.bands', '*.json') ) if not self._json_files: raise BandReferenceReaderError("No band definitions found") self._read_config() def _read_config(self): """Read configuration""" self.config = dict() for json_file in self._json_files: try: with open(json_file) as fd: config = json.load( fd, object_pairs_hook=OrderedDict ) except json.decoder.JSONDecodeError as e: raise BandReferenceReaderError( "Unable to parse '{}': {}".format( json_file, e )) # check if configuration is valid self._check_config(config) self.config[os.path.basename(json_file)] = config @staticmethod def _check_config(config): """Check if config is valid :todo: check shortcut uniqueness :param dict config: configuration to be validated """ for items in config.values(): for item in ('shortcut', 'bands'): if item not in items.keys(): raise BandReferenceReaderError( "Invalid band definition: <{}> is missing".format(item )) if len(items['bands']) < 1: raise BandReferenceReaderError( "Invalid band definition: no bands defined" ) @staticmethod def _print_band_extended(band, item): """Print band-specific metadata :param str band: band identifier :param str item: items to be printed out """ def print_kv(k, v, indent): if isinstance(v, OrderedDict): print ('{}{}:'.format(' ' * indent * 2, k)) for ki, vi in v.items(): print_kv(ki, vi, indent * 2) else: print ('{}{}: {}'.format(' ' * indent * 2, k, v)) indent = 4 print ('{}band: {}'.format( ' ' * indent, band )) for k, v in item[band].items(): print_kv(k, v, indent) def _print_band(self, shortcut, band, tag=None): sys.stdout.write(self._band_identifier(shortcut, band)) if tag: sys.stdout.write(' {}'.format(tag)) sys.stdout.write(os.linesep) def print_info(self, shortcut=None, band=None, extended=False): """Prints band reference information to stdout. Can be filtered by shortcut or band identifier. :param str shortcut: shortcut to filter (eg. S2) or None :param str band: band (eg. 1) or None :param bool extended: print also extended metadata """ found = False for root in self.config.values(): for item in root.values(): try: if shortcut and re.match(shortcut, item['shortcut']) is None: continue except re.error as e: raise BandReferenceReaderError( "Invalid pattern: {}".format(e) ) found = True if band and band not in item['bands']: raise BandReferenceReaderError( "Band <{}> not found in <{}>".format( band, shortcut )) # print generic information if extended: for subitem in item.keys(): if subitem == 'bands': # bands item is processed bellow continue print ('{}: {}'.format( subitem, item[subitem] )) # print detailed band information if band: self._print_band_extended(band, item['bands']) else: for iband in item['bands']: self._print_band_extended(iband, item['bands']) else: # basic information only if band: self._print_band( item['shortcut'], band, item['bands'][band].get('tag') ) else: for iband in item['bands']: self._print_band( item['shortcut'], iband, item['bands'][iband].get('tag') ) # raise error when defined shortcut not found if shortcut and not found: raise BandReferenceReaderError( "Band reference <{}> not found".format(shortcut) ) def find_file(self, band_reference): """Find file by band reference. Match is case-insensitive. :param str band_reference: band reference identifier to search for (eg. S2_1) :return str: file basename if found or None """ try: shortcut, band = band_reference.split('_') except ValueError: # raise BandReferenceReaderError("Invalid band identifier <{}>".format( # band_reference # )) shortcut = "unknown" band = band_reference for filename, config in self.config.items(): for root in config.keys(): if config[root]['shortcut'].upper() == shortcut.upper() and \ band.upper() in map(lambda x: x.upper(), config[root]['bands'].keys()): return filename return None def get_bands(self): """Get list of band identifiers. :return list: list of valid band identifiers """ bands = [] for root in self.config.values(): for item in root.values(): for band in item['bands']: bands.append( self._band_identifier(item['shortcut'], band) ) return bands @staticmethod def _band_identifier(shortcut, band): return '{}_{}'.format(shortcut, band)