methods/method_c.py

"""
Method C — Per-pole robust Z-score detection with road orientation.
===================================================================

Ported from the R prototype ``HK_summary_orientation.R``.

For each pole point:
  1. **Road Orientation** — finds the nearest road segment and calculates its azimuth.
  2. **Satellite Azimuth** — uses a provided satellite azimuth.
  3. **Alpha** — calculates the angular difference between road and satellite azimuth.
  4. **Pole sampling window** — a circle of radius ``pole_window_m / 2``
     around the pole.  Three radiance statistics are extracted:
       I_pole_max    = max(pole pixels)
       I_pole_median = median(pole pixels)
       I_pole_Q95    = 95th percentile(pole pixels)
  5. **Background ring (annulus)** — between ``bg_inner_m`` and
     ``bg_outer_m``.  Two statistics are extracted:
       I_bg_median = median(background pixels)
       MAD_bg      = median(|bg - I_bg_median|)
  6. **Robust Z-scores** (per-pole, using the *local* background):
       sigma_robust = max(1.4826 * MAD_bg, eps)
       Z_max    = (I_pole_max    - I_bg_median) / sigma_robust
       Z_median = (I_pole_median - I_bg_median) / sigma_robust
       Z_Q95    = (I_pole_Q95    - I_bg_median) / sigma_robust
  7. **Detection decisions** (Z >= T_z):
       det_max, det_median, det_Q95

Produced attributes
-------------------
road_id, dist_to_road_m, road_match_valid, road_azimuth_deg, road_orientation_deg, orient_class,
sat_azimuth_deg, sat_azimuth_axial, alpha_deg, alpha_class, orientation_weight_sin, orientation_weight_cos,
d_pole_nn_m, d_pole_m, d_proj_m, Wr_case1_m, Delta_case1_m, Wr_case2_m, Delta_case2_m, Wr_case3_m,
I_pole_max, I_pole_median, I_pole_Q95, I_bg_median, MAD_bg,
Z_max, Z_median, Z_Q95, det_max, det_median, det_Q95,
pole_window_m, bg_inner_m, bg_outer_m, T_z, method
"""

import math
import numpy as np
import geopandas as gpd
import pandas as pd
import rasterio
from rasterio.mask import mask as rasterio_mask
from shapely.geometry import Point, LineString
from scipy.spatial import cKDTree

def angular_difference_axial(a_deg, b_deg):
    alpha = abs(a_deg - b_deg)
    alpha = np.where(alpha > 90, 180 - alpha, alpha)
    return alpha

def azimuth_from_dxdy(dx, dy):
    az = np.arctan2(dx, dy) * 180 / np.pi
    return (az + 360) % 360

def get_segment_azimuth(point, line):
    if line is None or line.is_empty:
        return np.nan
    
    if line.geom_type == 'MultiLineString':
        line = line.geoms[0]
        
    coords = np.array(line.coords)
    if len(coords) < 2:
        return np.nan
        
    seg_midpoints = (coords[:-1] + coords[1:]) / 2
    pt_coords = np.array(point.coords)[0]
    
    dists = np.sqrt((seg_midpoints[:, 0] - pt_coords[0])**2 + (seg_midpoints[:, 1] - pt_coords[1])**2)
    idx = np.argmin(dists)
    
    seg_start = coords[idx]
    seg_end = coords[idx + 1]
    
    dx = seg_end[0] - seg_start[0]
    dy = seg_end[1] - seg_start[1]
    
    return azimuth_from_dxdy(dx, dy)

def make_orientation_class(orientation_deg):
    bins = [0, 22.5, 67.5, 112.5, 157.5, 180]
    labels = ["N-S", "NE-SW", "E-W", "NW-SE", "N-S"]
    # pandas cut doesn't support duplicate labels directly in the same way, so we map them
    temp_labels = ["L1", "L2", "L3", "L4", "L5"]
    res = pd.cut(orientation_deg, bins=bins, labels=temp_labels, include_lowest=True)
    mapping = {"L1": "N-S", "L2": "NE-SW", "L3": "E-W", "L4": "NW-SE", "L5": "N-S"}
    return res.map(mapping)

