|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from typing import TYPE_CHECKING, Literal, Self |
| 4 | + |
| 5 | +from async_tiff import TIFF, ImageFileDirectory, ObspecInput |
| 6 | +from async_tiff.store import ObjectStore |
| 7 | + |
| 8 | +from async_geotiff.enums import Compression, Interleaving, PhotometricInterp |
| 9 | + |
| 10 | +if TYPE_CHECKING: |
| 11 | + import pyproj |
| 12 | + from affine import Affine |
| 13 | + |
| 14 | + |
| 15 | +class GeoTIFF: |
| 16 | + """A class representing a GeoTIFF dataset.""" |
| 17 | + |
| 18 | + _tiff: TIFF |
| 19 | + """The underlying async-tiff TIFF instance that we wrap. |
| 20 | + """ |
| 21 | + |
| 22 | + def __init__(self, tiff: TIFF) -> None: |
| 23 | + """Create a GeoTIFF from an existing TIFF instance.""" |
| 24 | + # Validate that this is indeed a GeoTIFF |
| 25 | + if not has_geokeys(tiff.ifds[0]): |
| 26 | + raise ValueError("TIFF does not contain GeoTIFF keys") |
| 27 | + |
| 28 | + self._tiff = tiff |
| 29 | + |
| 30 | + @classmethod |
| 31 | + async def open( |
| 32 | + cls, |
| 33 | + path: str, |
| 34 | + *, |
| 35 | + store: ObjectStore | ObspecInput, |
| 36 | + prefetch: int = 32768, |
| 37 | + multiplier: int | float = 2.0, |
| 38 | + ) -> Self: |
| 39 | + """Open a new TIFF. |
| 40 | +
|
| 41 | + Args: |
| 42 | + path: The path within the store to read from. |
| 43 | + store: The backend to use for data fetching. |
| 44 | + prefetch: The number of initial bytes to read up front. |
| 45 | + multiplier: The multiplier to use for readahead size growth. Must be |
| 46 | + greater than 1.0. For example, for a value of `2.0`, the first metadata |
| 47 | + read will be of size `prefetch`, and then the next read will be of size |
| 48 | + `prefetch * 2`. |
| 49 | +
|
| 50 | + Returns: |
| 51 | + A TIFF instance. |
| 52 | + """ |
| 53 | + tiff = await TIFF.open( |
| 54 | + path=path, store=store, prefetch=prefetch, multiplier=multiplier |
| 55 | + ) |
| 56 | + return cls(tiff) |
| 57 | + |
| 58 | + @property |
| 59 | + def block_shapes(self) -> list[tuple[int, int]]: |
| 60 | + """An ordered list of block shapes for each bands |
| 61 | +
|
| 62 | + Shapes are tuples and have the same ordering as the dataset's shape: |
| 63 | +
|
| 64 | + - (count of image rows, count of image columns). |
| 65 | + """ |
| 66 | + raise NotImplementedError() |
| 67 | + |
| 68 | + def block_size(self, bidx: int, i: int, j: int) -> int: |
| 69 | + """Returns the size in bytes of a particular block. |
| 70 | +
|
| 71 | + Args: |
| 72 | + bidx: Band index, starting with 1. |
| 73 | + i: Row index of the block, starting with 0. |
| 74 | + j: Column index of the block, starting with 0. |
| 75 | + """ |
| 76 | + raise NotImplementedError() |
| 77 | + |
| 78 | + @property |
| 79 | + def bounds(self) -> tuple[float, float, float, float]: |
| 80 | + """Returns the bounds of the dataset in the units of its coordinate reference system. |
| 81 | +
|
| 82 | + Returns: |
| 83 | + (lower left x, lower left y, upper right x, upper right y) |
| 84 | + """ |
| 85 | + raise NotImplementedError() |
| 86 | + |
| 87 | + @property |
| 88 | + def colorinterp(self) -> list[str]: |
| 89 | + """The color interpretation of each band in index order.""" |
| 90 | + # TODO: we should return an enum here. The enum should match rasterio. |
| 91 | + raise NotImplementedError() |
| 92 | + |
| 93 | + def colormap(self, bidx: int) -> dict[int, tuple[int, int, int]]: |
| 94 | + """Returns a dict containing the colormap for a band. |
| 95 | +
|
| 96 | + Args: |
| 97 | + bidx: The 1-based index of the band whose colormap will be returned. |
| 98 | +
|
| 99 | + Returns: |
| 100 | + Mapping of color index value (starting at 0) to RGBA color as a |
| 101 | + 4-element tuple. |
| 102 | +
|
| 103 | + Raises: |
| 104 | + ValueError |
| 105 | + If no colormap is found for the specified band (NULL color table). |
| 106 | + IndexError |
| 107 | + If no band exists for the provided index. |
| 108 | +
|
| 109 | + """ |
| 110 | + raise NotImplementedError() |
| 111 | + |
| 112 | + @property |
| 113 | + def compression(self) -> Compression: |
| 114 | + """The compression algorithm used for the dataset.""" |
| 115 | + # TODO: should return an enum. The enum should match rasterio. |
| 116 | + # Also, is there ever a case where overviews have a different compression from |
| 117 | + # the base image? |
| 118 | + # Should we diverge from rasterio and not have this as a property returning a |
| 119 | + # single string? |
| 120 | + raise NotImplementedError() |
| 121 | + |
| 122 | + @property |
| 123 | + def count(self) -> int: |
| 124 | + """The number of raster bands in the dataset.""" |
| 125 | + raise NotImplementedError() |
| 126 | + |
| 127 | + @property |
| 128 | + def crs(self) -> pyproj.CRS: |
| 129 | + """The dataset’s coordinate reference system.""" |
| 130 | + raise NotImplementedError() |
| 131 | + |
| 132 | + @property |
| 133 | + def dtypes(self) -> list[str]: |
| 134 | + """The data types of each band in index order.""" |
| 135 | + # TODO: not sure what the return type should be. Perhaps we should define a |
| 136 | + # `DataType` enum? |
| 137 | + raise NotImplementedError() |
| 138 | + |
| 139 | + @property |
| 140 | + def height(self) -> int: |
| 141 | + """The height (number of rows) of the dataset.""" |
| 142 | + raise NotImplementedError() |
| 143 | + |
| 144 | + def index( |
| 145 | + self, |
| 146 | + x: float, |
| 147 | + y: float, |
| 148 | + op=None, |
| 149 | + ) -> tuple[int, int]: |
| 150 | + """Get the (row, col) index of the pixel containing (x, y). |
| 151 | +
|
| 152 | + Args: |
| 153 | + x: x value in coordinate reference system |
| 154 | + y: y value in coordinate reference system |
| 155 | + op: function, optional (default: numpy.floor) |
| 156 | + Function to convert fractional pixels to whole numbers |
| 157 | + (floor, ceiling, round) |
| 158 | +
|
| 159 | + Returns: |
| 160 | + (row index, col index) |
| 161 | + """ |
| 162 | + raise NotImplementedError() |
| 163 | + |
| 164 | + def indexes(self) -> list[int]: |
| 165 | + """The 1-based indexes of each band in the dataset |
| 166 | +
|
| 167 | + For a 3-band dataset, this property will be [1, 2, 3]. |
| 168 | + """ |
| 169 | + raise NotImplementedError() |
| 170 | + |
| 171 | + @property |
| 172 | + def interleaving(self) -> Interleaving: |
| 173 | + """The interleaving scheme of the dataset.""" |
| 174 | + # TODO: Should return an enum. |
| 175 | + # https://rasterio.readthedocs.io/en/stable/api/rasterio.enums.html#rasterio.enums.Interleaving |
| 176 | + raise NotImplementedError() |
| 177 | + |
| 178 | + @property |
| 179 | + def is_tiled(self) -> bool: |
| 180 | + """Check if the dataset is tiled.""" |
| 181 | + raise NotImplementedError() |
| 182 | + |
| 183 | + @property |
| 184 | + def nodata(self) -> float | int | None: |
| 185 | + """The dataset's single nodata value.""" |
| 186 | + raise NotImplementedError() |
| 187 | + |
| 188 | + @property |
| 189 | + def photometric(self) -> PhotometricInterp | None: |
| 190 | + """The photometric interpretation of the dataset.""" |
| 191 | + # TODO: should return enum |
| 192 | + # https://rasterio.readthedocs.io/en/stable/api/rasterio.enums.html#rasterio.enums.PhotometricInterp |
| 193 | + raise NotImplementedError() |
| 194 | + |
| 195 | + @property |
| 196 | + def res(self) -> tuple[float, float]: |
| 197 | + """Returns the (width, height) of pixels in the units of its coordinate reference system.""" |
| 198 | + raise NotImplementedError() |
| 199 | + |
| 200 | + @property |
| 201 | + def shape(self) -> tuple[int, int]: |
| 202 | + """Get the shape (height, width) of the full image.""" |
| 203 | + raise NotImplementedError() |
| 204 | + |
| 205 | + @property |
| 206 | + def transform(self) -> Affine: |
| 207 | + """The dataset's georeferencing transformation matrix |
| 208 | +
|
| 209 | + This transform maps pixel row/column coordinates to coordinates in the dataset's coordinate reference system. |
| 210 | + """ |
| 211 | + raise NotImplementedError() |
| 212 | + |
| 213 | + @property |
| 214 | + def width(self) -> int: |
| 215 | + """The width (number of columns) of the dataset.""" |
| 216 | + raise NotImplementedError() |
| 217 | + |
| 218 | + def xy( |
| 219 | + self, |
| 220 | + row: int, |
| 221 | + col: int, |
| 222 | + offset: Literal["center", "ul", "ur", "ll", "lr"] = "center", |
| 223 | + ) -> tuple[float, float]: |
| 224 | + """Get the coordinates x, y of a pixel at row, col. |
| 225 | +
|
| 226 | + The pixel's center is returned by default, but a corner can be returned |
| 227 | + by setting `offset` to one of `ul, ur, ll, lr`. |
| 228 | +
|
| 229 | + Parameters: |
| 230 | + row: Pixel row. |
| 231 | + col: Pixel column. |
| 232 | + offset: Determines if the returned coordinates are for the center of the |
| 233 | + pixel or for a corner. |
| 234 | +
|
| 235 | + """ |
| 236 | + raise NotImplementedError() |
| 237 | + |
| 238 | + |
| 239 | +def has_geokeys(ifd: ImageFileDirectory) -> bool: |
| 240 | + """Check if an IFD has GeoTIFF keys. |
| 241 | +
|
| 242 | + Args: |
| 243 | + ifd: The IFD to check. |
| 244 | +
|
| 245 | + """ |
| 246 | + return ifd.geo_key_directory is not None |
0 commit comments