Module delta.extensions.sources.worldview

Functions to support the WorldView satellites.

Functions

def get_files_from_unpack_folder(folder)

Return the image and header file paths from the given unpack folder. Returns (None, None) if the files were not found.

def toa_preprocess(image, calc_reflectance=False)

Set a WorldviewImage's preprocessing function to do worldview TOA correction. Using the reflectance calculation is slightly more complicated but may be more useful.

def unpack_wv_to_folder(zip_path, unpack_folder)

Classes

class WorldviewImage (paths, nodata_value=None)

Compressed WorldView image. Loads an image from a zip file with a tiff and a .imd file.

Opens a geotiff for reading.

Parameters

paths : str or List[str]
Either a single filename or a list. For a list, the images are opened in order as a multi-band image, assumed to overlap.
nodata_value : dtype of image
Value representing no data.
Expand source code 
class WorldviewImage(tiff.TiffImage):
    """Compressed WorldView image. Loads an image from a zip file with a tiff and a .imd file."""
    def __init__(self, paths, nodata_value=None):
        self._meta_path = None
        self._meta   = None
        self._sensor = None
        self._date   = None
        self._name   = None
        super().__init__(paths, nodata_value)

    def _unpack(self, zip_path):
        # Get the folder where this will be stored from the cache manager
        unpack_folder = config.io.cache.manager().register_item(self._name)
        return unpack_wv_to_folder(zip_path, unpack_folder)

    def _set_info_from_tif_name(self, tif_name):
        parts = os.path.basename(tif_name).split('_')
        self._sensor = parts[0][0:4]
        self._date   = parts[2][6:14]
        self._name   = os.path.splitext(os.path.basename(tif_name))[0]


    # This function is currently set up for the HDDS archived WV data, files from other
    #  locations will need to be handled differently.
    def _prep(self, paths):
        """Prepares a WorldView file from the archive for processing.
           Returns the path to the file ready to use.
           TODO: Apply TOA conversion!
        """
        assert isinstance(paths, str)
        (_, ext) = os.path.splitext(paths)
        tif_name = None

        if ext == '.zip': # Need to unpack

            zip_file = zipfile.ZipFile(paths, 'r')
            tif_names = list(filter(lambda x: x.lower().endswith('.tif'), zip_file.namelist()))
            assert len(tif_names) > 0, f'Error: no tif files in the file {paths}'
            assert len(tif_names) == 1, f'Error: too many tif files in {paths}: {tif_names}'
            tif_name = tif_names[0]

            self._set_info_from_tif_name(tif_name)

            (tif_path, imd_path) = self._unpack(paths)

        if ext == '.tif': # Already unpacked

            # Both files should be present in the same folder
            tif_name = paths
            unpack_folder = os.path.dirname(paths)
            (tif_path, imd_path) = get_files_from_unpack_folder(unpack_folder)

            if not (imd_path and tif_path):
                raise Exception('vendor_metadata not found in %s.' % (paths))
            self._set_info_from_tif_name(tif_name)

        assert tif_name is not None, f'Error: Unsupported extension {ext}'

        self._meta_path = imd_path
        self.__parse_meta_file(imd_path)

        return [tif_path]

    def meta_path(self):
        return self._meta_path

    def __parse_meta_file(self, meta_path):
        """Parse out the needed values from the IMD or XML file"""

        if not os.path.exists(meta_path):
            raise Exception('Metadata file not found: ' + meta_path)

        # TODO: Add more tags!
        # These are all the values we want to read in
        DESIRED_TAGS = ['ABSCALFACTOR', 'EFFECTIVEBANDWIDTH']

        data = {'ABSCALFACTOR':[],
                'EFFECTIVEBANDWIDTH':[]}

        with open(meta_path, 'r') as f:
            for line in f:

                upline = line.replace(';','').upper().strip()

                if 'MEANSUNEL = ' in upline:
                    value = upline.split('=')[-1]
                    data['MEANSUNEL'] = float(value)

                if 'SATID = ' in upline:
                    value = upline.split('=')[-1].replace('"','').strip()
                    data['SATID'] = value

                # Look for the other info we want
                for tag in DESIRED_TAGS:
                    if tag in upline:

                        # Add the value to the appropriate list
                        # -> If the bands are not in order we will need to be more careful here.
                        parts = upline.split('=')
                        value = parts[1]
                        data[tag].append(float(value))

        self._meta_path = meta_path
        self._meta = data

    def scale(self):
        return self._meta['ABSCALFACTOR']
    def bandwidth(self):
        return self._meta['EFFECTIVEBANDWIDTH']

Ancestors

Methods

def bandwidth(self)
def meta_path(self)
def scale(self)

Inherited members