Spaces:
Sleeping
Sleeping
File size: 6,820 Bytes
95cb050 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 |
"""
Batch processing functions for CSV file assessments
"""
from spatial_queries import get_terrain_metrics, distance_to_water
from vulnerability import calculate_vulnerability_index, calculate_multi_hazard_vulnerability
from height_predictor.inference import get_predictor
from height_predictor.get_height_gba import GlobalBuildingAtlasHeight
# Initialize GBA getter (singleton pattern)
gba_getter = GlobalBuildingAtlasHeight()
def process_single_row(row, use_predicted_height=False, use_gba_height=False):
"""Process a single row from CSV - used for parallel processing."""
try:
lat = row['latitude']
lon = row['longitude']
height = row.get('height', 0.0)
basement = row.get('basement', 0.0)
if use_gba_height:
try:
result = gba_getter.get_height_m(lat, lon, buffer_m=5.0)
if result.get('status') == 'success' and result.get('predicted_height') is not None:
h = result['predicted_height']
if h >= 0: # Only use valid positive heights
height = h
except Exception as e:
print(f"GBA height failed for {lat},{lon}: {e}")
elif use_predicted_height:
try:
predictor = get_predictor()
pred = predictor.predict_from_coordinates(lat, lon)
if pred['status'] == 'success' and pred['predicted_height'] is not None:
height = pred['predicted_height']
except Exception as e:
print(f"Height prediction failed for {lat},{lon}: {e}")
terrain = get_terrain_metrics(lat, lon)
water_dist = distance_to_water(lat, lon)
result = calculate_vulnerability_index(
lat=lat,
lon=lon,
height=height,
basement=basement,
terrain_metrics=terrain,
water_distance=water_dist
)
# CSV output - essential columns
return {
'latitude': lat,
'longitude': lon,
'height': height,
'basement': basement,
'vulnerability_index': result['vulnerability_index'],
'ci_lower_95': result['confidence_interval']['lower_bound_95'],
'ci_upper_95': result['confidence_interval']['upper_bound_95'],
'vulnerability_level': result['risk_level'],
'confidence': result['uncertainty_analysis']['confidence'],
'confidence_interpretation': result['uncertainty_analysis']['interpretation'],
'elevation_m': result['elevation_m'],
'tpi_m': result['relative_elevation_m'],
'slope_degrees': result['slope_degrees'],
'distance_to_water_m': result['distance_to_water_m'],
'quality_flags': ','.join(result['uncertainty_analysis']['data_quality_flags']) if result['uncertainty_analysis']['data_quality_flags'] else ''
}
except Exception as e:
return {
'latitude': row.get('latitude'),
'longitude': row.get('longitude'),
'height': row.get('height', 0.0),
'basement': row.get('basement', 0.0),
'error': str(e),
'vulnerability_index': None,
'ci_lower_95': None,
'ci_upper_95': None,
'risk_level': None,
'confidence': None,
'confidence_interpretation': None,
'elevation_m': None,
'tpi_m': None,
'slope_degrees': None,
'distance_to_water_m': None,
'quality_flags': ''
}
def process_single_row_multihazard(row, use_predicted_height=False, use_gba_height=False):
"""Process a single row with multi-hazard assessment."""
try:
lat = row['latitude']
lon = row['longitude']
height = row.get('height', 0.0)
basement = row.get('basement', 0.0)
if use_gba_height:
try:
result = gba_getter.get_height_m(lat, lon, buffer_m=5.0)
if result.get('status') == 'success' and result.get('predicted_height') is not None:
h = result['predicted_height']
if h >= 0: # Only use valid positive heights
height = h
except Exception as e:
print(f"GBA height failed for {lat},{lon}: {e}")
elif use_predicted_height:
try:
predictor = get_predictor()
pred = predictor.predict_from_coordinates(lat, lon)
if pred['status'] == 'success' and pred['predicted_height'] is not None:
height = pred['predicted_height']
except Exception as e:
print(f"Height prediction failed for {lat},{lon}: {e}")
terrain = get_terrain_metrics(lat, lon)
water_dist = distance_to_water(lat, lon)
result = calculate_multi_hazard_vulnerability(
lat=lat,
lon=lon,
height=height,
basement=basement,
terrain_metrics=terrain,
water_distance=water_dist
)
return {
'latitude': lat,
'longitude': lon,
'height': height,
'basement': basement,
'vulnerability_index': result['vulnerability_index'],
'ci_lower_95': result['confidence_interval']['lower_bound_95'],
'ci_upper_95': result['confidence_interval']['upper_bound_95'],
'vulnerability_level': result['risk_level'],
'confidence': result['uncertainty_analysis']['confidence'],
'confidence_interpretation': result['uncertainty_analysis']['interpretation'],
'elevation_m': result['elevation_m'],
'tpi_m': result['relative_elevation_m'],
'slope_degrees': result['slope_degrees'],
'distance_to_water_m': result['distance_to_water_m'],
'dominant_hazard': result['dominant_hazard'],
'fluvial_risk': result['hazard_breakdown']['fluvial_riverine'],
'coastal_risk': result['hazard_breakdown']['coastal_surge'],
'pluvial_risk': result['hazard_breakdown']['pluvial_drainage'],
'combined_risk': result['hazard_breakdown']['combined_index'],
'quality_flags': ','.join(result['uncertainty_analysis']['data_quality_flags'])
if result['uncertainty_analysis']['data_quality_flags'] else ''
}
except Exception as e:
return {
'latitude': row.get('latitude'),
'longitude': row.get('longitude'),
'height': row.get('height', 0.0),
'basement': row.get('basement', 0.0),
'error': str(e),
'vulnerability_index': None
}
|