""" 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