Python Script to Convert Drone Images to TIFF

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install "rasterio>=1.3.0" "Pillow>=10.0.0" "numpy>=1.24.0"

#!/usr/bin/env python3
"""
Production-grade drone image to TIFF converter.
Validates EXIF/GPS, enforces bit-depth, and writes tiled GeoTIFFs.
"""

import argparse
import logging
import os
import sys
from pathlib import Path
from typing import Optional, Tuple

import numpy as np
from PIL import Image, ExifTags
import rasterio

# ---------------------------------------------------------------------------
# Explicit Validation Thresholds
# ---------------------------------------------------------------------------
MAX_IMAGE_DIMENSION = 15000  # px (prevents memory exhaustion on oversized sensors)
MIN_VALID_COORD = -500.0     # meters (altitude/lat/lon sanity floor)
MAX_VALID_COORD = 10000.0    # meters (altitude/lat/lon sanity ceiling)
ALLOWED_COMPRESSIONS = {"lzw", "deflate", "zstd", "none"}
ALLOWED_BIT_DEPTHS = {8, 16}

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

def parse_gps_exif(exif) -> Optional[Tuple[float, float, Optional[float]]]:
    """Extract WGS84 lat/lon and optional altitude from the EXIF GPS sub-IFD."""
    # getexif().get(GPSInfo) returns only an offset; the GPS values live in the sub-IFD.
    gps_data = exif.get_ifd(ExifTags.IFD.GPSInfo)
    if not gps_data:
        return None

    def convert_to_degrees(value):
        d, m, s = value
        return float(d) + (float(m) / 60.0) + (float(s) / 3600.0)

    try:
        lat = convert_to_degrees(gps_data[ExifTags.GPS.GPSLatitude])
        lon = convert_to_degrees(gps_data[ExifTags.GPS.GPSLongitude])
        lat_ref = gps_data.get(ExifTags.GPS.GPSLatitudeRef, "N")
        lon_ref = gps_data.get(ExifTags.GPS.GPSLongitudeRef, "E")
        if lat_ref == "S": lat *= -1
        if lon_ref == "W": lon *= -1

        alt_tag = gps_data.get(ExifTags.GPS.GPSAltitude)
        alt = float(alt_tag) if alt_tag is not None else None
        alt_ref = gps_data.get(ExifTags.GPS.GPSAltitudeRef, 0)
        # 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 and alt is not None:
            alt *= -1

        return lat, lon, alt
    except (KeyError, TypeError, ValueError, ZeroDivisionError):
        return None

def validate_image(img: Image.Image, path: str) -> None:
    """Enforce dimension and bit-depth thresholds."""
    w, h = img.size
    if w > MAX_IMAGE_DIMENSION or h > MAX_IMAGE_DIMENSION:
        raise ValueError(f"Image exceeds {MAX_IMAGE_DIMENSION}px limit: {path} ({w}x{h})")

    # Check underlying bit depth (JPEGs omit the "bits" key, so guard against None)
    bits = img.info.get("bits")
    if bits is not None and bits not in ALLOWED_BIT_DEPTHS:
        logging.warning(f"Non-standard bit depth detected in {path}. Forcing 8-bit output.")

