support GDAL tags (#78)

* support GDAL_METADATA private tag

* update setup.py

* oops

* read more tiff tags included in gdal metadata

* improve geokey parsing (#79)

* improve GeoKeyDirectory parsing

* update test case

* update iter

* support AREA_OR_POINT tag

aiocogeo/cog.py

@@ -120,12 +120,7 @@ def profile(self) -> Dict[str, Any]:
     @property
     def epsg(self) -> int:
         """Return the EPSG code representing the crs of the image"""
-        ifd = self.ifds[0]
-        for idx in range(0, len(ifd.GeoKeyDirectoryTag), 4):
-            # 2048 is geographic crs
-            # 3072 is projected crs
-            if ifd.GeoKeyDirectoryTag[idx] in (2048, 3072):
-                return ifd.GeoKeyDirectoryTag[idx + 3]
+        return self.ifds[0].geo_keys.epsg
 
     @property
     def bounds(self) -> Tuple[float, float, float, float]:
@@ -233,8 +228,6 @@ def color_interp(self):
             interp = [ColorInterp.undefined for _ in range(self.profile['count'])]
         return interp
 
-
-
     @property
     def has_alpha(self) -> bool:
         """Check if the image has an alpha band"""
@@ -248,6 +241,10 @@ def has_alpha(self) -> bool:
     def nodata(self) -> Optional[int]:
         return self.ifds[0].nodata
 
+    @property
+    def gdal_metadata(self) -> Dict:
+        return self.ifds[0].gdal_metadata
+
     async def _read_header(self) -> None:
         """Internal method to read image header and parse into IFDs and Tags"""
         next_ifd_offset = 1

aiocogeo/constants.py

@@ -145,8 +145,19 @@ class ColorInterp(enum.IntEnum):
     258: "BitsPerSample",
     259: "Compression",
     262: "PhotometricInterpretation",
+    269: "DocumentName",
+    270: "ImageDescription",
     277: "SamplesPerPixel",
+    280: "MinSampleValue",
+    281: "MaxSampleValue",
+    282: "XResolution",
+    283: "YResolution",
     284: "PlanarConfiguration",
+    296: "ResolutionUnit",
+    305: "Software",
+    306: "DateTime",
+    315: "Artist",
+    316: "HostComputer",
     317: "Predictor",
     320: "ColorMap",
     322: "TileWidth",
@@ -156,12 +167,46 @@ class ColorInterp(enum.IntEnum):
     338: "ExtraSamples",
     339: "SampleFormat",
     347: "JPEGTables",
+    33432: "Copyright",
     33550: "ModelPixelScaleTag",
     33922: "ModelTiepointTag",
     34735: "GeoKeyDirectoryTag",
+    42112: "GdalMetadata",
     42113: "NoData"
 }
 
+
+GEO_KEYS = {
+    1025: "RasterType",
+    2048: "GeographicType",
+    3072: "ProjectedType",
+}
+
+
+# https://gdal.org/drivers/raster/gtiff.html#metadata
+GDAL_METADATA_TAGS = [
+    "DocumentName",
+    "ImageDescription",
+    "Software",
+    "DateTime",
+    "Artist",
+    "HostComputer",
+    "Copyright",
+    "XResolution",
+    "YResolution",
+    "ResolutionUnit",
+    "MinSampleValue",
+    "MaxSampleValue"
+]
+
+
+RASTER_TYPE = {
+    0: "Unknown",
+    1: "Area",
+    2: "Point",
+}
+
+
 class MaskFlags(enum.IntEnum):
     """https://github.com/mapbox/rasterio/blob/master/rasterio/enums.py#L80-L84"""
     all_valid = 1

aiocogeo/ifd.py

@@ -3,12 +3,12 @@
 from typing import Dict, Optional, Tuple, Union
 
 import numpy as np
+import xmltodict
 
 from .compression import Compression
-from .constants import COMPRESSIONS, INTERLEAVE, SAMPLE_DTYPES
-from .errors import TileNotFoundError
+from .constants import COMPRESSIONS, GDAL_METADATA_TAGS, INTERLEAVE, RASTER_TYPE, SAMPLE_DTYPES
 from .filesystems import Filesystem
-from .tag import Tag
+from .tag import GeoKeyDirectory, Tag
 from .utils import run_in_background
 
 @dataclass
@@ -17,6 +17,7 @@ class IFD:
     tag_count: int
     _file_reader: Filesystem
 
+
     @staticmethod
     def _is_masked(tiff_tags: Dict[str, Tag]) -> bool:
         """Check if an IFD is masked based on a dictionary of tiff tags"""
@@ -45,6 +46,9 @@ async def read(cls, file_reader: Filesystem) -> Union["ImageIFD", "MaskIFD"]:
         file_reader.seek(ifd_start + (12 * tag_count) + 2)
         next_ifd_offset = await file_reader.read(4, cast_to_int=True)
 
+        if 'GeoKeyDirectoryTag' in tiff_tags:
+            tiff_tags['geo_keys'] = GeoKeyDirectory.read(tiff_tags['GeoKeyDirectoryTag'])
+
         # Check if mask
         if cls._is_masked(tiff_tags):
             return MaskIFD(next_ifd_offset, tag_count, file_reader, **tiff_tags)
@@ -73,6 +77,18 @@ class OptionalTags:
     JPEGTables: Tag = None
     ExtraSamples: Tag = None
     ColorMap: Tag = None
+    ImageDescription: Tag = None
+    DocumentName: Tag = None
+    Software: Tag = None
+    DateTime: Tag = None
+    Artist: Tag = None
+    HostComputer: Tag = None
+    Copyright: Tag = None
+    XResolution: Tag = None
+    YResolution: Tag = None
+    ResolutionUnit: Tag = None
+    MinSampleValue: Tag = None
+    MaxSampleValue: Tag = None
 
     # GeoTiff
     GeoKeyDirectoryTag: Tag = None
@@ -81,11 +97,13 @@ class OptionalTags:
 
     # GDAL private tags
     NoData: Tag = None
+    GdalMetadata: Tag = None
 
 
 @dataclass
 class ImageIFD(OptionalTags, Compression, RequiredTags, IFD):
     _is_alpha: bool = False
+    geo_keys: Optional[GeoKeyDirectory] = None
 
     @property
     def is_alpha(self) -> bool:
@@ -158,10 +176,38 @@ def tile_count(self) -> Tuple[int, int]:
             math.ceil(self.ImageHeight.value / float(self.TileHeight.value)),
         )
 
