Add KmlPlaceNameTag and refactor ExifTag a bit

idle-code · idle-code · commit e8dbc5e92c4e · 2025-05-17T15:16:26.000+02:00
diff --git a/pyproject.toml b/pyproject.toml
@@ -36,6 +36,7 @@ isodate = "^0.6.1"
 gpxpy = "1.5.0"
 Pint = "^0.21.0"
 geopy = "^2.4.1"
+fastkml = "^1.1.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.5"
diff --git a/tempren/tags/geo.py b/tempren/tags/geo.py
@@ -1,11 +1,144 @@
-from tempren.primitives import Tag
+import itertools
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List, Optional
 
+import fastkml.kml
+from geopy import Point
+from geopy.distance import great_circle
 
-class PlaceNameTag(Tag):
-    """Convert"""
+from tempren.primitives import File, Tag
 
-    require_context = False
+
+@dataclass
+class Place:
+    name: str
+    latitude: float
+    longitude: float
+    radius: Optional[float]
+
+    @property
+    def position(self) -> Point:
+        return Point(self.latitude, self.longitude)
+
+    def __str__(self):
+        return f"{self.name}, {self.latitude}, {self.longitude}, {self.radius}"
+
+
+class KmlPlaceNameTag(Tag):
+    """Convert GPS latitude/longitude coordinates into names based on the file"""
+
+    require_context = True
+
+    log: logging.Logger
+
+    # TODO: Introduce global cache for storing data between different files processing
+    places: List[Place]
+    default: str
+
+    def __init__(self):
+        self.log = logging.getLogger(self.__class__.__name__)
+
+    def configure(
+        self,
+        kml: str,
+        use_look_at: bool = True,
+        use_folders: bool = True,
+        default: str = "",
+    ) -> None:
+        """
+        :param kml: Path to the KML file containing the places
+        :param use_look_at: Use LookAt attributes (instead of Point) of the placemark when looking for best matches
+        :param use_folders: Add folder itself (using its LookAt) as POI
+        :param default: Default value used when there is no matching POI
+        Given coordinates, select best matching (closest) placemark and return its name.
+        When placemarks are grouped into folders, returned name will be prefixed with it, i.e.
+           FolderName/PlacemarkName
+        """
+
+        if not Path(kml).exists():
+            raise FileNotFoundError(kml)
+
+        self.places = self._load_file(kml, use_look_at, use_folders)
+        self.default = default
+
+    def _load_file(
+        self, kml_path: str, use_look_at: bool, use_folders: bool
+    ) -> List[Place]:
+        self.log.debug("Loading %s", kml_path)
+        kml_file = fastkml.kml.KML.parse(kml_path)
+
+        def _collect_places(kml, prefix: List[str]) -> List[Place]:
+            if isinstance(kml, fastkml.containers.Folder):
+                if use_folders and kml.view is not None:
+                    folder_place = Place(
+                        "/".join(prefix + [kml.name]),
+                        kml.view.latitude,
+                        kml.view.longitude,
+                        radius=kml.view.range,
+                    )
+                    prefix = prefix + [kml.name]
+                    return [folder_place] + list(
+                        itertools.chain.from_iterable(
+                            [_collect_places(f, prefix) for f in kml.features]
+                        )
+                    )
+                prefix = prefix + [kml.name]
+            elif isinstance(kml, fastkml.containers.Placemark):
+                if use_look_at and kml.view is not None:
+                    return [
+                        Place(
+                            "/".join(prefix + [kml.name]),
+                            kml.view.latitude,
+                            kml.view.longitude,
+                            radius=kml.view.range,
+                        )
+                    ]
+                else:
+                    latitude = kml.kml_geometry.kml_coordinates.coords[0][0]
+                    longitude = kml.kml_geometry.kml_coordinates.coords[0][1]
+                    radius = kml.view.range if kml.view is not None else None
+                    return [
+                        Place(
+                            "/".join(prefix + [kml.name]),
+                            latitude,
+                            longitude,
+                            radius=radius,
+                        )
+                    ]
+
+            if hasattr(kml, "features"):
+                return list(
+                    itertools.chain.from_iterable(
+                        [_collect_places(f, prefix) for f in kml.features]
+                    )
+                )
+            return []
+
+        places = _collect_places(kml_file, [])
+
+        self.log.debug("Found %d places:", len(places))
+        for place in places:
+            self.log.debug("%s", place)
+
+        return places
 
     def process(self, file: File, context: Optional[str]) -> str:
-        assert context is None
-        return _calculate_hash(hashlib.md5(), file.absolute_path, CHUNK_SIZE)
+        assert context is not None
+        coordinates = Point.from_string(context)
+        place = self._find_best_match(self.places, coordinates)
+        if place is None:
+            return self.default
+        return place.name
+
+    @staticmethod
+    def _find_best_match(places: List[Place], point: Point) -> Optional[Place]:
+        gc = great_circle()
+        min_distance_place: Optional[Place] = None
+        for place in places:
+            if min_distance_place is None or gc.measure(
+                place.position, point
+            ) < gc.measure(min_distance_place.position, point):
+                min_distance_place = place
+        return min_distance_place
diff --git a/tempren/tags/image.py b/tempren/tags/image.py
@@ -5,6 +5,7 @@
 
 import piexif
 import PIL
