""" EIGIS FastAPI Application Microservices architecture with PostGIS integration High-concurrency endpoints for geospatial data collection & dashboard """ from fastapi import FastAPI, HTTPException, Depends, Query, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from typing import List, Optional from datetime import datetime, date, timedelta import json import logging import os from .models import ( ProjectCreate, ProjectResponse, ProjectMetrics, FieldTripCreate, FieldTripResponse, ObservationCreate, ObservationResponse, ObservationMapPoint, SoilProfileCreate, RockDescriptionCreate, DiscontinuityCreate, SlopeStabilityCreate, GeothermalManifestationCreate, SampleCreate, DashboardMetrics, ThemeDistribution, HazardDistribution, GeothermalStats, ) from .database import get_pool, close_pool, execute, fetch, fetchrow, fetchval from .services import email_service # ============================================================ # APP CONFIGURATION # ============================================================ app = FastAPI( title="EIGIS Engineering Geology & Geohazard Information System", description=""" ## EIGIS API Documentation A high-performance geospatial data collection and analytics API for engineering geology, structural features, and geothermal monitoring. ### Themes - **Surface Geological**: Soil profiles, rock descriptions, lithological units - **Structural**: Discontinuities, slope stability, rock mass classification - **Geothermal**: Hot springs, fumaroles, geothermal gradients, geochemistry ### Features - PostGIS spatial queries with complex indexing - Concurrent data ingestion with connection pooling - Real-time dashboard metrics via materialized views - Email alerts for anomalous progress detection - Mobile data collection sync support - Third-party integration endpoints ### Architecture - Microservices-ready with stateless design - PostgreSQL + PostGIS backend - Async connection pooling (asyncpg) - Event-driven alert triggers """, version="2.0.0", docs_url="/api/docs", redoc_url="/api/redoc", openapi_url="/api/openapi.json", contact={ "name": "GIE Engineering Geology Division", "email": "eigis@gie.gov", "url": "https://www.gie.gov/eigis", }, license_info={ "name": "MIT", "url": "https://opensource.org/licenses/MIT", }, ) # CORS for React frontend and third-party integrations app.add_middleware( CORSMiddleware, allow_origins=[ os.getenv("FRONTEND_URL", "http://localhost:3000"), os.getenv("FRONTEND_URL", "http://localhost:5173"), "https://eigis.pages.dev", "*", ], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("eigis.api") # ============================================================ # STARTUP / SHUTDOWN # ============================================================ @app.on_event("startup") async def startup(): logger.info("EIGIS API starting up... Initializing connection pool.") await get_pool() logger.info("Connection pool ready.") @app.on_event("shutdown") async def shutdown(): logger.info("EIGIS API shutting down... Closing connection pool.") await close_pool() # ============================================================ # HEALTH CHECK # ============================================================ @app.get("/api/health", tags=["System"]) async def health_check(): """System health check endpoint for monitoring and load balancers.""" try: db_ok = await fetchval("SELECT 1") postgis_ver = await fetchval("SELECT PostGIS_Version()") return { "status": "healthy", "database": "connected" if db_ok else "error", "postgis_version": postgis_ver, "timestamp": datetime.utcnow().isoformat(), } except Exception as e: return JSONResponse( status_code=503, content={"status": "unhealthy", "error": str(e)}, ) # ============================================================ # PROJECT ENDPOINTS # ============================================================ @app.post("/api/projects", response_model=ProjectResponse, tags=["Projects"], status_code=201) async def create_project(data: ProjectCreate): """Create a new EIGIS project with optional spatial boundary.""" geom_wkt = None if data.boundary_geom: coords = data.boundary_geom.get("coordinates") if coords: geom_wkt = f"ST_MakePolygon(ST_GeomFromText('LINESTRING({','.join([f'{c[0]} {c[1]}' for c in coords[0]])})', 4326))" query = """ INSERT INTO projects (project_code, project_name, description, client_name, client_email, receptionist_email, region, start_date, end_date, status, boundary_geom) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, ST_SetSRID(ST_MakePoint(0,0), 4326)) RETURNING project_id, project_code, project_name, description, client_name, client_email, region, status, start_date, end_date, created_at """ row = await fetchrow( query, data.project_code, data.project_name, data.description, data.client_name, data.client_email, data.receptionist_email, data.region, data.start_date, data.end_date, data.status.value, ) return ProjectResponse( project_id=str(row["project_id"]), project_code=row["project_code"], project_name=row["project_name"], description=row["description"], client_name=row["client_name"], client_email=row["client_email"], region=row["region"], status=row["status"], start_date=row["start_date"], end_date=row["end_date"], created_at=row["created_at"], ) @app.get("/api/projects", response_model=List[ProjectResponse], tags=["Projects"]) async def list_projects( status: Optional[str] = Query(None, description="Filter by project status"), limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0), ): """List all projects with optional status filtering.""" if status: rows = await fetch( """SELECT project_id, project_code, project_name, description, client_name, client_email, region, status, start_date, end_date, created_at FROM projects WHERE status = $1 ORDER BY created_at DESC LIMIT $2 OFFSET $3""", status, limit, offset, ) else: rows = await fetch( """SELECT project_id, project_code, project_name, description, client_name, client_email, region, status, start_date, end_date, created_at FROM projects ORDER BY created_at DESC LIMIT $1 OFFSET $2""", limit, offset, ) return [ ProjectResponse( project_id=str(r["project_id"]), project_code=r["project_code"], project_name=r["project_name"], description=r["description"], client_name=r["client_name"], client_email=r["client_email"], region=r["region"], status=r["status"], start_date=r["start_date"], end_date=r["end_date"], created_at=r["created_at"], ) for r in rows ] @app.get("/api/projects/{project_id}/metrics", response_model=ProjectMetrics, tags=["Projects"]) async def get_project_metrics(project_id: str): """Get real-time project metrics from materialized view.""" row = await fetchrow( "SELECT * FROM mv_dashboard_metrics WHERE project_id = $1", project_id, ) if not row: raise HTTPException(404, "Project not found") return ProjectMetrics( project_id=str(row["project_id"]), project_code=row["project_code"], project_name=row["project_name"], project_status=row["project_status"], total_observations=row["total_observations"] or 0, surface_geological_count=row["surface_geological_count"] or 0, structural_count=row["structural_count"] or 0, geothermal_count=row["geothermal_count"] or 0, soil_profiles=row["soil_profiles"] or 0, soil_horizons=row["soil_horizons"] or 0, rock_descriptions=row["rock_descriptions"] or 0, discontinuity_sets=row["discontinuity_sets"] or 0, slope_assessments=row["slope_assessments"] or 0, rock_mass_classifications=row["rock_mass_classifications"] or 0, geothermal_manifestations=row["geothermal_manifestations"] or 0, total_samples=row["total_samples"] or 0, samples_completed=row["samples_completed"] or 0, total_photos=row["total_photos"] or 0, high_hazard_count=row["high_hazard_count"] or 0, very_high_hazard_count=row["very_high_hazard_count"] or 0, extreme_hazard_count=row["extreme_hazard_count"] or 0, max_geothermal_temp=row["max_geothermal_temp"], avg_geothermal_temp=row["avg_geothermal_temp"], landslide_count=row["landslide_count"] or 0, last_observation_at=row["last_observation_at"], ) # ============================================================ # FIELD TRIP ENDPOINTS # ============================================================ @app.post("/api/field-trips", response_model=FieldTripResponse, tags=["Field Trips"], status_code=201) async def create_field_trip(data: FieldTripCreate): """Create a new field trip record.""" query = """ INSERT INTO field_trips (project_id, trip_code, trip_date, weather_condition, area_visited, device_id) VALUES ($1, $2, $3, $4, $5, $6) RETURNING field_trip_id, project_id, trip_code, trip_date, weather_condition, area_visited, sync_status, created_at """ row = await fetchrow( query, data.project_id, data.trip_code, data.trip_date, data.weather_condition, data.area_visited, data.device_id, ) return FieldTripResponse( field_trip_id=str(row["field_trip_id"]), project_id=str(row["project_id"]), trip_code=row["trip_code"], trip_date=row["trip_date"], weather_condition=row["weather_condition"], area_visited=row["area_visited"], sync_status=row["sync_status"], created_at=row["created_at"], ) # ============================================================ # OBSERVATION ENDPOINTS # ============================================================ @app.post("/api/observations", response_model=ObservationResponse, tags=["Observations"], status_code=201) async def create_observation(data: ObservationCreate, background_tasks: BackgroundTasks): """ Create a new observation with spatial geometry. Auto-populates admin unit, watershed, and geological formation via triggers. """ query = """ INSERT INTO observations ( field_trip_id, site_id, observation_type, exposure_type, geom, elevation_m, easting, northing, utm_zone, exposure_length_m, exposure_height_m, groundwater_level_m, weather_condition, excavation_method, accessibility, remarks, logger_id ) VALUES ( $1, $2, $3, $4, ST_SetSRID(ST_MakePoint($5, $6), 4326), $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 ) RETURNING observation_id, field_trip_id, site_id, observation_type, ST_Y(geom) as latitude, ST_X(geom) as longitude, elevation_m, admin_unit, watershed, geological_formation, status, created_at """ row = await fetchrow( query, data.field_trip_id, data.site_id, data.observation_type.value, data.exposure_type, data.longitude, data.latitude, # ST_MakePoint(lon, lat) data.elevation_m, data.easting, data.northing, data.utm_zone, data.exposure_length_m, data.exposure_height_m, data.groundwater_level_m, data.weather_condition, data.excavation_method, data.accessibility, data.remarks, data.logger_id, ) return ObservationResponse( observation_id=str(row["observation_id"]), field_trip_id=str(row["field_trip_id"]), site_id=row["site_id"], observation_type=row["observation_type"], latitude=float(row["latitude"]), longitude=float(row["longitude"]), elevation_m=row["elevation_m"], admin_unit=row["admin_unit"], watershed=row["watershed"], geological_formation=row["geological_formation"], status=row["status"], created_at=row["created_at"], ) @app.get("/api/observations", response_model=List[ObservationResponse], tags=["Observations"]) async def list_observations( project_id: Optional[str] = None, observation_type: Optional[str] = None, status: Optional[str] = None, bbox: Optional[str] = Query(None, description="Bounding box: xmin,ymin,xmax,ymax"), limit: int = Query(100, ge=1, le=1000), offset: int = Query(0, ge=0), ): """List observations with spatial and attribute filtering.""" conditions = [] params = [] idx = 1 if project_id: conditions.append(f"ft.project_id = ${idx}") params.append(project_id) idx += 1 if observation_type: conditions.append(f"o.observation_type = ${idx}") params.append(observation_type) idx += 1 if status: conditions.append(f"o.status = ${idx}") params.append(status) idx += 1 if bbox: parts = bbox.split(",") if len(parts) == 4: conditions.append( f"o.geom && ST_MakeEnvelope(${idx}, ${idx+1}, ${idx+2}, ${idx+3}, 4326)" ) params.extend([float(p) for p in parts]) idx += 4 where = " AND ".join(conditions) if conditions else "TRUE" query = f""" SELECT o.observation_id, o.field_trip_id, o.site_id, o.observation_type, ST_Y(o.geom) as latitude, ST_X(o.geom) as longitude, o.elevation_m, o.admin_unit, o.watershed, o.geological_formation, o.status, o.created_at FROM observations o JOIN field_trips ft ON ft.field_trip_id = o.field_trip_id WHERE {where} ORDER BY o.created_at DESC LIMIT ${idx} OFFSET ${idx+1} """ params.extend([limit, offset]) rows = await fetch(query, *params) return [ ObservationResponse( observation_id=str(r["observation_id"]), field_trip_id=str(r["field_trip_id"]), site_id=r["site_id"], observation_type=r["observation_type"], latitude=float(r["latitude"]), longitude=float(r["longitude"]), elevation_m=r["elevation_m"], admin_unit=r["admin_unit"], watershed=r["watershed"], geological_formation=r["geological_formation"], status=r["status"], created_at=r["created_at"], ) for r in rows ] @app.get("/api/observations/map", response_model=List[ObservationMapPoint], tags=["WebGIS Map"]) async def get_map_points( project_id: Optional[str] = None, theme: Optional[str] = Query(None, description="Filter: surface_geological, structural, geothermal"), bbox: Optional[str] = Query(None, description="Bounding box filter for vector tiles"), ): """ Get observation points for WebGIS map rendering. Optimized for low-latency vector tile generation. Includes theme flags for client-side styling. """ conditions = [] params = [] idx = 1 if project_id: conditions.append(f"ft.project_id = ${idx}") params.append(project_id) idx += 1 if bbox: parts = bbox.split(",") if len(parts) == 4: conditions.append( f"o.geom && ST_MakeEnvelope(${idx}, ${idx+1}, ${idx+2}, ${idx+3}, 4326)" ) params.extend([float(p) for p in parts]) idx += 4 where = " AND ".join(conditions) if conditions else "TRUE" query = f""" SELECT o.observation_id, o.site_id, o.observation_type, ST_Y(o.geom) as latitude, ST_X(o.geom) as longitude, o.elevation_m, o.status, o.created_at, CASE WHEN sp.soil_profile_id IS NOT NULL OR rd.rock_desc_id IS NOT NULL THEN TRUE ELSE FALSE END AS has_surface_geological, CASE WHEN dm.disc_id IS NOT NULL OR sl.slope_id IS NOT NULL THEN TRUE ELSE FALSE END AS has_structural, CASE WHEN gm.manifestation_id IS NOT NULL THEN TRUE ELSE FALSE END AS has_geothermal, (SELECT COUNT(*) FROM samples s WHERE s.observation_id = o.observation_id) AS sample_count, (SELECT COUNT(*) FROM photos ph WHERE ph.observation_id = o.observation_id) AS photo_count FROM observations o JOIN field_trips ft ON ft.field_trip_id = o.field_trip_id LEFT JOIN soil_profiles sp ON sp.observation_id = o.observation_id LEFT JOIN rock_descriptions rd ON rd.observation_id = o.observation_id LEFT JOIN discontinuity_measurements dm ON dm.observation_id = o.observation_id LEFT JOIN slope_stability sl ON sl.observation_id = o.observation_id LEFT JOIN geothermal_manifestations gm ON gm.observation_id = o.observation_id WHERE {where} ORDER BY o.created_at DESC """ rows = await fetch(query, *params) results = [] for r in rows: point = ObservationMapPoint( observation_id=str(r["observation_id"]), site_id=r["site_id"], observation_type=r["observation_type"], latitude=float(r["latitude"]), longitude=float(r["longitude"]), elevation_m=r["elevation_m"], has_surface_geological=r["has_surface_geological"], has_structural=r["has_structural"], has_geothermal=r["has_geothermal"], sample_count=int(r["sample_count"]), photo_count=int(r["photo_count"]), status=r["status"], created_at=r["created_at"], ) # Apply theme filter if specified if theme == "surface_geological" and not point.has_surface_geological: continue if theme == "structural" and not point.has_structural: continue if theme == "geothermal" and not point.has_geothermal: continue results.append(point) return results # ============================================================ # SURFACE GEOLOGICAL ENDPOINTS # ============================================================ @app.post("/api/soil-profiles", tags=["Surface Geological"], status_code=201) async def create_soil_profile(data: SoilProfileCreate): """Create a soil profile with multiple horizons for an observation.""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): # Create profile profile_id = await conn.fetchval( """INSERT INTO soil_profiles (observation_id, profile_code, total_depth_m, groundwater_depth_m, dilatancy, sorting, notes) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING soil_profile_id""", data.observation_id, data.profile_code, data.total_depth_m, data.groundwater_depth_m, data.dilatancy, data.sorting, data.notes, ) # Create horizons for h in data.horizons: await conn.execute( """INSERT INTO soil_horizons (soil_profile_id, horizon_label, depth_from_m, depth_to_m, color, consistency, moisture, grain_size, plasticity, uscs_class, organic_content, boundary_type, structure_type, notes, sort_order) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)""", profile_id, h.horizon_label, h.depth_from_m, h.depth_to_m, h.color, h.consistency, h.moisture, h.grain_size, h.plasticity, h.uscs_class, h.organic_content, h.boundary_type, h.structure_type, h.notes, h.sort_order, ) return {"soil_profile_id": str(profile_id), "horizons_created": len(data.horizons)} @app.post("/api/rock-descriptions", tags=["Surface Geological"], status_code=201) async def create_rock_description(data: RockDescriptionCreate): """Create a rock description for an observation.""" query = """ INSERT INTO rock_descriptions (observation_id, rock_name, rock_type, color, grain_size, texture, weathering_grade, intact_strength, structure, mineralogy, alteration, rock_description) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING rock_desc_id """ rid = await fetchval( query, data.observation_id, data.rock_name, data.rock_type, data.color, data.grain_size, data.texture, data.weathering_grade, data.intact_strength, data.structure, data.mineralogy, data.alteration, data.rock_description, ) return {"rock_desc_id": str(rid)} # ============================================================ # STRUCTURAL ENDPOINTS # ============================================================ @app.post("/api/discontinuities", tags=["Structural"], status_code=201) async def create_discontinuity(data: DiscontinuityCreate): """Add a discontinuity measurement to an observation.""" query = """ INSERT INTO discontinuity_measurements (observation_id, set_label, disc_type, dip_direction_deg, dip_deg, spacing_m, aperture_mm, persistence_m, roughness, infill_material, infill_thickness_mm, water_condition, wall_weathering, notes) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) RETURNING disc_id """ did = await fetchval( query, data.observation_id, data.set_label, data.disc_type, data.dip_direction_deg, data.dip_deg, data.spacing_m, data.aperture_mm, data.persistence_m, data.roughness, data.infill_material, data.infill_thickness_mm, data.water_condition, data.wall_weathering, data.notes, ) return {"disc_id": str(did)} @app.post("/api/discontinuities/bulk", tags=["Structural"], status_code=201) async def create_discontinuities_bulk(data: List[DiscontinuityCreate]): """Bulk insert discontinuity measurements for efficient concurrent ingestion.""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): ids = [] for d in data: rid = await conn.fetchval( """INSERT INTO discontinuity_measurements (observation_id, set_label, disc_type, dip_direction_deg, dip_deg, spacing_m, aperture_mm, persistence_m, roughness, infill_material, infill_thickness_mm, water_condition, wall_weathering, notes) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) RETURNING disc_id""", d.observation_id, d.set_label, d.disc_type, d.dip_direction_deg, d.dip_deg, d.spacing_m, d.aperture_mm, d.persistence_m, d.roughness, d.infill_material, d.infill_thickness_mm, d.water_condition, d.wall_weathering, d.notes, ) ids.append(str(rid)) return {"created": len(ids), "disc_ids": ids} @app.post("/api/slope-stability", tags=["Structural"], status_code=201) async def create_slope_stability(data: SlopeStabilityCreate, background_tasks: BackgroundTasks): """ Create slope stability assessment. Triggers email alert for high hazard levels. """ query = """ INSERT INTO slope_stability (observation_id, has_existing_slope, slope_height_m, slope_angle_deg, aspect_deg, slope_form, stability_condition, hazard_level, failure_mode, factor_of_safety, trigger_mechanism, recommended_actions, mitigation_measures, monitoring_required) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) RETURNING slope_id """ sid = await fetchval( query, data.observation_id, data.has_existing_slope, data.slope_height_m, data.slope_angle_deg, data.aspect_deg, data.slope_form, data.stability_condition, data.hazard_level.value, data.failure_mode, data.factor_of_safety, data.trigger_mechanism, data.recommended_actions, data.mitigation_measures, data.monitoring_required, ) # If hazard is high+, trigger alert in background if data.hazard_level in (HazardLevel.high, HazardLevel.very_high, HazardLevel.extreme): background_tasks.add_task( _send_hazard_alert, data.observation_id, data.hazard_level.value ) return {"slope_id": str(sid)} async def _send_hazard_alert(observation_id: str, hazard_level: str): """Background task to send hazard alert emails.""" try: row = await fetchrow( """SELECT p.client_email, p.receptionist_email, p.project_code, o.site_id FROM observations o JOIN field_trips ft ON ft.field_trip_id = o.field_trip_id JOIN projects p ON p.project_id = ft.project_id WHERE o.observation_id = $1""", observation_id, ) if row and row["client_email"]: await email_service.send_hazard_alert( client_email=row["client_email"], receptionist_email=row["receptionist_email"] or "", project_code=row["project_code"], site_id=row["site_id"], hazard_level=hazard_level, details=f"Slope stability assessment indicates {hazard_level} hazard level requiring immediate attention.", ) except Exception as e: logger.error(f"Failed to send hazard alert: {e}") # ============================================================ # GEOTHERMAL ENDPOINTS # ============================================================ @app.post("/api/geothermal-manifestations", tags=["Geothermal"], status_code=201) async def create_geothermal_manifestation( data: GeothermalManifestationCreate, background_tasks: BackgroundTasks ): """ Create a geothermal manifestation record. Triggers anomaly alerts for high temperatures or extreme pH values. """ query = """ INSERT INTO geothermal_manifestations (observation_id, manifestation_type, surface_temp_c, discharge_rate_lps, ph_value, electrical_conductivity_us, total_dissolved_solids, fluid_color, odor, deposit_type, alteration_zone, alteration_intensity, structural_control, host_rock, elevation_m, usage_current, usage_potential, notes) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) RETURNING manifestation_id """ mid = await fetchval( query, data.observation_id, data.manifestation_type.value, data.surface_temp_c, data.discharge_rate_lps, data.ph_value, data.electrical_conductivity_us, data.total_dissolved_solids, data.fluid_color, data.odor, data.deposit_type, data.alteration_zone, data.alteration_intensity, data.structural_control, data.host_rock, data.elevation_m, data.usage_current, data.usage_potential, data.notes, ) # Check for anomalies anomalies = [] if data.surface_temp_c and data.surface_temp_c > 95.0: anomalies.append("high_temperature") if data.ph_value and (data.ph_value < 2.0 or data.ph_value > 10.0): anomalies.append("ph_anomaly") if anomalies: background_tasks.add_task( _send_geothermal_alert, data.observation_id, anomalies ) return {"manifestation_id": str(mid), "anomalies_detected": anomalies} async def _send_geothermal_alert(observation_id: str, anomalies: List[str]): """Background task to send geothermal anomaly alerts.""" try: row = await fetchrow( """SELECT p.client_email, p.receptionist_email, p.project_code, o.site_id FROM observations o JOIN field_trips ft ON ft.field_trip_id = o.field_trip_id JOIN projects p ON p.project_id = ft.project_id WHERE o.observation_id = $1""", observation_id, ) if row and row["client_email"]: for anomaly in anomalies: desc = { "high_temperature": "Surface temperature exceeds 95°C threshold", "ph_anomaly": "pH value outside normal range (2-10)", }.get(anomaly, anomaly) await email_service.send_geothermal_anomaly_alert( client_email=row["client_email"], receptionist_email=row["receptionist_email"] or "", project_code=row["project_code"], site_id=row["site_id"], anomaly_type=anomaly, anomaly_details=desc, ) except Exception as e: logger.error(f"Failed to send geothermal alert: {e}") # ============================================================ # SAMPLING ENDPOINTS # ============================================================ @app.post("/api/samples", tags=["Sampling"], status_code=201) async def create_sample(data: SampleCreate): """Create a new sample linked to an observation.""" query = """ INSERT INTO samples (observation_id, sample_code, sample_type, depth_m, horizon_ref, weight_kg, sample_condition, storage_method, tests_requested, notes) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING sample_id """ sid = await fetchval( query, data.observation_id, data.sample_code, data.sample_type.value, data.depth_m, data.horizon_ref, data.weight_kg, data.sample_condition, data.storage_method, data.tests_requested, data.notes, ) return {"sample_id": str(sid)} @app.post("/api/samples/bulk", tags=["Sampling"], status_code=201) async def create_samples_bulk(data: List[SampleCreate]): """Bulk insert samples for efficient concurrent ingestion.""" pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): ids = [] for s in data: rid = await conn.fetchval( """INSERT INTO samples (observation_id, sample_code, sample_type, depth_m, horizon_ref, weight_kg, sample_condition, storage_method, tests_requested, notes) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING sample_id""", s.observation_id, s.sample_code, s.sample_type.value, s.depth_m, s.horizon_ref, s.weight_kg, s.sample_condition, s.storage_method, s.tests_requested, s.notes, ) ids.append(str(rid)) return {"created": len(ids), "sample_ids": ids} # ============================================================ # DASHBOARD ENDPOINTS # ============================================================ @app.get("/api/dashboard/metrics", response_model=DashboardMetrics, tags=["Dashboard"]) async def get_dashboard_metrics(project_id: Optional[str] = None): """ Get real-time dashboard metrics for visualization. Uses materialized views for low-latency rendering. """ # Refresh materialized view for latest data await execute("REFRESH MATERIALIZED VIEW CONCURRENTLY mv_dashboard_metrics") if project_id: row = await fetchrow( "SELECT * FROM mv_dashboard_metrics WHERE project_id = $1", project_id ) else: # Aggregate across all projects rows = await fetch("SELECT * FROM mv_dashboard_metrics") # Get recent observations for map recent = await fetch( """SELECT o.observation_id, o.site_id, o.observation_type, ST_Y(o.geom) as latitude, ST_X(o.geom) as longitude, o.elevation_m, o.status, o.created_at, CASE WHEN sp.soil_profile_id IS NOT NULL THEN TRUE ELSE FALSE END AS has_surface_geological, CASE WHEN dm.disc_id IS NOT NULL THEN TRUE ELSE FALSE END AS has_structural, CASE WHEN gm.manifestation_id IS NOT NULL THEN TRUE ELSE FALSE END AS has_geothermal, 0 AS sample_count, 0 AS photo_count FROM observations o LEFT JOIN soil_profiles sp ON sp.observation_id = o.observation_id LEFT JOIN discontinuity_measurements dm ON dm.observation_id = o.observation_id LEFT JOIN geothermal_manifestations gm ON gm.observation_id = o.observation_id ORDER BY o.created_at DESC LIMIT 50""" ) recent_points = [ ObservationMapPoint( observation_id=str(r["observation_id"]), site_id=r["site_id"], observation_type=r["observation_type"], latitude=float(r["latitude"]), longitude=float(r["longitude"]), elevation_m=r["elevation_m"], has_surface_geological=r["has_surface_geological"], has_structural=r["has_structural"], has_geothermal=r["has_geothermal"], sample_count=r["sample_count"], photo_count=r["photo_count"], status=r["status"], created_at=r["created_at"], ) for r in recent ] # Compute aggregate metrics if project_id and row: metrics = row else: # Sum across all projects agg = await fetchrow(""" SELECT COUNT(*) as total_projects, COUNT(*) FILTER (WHERE project_status = 'active') as active_projects, SUM(total_observations) as total_observations, SUM(surface_geological_count) as surface_geological_total, SUM(structural_count) as structural_total, SUM(geothermal_count) as geothermal_total, SUM(total_samples) as total_samples, SUM(samples_completed) as samples_completed, SUM(high_hazard_count) as high_hazard_count, SUM(landslide_count) as landslide_count, MAX(max_geothermal_temp) as max_geothermal_temp, AVG(avg_geothermal_temp) as avg_geothermal_temp FROM mv_dashboard_metrics """) metrics = agg or {} # Count today's observations obs_today = await fetchval( "SELECT COUNT(*) FROM observations WHERE created_at::date = CURRENT_DATE" ) # Count pending lab samples pending_lab = await fetchval( "SELECT COUNT(*) FROM samples WHERE lab_status = 'pending'" ) # Count critical alerts critical_alerts = await fetchval( "SELECT COUNT(*) FROM alert_queue WHERE is_sent = FALSE AND severity IN ('critical','emergency')" ) return DashboardMetrics( total_projects=int(metrics.get("total_projects", 0) or 0), active_projects=int(metrics.get("active_projects", 0) or 0), total_observations=int(metrics.get("total_observations", 0) or 0), observations_today=obs_today or 0, surface_geological_total=int(metrics.get("surface_geological_total", 0) or 0), structural_total=int(metrics.get("structural_total", 0) or 0), geothermal_total=int(metrics.get("geothermal_total", 0) or 0), total_samples=int(metrics.get("total_samples", 0) or 0), samples_pending_lab=pending_lab or 0, high_hazard_count=int(metrics.get("high_hazard_count", 0) or 0), critical_alerts=critical_alerts or 0, landslide_count=int(metrics.get("landslide_count", 0) or 0), max_geothermal_temp=metrics.get("max_geothermal_temp"), avg_geothermal_temp=metrics.get("avg_geothermal_temp"), recent_observations=recent_points, ) @app.get("/api/dashboard/theme-distribution", response_model=List[ThemeDistribution], tags=["Dashboard"]) async def get_theme_distribution(): """Get observation distribution across themes for pie/donut charts.""" rows = await fetch(""" SELECT SUM(surface_geological_count) as surface_geological, SUM(structural_count) as structural, SUM(geothermal_count) as geothermal FROM mv_dashboard_metrics """) if not rows: return [] r = rows[0] if rows else {} total = (r.get("surface_geological", 0) or 0) + (r.get("structural", 0) or 0) + (r.get("geothermal", 0) or 0) if total == 0: total = 1 return [ ThemeDistribution( theme="Surface Geological", count=r.get("surface_geological", 0) or 0, percentage=round((r.get("surface_geological", 0) or 0) / total * 100, 1), color="#0F6E56", ), ThemeDistribution( theme="Structural", count=r.get("structural", 0) or 0, percentage=round((r.get("structural", 0) or 0) / total * 100, 1), color="#BA7517", ), ThemeDistribution( theme="Geothermal", count=r.get("geothermal", 0) or 0, percentage=round((r.get("geothermal", 0) or 0) / total * 100, 1), color="#DC2626", ), ] @app.get("/api/dashboard/hazard-distribution", response_model=List[HazardDistribution], tags=["Dashboard"]) async def get_hazard_distribution(): """Get hazard level distribution for risk charts.""" rows = await fetch(""" SELECT hazard_level, COUNT(*) as count FROM slope_stability GROUP BY hazard_level ORDER BY count DESC """) colors = {"low": "#22C55E", "moderate": "#F59E0B", "high": "#F97316", "very_high": "#EF4444", "extreme": "#7F1D1D"} return [ HazardDistribution(level=r["hazard_level"], count=r["count"], color=colors.get(r["hazard_level"], "#999")) for r in rows ] @app.get("/api/dashboard/geothermal-stats", response_model=List[GeothermalStats], tags=["Dashboard"]) async def get_geothermal_stats(): """Get geothermal manifestation statistics.""" rows = await fetch(""" SELECT manifestation_type, COUNT(*) as count, AVG(surface_temp_c) as avg_temp, MAX(surface_temp_c) as max_temp, AVG(ph_value) as avg_ph FROM geothermal_manifestations GROUP BY manifestation_type ORDER BY count DESC """) return [ GeothermalStats( manifestation_type=r["manifestation_type"], count=r["count"], avg_temp=r["avg_temp"], max_temp=r["max_temp"], avg_ph=r["avg_ph"], ) for r in rows ] # ============================================================ # ALERT ENDPOINTS # ============================================================ @app.get("/api/alerts", tags=["Alerts"]) async def list_alerts( is_sent: Optional[bool] = None, severity: Optional[str] = None, limit: int = Query(50, ge=1, le=200), ): """List alert queue entries.""" conditions = [] params = [] idx = 1 if is_sent is not None: conditions.append(f"is_sent = ${idx}") params.append(is_sent) idx += 1 if severity: conditions.append(f"severity = ${idx}") params.append(severity) idx += 1 where = " AND ".join(conditions) if conditions else "TRUE" query = f""" SELECT alert_id, project_id, alert_type, severity, message, recipients, is_sent, created_at FROM alert_queue WHERE {where} ORDER BY created_at DESC LIMIT ${idx} """ params.append(limit) rows = await fetch(query, *params) return [ { "alert_id": r["alert_id"], "project_id": str(r["project_id"]), "alert_type": r["alert_type"], "severity": r["severity"], "message": r["message"], "recipients": r["recipients"], "is_sent": r["is_sent"], "created_at": r["created_at"].isoformat(), } for r in rows ] @app.post("/api/alerts/process", tags=["Alerts"]) async def process_pending_alerts(background_tasks: BackgroundTasks): """Process all pending alerts and send email notifications.""" rows = await fetch( "SELECT alert_id, project_id, alert_type, severity, message, recipients FROM alert_queue WHERE is_sent = FALSE" ) sent_count = 0 for r in rows: # Send email in background background_tasks.add_task( email_service.send_alert, r["recipients"], f"{r['alert_type']} - {r['severity'].upper()}", f"