def make_alpha_class(alpha_deg):
    bins = [0, 15, 30, 45, 60, 75, 90]
    labels = ["(0, 15]", "(15, 30]", "(30, 45]", "(45, 60]", "(60, 75]", "(75, 90]"]
    return pd.cut(alpha_deg, bins=bins, include_lowest=True)

def estimate_nearest_pole_spacing(gdf):
    if len(gdf) < 2:
        return np.full(len(gdf), np.nan)
    
    coords = np.array([(geom.x, geom.y) for geom in gdf.geometry])
    tree = cKDTree(coords)
    dists, _ = tree.query(coords, k=2)
    return dists[:, 1]

import xml.etree.ElementTree as ET

def read_sgdsat_azimuth(meta_file):
    if not meta_file:
        return 0.0, 0.0
        
    try:
        tree = ET.parse(meta_file)
        root = tree.getroot()
        
        def get_val(xpath):
            node = root.find(f".//{xpath}")
            if node is not None and node.text:
                return float(node.text)
            return np.nan
            
        yaw = get_val("YawSatelliteAngle")
        if np.isnan(yaw):
            yaw = 0.0
            
        lat_tl = get_val("TopLeftLatitude")
        lon_tl = get_val("TopLeftLongitude")
        lat_tr = get_val("TopRightLatitude")
        lon_tr = get_val("TopRightLongitude")
        
        if np.isnan(lat_tl) or np.isnan(lon_tl) or np.isnan(lat_tr) or np.isnan(lon_tr):
            return 0.0, yaw
            
        dlat = lat_tr - lat_tl
        dlon = lon_tr - lon_tl
        
        sat_azimuth_deg = azimuth_from_dxdy(dlon, dlat)
        sat_azimuth_deg = (sat_azimuth_deg + yaw) % 360
        
        return sat_azimuth_deg, yaw
    except Exception as e:
        print(f"Warning: Could not read metadata file: {e}")
        return 0.0, 0.0