+    @property
+    def gdal_metadata(self) -> Dict:
+        """Return gdal metadata"""
+        meta = {}
+        for tag in GDAL_METADATA_TAGS:
+            inst = getattr(self, tag)
+            if inst is not None:
+                if isinstance(inst.value, tuple):
+                    # TODO: Maybe we are reading one extra byte
+                    val = b"".join(inst.value)[:-1].decode('utf-8')
+                else:
+                    val = inst.value
+                meta[f"TIFFTAG_{tag.upper()}"] = val
+
+        if self.GdalMetadata:
+            xml = b''.join(self.GdalMetadata.value[:-1]).decode('utf-8')
+            parsed = xmltodict.parse(xml)
+            tags = parsed['GDALMetadata']['Item']
+            if isinstance(tags, list):
+                meta.update({tag['@name']:tag['#text'] for tag in tags})
+            else:
+                meta.update({tags['@name']:tags['#text']})
+
+        if self.geo_keys:
+            meta['AREA_OR_POINT'] = RASTER_TYPE[self.geo_keys.RasterType.value]
+
+        return meta
+
     def __iter__(self):
         """Iterate through TIFF Tags"""
         for (k, v) in self.__dict__.items():
-            if k not in ("next_ifd_offset", "tag_count", "_file_reader") and v:
+            if k not in ("next_ifd_offset", "tag_count", "_file_reader", "geo_keys") and v:
                 yield v
 
 

aiocogeo/tag.py

