app/dda/preflight.py

"""Preflight image-pair validation: CRS, GSD, RGB bands, overlap, georeferencing.

Runs once per detection request, right after library paths are resolved but
before either image is pixel-decoded — all checks here are cheap header-only
reads, so a genuinely unusable pair (unreadable file, unsupported band
layout, no ground overlap) is rejected in milliseconds instead of after
minutes of registration + model inference. Lesser issues (grayscale input,
manual-bounds-only georeferencing, GSD mismatch, weak overlap) don't block
detection — they're collected as warnings and surfaced on the result instead.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional, Tuple

from sqlalchemy.orm import Session

from ..detection_config import (
    get_gsd_tolerance,
    get_max_aspect_ratio_diff,
    get_min_overlap_hard,
    get_min_overlap_warn,
)
from .geo_regions import BoundsWGS84, resolve_geo_context
from .geotiff_io import inspect_image, read_georef, read_gsd_meters

import logging

logger = logging.getLogger(__name__)

_MODE_BAND_COUNTS = {
    "1": 1, "L": 1, "I": 1, "F": 1,
    "LA": 2,
    "RGB": 3, "YCbCr": 3, "P": 3,
    "RGBA": 4, "CMYK": 4,
}

_NON_RGB_BAND_KEYWORDS = ("nir", "infrared", "swir", "thermal", "alpha", "panchromatic")


@dataclass
class CheckResult:
    name: str        # readability|bands|georef|crs|gsd|overlap
    status: str       # pass|warn|fail
    message: str = ""


@dataclass
class PreflightResult:
    hard_fail: bool = False
    fail_reason: str = ""
    warnings: List[str] = field(default_factory=list)
    checks: List[CheckResult] = field(default_factory=list)


def read_band_count(path: Path) -> Optional[int]:
    """Raster band count via rasterio, falling back to a Pillow mode→band map."""
    try:
        import rasterio
        with rasterio.open(path) as src:
            return int(src.count)
    except ImportError:
        pass
    except Exception as exc:
        logger.warning("read_band_count rasterio failed for %s: %s", path.name, exc)
    try:
        from PIL import Image
        with Image.open(path) as img:
            return _MODE_BAND_COUNTS.get(img.mode, len(img.getbands()))
    except Exception as exc:
        logger.warning("read_band_count Pillow failed for %s: %s", path.name, exc)
        return None


def _rect_overlap_frac(
    before_bounds: Optional[BoundsWGS84],
    after_bounds: Optional[BoundsWGS84],
) -> Optional[float]:
    """Intersection area / after-footprint area, as a plain WGS84 rectangle."""
    if not before_bounds or not after_bounds:
        return None
    bw, bs, be, bn = before_bounds
    aw, asf, ae, an = after_bounds
    ix0, iy0 = max(bw, aw), max(bs, asf)
    ix1, iy1 = min(be, ae), min(bn, an)
    inter_w, inter_h = max(0.0, ix1 - ix0), max(0.0, iy1 - iy0)
    inter_area = inter_w * inter_h
    after_area = max(1e-12, abs(ae - aw) * abs(an - asf))
    return inter_area / after_area


def _apply_roi_to_bounds(
    bounds: Optional[BoundsWGS84], roi: Optional[dict],
) -> Optional[BoundsWGS84]:
    """Crop a WGS84 rectangle to a fractional {x,y,w,h} ROI window.

    Mirrors ``load_rgb_roi``'s convention: x/y are the top-left corner and
    y grows downward in the image's own pixel grid, i.e. y=0 is the image's
    top edge (north when north-up). The same fractional window is applied to
    both images by the caller (matching how detection itself crops both
    images to the same fractional ROI), so overlap is checked on the actual
    selected area rather than the full footprint.
    """
    if not bounds or not roi:
        return bounds
    west, south, east, north = bounds
    x, y, w, h = roi["x"], roi["y"], roi["w"], roi["h"]
    roi_west = west + x * (east - west)
    roi_east = west + (x + w) * (east - west)
    roi_north = north - y * (north - south)
    roi_south = north - (y + h) * (north - south)
    return (roi_west, roi_south, roi_east, roi_north)


def read_band_descriptions(path: Path) -> Optional[List[str]]:
    """First-3-band GDAL band descriptions, when the file has any set."""
    try:
        import rasterio
        with rasterio.open(path) as src:
            return list(src.descriptions[:3])
    except Exception:
        return None


def _suspicious_band_order(descriptions: Optional[List[Optional[str]]]) -> bool:
    """Best-effort hint only — most files carry no band descriptions at all,
    so this stays silent far more often than it fires. There is no reliable
    way to verify true RGB band *order* from metadata alone; this only flags
    the case where a sensor explicitly labelled a leading band as non-visible
    (e.g. NIR), which is worth a second look before trusting on-screen color.
    """
    if not descriptions:
        return False
    for d in descriptions:
        if not d:
            continue
        low = d.lower()
        if any(kw in low for kw in _NON_RGB_BAND_KEYWORDS):
            return True
    return False


def run_preflight_checks(
    db: Session,
    before_path: Path,
    after_path: Path,
    before_rel: str,
    after_rel: str,
    roi: Optional[dict] = None,
) -> PreflightResult:
    """Validate a (before, after) library image pair before decoding pixels.

    ``roi`` (optional fractional ``{x,y,w,h}``) narrows the overlap check to
    the actually-selected crop window instead of the full image footprints,
    matching what detection itself will run on when a ROI is set.
    """
    result = PreflightResult()

    # 1. Readability
    before_meta = inspect_image(before_path)
    after_meta = inspect_image(after_path)
    if before_meta.width <= 0 or before_meta.height <= 0:
        result.hard_fail = True
        result.fail_reason = f"Before image could not be read: {before_path.name}"
        result.checks.append(CheckResult("readability", "fail", result.fail_reason))
        return result
    if after_meta.width <= 0 or after_meta.height <= 0:
        result.hard_fail = True
        result.fail_reason = f"After image could not be read: {after_path.name}"
        result.checks.append(CheckResult("readability", "fail", result.fail_reason))
        return result
    result.checks.append(CheckResult("readability", "pass"))

    # 2. RGB bands
    before_bands = read_band_count(before_path)
    after_bands = read_band_count(after_path)
    for label, bands, path in (
        ("Before", before_bands, before_path),
        ("After", after_bands, after_path),
    ):
        if not bands:
            result.hard_fail = True
            result.fail_reason = f"{label} image band layout could not be determined: {path.name}"
            result.checks.append(CheckResult("bands", "fail", result.fail_reason))
            return result
        if bands == 2:
            result.hard_fail = True
            result.fail_reason = f"{label} image has an unsupported 2-band layout: {path.name}"
            result.checks.append(CheckResult("bands", "fail", result.fail_reason))
            return result
    band_msgs = []
    if before_bands == 1 or after_bands == 1:
        band_msgs.append(
            "One or both images are single-band (grayscale) — detection runs on a tripled grayscale channel.")
    extra_band_labels = [
        label for label, bands in (("Before", before_bands), ("After", after_bands)) if bands > 3
    ]
    if extra_band_labels:
        band_msgs.append(
            f"{' and '.join(extra_band_labels)} image has more than 3 bands — "
            "only the first 3 are used for detection.")
    suspicious_labels = [
        label for label, path in (("Before", before_path), ("After", after_path))
        if _suspicious_band_order(read_band_descriptions(path))
    ]
    if suspicious_labels:
        band_msgs.append(
            f"{' and '.join(suspicious_labels)} image band descriptions suggest non-RGB content "
            "(e.g. NIR/infrared) in the first bands — verify band order before relying on colors.")
    if band_msgs:
        for msg in band_msgs:
            result.warnings.append(msg)
            result.checks.append(CheckResult("bands", "warn", msg))
    else:
        result.checks.append(CheckResult("bands", "pass"))

    # 3. Georeferencing (also feeds the CRS/GSD/overlap checks below)
    before_geo = resolve_geo_context(db, before_rel, before_path)
    after_geo = resolve_geo_context(db, after_rel, after_path)

    if before_geo.source == "none" and after_geo.source == "none":
        # Plain, non-georeferenced photo pair — an already-supported use case
        # (pair_align's resize_only fallback). Skip CRS/GSD/overlap entirely
        # rather than penalize a mode the app deliberately allows.
        result.checks.append(CheckResult(
            "georef", "pass", "Neither image is georeferenced — treated as a plain photo pair."))
        result.checks.append(CheckResult("crs", "pass"))
        result.checks.append(CheckResult("gsd", "pass"))
        # With no georeferencing, overlap can't be checked geographically — but
        # detection still force-resizes "after" to "before"'s pixel dimensions
        # regardless, so a large aspect-ratio gap is worth flagging here instead.
        ratio_b = before_meta.width / before_meta.height if before_meta.height else 0.0
        ratio_a = after_meta.width / after_meta.height if after_meta.height else 0.0
        if ratio_b > 0 and ratio_a > 0:
            rel_diff = abs(ratio_b - ratio_a) / max(ratio_b, ratio_a)
            if rel_diff > get_max_aspect_ratio_diff():
                msg = (
                    f"Images have different aspect ratios (before {before_meta.width}x{before_meta.height}, "
                    f"after {after_meta.width}x{after_meta.height}) — the after image will be stretched "
                    "to match, which may distort content.")
                result.warnings.append(msg)
                result.checks.append(CheckResult("aspect_ratio", "warn", msg))
            else:
                result.checks.append(CheckResult("aspect_ratio", "pass"))
        result.checks.append(CheckResult("overlap", "pass"))
        return result

    if before_geo.source == "none" or after_geo.source == "none":
        msg = "Only one image is georeferenced — geographic overlap could not be verified."
        result.warnings.append(msg)
        result.checks.append(CheckResult("georef", "warn", msg))
    elif before_geo.source == "manual" or after_geo.source == "manual":
        msg = "One or both images rely on manually-entered bounds rather than embedded georeferencing."
        result.warnings.append(msg)
        result.checks.append(CheckResult("georef", "warn", msg))
    else:
        result.checks.append(CheckResult("georef", "pass"))

    # 4. CRS — present but unreprojectable is distinguishable from absent
    before_georef = read_georef(before_path)
    after_georef = read_georef(after_path)
    broken_crs = [
        label for label, gi in (("Before", before_georef), ("After", after_georef))
        if gi is not None and gi.crs is not None and gi.bounds_wgs84 is None
    ]
    if broken_crs:
        msg = (f"{' and '.join(broken_crs)} image CRS is present but could not be reprojected "
               "to WGS84 — verify the coordinate system.")
        result.warnings.append(msg)
        result.checks.append(CheckResult("crs", "warn", msg))
    else:
        result.checks.append(CheckResult("crs", "pass"))

    # 5. GSD — never a hard fail, only harmonization is affected downstream
    gsd_b = read_gsd_meters(before_path)
    gsd_a = read_gsd_meters(after_path)
    if gsd_b and gsd_a:
        rel_diff = abs(gsd_b - gsd_a) / max(gsd_b, gsd_a)
        tolerance = get_gsd_tolerance()
        if rel_diff > tolerance:
            msg = (f"Ground sample distance differs by {rel_diff * 100:.0f}% "
                   f"(before {gsd_b:.2f} m/px, after {gsd_a:.2f} m/px) — "
                   "the finer image will be resampled to match.")
            result.warnings.append(msg)
            result.checks.append(CheckResult("gsd", "warn", msg))
        else:
            result.checks.append(CheckResult("gsd", "pass"))
    else:
        result.checks.append(CheckResult("gsd", "pass"))

    # 6. Overlap — cropped to the selected ROI when one is set, so the check
    # reflects the area detection will actually run on, not the full images.
    before_bounds = _apply_roi_to_bounds(before_geo.bounds, roi)
    after_bounds = _apply_roi_to_bounds(after_geo.bounds, roi)
    roi_note = " within the selected area" if roi else ""
    overlap_frac = _rect_overlap_frac(before_bounds, after_bounds)
    if overlap_frac is not None:
        hard_min = get_min_overlap_hard()
        warn_min = get_min_overlap_warn()
        if overlap_frac < hard_min:
            result.hard_fail = True
            result.fail_reason = (
                f"Images do not overlap on the ground{roi_note} (overlap {overlap_frac * 100:.1f}% "
                "of the after-image footprint) — not suitable for change detection.")
            result.checks.append(CheckResult("overlap", "fail", result.fail_reason))
            return result
        if overlap_frac < warn_min:
            msg = f"Weak geographic overlap{roi_note} ({overlap_frac * 100:.1f}% of the after-image footprint)."
            result.warnings.append(msg)
            result.checks.append(CheckResult("overlap", "warn", msg))
        else:
            result.checks.append(CheckResult("overlap", "pass"))
    else:
        result.checks.append(CheckResult("overlap", "pass"))

    return result