RoadTripsGenerator/rtg/extraction.py

"""Positions GPS des photos (EXIF) et des rushs GoPro (GPS embarqué).

Produit, comme l'ancien extraction_infos.sh, deux CSV sans en-tête :
    export_photos.csv : fichier, date, lat, lon, alt
    export_videos.csv : fichier, date GPS, lat, lon, alt, vitesse

Une photo sans GPS mais datée peut recevoir la position de la trace à son
heure (réglage photos.positions_deduites), si la trace est horodatée.
"""
import bisect
import os
import subprocess
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo

from rtg import corrections
from rtg.outils import dire, exiger

FMT_PHOTOS = '"$FileName",$DateTimeOriginal,$GPSLatitude,$GPSLongitude,$GPSAltitude'
FMT_PHOTOS_MIN = '"$FileName",$DateTimeOriginal,$GPSLatitude,$GPSLongitude'
FMT_PHOTOS_DATE = '"$FileName",$DateTimeOriginal'
TOLERANCE_POSITION = timedelta(minutes=5)


def _exiftool(args):
    r = subprocess.run(["exiftool"] + args, capture_output=True, text=True, errors="replace")
    return [l for l in r.stdout.splitlines() if l.strip()]


# --- Photos ------------------------------------------------------------------
def extraire_photos(voyage, cfg, analyse):
    dossier = os.path.join(voyage, "photos")
    csv = os.path.join(voyage, "export_photos.csv")
    if not os.path.isdir(dossier):
        open(csv, "w").close()
        dire("⚠️  Dossier photos non trouvé, étape ignorée.")
        return
    exiger("exiftool", "Installe-le : https://exiftool.org (brew install exiftool sous macOS).")
    # Un seul appel par passe pour tout le dossier : chaque lancement d'exiftool
    # coûte ~200 ms. Seconde passe au format réduit pour les photos sans altitude.
    commun = ["-n", "-q", "-ext", "jpg", "-ext", "jpeg"]
    p1 = _exiftool(commun + ["-p", FMT_PHOTOS, dossier])
    p2 = [l + "," for l in _exiftool(commun + ["-p", FMT_PHOTOS_MIN, dossier])]
    vus = {l.split(",")[0] for l in p1}
    lignes = p1 + [l for l in p2 if l.split(",")[0] not in vus]

    deduites = 0
    if cfg["photos"].get("positions_deduites", True):
        vus = {l.split(",")[0] for l in lignes}
        sans_gps = [l for l in _exiftool(commun + ["-p", FMT_PHOTOS_DATE, dossier])
                    if l.split(",")[0] not in vus]
        if sans_gps:
            index = IndexTrace(analyse, cfg)
            for l in sans_gps:
                nom, date = l.split(",", 1)
                pos = index.position(date)
                if pos:
                    lignes.append(f"{nom},{date},{pos[0]},{pos[1]},")
                    deduites += 1
    # Photos datées restées sans position : lignes sans coordonnées, ignorées
    # par les pages mais listées par l'interface pour être placées à la main.
    vus = {l.split(",")[0] for l in lignes}
    sans_position = [l.split(",", 1) for l in _exiftool(commun + ["-p", FMT_PHOTOS_DATE, dossier])
                     if l.split(",")[0] not in vus]
    lignes += [f"{nom},{date},,," for nom, date in sans_position]

    with open(csv, "w", encoding="utf-8") as f:
        f.write("".join(l + "\n" for l in lignes))
    localisees = len(lignes) - len(sans_position)
    msg = f"📸 Photos  : {localisees} localisée(s)"
    if localisees > len(p1) + deduites:
        msg += f" (dont {localisees - len(p1) - deduites} sans altitude)"
    if deduites:
        msg += f", {deduites} position(s) déduite(s) de la trace"
    if sans_position:
        placees = corrections.positions(cfg)
        a_la_main = sum(1 for nom, _ in sans_position if nom.strip('"') in placees)
        reste = len(sans_position) - a_la_main
        if a_la_main:
            msg += f", {a_la_main} placée(s) à la main"
        if reste:
            msg += f", {reste} sans position (à placer dans l'interface)"
    dire(msg)


