Geografic-Information-System/lib/python/bandref/reader.py

import os
import sys
import json
import glob
import re
from collections import OrderedDict


# band reference should be required to have the format
# <shortcut>_<band>
# 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)