blockutils.raster
¶
Common raster handling methods shared between blocks
create_multiband_tif(list_tif_files, filename_path, band_descriptions=None, drop_nodata=False, colorinterp=None, return_cog=True)
¶
This method combines list of tif files and save them in one tif with each of tif files to be a separate band
Parameters:
Name | Type | Description | Default |
---|---|---|---|
list_tif_files |
List[pathlib.Path] |
List of separate tif files. |
required |
filename_path |
Path |
Final path for a single multi-bands tif file. |
required |
band_descriptions |
List[str] |
Description of each band. |
None |
drop_nodata |
bool |
Ignore nodata in final output. |
False |
use_colorinterp |
Add color interpretation for each band. |
required | |
return_cog |
bool |
Whether convert a tif to cog tif. Default is true. |
True |
Source code in blockutils/raster.py
def create_multiband_tif(
list_tif_files: List[Path],
filename_path: Path,
band_descriptions: List[str] = None,
drop_nodata: bool = False,
colorinterp: List[ColorInterp] = None,
return_cog: bool = True,
):
"""
This method combines list of tif files and save them
in one tif with each of tif files to be a separate band
Args:
list_tif_files: List of separate tif files.
filename_path: Final path for a single multi-bands tif file.
band_descriptions: Description of each band.
drop_nodata: Ignore nodata in final output.
use_colorinterp: Add color interpretation for each band.
return_cog: Whether convert a tif to cog tif. Default is true.
"""
with rio.Env():
list_tif_files_dataset = [rio.open(tif_file) for tif_file in list_tif_files]
raster_profile = _get_profile(
list_tif_files_dataset, band_descriptions, drop_nodata, colorinterp
)
with rio.open(filename_path, "w", **raster_profile) as dst:
for _, window in list_tif_files_dataset[0].block_windows(1):
src = []
for tif_dataset in list_tif_files_dataset:
src.append(tif_dataset.read(window=window))
dst.write(np.concatenate(src, axis=0), window=window)
if band_descriptions:
for b_id, layer in enumerate(band_descriptions):
dst.set_band_description(b_id + 1, layer)
if colorinterp:
dst.colorinterp = colorinterp
if return_cog:
to_cog(filename_path)
is_empty(path_to_image, nodataval=0)
¶
Tests if a created geotiff image only consists of nodata or NaN values Converts NaN to nodata values as a side effect
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path_to_image |
Path |
Path object pointing to geotiff image |
required |
nodataval |
no data value, default is 0 |
0 |
Source code in blockutils/raster.py
def is_empty(path_to_image: Path, nodataval=0) -> bool:
"""
Tests if a created geotiff image only consists of nodata or NaN values
Converts NaN to nodata values as a side effect
Args:
path_to_image: Path object pointing to geotiff image
nodataval: no data value, default is 0
Returns: True if image is empty, False otherwise
"""
with rio.open(str(path_to_image)) as img_file:
data = img_file.read()
np.nan_to_num(data, nan=nodataval, copy=False)
return not np.any(data - nodataval)
to_cog(path_to_image, profile='deflate', **options)
¶
Converts a GeoTIFF into a Cloud-optimized GeoTIFF :param path_to_image: path to GeoTIFF :param profile: compression profile :param options: additional kwargs :return: True if all went well
Source code in blockutils/raster.py
def to_cog(path_to_image: Path, profile: str = "deflate", **options) -> bool:
"""
Converts a GeoTIFF into a Cloud-optimized GeoTIFF
:param path_to_image: path to GeoTIFF
:param profile: compression profile
:param options: additional kwargs
:return: True if all went well
"""
logger.info("Now converting to COG")
tmp_file_path = Path(str(path_to_image) + ".tmp")
path_to_image.rename(tmp_file_path)
# Format creation option (see gdalwarp `-co` option)
output_profile = cog_profiles.get(profile)
output_profile.update(dict(BIGTIFF="IF_SAFER"))
# Dataset Open option (see gdalwarp `-oo` option)
config = dict(
GDAL_NUM_THREADS="ALL_CPUS",
GDAL_TIFF_INTERNAL_MASK=True,
GDAL_TIFF_OVR_BLOCKSIZE="128",
)
cog_translate(
str(tmp_file_path),
str(path_to_image),
output_profile,
config=config,
in_memory=False,
quiet=False,
**options,
)
tmp_file_path.unlink()
return True