def run(tiff_path, vector_path, roads_path=None, meta_path=None, band=1, threshold=3.0,
        pole_window_m=30.0, bg_inner_m=40.0, bg_outer_m=80.0, max_road_distance_m=30.0, default_pole_spacing_m=35.0):
    """
    Run the Method C analysis.
    """
    T_z = float(threshold)
    eps = 1e-6
    pole_radius = pole_window_m / 2.0
    k1 = 0.4
    k2 = 0.8

    # ------------------------------------------------------------------
    # Read vector data
    # ------------------------------------------------------------------
    gdf = gpd.read_file(vector_path)

    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
        gdf_raster_crs = gdf.to_crs(raster_crs)

        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)
        
        # ------------------------------------------------------------------
        # Road Orientation
        # ------------------------------------------------------------------
        n = len(gdf_proj)
        road_id_arr = np.full(n, -1, dtype=int)
        dist_to_road_m_arr = np.full(n, np.nan)
        road_azimuth_deg_arr = np.full(n, np.nan)
        road_match_valid_arr = np.zeros(n, dtype=bool)
        
        if roads_path is not None and roads_path != "":
            try:
                roads_gdf = gpd.read_file(roads_path)
                roads_gdf = roads_gdf[roads_gdf.geometry.type.isin(['LineString', 'MultiLineString'])]
                if len(roads_gdf) > 0:
                    roads_proj = roads_gdf.to_crs(utm_crs)
                    
                    # Nearest road
                    roads_sindex = roads_proj.sindex
                    for i, point in enumerate(gdf_proj.geometry):
                        nearest_idx = list(roads_sindex.nearest(point))[1][0]
                        nearest_road = roads_proj.iloc[nearest_idx]
                        dist = point.distance(nearest_road.geometry)
                        
                        road_id_arr[i] = nearest_idx
                        dist_to_road_m_arr[i] = dist
                        
                        if dist <= max_road_distance_m:
                            road_match_valid_arr[i] = True
                            road_azimuth_deg_arr[i] = get_segment_azimuth(point, nearest_road.geometry)
            except Exception as e:
                print(f"Warning: Could not process roads file: {e}")
                
        road_orientation_deg_arr = road_azimuth_deg_arr % 180
        orient_class_arr = make_orientation_class(road_orientation_deg_arr)
        
        sat_azimuth_deg, sat_yaw = read_sgdsat_azimuth(meta_path)
        sat_azimuth_axial = sat_azimuth_deg % 180
        road_azimuth_axial = road_azimuth_deg_arr % 180
        alpha_deg_arr = angular_difference_axial(road_azimuth_axial, sat_azimuth_axial)
        alpha_class_arr = make_alpha_class(alpha_deg_arr)
        
        orientation_weight_sin_arr = np.sin(alpha_deg_arr * np.pi / 180)
        orientation_weight_cos_arr = np.cos(alpha_deg_arr * np.pi / 180)
        
        d_pole_nn_m_arr = estimate_nearest_pole_spacing(gdf_proj)
        d_pole_m_arr = np.where(np.isnan(d_pole_nn_m_arr) | np.isinf(d_pole_nn_m_arr), default_pole_spacing_m, d_pole_nn_m_arr)
        d_pole_m_arr = np.minimum(d_pole_m_arr, 3 * default_pole_spacing_m)
        
        d_proj_m_arr = d_pole_m_arr * np.sin(alpha_deg_arr * np.pi / 180)
        
        Wr_case1_m_arr = np.full(n, 35.0)
        Delta_case1_m_arr = np.full(n, 60.0)
        Wr_case2_m_arr = k1 * d_pole_m_arr
        Delta_case2_m_arr = k2 * d_pole_m_arr
        Wr_case3_m_arr = k1 * d_proj_m_arr

        # ------------------------------------------------------------------
        # Per-pole extraction
        # ------------------------------------------------------------------
        I_pole_max_arr    = np.full(n, np.nan)
        I_pole_median_arr = np.full(n, np.nan)
        I_pole_Q95_arr    = np.full(n, np.nan)
        I_bg_median_arr   = np.full(n, np.nan)
        MAD_bg_arr        = np.full(n, np.nan)
        Z_max_arr         = np.full(n, np.nan)
        Z_median_arr      = np.full(n, np.nan)
        Z_Q95_arr         = np.full(n, np.nan)
        det_max_arr       = np.zeros(n, dtype=int)
        det_median_arr    = np.zeros(n, dtype=int)
        det_Q95_arr       = np.zeros(n, dtype=int)

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

            # --- 6.1 Pole sampling window ---
            pole_circle = point.buffer(pole_radius)
            pole_gdf = gpd.GeoDataFrame(
                geometry=[pole_circle], crs=utm_crs
            ).to_crs(raster_crs)
            pole_geom = [pole_gdf.geometry.iloc[0]]

            try:
                pole_data, _ = rasterio_mask(
                    src, pole_geom, crop=True, nodata=0, filled=True
                )
                pole_band = pole_data[band - 1]
                pole_vals = pole_band[pole_band != 0]
            except Exception:
                pole_vals = np.array([])

            if len(pole_vals) > 0:
                I_pole_max_arr[i]    = float(np.max(pole_vals))
                I_pole_median_arr[i] = float(np.median(pole_vals))
                I_pole_Q95_arr[i]    = float(np.percentile(pole_vals, 95))

            # --- 6.2 Background ring (annulus) ---
            bg_outer_circle = point.buffer(bg_outer_m)
            bg_inner_circle = point.buffer(bg_inner_m)
            bg_ring = bg_outer_circle.difference(bg_inner_circle)

            bg_gdf = gpd.GeoDataFrame(
                geometry=[bg_ring], crs=utm_crs
            ).to_crs(raster_crs)
            bg_geom = [bg_gdf.geometry.iloc[0]]

            try:
                bg_data, _ = rasterio_mask(
                    src, bg_geom, crop=True, nodata=0, filled=True
                )
                bg_band = bg_data[band - 1]
                bg_vals = bg_band[bg_band != 0]
            except Exception:
                bg_vals = np.array([])

            # Need at least 5 background pixels
            if len(bg_vals) >= 5:
                bg_median = float(np.median(bg_vals))
                bg_mad = float(np.median(np.abs(bg_vals - bg_median)))
                sigma_robust = max(1.4826 * bg_mad, eps)

                I_bg_median_arr[i] = bg_median
                MAD_bg_arr[i]      = bg_mad

                # --- 6.3 Robust Z-scores ---
                if not np.isnan(I_pole_max_arr[i]):
                    Z_max_arr[i]    = (I_pole_max_arr[i]    - bg_median) / sigma_robust
                    Z_median_arr[i] = (I_pole_median_arr[i] - bg_median) / sigma_robust
                    Z_Q95_arr[i]    = (I_pole_Q95_arr[i]    - bg_median) / sigma_robust

                    # --- 6.4 Detection decisions ---
                    det_max_arr[i]    = int(Z_max_arr[i]    >= T_z)
                    det_median_arr[i] = int(Z_median_arr[i] >= T_z)
                    det_Q95_arr[i]    = int(Z_Q95_arr[i]    >= T_z)

    # ------------------------------------------------------------------
    # Build output GeoDataFrame
    # ------------------------------------------------------------------
    result_gdf = gdf.copy()
    
    result_gdf['road_id'] = road_id_arr
    result_gdf['dist_to_road_m'] = dist_to_road_m_arr
    result_gdf['road_match_valid'] = road_match_valid_arr
    result_gdf['road_azimuth_deg'] = road_azimuth_deg_arr
    result_gdf['road_orientation_deg'] = road_orientation_deg_arr
    result_gdf['orient_class'] = orient_class_arr
    
    result_gdf['sat_azimuth_deg'] = sat_azimuth_deg
    result_gdf['sat_azimuth_axial'] = sat_azimuth_axial
    result_gdf['sat_yaw'] = sat_yaw
    result_gdf['alpha_deg'] = alpha_deg_arr
    result_gdf['alpha_class'] = alpha_class_arr
    result_gdf['orientation_weight_sin'] = orientation_weight_sin_arr
    result_gdf['orientation_weight_cos'] = orientation_weight_cos_arr
    
    result_gdf['d_pole_nn_m'] = d_pole_nn_m_arr
    result_gdf['d_pole_m'] = d_pole_m_arr
    result_gdf['d_proj_m'] = d_proj_m_arr
    
    result_gdf['Wr_case1_m'] = Wr_case1_m_arr
    result_gdf['Delta_case1_m'] = Delta_case1_m_arr
    result_gdf['Wr_case2_m'] = Wr_case2_m_arr
    result_gdf['Delta_case2_m'] = Delta_case2_m_arr
    result_gdf['Wr_case3_m'] = Wr_case3_m_arr
    
    result_gdf['I_pole_max']    = I_pole_max_arr
    result_gdf['I_pole_median'] = I_pole_median_arr
    result_gdf['I_pole_Q95']    = I_pole_Q95_arr
    result_gdf['I_bg_median']   = I_bg_median_arr
    result_gdf['MAD_bg']        = MAD_bg_arr
    result_gdf['Z_max']         = Z_max_arr
    result_gdf['Z_median']      = Z_median_arr
    result_gdf['Z_Q95']         = Z_Q95_arr
    result_gdf['det_max']       = det_max_arr
    result_gdf['det_median']    = det_median_arr
    result_gdf['det_Q95']       = det_Q95_arr
    result_gdf['pole_window_m'] = pole_window_m
    result_gdf['bg_inner_m']    = bg_inner_m
    result_gdf['bg_outer_m']    = bg_outer_m
    result_gdf['T_z']           = T_z
    result_gdf['method']        = 'Method C'

    return result_gdf