immich_owntracks/exif2gpx.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
exif2gpx.py — generuje plik GPX z danych GPS w EXIF zdjęć (JPG/JPEG).

Użycie:
    pip install Pillow
    python exif2gpx.py /sciezka/do/folderu_lub_pliku output.gpx --tz +02:00

Opcje:
    --tz  Strefa czasowa, w której interpretować czasy EXIF (gdy EXIF nie ma TZ),
          format +HH:MM lub -HH:MM (np. +02:00). Jeśli podasz, czasy zostaną zapisane
          w GPX w UTC (z literą 'Z'). Jeśli nie podasz, <time> będzie pominięty.
"""
import argparse
import os
import sys
from pathlib import Path
from datetime import datetime, timezone, timedelta
from typing import Optional, Tuple, List
from numbers import Real

from PIL import Image, ExifTags
import xml.etree.ElementTree as ET

# Mapy tagów EXIF
TAGS = {v: k for k, v in ExifTags.TAGS.items()}
GPSTAGS = {v: k for k, v in ExifTags.GPSTAGS.items()}

# Akceptowane rozszerzenia zdjęć
IMAGE_EXTS = {".jpg", ".jpeg"}

def rational_to_float(x) -> float:
    """
    Zamienia wartość EXIF na float.
    Obsługuje: liczby (int/float), IFDRational (numerator/denominator),
    oraz krotki/listy (num, den).
    """
    if x is None:
        raise ValueError("Brak wartości rational")

    # 1) już liczba
    if isinstance(x, Real):
        return float(x)

    # 2) IFDRational (Pillow) albo coś z numerator/denominator
    if hasattr(x, "numerator") and hasattr(x, "denominator"):
        den = x.denominator
        if den == 0:
            raise ValueError("Mianownik równy zero w rational")
        return float(x.numerator) / float(den)

    # 3) krotka/lista (num, den)
    if isinstance(x, (tuple, list)) and len(x) == 2:
        num, den = x
        den = float(den)
        if den == 0:
            raise ValueError("Mianownik równy zero w rational")
        return float(num) / den

    raise TypeError(f"Nieobsługiwany typ rational: {type(x)}")


def dms_to_decimal(dms, ref: str) -> float:
    """
    Konwersja DMS -> stopnie dziesiętne.
    dms: (deg, min, sec) — każdy element może być float/int/IFDRational/(num, den)
    ref: 'N'/'S' lub 'E'/'W'
    """
    if not isinstance(dms, (tuple, list)) or len(dms) != 3:
        raise ValueError(f"DMS musi być 3-elementową krotką/listą, dostałem: {dms!r}")

    deg = rational_to_float(dms[0])
    minutes = rational_to_float(dms[1])
    seconds = rational_to_float(dms[2])

    if not isinstance(ref, str) or ref.upper() not in ("N", "S", "E", "W"):
        raise ValueError(f"Nieprawidłowy ref: {ref!r} (oczekiwano 'N'/'S'/'E'/'W')")

    dec = abs(float(deg)) + float(minutes) / 60.0 + float(seconds) / 3600.0
    if ref.upper() in ("S", "W"):
        dec = -dec
    return dec


def parse_exif_datetime(raw_dt: Optional[str], subsec: Optional[str]) -> Optional[datetime]:
    """
    Parsuje DateTimeOriginal w formacie 'YYYY:MM:DD HH:MM:SS' i składa z sub-sekundami.
    Zwraca naive datetime (bez strefy) – później ewentualnie przeliczane wg --tz.
    """
    if not raw_dt:
        return None
    try:
        base = datetime.strptime(raw_dt, "%Y:%m:%d %H:%M:%S")
        # Subsekundy bywają różne: np. "123", "12", czasem z innymi znakami
        if subsec:
            only_digits = "".join(ch for ch in str(subsec) if ch.isdigit())
            if only_digits:
                micro = int((only_digits + "000000")[:6])
                base = base.replace(microsecond=micro)
        return base
    except Exception as e:
        raise ValueError(f"Nie można sparsować daty EXIF '{raw_dt}' ({subsec=}): {e}")


def apply_tz_and_to_utc(dt: datetime, tz_offset: Optional[str]) -> Optional[datetime]:
    """
    Jeśli podano --tz w formacie +HH:MM lub -HH:MM, interpretuj naive dt w tej strefie,
    a następnie zwróć dt w UTC.
    Jeśli --tz nie podano, zwróć None (żeby pominąć <time>).
    """
    if tz_offset is None:
        return None
    try:
        if not (len(tz_offset) == 6 and tz_offset[0] in "+-" and tz_offset[3] == ":"):
            raise ValueError("Błędny format strefy (oczekiwana postać +HH:MM lub -HH:MM)")
        sign = 1 if tz_offset.startswith("+") else -1
        hh = int(tz_offset[1:3])
        mm = int(tz_offset[4:6])
        offset = timedelta(hours=hh, minutes=mm)
        tzinfo = timezone(sign * offset)
        aware = dt.replace(tzinfo=tzinfo)
        return aware.astimezone(timezone.utc)
    except Exception as e:
        raise ValueError(f"Nieprawidłowa strefa czasowa '{tz_offset}': {e}")


def extract_exif_gps_and_time(img_path: Path) -> Tuple[Optional[float], Optional[float], Optional[float], Optional[datetime]]:
    """
    Zwraca: (lat, lon, ele, naive_time)
    Rzuca wyjątek, jeśli plik nie jest czytelny lub ma wadliwe EXIF.
    """
    try:
        with Image.open(img_path) as im:
            exif = im._getexif() or {}
    except Exception as e:
        raise ValueError(f"Nie można otworzyć lub odczytać EXIF: {e}")

    # Zamiana kluczy numerycznych na nazwy
    exif_readable = {}
    for tag, val in exif.items():
        name = ExifTags.TAGS.get(tag, tag)
        exif_readable[name] = val

    gps_info = exif_readable.get("GPSInfo")
    lat = lon = ele = None
    if isinstance(gps_info, dict):
        gps = {ExifTags.GPSTAGS.get(k, k): v for k, v in gps_info.items()}

        if "GPSLatitude" in gps and "GPSLatitudeRef" in gps and \
           "GPSLongitude" in gps and "GPSLongitudeRef" in gps:
            lat = dms_to_decimal(gps["GPSLatitude"], gps["GPSLatitudeRef"])
            lon = dms_to_decimal(gps["GPSLongitude"], gps["GPSLongitudeRef"])
        # Wysokość (opcjonalna)
        if "GPSAltitude" in gps:
            try:
                ele = rational_to_float(gps["GPSAltitude"])
                if gps.get("GPSAltitudeRef") == 1:
                    ele = -ele
            except Exception as e:
                raise ValueError(f"Nie można zinterpretować wysokości: {e}")

    # Czas
    raw_dt = exif_readable.get("DateTimeOriginal") or exif_readable.get("DateTime") or exif_readable.get("DateTimeDigitized")
    subsec = exif_readable.get("SubsecTimeOriginal") or exif_readable.get("SubSecTimeOriginal") or exif_readable.get("SubsecTime")
    naive_dt = parse_exif_datetime(raw_dt, subsec) if raw_dt else None

    return lat, lon, ele, naive_dt


def collect_images(input_path: Path) -> List[Path]:
    files = []
    if input_path.is_dir():
        for root, _, fns in os.walk(input_path):
            for fn in fns:
                p = Path(root) / fn
                if p.suffix.lower() in IMAGE_EXTS:
                    files.append(p)
    else:
        if input_path.suffix.lower() in IMAGE_EXTS:
            files.append(input_path)
    return files


def generate_gpx(points: List[dict], creator: str = "exif2gpx") -> ET.ElementTree:
    ET.register_namespace("", "http://www.topografix.com/GPX/1/1")
    ET.register_namespace("xsi", "http://www.w3.org/2001/XMLSchema-instance")
    gpx = ET.Element(
        "gpx",
        {
            "version": "1.1",
            "creator": creator,
            "xmlns": "http://www.topografix.com/GPX/1/1",
            "xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance",
            "xsi:schemaLocation": "http://www.topografix.com/GPX/1/1 http://www.topografix.com/GPX/1/1/gpx.xsd",
        },
    )
    trk = ET.SubElement(gpx, "trk")
    name = ET.SubElement(trk, "name")
    name.text = "Track from EXIF"
    seg = ET.SubElement(trk, "trkseg")

    for p in points:
        trkpt = ET.SubElement(seg, "trkpt", {"lat": f"{p['lat']:.7f}", "lon": f"{p['lon']:.7f}"})
        if p.get("ele") is not None:
            ele = ET.SubElement(trkpt, "ele")
            ele.text = f"{p['ele']:.2f}"
        if p.get("time"):
            # GPX wymaga ISO 8601 w UTC z 'Z'
            t = ET.SubElement(trkpt, "time")
            iso = p["time"].strftime("%Y-%m-%dT%H:%M:%S.%f")
            iso = iso.rstrip("0").rstrip(".") + "Z"
            t.text = iso
        desc = ET.SubElement(trkpt, "desc")
        desc.text = p.get("source", "")

    return ET.ElementTree(gpx)


def main():
    ap = argparse.ArgumentParser(description="Z EXIF zdjęć (JPG) generuje GPX z trackiem.")
    ap.add_argument("input", help="Plik JPG lub folder ze zdjęciami")
    ap.add_argument("output", help="Ścieżka wyjściowa pliku GPX")
    ap.add_argument("--tz", help="Strefa czasowa EXIF w formacie +HH:MM lub -HH:MM (np. +02:00)", default=None)
    args = ap.parse_args()

    src = Path(args.input)
    out = Path(args.output)

    files = collect_images(src)
    if not files:
        print("Brak zdjęć JPG/JPEG w podanej lokalizacji.", file=sys.stderr)
        return

    points = []
    for fp in files:
        try:
            lat, lon, ele, naive_dt = extract_exif_gps_and_time(fp)
        except Exception as e:
            print(f"Pomijam {fp}: {e}", file=sys.stderr)
            continue

        if lat is None or lon is None:
            print(f"Pomijam {fp}: brak prawidłowych współrzędnych GPS", file=sys.stderr)
            continue

        # Wyznacz czas w UTC jeśli podano --tz
        time_utc = None
        if naive_dt is not None and args.tz:
            try:
                time_utc = apply_tz_and_to_utc(naive_dt, args.tz)
            except Exception as e:
                print(f"Ostrzeżenie: nieprawidłowa strefa czasowa dla {fp.name}: {e}", file=sys.stderr)

        points.append({
            "lat": lat,
            "lon": lon,
            "ele": ele,
            "naive_time": naive_dt,
            "time": time_utc,     # tylko jeśli mamy --tz
            "source": fp.name,
        })

    if not points:
        print("Nie znaleziono zdjęć z danymi GPS.", file=sys.stderr)
        return

    # Sortowanie: po czasie jeśli dostępny; potem po nazwie pliku
    def sort_key(p):
        # preferuj p['time'] (UTC) gdy jest; w przeciwnym razie naive_time; w ostateczności nazwa
        primary = p["time"] if p["time"] is not None else p["naive_time"]
        return (primary or datetime.max, p["source"])

    points.sort(key=sort_key)

    tree = generate_gpx(points)
    out.parent.mkdir(parents=True, exist_ok=True)
    tree.write(out, encoding="utf-8", xml_declaration=True)
    print(f"Zapisano GPX: {out}")

if __name__ == "__main__":
    main()