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 --from-date 2024-01-01 --to-date 2024-12-31

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.
    --from-date  Uwzględniaj punkty od tej daty (włącznie). Format: YYYY-MM-DD[THH:MM[:SS]]
    --to-date    Uwzględniaj punkty do tej daty (włącznie). Format: YYYY-MM-DD[THH:MM[:SS]]
"""
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")

    if isinstance(x, Real):
        return float(x)

    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)

    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")
        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 parse_cli_datetime(s: str) -> datetime:
    """
    Parsuje datę z CLI. Akceptowane formaty (naive, bez strefy):
      - YYYY-MM-DD
      - YYYY-MM-DD HH:MM
      - YYYY-MM-DD HH:MM:SS
      - YYYY-MM-DDTHH:MM
      - YYYY-MM-DDTHH:MM:SS
    Zwraca naive datetime.
    """
    if not s:
        raise ValueError("Pusta data")
    formats = [
        "%Y-%m-%d",
        "%Y-%m-%d %H:%M",
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%dT%H:%M",
        "%Y-%m-%dT%H:%M:%S",
    ]
    last_err = None
    for fmt in formats:
        try:
            dt = datetime.strptime(s, fmt)
            if fmt == "%Y-%m-%d":
                return dt.replace(hour=0, minute=0, second=0, microsecond=0)
            return dt
        except Exception as e:
            last_err = e
            continue
    raise ValueError(f"Nie rozpoznano formatu daty: {s!r} ({last_err})")


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}")

    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"])
        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}")

    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"):
            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)
    ap.add_argument("--from-date", dest="from_date", help="Uwzględniaj punkty od tej daty (włącznie). Format: YYYY-MM-DD[THH:MM[:SS]]", default=None)
    ap.add_argument("--to-date", dest="to_date", help="Uwzględniaj punkty do tej daty (włącznie). Format: YYYY-MM-DD[THH:MM[:SS]]", default=None)
    args = ap.parse_args()

    # Parsowanie filtrów daty
    from_dt = to_dt = None
    if args.from_date:
        try:
            from_dt = parse_cli_datetime(args.from_date)
            if args.tz:
                from_dt = apply_tz_and_to_utc(from_dt, args.tz)
        except Exception as e:
            print(f"Ostrzeżenie: nie można zinterpretować --from-date: {e}", file=sys.stderr)
            from_dt = None
    if args.to_date:
        try:
            to_dt = parse_cli_datetime(args.to_date)
            if args.tz:
                to_dt = apply_tz_and_to_utc(to_dt, args.tz)
        except Exception as e:
            print(f"Ostrzeżenie: nie można zinterpretować --to-date: {e}", file=sys.stderr)
            to_dt = None

    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)

        # Filtrowanie po datach (inclusive). Jeśli filtry ustawione, wymagamy czasu w zdjęciu.
        if args.from_date or args.to_date:
            base_dt = time_utc if args.tz else naive_dt
            if base_dt is None:
                print(f"Pomijam {fp}: brak czasu przy aktywnych filtrach daty", file=sys.stderr)
                continue
            if from_dt and (base_dt < from_dt):
                continue
            if to_dt and (base_dt > to_dt):
                continue

        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):
        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()