Cumulus specific gdal utilities

GTiff Creation Options to be a COG:

"-co",
"COMPRESS=LZW",
"-co",
"COPY_SRC_OVERVIEWS=YES",
"-co",
"TILE=YES",

find_band(data_set, attr={})

Return the band number

Parameters:

Name Type Description Default
data_set gdal.Dataset

gdal dataset

required
attr dict, optional

attributes matching those in the metadata, by default {}

{}

Returns:

Type Description
int

band number

Source code in cumulus_geoproc/utils/cgdal.py
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def find_band(data_set: "gdal.Dataset", attr: dict = {}):
    """Return the band number

    Parameters
    ----------
    data_set : gdal.Dataset
        gdal dataset
    attr : dict, optional
        attributes matching those in the metadata, by default {}

    Returns
    -------
    int
        band number
    """
    count = data_set.RasterCount
    for b in range(1, count + 1):
        try:
            raster = data_set.GetRasterBand(b)
            meta = raster.GetMetadata_Dict()
            has_attr = [
                True
                for key, val in attr.items()
                if (key in meta and val in raster.GetMetadataItem(key))
            ]
            if len(has_attr) == len(attr):
                logger.debug(f"{has_attr=}")
                return b

        except RuntimeError as ex:
            logger.error(f"{type(ex).__name__}: {this}: {ex}")
            continue
        finally:
            raster = None

    return None

gdal_calculate(*args)

Implement gdal-utils gdal_calc CLI utility

gdal_translate documentation:

https://gdal.org/programs/gdal_translate.html

Source code in cumulus_geoproc/utils/cgdal.py
203
204
205
206
207
208
209
210
211
212
213
214
215
def gdal_calculate(*args):
    """Implement gdal-utils gdal_calc CLI utility

    gdal_translate documentation:

    https://gdal.org/programs/gdal_translate.html
    """
    argv = [gdal_calc.__file__]
    argv.extend(list(args))

    logger.debug(f"Argvs: {argv=}")

    gdal_calc.main(argv)

gdal_fillnodataval(src, dst, /, *args)

Implement gdal-utils gdal_fillnodata CLI utility as a subprocess

gdal_fillnodata documentation:

https://gdal.org/programs/gdal_fillnodata.html

Source code in cumulus_geoproc/utils/cgdal.py
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def gdal_fillnodataval(src: str, dst: str, /, *args):
    """Implement gdal-utils gdal_fillnodata CLI utility as a subprocess

    gdal_fillnodata documentation:

    https://gdal.org/programs/gdal_fillnodata.html
    """
    argv = ["gdal_fillnodata.py"]
    argv.extend(list(args))
    argv.append(src)
    argv.append(dst)

    logger.debug(f"Argvs: {argv=}")

    try:
        result = subprocess.check_call(argv, cwd=pathlib.PurePath(src).parent)
        return result
    except subprocess.CalledProcessError as ex:
        logger.error(f"{type(ex).__name__}: {this}: {ex}")
        return result

gdal_translate_options(**kwargs)

Return gdal translate options

Add dictionary attributes to use those options for translate

Adding an existing attribute in 'base' will overwright that option

Returns:

Type Description
dict

dictionary of gdal translate options with base option(s)

base = { "format": "COG", }

Source code in cumulus_geoproc/utils/cgdal.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def gdal_translate_options(**kwargs):
    """
    # Return gdal translate options

    Add dictionary attributes to use those options for translate

    Adding an existing attribute in 'base' will overwright that option

    Returns
    -------
    dict
        dictionary of gdal translate options with base option(s)

    base = {
        "format": "COG",
    }
    """
    # COG driver generates overviews while GTiff uses seperate step to build them
    base = {
        "format": "COG",
    }
    return {**base, **kwargs}

gdal_translate_w_options(dst, src, **kwargs)

GDAL Translate wrapper with base configurations

Parameters:

Name Type Description Default
dst str

Output dataset

required
src gdal.Dataset

Dataset object or a filename

required
**kwargs

User defined keyword arguments

{}
Source code in cumulus_geoproc/utils/cgdal.py
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def gdal_translate_w_options(
    dst: str,
    src: gdal.Dataset,
    **kwargs,
):
    """
    # GDAL Translate wrapper with base configurations

    Parameters
    ----------
    dst : str
        Output dataset
    src : gdal.Dataset
        Dataset object or a filename
    **kwargs
        User defined keyword arguments
    """
    base = {
        "format": "COG",
        "bandList": [1],
        "creationOptions": [
            "RESAMPLING=BILINEAR",
            "OVERVIEWS=IGNORE_EXISTING",
            "OVERVIEW_RESAMPLING=BILINEAR",
        ],
    }
    """dict: base (default) options but can be re-asigned"""
    _kwargs = {**base, **kwargs}
    gdal.Translate(
        dst,
        src,
        **_kwargs,
    )

gdal_translate_w_overviews(dst, src, translate_options, resampling=None, overviewlist=[2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048])

Build overviews for the gdal dataset with the resampling algorithm provided

If no sampling algorithm is given, only gdal.Translate() executed

allowable resampling algorithms: nearest,average,rms,bilinear,gauss,cubic,cubicspline,lanczos,average_magphase,mode

Parameters:

Name Type Description Default
dst str

Output dataset name

required
src gdal.Dataset

Dataset object or a filename

required
translate_options dict

Dictionary of creation options

required
resampling str, optional

resampling algorithm, by default None

None
overviewlist List[int], optional

list of integers, by default [2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048]

[2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048]
Source code in cumulus_geoproc/utils/cgdal.py
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def gdal_translate_w_overviews(
    dst: str,
    src: gdal.Dataset,
    translate_options: dict,
    resampling: str = None,
    overviewlist: List[int] = [2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048],
):
    """Build overviews for the gdal dataset with the resampling algorithm provided

    If no sampling algorithm is given, only gdal.Translate() executed

    allowable resampling algorithms:
        nearest,average,rms,bilinear,gauss,cubic,cubicspline,lanczos,average_magphase,mode

    Parameters
    ----------
    dst : str
        Output dataset name
    src : gdal.Dataset
        Dataset object or a filename
    translate_options : dict
        Dictionary of creation options
    resampling : str, optional
        resampling algorithm, by default None
    overviewlist : List[int], optional
        list of integers, by default [2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048]
    """

    resampling_algo = (
        "nearest",
        "average",
        "rms",
        "bilinear",
        "gauss",
        "cubic",
        "cubicspline",
        "lanczos",
        "average_magphase",
        "mode",
    )
    if resampling is not None and resampling not in resampling_algo:
        logger.debug(f"Resampling algorithm {resampling} not available")
        return False
    try:
        if resampling:
            gdal.Translate(
                f"/vsimem/{dst}",
                src,
                format="GTiff",
                creationOptions=[
                    "COMPRESS=LZW",
                    "TILED=YES",
                ],
            )
            _ds = gdal.Open(f"/vsimem/{dst}", gdal.GA_Update)
            _ds.BuildOverviews(resampling=resampling, overviewlist=overviewlist)
            gdal.Translate(
                dst,
                _ds,
                **translate_options,
            )
        else:
            gdal.Translate(
                dst,
                src,
                **translate_options,
            )
        return True
    except RuntimeError as ex:
        logger.error(f"{type(ex).__name__}: {this}: {ex}")
    finally:
        _ds = None
    return False