Validating and Repairing EXIF Metadata at Scale

import exiftool
import pandas as pd


TAGS = [
    "-EXIF:DateTimeOriginal", "-EXIF:SubSecTimeOriginal", "-EXIF:OffsetTimeOriginal",
    "-EXIF:GPSLatitude", "-EXIF:GPSLongitude", "-EXIF:GPSAltitude",
    "-EXIF:GPSAltitudeRef", "-EXIF:FocalLength", "-EXIF:ExposureTime",
    "-EXIF:ISO", "-EXIF:Model", "-EXIF:Make",
    "-XMP:GimbalYawDegree", "-XMP:GimbalPitchDegree", "-XMP:GimbalRollDegree",
    "-XMP:FlightYawDegree", "-XMP:RelativeAltitude", "-XMP:RtkFlag",
    "-XMP:RtkStdLon", "-XMP:RtkStdLat", "-XMP:RtkStdHgt",
]


def read_survey_metadata(paths: list[str], batch: int = 500) -> pd.DataFrame:
    """One structured record per image, from a small number of ExifTool calls.

    ExifTool's startup cost dominates when it is invoked per file, so batching
    turns an operation that takes twenty minutes on ten thousand images into
    one that takes under a minute.
    """
    rows = []
    with exiftool.ExifToolHelper() as et:
        for i in range(0, len(paths), batch):
            chunk = paths[i:i + batch]
            for meta in et.get_tags(chunk, TAGS):
                rows.append({k.split(":")[-1]: v for k, v in meta.items()})
    df = pd.DataFrame(rows)
    df["path"] = [m.get("SourceFile") for m in rows] if "SourceFile" in df else paths
    return df

import numpy as np
import pandas as pd


def validate_survey(df: pd.DataFrame) -> dict:
    """Dataset-level checks that no per-image validation would catch."""
    problems = []

    if df["Model"].nunique() > 1:
        problems.append(f"{df['Model'].nunique()} camera models in one set: "
                        f"{sorted(df['Model'].dropna().unique())}")

    if "GPSAltitudeRef" in df and df["GPSAltitudeRef"].nunique() > 1:
        problems.append("GPS altitude reference changes within the set — "
                        "some images are above the ellipsoid and some above sea level")

    times = pd.to_datetime(df["DateTimeOriginal"], errors="coerce",
                           format="%Y:%m:%d %H:%M:%S")
    gaps = times.sort_values().diff().dt.total_seconds().dropna()
    if (gaps > 3000).any():
        problems.append(f"a gap of {gaps.max() / 60:.0f} minutes between consecutive "
                        "images — two flights merged, or a clock jump")
    if (gaps < 0).any():
        problems.append("timestamps are not monotonic — a clock reset mid-flight")

    if "FocalLength" in df and df["FocalLength"].nunique() > 1:
        problems.append(f"focal length varies: {sorted(df['FocalLength'].dropna().unique())}")

    missing_gps = df["GPSLatitude"].isna().sum()
    if missing_gps:
        problems.append(f"{missing_gps} images have no GPS position")

    attitude = [c for c in ("GimbalYawDegree", "GimbalPitchDegree") if c in df]
    if not attitude:
        problems.append("no gimbal attitude in the XMP — oblique handling will be blind")

    return {"images": len(df), "problems": problems}

import pandas as pd


def normalise_timestamps(df: pd.DataFrame, *, assume_utc_offset_hours: float | None = None
                         ) -> pd.DataFrame:
    """One timezone-aware timestamp column from whatever the camera wrote.

    Cameras variously write local time with no offset, local time with an
    offset tag, or UTC. Guessing is unavoidable when no offset is present, but
    the guess must be recorded rather than applied silently.
    """
    out = df.copy()
    naive = pd.to_datetime(out["DateTimeOriginal"], errors="coerce",
                           format="%Y:%m:%d %H:%M:%S")

    if "OffsetTimeOriginal" in out and out["OffsetTimeOriginal"].notna().any():
        offsets = out["OffsetTimeOriginal"].fillna("+00:00")
        out["timestamp"] = pd.to_datetime(
            naive.dt.strftime("%Y-%m-%d %H:%M:%S") + offsets, utc=True, format="mixed")
        out["timestamp_source"] = "offset tag"
    elif assume_utc_offset_hours is not None:
        out["timestamp"] = (naive - pd.Timedelta(hours=assume_utc_offset_hours)
                            ).dt.tz_localize("UTC")
        out["timestamp_source"] = f"assumed UTC{assume_utc_offset_hours:+g}"
    else:
        out["timestamp"] = naive.dt.tz_localize("UTC")
        out["timestamp_source"] = "assumed already UTC — UNVERIFIED"

    if "SubSecTimeOriginal" in out:
        subsec = pd.to_numeric(out["SubSecTimeOriginal"], errors="coerce").fillna(0)
        out["timestamp"] = out["timestamp"] + pd.to_timedelta(subsec / 1000, unit="s")
    return out