{r['message']}

Auto-generated by EIGIS Alert System

", priority=r["severity"], ) await execute( "UPDATE alert_queue SET is_sent = TRUE, sent_at = NOW() WHERE alert_id = $1", r["alert_id"], ) sent_count += 1 return {"processed": sent_count} # ============================================================ # MOBILE SYNC ENDPOINTS # ============================================================ @app.post("/api/sync/observations", tags=["Mobile Sync"], status_code=201) async def sync_observations(data: List[ObservationCreate]): """ Mobile data collection sync endpoint. Handles batched observations from field devices. """ pool = await get_pool() async with pool.acquire() as conn: async with conn.transaction(): ids = [] for obs in data: rid = await conn.fetchval( """INSERT INTO observations (field_trip_id, site_id, observation_type, exposure_type, geom, elevation_m, easting, northing, utm_zone, exposure_length_m, exposure_height_m, groundwater_level_m, weather_condition, excavation_method, accessibility, remarks, logger_id, sync_status) VALUES ($1, $2, $3, $4, ST_SetSRID(ST_MakePoint($5, $6), 4326), $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, 'synced') RETURNING observation_id""", obs.field_trip_id, obs.site_id, obs.observation_type.value, obs.exposure_type, obs.longitude, obs.latitude, obs.elevation_m, obs.easting, obs.northing, obs.utm_zone, obs.exposure_length_m, obs.exposure_height_m, obs.groundwater_level_m, obs.weather_condition, obs.excavation_method, obs.accessibility, obs.remarks, obs.logger_id, ) ids.append(str(rid)) return {"synced": len(ids), "observation_ids": ids} # ============================================================ # SPATIAL QUERY ENDPOINTS # ============================================================ @app.get("/api/spatial/nearby", tags=["Spatial Queries"]) async def find_nearby_observations( latitude: float = Query(..., ge=-90, le=90), longitude: float = Query(..., ge=-180, le=180), radius_m: float = Query(1000, ge=10, le=100000, description="Search radius in meters"), limit: int = Query(20, ge=1, le=100), ): """ Find observations near a point using PostGIS spatial indexing. Uses ST_DWithin for efficient radius search with spatial index. """ rows = await fetch( """SELECT observation_id, site_id, observation_type, status, ST_Y(geom) as lat, ST_X(geom) as lon, elevation_m, ST_Distance(geom::geography, ST_SetSRID(ST_MakePoint($1, $2), 4326)::geography) as distance_m FROM observations WHERE ST_DWithin(geom::geography, ST_SetSRID(ST_MakePoint($1, $2), 4326)::geography, $3) ORDER BY distance_m LIMIT $4""", longitude, latitude, radius_m, limit, ) return [ { "observation_id": str(r["observation_id"]), "site_id": r["site_id"], "observation_type": r["observation_type"], "status": r["status"], "latitude": float(r["lat"]), "longitude": float(r["lon"]), "elevation_m": r["elevation_m"], "distance_m": round(float(r["distance_m"]), 2), } for r in rows ] @app.get("/api/spatial/cluster", tags=["Spatial Queries"]) async def get_spatial_clusters( bbox: str = Query(..., description="Bounding box: xmin,ymin,xmax,ymax"), zoom: int = Query(10, ge=1, le=20, description="Map zoom level for cluster resolution"), ): """ Generate spatial clusters for map visualization at different zoom levels. Uses PostGIS ST_ClusterDBSCAN for density-based clustering. """ parts = [float(p) for p in bbox.split(",")] eps = max(0.001, 0.01 * (20 - zoom) / 20) # Adaptive epsilon rows = await fetch( """SELECT unnest(array_agg(observation_id)) as observation_id, unnest(array_agg(site_id)) as site_id, ST_Centroid(ST_Collect(geom)) as center, COUNT(*) as cluster_size FROM ( SELECT observation_id, site_id, geom, ST_ClusterDBSCAN(geom, $1, 2) OVER () AS cid FROM observations WHERE geom && ST_MakeEnvelope($2, $3, $4, $5, 4326) ) sub WHERE cid IS NOT NULL GROUP BY cid""", eps, parts[0], parts[1], parts[2], parts[3], ) return [ { "observation_id": str(r["observation_id"]), "site_id": r["site_id"], "latitude": float(r["center"].y) if r["center"] else 0, "longitude": float(r["center"].x) if r["center"] else 0, "cluster_size": r["cluster_size"], } for r in rows ]