+from geopy import Point
 from piexif import TAGS
 from piexif import TYPES as TAG_TYPES
 from PIL.Image import Image
@@ -122,48 +123,62 @@ def extract_metadata(self, image: Image) -> bool:
         raise NotImplementedError()
 
 
+def extract_exif_value(exif_dict, tag_name: str):
+    tag_id, tag_type = tag_name_to_id_type(tag_name)
+    for src in exif_dict.values():
+        if isinstance(src, dict) and tag_id in src:
+            tag_value = src[tag_id]
+            return convert_tag_value(tag_type, tag_value)
+    else:
+        raise MissingMetadataError()
+
+
+def tag_name_to_id_type(tag_name: str) -> Tuple[int, int]:
+    for _, tags in TAGS.items():
+        for tag_id, description in tags.items():
+            if description["name"] == tag_name:
+                return tag_id, description["type"]
+    raise ValueError(f"Could not find tag id for '{tag_name}'")
+
+
+def convert_tag_value(tag_type, tag_value):
+    if tag_type in (TAG_TYPES.Rational, TAG_TYPES.SRational):
+        if isinstance(tag_value[0], tuple):
+            if len(tag_value) == 3:
+                # We are probably dealing with GPS coordinates
+                (
+                    degrees,
+                    minutes,
+                    seconds,
+                ) = (convert_tag_value(tag_type, v) for v in tag_value)
+                return f"{degrees}°{minutes}′{seconds}″"
+            return " ".join(str(convert_tag_value(tag_type, v)) for v in tag_value)
+        else:
+            if tag_value[1] == 1:
+                return tag_value[0]
+            else:
+                return tag_value[0] / tag_value[1]
+    if tag_type == TAG_TYPES.Ascii:
+        return tag_value.decode("ascii")
+    return tag_value
+
+
 class ExifTag(Tag):
     """Extract value of any EXIF tag"""
 
-    tag_id: int
-    tag_type: int
+    tag_name: str
 
     def configure(self, tag_name: str):  # type: ignore
         """
         :param tag_name: name of the tag to extract (e.g. 'FocalLength', 'DateTime', 'FNumber')
         """
         # TODO: generate list of supported tags dynamically
         assert tag_name, "expected non empty tag name"
-        self.tag_id, self.tag_type = self._tag_name_to_id_type(tag_name)
-
-    @staticmethod
-    def _tag_name_to_id_type(tag_name: str) -> Tuple[int, int]:
-        for _, tags in TAGS.items():
-            for tag_id, description in tags.items():
-                if description["name"] == tag_name:
-                    return tag_id, description["type"]
-        raise ValueError(f"Could not find tag id for '{tag_name}'")
+        self.tag_name = tag_name
 
     def process(self, file: File, context: Optional[str]) -> Any:
         exif_dict = piexif.load(str(file.absolute_path))
-        for src in exif_dict.values():
-            if isinstance(src, dict) and self.tag_id in src:
-                return self._extract_value(self.tag_type, src[self.tag_id])
-        else:
-            raise MissingMetadataError()
-
-    def _extract_value(self, tag_type, tag_value):
-        if tag_type in (TAG_TYPES.Rational, TAG_TYPES.SRational):
-            if isinstance(tag_value[0], tuple):
-                return " ".join(str(self._extract_value(v)) for v in tag_value)
-            else:
-                if tag_value[1] == 1:
-                    return tag_value[0]
-                else:
-                    return round(tag_value[0] / tag_value[1], 1)
-        if tag_type == TAG_TYPES.Ascii:
-            return tag_value.decode("ascii")
-        return tag_value
+        return extract_exif_value(exif_dict, self.tag_name)
 
 
 _tag_type_map = {
@@ -195,13 +210,32 @@ class ResolutionTagAlias(TagAlias):
     """%Image.Width()x%Image.Height()"""
 
 
-class GpsPositionTag(ExifTag):
+class GpsPositionTag(Tag):
     """Latitude and longitude of place where photo was taken"""
 
-    def configure(self):
-        self.latitude_tag_id, self.latitude_tag_type = self._tag_name_to_id_type(
-            "GpsLatitude"
-        )
+    require_context = False
+
+    use_decimal: bool
+
+    def configure(self, decimal: bool = True):
+        """
+        :param decimal: Use simpler decimal notation instead of degrees-minutes-seconds
+        """
+        self.use_decimal = decimal
+
+    def process(self, file: File, context: Optional[str]) -> Any:
+        assert context is None
+
+        exif_dict = piexif.load(str(file.absolute_path))
+        latitude = extract_exif_value(exif_dict, "GPSLatitude")
+        latitude_ref = extract_exif_value(exif_dict, "GPSLatitudeRef")
+        longitude = extract_exif_value(exif_dict, "GPSLongitude")
+        longitude_ref = extract_exif_value(exif_dict, "GPSLongitudeRef")
+
+        degrees_notation = f"{latitude}{latitude_ref}, {longitude}{longitude_ref}"
+        if self.use_decimal:
+            return Point.from_string(degrees_notation).format_decimal()
+        return degrees_notation
 
 
 class HasGpsPositionTagAlias(TagAlias):