@@ -5,7 +5,7 @@
 from typing import Any, Optional, Tuple, Union
 
 from .config import INGESTED_BYTES_AT_OPEN, LOG_LEVEL
-from .constants import TIFF_TAGS
+from .constants import GEO_KEYS, TIFF_TAGS
 from .filesystems import Filesystem
 
 
@@ -34,13 +34,15 @@ class TagType:
     16: TagType(format="Q", size=8),  # TIFFlong8
 }
 
-
 @dataclass
-class Tag:
+class BaseTag:
     code: int
     name: str
-    tag_type: TagType
     count: int
+
+@dataclass
+class Tag(BaseTag):
+    tag_type: TagType
     length: int
     value: Union[Any, Tuple[Any]]
 
@@ -99,3 +101,43 @@ async def read(cls, reader: Filesystem) -> Optional["Tag"]:
             value=value,
         )
         return tag
+
+
+@dataclass
+class GeoKey(BaseTag):
+    """http://docs.opengeospatial.org/is/19-008r4/19-008r4.html#_geokey"""
+    tag_location: int
+    value: Any
+
+    @classmethod
+    def read(cls, key: Tuple[int, int, int, int]):
+        return cls(
+            code=key[0],
+            tag_location=key[1],
+            count=key[2],
+            value=key[3],
+            name=GEO_KEYS[key[0]]
+        )
+
+
+@dataclass
+class GeoKeyDirectory:
+    """http://docs.opengeospatial.org/is/19-008r4/19-008r4.html#_requirements_class_geokeydirectorytag"""
+    RasterType: GeoKey
+    GeographicType: Optional[GeoKey] = None
+    ProjectedType: Optional[GeoKey] = None
+
+    @classmethod
+    def read(cls, tag: Tag) -> "GeoKeyDirectory":
+        """Parse GeoKeyDirectoryTag"""
+        geokeys = {}
+        assert tag.name == 'GeoKeyDirectoryTag'
+        for idx in range(0, len(tag), 4):
+            if tag[idx] in list(GEO_KEYS):
+                geokeys[GEO_KEYS[tag[idx]]] = GeoKey.read(tag[idx:idx+4])
+        return cls(**geokeys)
+
+    @property
+    def epsg(self) -> int:
+        """Return the EPSG code representing the crs of the image"""
+        return self.ProjectedType.value if self.ProjectedType else self.GeographicType.value

setup.py

@@ -53,7 +53,8 @@
         "typer",
         "Pillow",
         "stac-pydantic>=1.3.*",
-        "geojson-pydantic==0.1.0"
+        "geojson-pydantic==0.1.0",
+        "xmltodict"
     ],
     test_suite="tests",
     setup_requires=[

tests/test_cog_reader.py

@@ -18,7 +18,7 @@
 
 from aiocogeo import config, COGReader
 from aiocogeo.ifd import IFD
-from aiocogeo.tag import Tag
+from aiocogeo.tag import Tag, BaseTag
 from aiocogeo.tiler import COGTiler
 from aiocogeo.errors import InvalidTiffError, TileNotFoundError
 from aiocogeo.constants import MaskFlags
@@ -42,6 +42,14 @@ async def test_cog_metadata(infile, create_cog_reader):
             assert rio_profile == cog_profile
             assert ds.overviews(1) == cog.overviews
 
+            rio_tags = ds.tags()
+            gdal_metadata = cog.gdal_metadata
+
+            for (k,v) in rio_tags.items():
+                if k in ("TIFFTAG_XRESOLUTION", "TIFFTAG_YRESOLUTION", "TIFFTAG_RESOLUTIONUNIT"):
+                    continue
+                assert str(gdal_metadata[k]) == v
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("infile", TEST_DATA)
@@ -522,7 +530,7 @@ async def test_cog_metadata_iter(infile, create_cog_reader):
         for ifd in cog:
             assert isinstance(ifd, IFD)
             for tag in ifd:
-                assert isinstance(tag, Tag)
+                assert isinstance(tag, BaseTag)
 
 
 @pytest.mark.asyncio