Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| } | |