class IndexTrace:
    """Points horodatés des traces du voyage, pour retrouver une position à
    une heure donnée."""

    def __init__(self, analyse, cfg):
        self.tz = ZoneInfo(cfg["journees"]["fuseau"])
        pts = []
        for pistes in analyse.journees.values():
            for t in pistes:
                for s in t.segments:
                    for p in s.points:
                        if p.time:
                            h = p.time if p.time.tzinfo else p.time.replace(tzinfo=timezone.utc)
                            pts.append((h.timestamp(), p.latitude, p.longitude))
        pts.sort()
        self.temps = [p[0] for p in pts]
        self.pts = pts

    def position(self, date_exif):
        """Date EXIF « AAAA:MM:JJ HH:MM:SS » (heure locale du voyage) → (lat, lon)."""
        if not self.pts:
            return None
        try:
            locale = datetime.strptime(date_exif.split(".")[0][:19], "%Y:%m:%d %H:%M:%S")
        except ValueError:
            return None
        t = locale.replace(tzinfo=self.tz).timestamp()
        i = bisect.bisect_left(self.temps, t)
        voisins = [j for j in (i - 1, i) if 0 <= j < len(self.pts)]
        j = min(voisins, key=lambda k: abs(self.temps[k] - t))
        if abs(self.temps[j] - t) > TOLERANCE_POSITION.total_seconds():
            return None
        return self.pts[j][1], self.pts[j][2]


# --- Vidéos ------------------------------------------------------------------
def _premier_point(video):
    """(date, lat, lon) du premier point GPS d'un rush GoPro, ou None.

    Lit les tags un par un : `exiftool -p` ne voit que le document principal,
    qui porte la position mais pas GPSDateTime (rangé dans [Doc1]). On arrête
    exiftool dès qu'on a une date et une position, sans lire tout le fichier.
    """
    cmd = ["exiftool", "-ee", "-n", "-q", "-s", "-api", "largefilesupport=1",
           "-GPSDateTime", "-GPSLatitude", "-GPSLongitude", "-CreateDate", video]
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                            text=True, errors="replace")
    dt = cd = lat = None
    premier = None
    try:
        for ligne in proc.stdout:
            if ":" not in ligne:
                continue
            ligne = ligne.split("]", 1)[-1] if ligne.startswith("[") else ligne
            tag, _, val = ligne.partition(":")
            tag, val = tag.strip().replace(" ", ""), val.strip()
            if tag == "GPSDateTime" and dt is None:
                dt = val
            elif tag == "CreateDate" and cd is None:
                cd = val
            elif tag == "GPSLatitude":
                lat = val
            elif tag == "GPSLongitude" and lat is not None:
                if premier is None:
                    premier = (lat, val)
                lat = None
            if premier and dt:
                break
    finally:
        proc.kill()
        proc.wait()
    date = dt or cd
    if premier and date:
        return date, premier[0], premier[1]
    if date:
        return date, None, None      # datée mais sans GPS : à placer à la main
    return None


def extraire_videos(voyage, analyse):
    csv = os.path.join(voyage, "export_videos.csv")
    if not analyse.videos:
        open(csv, "w").close()
        dire("⚠️  Aucune vidéo source trouvée.")
        return
    exiger("exiftool", "Installe-le : https://exiftool.org (brew install exiftool sous macOS).")
    # Lignes déjà connues : un rush non modifié depuis le dernier export n'est
    # pas relu (gain important sur un disque externe).
    connues, date_csv = {}, 0
    if os.path.exists(csv) and os.path.getsize(csv) > 0:
        date_csv = os.path.getmtime(csv)
        with open(csv, encoding="utf-8") as f:
            for l in f:
                if l.strip():
                    connues[l.split(",")[0].strip('"')] = l.rstrip("\n")
    dire(f"🎬 Vidéos  : {len(analyse.videos)} rush(s), premier point GPS")
    lignes, sans, reprises = [], 0, 0
    for i, v in enumerate(analyse.videos, 1):
        nom = os.path.basename(v)
        if nom in connues and os.path.getmtime(v) <= date_csv:
            lignes.append(connues[nom])
            reprises += 1
            continue
        r = _premier_point(v)
        if r and r[1] is not None:
            lignes.append(f'"{nom}",{r[0]},{r[1]},{r[2]},,')
            etat = "✓"
        else:
            sans += 1
            etat = "— pas de GPS"
            if r:
                lignes.append(f'"{nom}",{r[0]},,,,')
        dire(f"   [{i:3d}/{len(analyse.videos):3d}] {nom:<28} {etat}")
    if reprises:
        dire(f"   {reprises} rush(s) inchangé(s), positions reprises du dernier export")
    with open(csv, "w", encoding="utf-8") as f:
        f.write("".join(l + "\n" for l in lignes))
    sans_pos = sum(1 for l in lignes if l.split(",")[2:4] == ["", ""])
    dire(f"🎬 Vidéos  : {len(lignes) - sans_pos} rush(s) localisé(s)"
         + (f", {sans_pos} sans GPS (à placer dans l'interface)" if sans_pos else ""))