methods/method_rast.py

"""
Method RAST — Raster-only observability scoring.
=================================================

Uses only the point geometries and the raster pixel values; no attribute
fields are required.

For each pole point:
  1. An *inner circle* (radius = ``inner_radius``) is drawn around the pole.
     The maximum pixel value inside this circle is ``l_pole``.
  2. An *annulus* (ring between ``inner_radius`` and ``outer_radius``) is
     drawn around the pole.  The maximum pixel value inside the annulus is
     ``l_bg``.
  3. Global background statistics are computed across **all** poles:
       mu_bg    = median(l_bg)
       sigma_bg = MAD(l_bg)
  4. Per-pole observability score:
       s = (l_pole - mu_bg) / sigma_bg
  5. Detection:  detected = (s >= threshold)

Produced attributes
-------------------
l_pole, l_bg, mu_bg, sigma_bg, s,
img_radiance_near_mean, img_radiance_near_minus_far,
method, detected
"""

import math
import numpy as np
import geopandas as gpd
import rasterio
from rasterio.mask import mask as rasterio_mask
from scipy.stats import median_abs_deviation


def run(tiff_path, vector_path, band=1, threshold=3.0,
        inner_radius=5, outer_radius=15):
    """
    Run the RAST analysis.

    Parameters
    ----------
    tiff_path : str
        Path to the input GeoTIFF (night-time light raster).
    vector_path : str
        Path to the input vector file (pole points, GeoJSON / GPKG / SHP).
    band : int
        Raster band to use (1, 2 or 3).
    threshold : float
        Observability threshold.  A point is *detected* when ``s >= threshold``.
    inner_radius : int
        Inner radius of the annulus in metres (also the pole-circle radius).
    outer_radius : int
        Outer radius of the annulus in metres.

    Returns
    -------
    geopandas.GeoDataFrame
        Original poles enriched with RAST attributes.
    """
    # ------------------------------------------------------------------
    # Read vector data
    # ------------------------------------------------------------------
    gdf = gpd.read_file(vector_path)

    # Convert any datetime columns to ISO strings (avoids JSON issues)
    for col in gdf.columns:
        if str(gdf[col].dtype).startswith('datetime'):
            gdf[col] = gdf[col].apply(
                lambda x: x.isoformat()
                if x is not None and not (isinstance(x, float) and math.isnan(x))
                else None
            )

    # ------------------------------------------------------------------
    # Open raster and reproject vector
    # ------------------------------------------------------------------
    with rasterio.open(tiff_path) as src:
        raster_crs = src.crs

        # Reproject vector to raster CRS
        gdf_raster_crs = gdf.to_crs(raster_crs)

        # Determine a projected CRS (UTM) for metre-accurate buffering
        centroid = gdf_raster_crs.geometry.unary_union.centroid
        lon, lat = centroid.x, centroid.y
        if gdf_raster_crs.crs.is_geographic:
            utm_zone = int((lon + 180) / 6) + 1
            hemisphere = 'north' if lat >= 0 else 'south'
            utm_crs = f'+proj=utm +zone={utm_zone} +{hemisphere} +datum=WGS84'
        else:
            utm_crs = raster_crs

        gdf_proj = gdf_raster_crs.to_crs(utm_crs)

        # ------------------------------------------------------------------
        # Per-pole pixel extraction
        # ------------------------------------------------------------------
        l_pole_list = []
        l_bg_list = []
        near_mean_list = []
        near_minus_far_list = []

        for _idx, row in gdf_proj.iterrows():
            point = row.geometry

            inner_circle = point.buffer(inner_radius)
            outer_circle = point.buffer(outer_radius)
            annulus = outer_circle.difference(inner_circle)

            # Reproject geometries back to raster CRS for masking
            inner_gdf = gpd.GeoDataFrame(
                geometry=[inner_circle], crs=utm_crs
            ).to_crs(raster_crs)
            annulus_gdf = gpd.GeoDataFrame(
                geometry=[annulus], crs=utm_crs
            ).to_crs(raster_crs)

            inner_geom = [inner_gdf.geometry.iloc[0]]
            annulus_geom = [annulus_gdf.geometry.iloc[0]]

            # Extract pixels inside inner circle
            try:
                inner_data, _ = rasterio_mask(
                    src, inner_geom, crop=True, nodata=0, filled=True
                )
                band_data_inner = inner_data[band - 1]
                valid_inner = band_data_inner[band_data_inner != 0]
                l_pole = float(np.max(valid_inner)) if len(valid_inner) > 0 else 0.0
                near_mean = float(np.mean(valid_inner)) if len(valid_inner) > 0 else 0.0
            except Exception:
                l_pole = 0.0
                near_mean = 0.0

            # Extract pixels inside annulus
            try:
                annulus_data, _ = rasterio_mask(
                    src, annulus_geom, crop=True, nodata=0, filled=True
                )
                band_data_annulus = annulus_data[band - 1]
                valid_annulus = band_data_annulus[band_data_annulus != 0]
                l_bg = float(np.max(valid_annulus)) if len(valid_annulus) > 0 else 0.0
                far_mean = float(np.mean(valid_annulus)) if len(valid_annulus) > 0 else 0.0
            except Exception:
                l_bg = 0.0
                far_mean = 0.0

            l_pole_list.append(l_pole)
            l_bg_list.append(l_bg)
            near_mean_list.append(near_mean)
            near_minus_far_list.append(near_mean - far_mean)

    # ------------------------------------------------------------------
    # Global statistics
    # ------------------------------------------------------------------
    l_bg_array = np.array(l_bg_list)
    mu_bg = float(np.median(l_bg_array))
    sigma_bg = float(median_abs_deviation(l_bg_array))

    # ------------------------------------------------------------------
    # Per-point scoring
    # ------------------------------------------------------------------
    s_list = []
    for l_pole in l_pole_list:
        if sigma_bg > 0:
            s = (l_pole - mu_bg) / sigma_bg
        else:
            s = 0.0
        s_list.append(s)

    # ------------------------------------------------------------------
    # Build output GeoDataFrame
    # ------------------------------------------------------------------
    result_gdf = gdf.copy()
    result_gdf['l_pole'] = l_pole_list
    result_gdf['l_bg'] = l_bg_list
    result_gdf['mu_bg'] = mu_bg
    result_gdf['sigma_bg'] = sigma_bg
    result_gdf['s'] = s_list
    result_gdf['img_radiance_near_mean'] = near_mean_list
    result_gdf['img_radiance_near_minus_far'] = near_minus_far_list
    result_gdf['method'] = 'RAST'
    result_gdf['detected'] = [bool(s >= threshold) for s in s_list]

    return result_gdf