How to Validate EXIF GPS Data Before Processing

import os
import sys
import argparse
from pathlib import Path
from PIL import Image
from PIL.ExifTags import TAGS, GPSTAGS
from fractions import Fraction
from typing import Dict, Tuple, Optional, List
import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

def parse_rational(value) -> float:
    """Convert an EXIF rational to float.

    Pillow's get_ifd returns rationals as IFDRational scalars (a float subclass),
    while raw EXIF or other libraries may expose a (numerator, denominator) tuple.
    Both forms are handled, with a zero-denominator guard.
    """
    if isinstance(value, (tuple, list)) and len(value) == 2:
        num, den = value
        if den == 0:
            raise ZeroDivisionError("EXIF rational denominator is zero")
        return float(num) / float(den)
    try:
        return float(value)
    except (TypeError, ValueError):
        raise ValueError(f"Invalid EXIF rational format: {value!r}")

def dms_to_decimal(dms: Tuple[Tuple[int, int], Tuple[int, int], Tuple[int, int]]) -> float:
    """Convert DMS rational tuple to decimal degrees."""
    degrees = parse_rational(dms[0])
    minutes = parse_rational(dms[1])
    seconds = parse_rational(dms[2])
    return degrees + (minutes / 60.0) + (seconds / 3600.0)

def validate_image_gps(image_path: Path, 
                       lat_bounds: Tuple[float, float] = (-90.0, 90.0),
                       lon_bounds: Tuple[float, float] = (-180.0, 180.0),
                       alt_bounds: Tuple[float, float] = (-500.0, 15000.0),
                       reject_null_alt: bool = True) -> Dict[str, any]:
    """Validate EXIF GPS data against explicit thresholds. Returns structured report."""
    report = {"path": str(image_path), "valid": False, "coords": None, "error": None}
    
    try:
        with Image.open(image_path) as img:
            exif = img.getexif()
            gps_ifd = exif.get_ifd(0x8825)  # GPS IFD tag ID
            
            if not gps_ifd:
                report["error"] = "MISSING_GPS_IFD"
                return report

            # Extract tags using GPSTAGS mapping
            gps_tags = {GPSTAGS.get(k, k): v for k, v in gps_ifd.items()}
            
            lat_ref = gps_tags.get("GPSLatitudeRef", "N")
            lon_ref = gps_tags.get("GPSLongitudeRef", "E")
            lat_dms = gps_tags.get("GPSLatitude")
            lon_dms = gps_tags.get("GPSLongitude")
            alt_data = gps_tags.get("GPSAltitude")
            alt_ref = gps_tags.get("GPSAltitudeRef", 0)

            if not lat_dms or not lon_dms:
                report["error"] = "MISSING_COORD_TAGS"
                return report

            lat = dms_to_decimal(lat_dms)
            lon = dms_to_decimal(lon_dms)
            
            # Apply hemisphere reference
            if lat_ref == "S": lat *= -1
            if lon_ref == "W": lon *= -1

            # Altitude parsing & reference application
            alt = None
            if alt_data:
                alt = parse_rational(alt_data)
                # GPSAltitudeRef is returned as bytes (b"\x01") or an int by Pillow
                ref_val = int.from_bytes(alt_ref, "big") if isinstance(alt_ref, bytes) else int(alt_ref)
                if ref_val == 1:  # Below sea level
                    alt *= -1
                if reject_null_alt and abs(alt) < 0.001:
                    report["error"] = "ZERO_ALTITUDE_REJECTED"
                    return report

            # Threshold enforcement
            if not (lat_bounds[0] <= lat <= lat_bounds[1]):
                report["error"] = f"OUT_OF_LAT_BOUNDS:{lat}"
                return report
            if not (lon_bounds[0] <= lon <= lon_bounds[1]):
                report["error"] = f"OUT_OF_LON_BOUNDS:{lon}"
                return report
            if alt is not None and not (alt_bounds[0] <= alt <= alt_bounds[1]):
                report["error"] = f"OUT_OF_ALT_BOUNDS:{alt}"
                return report

            report["valid"] = True
            report["coords"] = {"lat": lat, "lon": lon, "alt": alt}
            
    except Exception as e:
        report["error"] = f"PARSE_ERROR:{type(e).__name__}:{str(e)}"
        
    return report

def batch_validate(image_dir: Path, output_csv: Path, min_valid_ratio: float = 0.95) -> None:
    """Run validation across directory, enforce minimum valid image ratio, write CSV."""
    images = sorted(list(image_dir.glob("*.JPG")) + list(image_dir.glob("*.jpg")))
    if not images:
        logging.error("No JPEG images found in target directory.")
        sys.exit(1)

    valid_count = 0
    results = []
    for img in images:
        res = validate_image_gps(img)
        results.append(res)
        if res["valid"]:
            valid_count += 1

    valid_ratio = valid_count / len(images)
    logging.info(f"Validation complete: {valid_count}/{len(images)} valid ({valid_ratio:.2%})")
    
    if valid_ratio < min_valid_ratio:
        logging.critical(f"Valid image ratio {valid_ratio:.2%} below threshold {min_valid_ratio:.2%}. Aborting pipeline.")
        sys.exit(2)
    
    with open(output_csv, "w") as f:
        f.write("path,valid,lat,lon,alt,error\n")
        for r in results:
            c = r.get("coords") or {}
            f.write(f"{r['path']},{r['valid']},{c.get('lat','')},{c.get('lon','')},{c.get('alt','')},{r.get('error','')}\n")
    logging.info(f"Validation report written to {output_csv}")

python validate_gps.py \
  --input-dir ./raw_flight_data/ \
  --output-report ./qc/gps_validation.csv \
  --min-valid-ratio 0.95 \
  --reject-null-alt \
  --max-altitude 8000.0 \
  --min-altitude -200.0