improve geokey parsing

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]:

aiocogeo/constants.py

@@ -176,6 +176,13 @@ class ColorInterp(enum.IntEnum):
 }
 
 
+GEO_KEYS = {
+    1025: "RasterType",
+    2048: "GeographicType",
+    3072: "ProjectedType",
+}
+
+
 # https://gdal.org/drivers/raster/gtiff.html#metadata
 GDAL_METADATA_TAGS = [
     "DocumentName",

aiocogeo/ifd.py

@@ -8,7 +8,7 @@
 from .compression import Compression
 from .constants import COMPRESSIONS, GDAL_METADATA_TAGS, INTERLEAVE, 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)
@@ -86,7 +90,6 @@ class OptionalTags:
     MinSampleValue: Tag = None
     MaxSampleValue: Tag = None
 
-
     # GeoTiff
     GeoKeyDirectoryTag: Tag = None
     ModelPixelScaleTag: Tag = None
@@ -100,6 +103,7 @@ class OptionalTags:
 @dataclass
 class ImageIFD(OptionalTags, Compression, RequiredTags, IFD):
     _is_alpha: bool = False
+    geo_keys: Optional[GeoKeyDirectory] = None
 
     @property
     def is_alpha(self) -> bool:
@@ -200,7 +204,7 @@ def gdal_metadata(self) -> Dict:
     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

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
@@ -530,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