def convert_to_tiff(
    input_dir: str,
    output_dir: str,
    compression: str = "lzw",
    block_size: int = 512,
    crs: str = "EPSG:4326",
    validate_gps: bool = True
) -> None:
    if compression not in ALLOWED_COMPRESSIONS:
        raise ValueError(f"Unsupported compression: {compression}. Allowed: {ALLOWED_COMPRESSIONS}")

    Path(output_dir).mkdir(parents=True, exist_ok=True)
    extensions = ("*.jpg", "*.jpeg", "*.JPG", "*.JPEG", "*.png", "*.PNG")
    image_paths = []
    for ext in extensions:
        image_paths.extend(Path(input_dir).glob(ext))

    if not image_paths:
        logging.error("No supported images found in input directory.")
        sys.exit(1)

    success_count = 0
    for img_path in image_paths:
        try:
            with Image.open(img_path) as pil_img:
                validate_image(pil_img, str(img_path))
                # Normalize palette/grayscale to RGB for photogrammetric consistency
                if pil_img.mode not in ("RGB", "RGBA"):
                    pil_img = pil_img.convert("RGB")
                arr = np.array(pil_img)
                if arr.ndim == 2:
                    arr = np.stack([arr, arr, arr], axis=-1)
                
                out_path = Path(output_dir) / f"{img_path.stem}.tif"
                gps = parse_gps_exif(pil_img.getexif())

                if validate_gps and gps:
                    lat, lon, alt = gps
                    if not (MIN_VALID_COORD <= lat <= 90.0 and 
                            -180.0 <= lon <= 180.0 and 
                            (alt is None or MIN_VALID_COORD <= alt <= MAX_VALID_COORD)):
                        logging.warning(f"GPS coordinates out of bounds for {img_path.name}. Skipping georeferencing.")
                        gps = None

                # Build rasterio profile
                profile = {
                    "driver": "GTiff",
                    "height": arr.shape[0],
                    "width": arr.shape[1],
                    "count": arr.shape[2],
                    "dtype": arr.dtype,
                    "compress": compression,
                    "photometric": "RGB",
                    "tiled": True,
                    "blockxsize": block_size,
                    "blockysize": block_size,
                    "bigtiff": "IF_SAFER",
                }

                arr_transposed = np.moveaxis(arr, -1, 0)

                with rasterio.open(out_path, "w", **profile) as dst:
                    dst.write(arr_transposed)
                    # A single frame is not an orthomosaic, so we do not fabricate a
                    # geotransform. Persist the camera position as metadata instead;
                    # georeferencing happens later in the SfM/ODM stage.
                    if gps:
                        lat, lon, alt = gps
                        dst.update_tags(
                            GPS_LATITUDE=f"{lat:.8f}",
                            GPS_LONGITUDE=f"{lon:.8f}",
                            GPS_ALTITUDE="" if alt is None else f"{alt:.3f}",
                            GPS_CRS=crs,
                        )
                
                logging.info(f"Converted: {img_path.name} -> {out_path.name}")
                success_count += 1

        except Exception as e:
            logging.error(f"Failed {img_path.name}: {e}")
            continue

    logging.info(f"Pipeline complete. {success_count}/{len(image_paths)} images converted successfully.")

def main():
    parser = argparse.ArgumentParser(
        description="Convert drone imagery to tiled, compressed GeoTIFFs with EXIF/GPS validation."
    )
    parser.add_argument("--input-dir", required=True, help="Directory containing raw drone images.")
    parser.add_argument("--output-dir", required=True, help="Target directory for converted TIFFs.")
    parser.add_argument("--compression", default="lzw", choices=ALLOWED_COMPRESSIONS,
                        help="TIFF compression algorithm (default: lzw).")
    parser.add_argument("--block-size", type=int, default=512,
                        help="Tile block size in pixels (default: 512).")
    parser.add_argument("--crs", default="EPSG:4326",
                        help="Coordinate Reference System for output (default: EPSG:4326).")
    parser.add_argument("--no-validate-gps", action="store_true",
                        help="Disable GPS coordinate threshold validation.")
    
    args = parser.parse_args()
    convert_to_tiff(
        input_dir=args.input_dir,
        output_dir=args.output_dir,
        compression=args.compression,
        block_size=args.block_size,
        crs=args.crs,
        validate_gps=not args.no_validate_gps
    )

if __name__ == "__main__":
    main()

python drone_to_tiff.py \
  --input-dir ./raw_flight_data \
  --output-dir ./processed_tiffs \
  --compression zstd \
  --block-size 1024 \
  --crs EPSG:4326