import numpy as np
import pandas as pd


def repair_missing_positions(df: pd.DataFrame, *, max_gap: int = 3) -> pd.DataFrame:
    """Interpolate positions for a few isolated images with no GPS fix.

    Only for short gaps within a flight line, where the aircraft was flying a
    straight, constant-speed path. A longer gap, or one spanning a turn, is
    not interpolable and the images should be excluded instead.
    """
    out = df.sort_values("timestamp").copy()
    missing = out["GPSLatitude"].isna()
    if not missing.any():
        return out

    runs = (missing != missing.shift()).cumsum()[missing]
    for _, idx in out[missing].groupby(runs).groups.items():
        if len(idx) > max_gap:
            out.loc[idx, "position_status"] = "excluded: gap too long to interpolate"
            continue
        out.loc[idx, "position_status"] = f"interpolated across {len(idx)} images"

    for col in ("GPSLatitude", "GPSLongitude", "GPSAltitude"):
        short_gaps = out[col].isna() & (out.get("position_status", "").astype(str)
                                        .str.startswith("interpolated"))
        out.loc[short_gaps, col] = out[col].interpolate(limit=max_gap)[short_gaps]
    return out

def gate_survey(report: dict, *, require_attitude: bool = False,
                min_gps_fraction: float = 0.98,
                max_models: int = 1) -> None:
    """Refuse a survey whose metadata cannot support the pipeline downstream."""
    if len(report["models"]) > max_models:
        raise ValueError(f"{len(report['models'])} camera models: {report['models']} "
                         "— split the dataset before processing")
    if len(report["altitude_refs"]) > 1:
        raise ValueError("GPS altitude reference is not constant across the survey — "
                         "resolve before the reconstruction absorbs it as a step")
    gps_fraction = report["with_gps"] / max(report["images"], 1)
    if gps_fraction < min_gps_fraction:
        raise ValueError(f"only {gps_fraction:.1%} of images carry a GPS position")
    if require_attitude and report["with_gimbal_attitude"] < report["images"]:
        raise ValueError("gimbal attitude is missing on some images and this job needs it")
    if report["timestamp_source"] and "UNVERIFIED" in str(report["timestamp_source"]):
        raise ValueError("timestamps were assumed to be UTC without evidence — "
                         "confirm before matching against a trajectory")

from pathlib import Path
import subprocess
import pandas as pd


def persist_metadata(df: pd.DataFrame, survey_dir: str) -> str:
    """Store the metadata table with the survey, with its provenance."""
    version = subprocess.run(["exiftool", "-ver"], capture_output=True,
                             text=True, check=True).stdout.strip()
    out = Path(survey_dir) / "metadata.csv"
    df.attrs["exiftool_version"] = version
    df.to_csv(out, index=False)
    (Path(survey_dir) / "metadata.provenance.txt").write_text(
        f"exiftool {version}
images {len(df)}
columns {list(df.columns)}
")
    return str(out)

import numpy as np
import pandas as pd


def metadata_report(df: pd.DataFrame) -> dict:
    """A summary that belongs in the run manifest for every survey."""
    times = pd.to_datetime(df["timestamp"], errors="coerce", utc=True)
    intervals = times.sort_values().diff().dt.total_seconds().dropna()

    return {
        "images": len(df),
        "models": sorted(df["Model"].dropna().unique().tolist()),
        "duration_min": float((times.max() - times.min()).total_seconds() / 60),
        "median_interval_s": float(intervals.median()) if len(intervals) else None,
        "timestamp_source": df["timestamp_source"].mode().iat[0]
        if "timestamp_source" in df else None,
        "with_gps": int(df["GPSLatitude"].notna().sum()),
        "with_gimbal_attitude": int(df.get("GimbalPitchDegree",
                                           pd.Series(dtype=float)).notna().sum()),
        "rtk_fixed": int((df.get("RtkFlag", pd.Series(dtype=float)) == 50).sum()),
        "altitude_refs": sorted(df.get("GPSAltitudeRef",
                                       pd.Series(dtype=object)).dropna().unique().tolist()),
    }