Module delta.extensions.sources.landsat

Read Landsat images.

Functions

def get_scene_info(path)

Extract information about the landsat scene from the file name

def toa_preprocess(image, calc_reflectance=False)

Convert landsat files in one folder to TOA corrected files in the output folder. Using the reflectance calculation is slightly more complicated but may be more useful. Multiprocessing is used if multiple processes are specified.

Classes

class LandsatImage (paths, nodata_value=None, bands=None)

Compressed Landsat image. Loads a compressed zip or tar file with a .mtl 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 LandsatImage(tiff.TiffImage):
    """Compressed Landsat image. Loads a compressed zip or tar file with a .mtl file."""

    def __init__(self, paths, nodata_value=None, bands=None):
        self._bands = bands
        super().__init__(paths, nodata_value)

    def _prep(self, paths):
        """Prepares a Landsat file from the archive for processing.
           Returns [band, paths, in, order, ...]
           Uses the bands specified in _get_landsat_bands_to_use()
           TODO: Handle bands which are not 30 meters!
           TODO: Apply TOA conversion!
        """
        scene_info = get_scene_info(paths)
        self._sensor = scene_info['sensor']
        self._lpath = scene_info['lpath']
        self._lrow = scene_info['lrow']
        self._date = scene_info['date']

        # Get the folder where this will be stored from the cache manager
        name = '_'.join([self._sensor, self._lpath, self._lrow, self._date])
        untar_folder = config.io.cache.manager().register_item(name)

        # Check if we already unpacked this data
        all_files_present = False
        if os.path.exists(untar_folder):
            mtl_path = _find_mtl_file(untar_folder)
            if mtl_path:
                mtl_data = _parse_mtl_file(mtl_path)
                all_files_present = _check_if_files_present(mtl_data, untar_folder)

        if all_files_present:
            print('Already have unpacked files in ' + untar_folder)
        else:
            print('Unpacking tar file ' + paths + ' to folder ' + untar_folder)
            utilities.unpack_to_folder(paths, untar_folder)

        bands_to_use = _get_landsat_bands_to_use(self._sensor) if self._bands is None else self._bands

        # Generate all the band file names (the MTL file is not returned)
        self._mtl_path = _find_mtl_file(untar_folder)
        self._mtl_data = _parse_mtl_file(self._mtl_path)
        output_paths = _get_band_paths(self._mtl_data, untar_folder, bands_to_use)

        # Check that the files exist
        for p in output_paths:
            if not os.path.exists(p):
                raise Exception('Did not find expected file: ' + p
                                + ' after unpacking tar file ' + paths)

        return output_paths

    def radiance_mult(self):
        return self._mtl_data['RADIANCE_MULT']
    def radiance_add(self):
        return self._mtl_data['RADIANCE_ADD']
    def reflectance_mult(self):
        return self._mtl_data['REFLECTANCE_MULT']
    def reflectance_add(self):
        return self._mtl_data['REFLECTANCE_ADD']
    def k1_constant(self):
        return self._mtl_data['K1_CONSTANT']
    def k2_constant(self):
        return self._mtl_data['K2_CONSTANT']
    def sun_elevation(self):
        return self._mtl_data['SUN_ELEVATION']

Ancestors

Methods

def k1_constant(self)
def k2_constant(self)
def radiance_add(self)
def radiance_mult(self)
def reflectance_add(self)
def reflectance_mult(self)
def sun_elevation(self)

Inherited members