"""Visualization API - GeoJSON endpoints for 3D Earth display Unified API for all visualization data sources. Returns GeoJSON format compatible with Three.js, CesiumJS, and Unreal Cesium. """ from datetime import UTC, datetime import math import httpx from fastapi import APIRouter, HTTPException, Depends, Query, Response from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from typing import List, Dict, Any, Optional from app.core.collected_data_fields import get_record_field from app.core.satellite_tle import build_tle_lines_from_elements from app.core.time import to_iso8601_utc from app.db.session import get_db from app.models.bgp_anomaly import BGPAnomaly from app.models.bgp_incident import BGPIncident from app.models.collected_data import CollectedData from app.services.bgp_collectors import build_bgp_collector_coverage from app.services.cable_graph import build_graph_from_data, CableGraph, haversine_distance from app.services.collectors.bgp_common import RIPE_RIS_COLLECTOR_COORDS router = APIRouter() TERRAIN_TILE_URL_TEMPLATE = ( "https://s3.amazonaws.com/elevation-tiles-prod/terrarium/{z}/{x}/{y}.png" ) # ============== Converter Functions ============== def convert_cable_to_geojson(records: List[CollectedData]) -> Dict[str, Any]: """Convert cable records to GeoJSON FeatureCollection""" features = [] for record in records: metadata = record.extra_data or {} route_coords = metadata.get("route_coordinates", []) if not route_coords: continue all_lines = [] # Handle both old format (flat array) and new format (array of arrays) if route_coords and isinstance(route_coords[0], list): # New format: array of arrays (MultiLineString structure) if route_coords and isinstance(route_coords[0][0], list): # Array of arrays of arrays - multiple lines for line in route_coords: line_coords = [] for point in line: if len(point) >= 2: try: lon = float(point[0]) lat = float(point[1]) line_coords.append([lon, lat]) except (ValueError, TypeError): continue if len(line_coords) >= 2: all_lines.append(line_coords) else: # Old format: flat array of points - treat as single line line_coords = [] for point in route_coords: if len(point) >= 2: try: lon = float(point[0]) lat = float(point[1]) line_coords.append([lon, lat]) except (ValueError, TypeError): continue if len(line_coords) >= 2: all_lines.append(line_coords) if not all_lines: continue # Use MultiLineString format to preserve cable segments features.append( { "type": "Feature", "geometry": {"type": "MultiLineString", "coordinates": all_lines}, "properties": { "id": record.id, "cable_id": record.name, "source_id": record.source_id, "Name": record.name, "name": record.name, "owner": metadata.get("owners"), "owners": metadata.get("owners"), "rfs": metadata.get("rfs"), "RFS": metadata.get("rfs"), "status": metadata.get("status", "active"), "length": get_record_field(record, "value"), "length_km": get_record_field(record, "value"), "SHAPE__Length": get_record_field(record, "value"), "url": metadata.get("url"), "color": metadata.get("color"), "year": metadata.get("year"), }, } ) return {"type": "FeatureCollection", "features": features} def convert_landing_point_to_geojson(records: List[CollectedData], city_to_cable_ids_map: Dict[int, List[int]] = None, cable_id_to_name_map: Dict[int, str] = None) -> Dict[str, Any]: features = [] for record in records: try: latitude = get_record_field(record, "latitude") longitude = get_record_field(record, "longitude") lat = float(latitude) if latitude else None lon = float(longitude) if longitude else None except (ValueError, TypeError): continue if lat is None or lon is None: continue metadata = record.extra_data or {} city_id = metadata.get("city_id") props = { "id": record.id, "source_id": record.source_id, "name": record.name, "country": get_record_field(record, "country"), "city": get_record_field(record, "city"), "is_tbd": metadata.get("is_tbd", False), } cable_names = [] if city_to_cable_ids_map and city_id in city_to_cable_ids_map: for cable_id in city_to_cable_ids_map[city_id]: if cable_id_to_name_map and cable_id in cable_id_to_name_map: cable_names.append(cable_id_to_name_map[cable_id]) if cable_names: props["cable_names"] = cable_names features.append( { "type": "Feature", "geometry": {"type": "Point", "coordinates": [lon, lat]}, "properties": props, } ) return {"type": "FeatureCollection", "features": features} def convert_satellite_to_geojson(records: List[CollectedData]) -> Dict[str, Any]: """Convert satellite TLE records to GeoJSON""" features = [] for record in records: metadata = record.extra_data or {} norad_id = metadata.get("norad_cat_id") if not norad_id: continue tle_line1 = metadata.get("tle_line1") tle_line2 = metadata.get("tle_line2") if not tle_line1 or not tle_line2: tle_line1, tle_line2 = build_tle_lines_from_elements( norad_cat_id=norad_id, epoch=metadata.get("epoch"), inclination=metadata.get("inclination"), raan=metadata.get("raan"), eccentricity=metadata.get("eccentricity"), arg_of_perigee=metadata.get("arg_of_perigee"), mean_anomaly=metadata.get("mean_anomaly"), mean_motion=metadata.get("mean_motion"), ) features.append( { "type": "Feature", "id": norad_id, "geometry": {"type": "Point", "coordinates": [0, 0, 0]}, "properties": { "id": record.id, "norad_cat_id": norad_id, "name": record.name, "international_designator": metadata.get("international_designator"), "epoch": metadata.get("epoch"), "inclination": metadata.get("inclination"), "raan": metadata.get("raan"), "eccentricity": metadata.get("eccentricity"), "arg_of_perigee": metadata.get("arg_of_perigee"), "mean_anomaly": metadata.get("mean_anomaly"), "mean_motion": metadata.get("mean_motion"), "bstar": metadata.get("bstar"), "classification_type": metadata.get("classification_type"), "tle_line1": tle_line1, "tle_line2": tle_line2, "data_type": "satellite_tle", }, } ) return {"type": "FeatureCollection", "features": features} def _current_collected_data_stmt(source: str): return ( select(CollectedData) .where(CollectedData.source == source) .where(CollectedData.is_current.is_(True)) .order_by(CollectedData.id.desc()) ) async def _load_current_collected_data( db: AsyncSession, source: str, *, exclude_unknown_name: bool = False, limit: Optional[int] = None, ) -> List[CollectedData]: stmt = _current_collected_data_stmt(source) if exclude_unknown_name: stmt = stmt.where(CollectedData.name != "Unknown") if limit is not None: stmt = stmt.limit(limit) result = await db.execute(stmt) return list(result.scalars().all()) async def _load_current_collected_data_by_sources( db: AsyncSession, sources: List[str], ) -> Dict[str, List[CollectedData]]: if not sources: return {} stmt = ( select(CollectedData) .where(CollectedData.source.in_(sources)) .where(CollectedData.is_current.is_(True)) .order_by(CollectedData.source.asc(), CollectedData.id.desc()) ) result = await db.execute(stmt) grouped_records: Dict[str, List[CollectedData]] = {source: [] for source in sources} for record in result.scalars().all(): grouped_records.setdefault(record.source, []).append(record) return grouped_records def _build_landing_point_cable_maps( relation_records: List[CollectedData], cable_records: List[CollectedData], ) -> tuple[Dict[int, List[int]], Dict[int, str]]: city_to_cable_ids_map: Dict[int, List[int]] = {} for relation_record in relation_records: if not relation_record.extra_data: continue city_id = relation_record.extra_data.get("city_id") cable_id = relation_record.extra_data.get("cable_id") if city_id is None or cable_id is None: continue city_to_cable_ids_map.setdefault(city_id, []) if cable_id not in city_to_cable_ids_map[city_id]: city_to_cable_ids_map[city_id].append(cable_id) cable_id_to_name_map: Dict[int, str] = {} for cable_record in cable_records: if not cable_record.extra_data: continue cable_id = cable_record.extra_data.get("cable_id") cable_name = cable_record.name if cable_id and cable_name: cable_id_to_name_map[cable_id] = cable_name return city_to_cable_ids_map, cable_id_to_name_map def _filter_known_records(records: List[CollectedData]) -> List[CollectedData]: return [record for record in records if record.name != "Unknown"] def convert_supercomputer_to_geojson(records: List[CollectedData]) -> Dict[str, Any]: """Convert TOP500 supercomputer records to GeoJSON""" features = [] for record in records: try: latitude = get_record_field(record, "latitude") longitude = get_record_field(record, "longitude") lat = float(latitude) if latitude and latitude != "0.0" else None lon = ( float(longitude) if longitude and longitude != "0.0" else None ) except (ValueError, TypeError): lat, lon = None, None metadata = record.extra_data or {} features.append( { "type": "Feature", "id": record.id, "geometry": {"type": "Point", "coordinates": [lon or 0, lat or 0]}, "properties": { "id": record.id, "name": record.name, "rank": metadata.get("rank"), "r_max": get_record_field(record, "rmax"), "r_peak": get_record_field(record, "rpeak"), "cores": get_record_field(record, "cores"), "power": get_record_field(record, "power"), "country": get_record_field(record, "country"), "city": get_record_field(record, "city"), "data_type": "supercomputer", }, } ) return {"type": "FeatureCollection", "features": features} def convert_gpu_cluster_to_geojson(records: List[CollectedData]) -> Dict[str, Any]: """Convert GPU cluster records to GeoJSON""" features = [] for record in records: try: latitude = get_record_field(record, "latitude") longitude = get_record_field(record, "longitude") lat = float(latitude) if latitude else None lon = float(longitude) if longitude else None except (ValueError, TypeError): lat, lon = None, None metadata = record.extra_data or {} features.append( { "type": "Feature", "id": record.id, "geometry": {"type": "Point", "coordinates": [lon or 0, lat or 0]}, "properties": { "id": record.id, "name": record.name, "country": get_record_field(record, "country"), "city": get_record_field(record, "city"), "metadata": metadata, "data_type": "gpu_cluster", }, } ) return {"type": "FeatureCollection", "features": features} def convert_bgp_anomalies_to_geojson( records: List[BGPAnomaly], geography_hints: Optional[Dict[str, Dict[str, Any]]] = None, ) -> Dict[str, Any]: features = [] geography_hints = geography_hints or {} for record in records: evidence = record.evidence or {} hint = geography_hints.get(str(record.entity_key or record.id), {}) collectors = evidence.get("collectors") or record.peer_scope or [] if not collectors: nested = evidence.get("events") or [] collectors = [ str((item or {}).get("collector") or "").strip() for item in nested if (item or {}).get("collector") ] collectors = [collector for collector in collectors if collector] if not collectors: collectors = [] as_path = [] if isinstance(evidence.get("as_path"), list): as_path = evidence.get("as_path") or [] if not as_path: nested = evidence.get("events") or [] for item in nested: candidate_path = (item or {}).get("as_path") if isinstance(candidate_path, list) and candidate_path: as_path = candidate_path break impacted_regions = [] seen_regions = set() for collector_name in collectors: collector_location = RIPE_RIS_COLLECTOR_COORDS.get(str(collector_name)) if not collector_location: continue region_key = ( collector_location.get("country"), collector_location.get("city"), ) if region_key in seen_regions: continue seen_regions.add(region_key) impacted_regions.append( { "collector": collector_name, "country": collector_location.get("country"), "city": collector_location.get("city"), "latitude": collector_location.get("latitude"), "longitude": collector_location.get("longitude"), } ) geography_regions = _normalize_geo_regions(hint.get("regions") or []) geography_mode = hint.get("geography_mode") or "collector_centroid" collector = collectors[0] if collectors else None location = geography_regions[0] if geography_regions else None if location is None and collector: location = RIPE_RIS_COLLECTOR_COORDS.get(str(collector)) if location is None: nested = evidence.get("events") or [] for item in nested: collector_name = (item or {}).get("collector") if collector_name and collector_name in RIPE_RIS_COLLECTOR_COORDS: location = RIPE_RIS_COLLECTOR_COORDS[collector_name] collector = collector_name geography_mode = "collector_centroid" break if location is None: continue features.append( { "type": "Feature", "geometry": { "type": "Point", "coordinates": [location["longitude"], location["latitude"]], }, "properties": { "id": record.id, "collector": collector, "city": location.get("city"), "country": location.get("country"), "source": record.source, "anomaly_type": record.anomaly_type, "severity": record.severity, "status": record.status, "prefix": record.prefix, "origin_asn": record.origin_asn, "new_origin_asn": record.new_origin_asn, "collectors": collectors, "collector_count": len(collectors) or 1, "as_path": as_path, "impacted_regions": impacted_regions, "geography_mode": geography_mode, "confidence": record.confidence, "summary": record.summary, "created_at": to_iso8601_utc(record.created_at), }, } ) return {"type": "FeatureCollection", "features": features} async def build_anomaly_geography_hints( db: AsyncSession, records: List[BGPAnomaly], ) -> Dict[str, Dict[str, Any]]: hints: Dict[str, Dict[str, Any]] = {} for record in records: hint = _extract_evidence_geography_hint(record.evidence or {}) if hint: hints[str(record.entity_key or record.id)] = hint return hints def convert_bgp_collectors_to_geojson( coverage_by_collector: Dict[str, Dict[str, Any]] | None = None, ) -> Dict[str, Any]: features = [] coverage_by_collector = coverage_by_collector or {} for collector, location in sorted(RIPE_RIS_COLLECTOR_COORDS.items()): coverage = coverage_by_collector.get(collector, {}) features.append( { "type": "Feature", "geometry": { "type": "Point", "coordinates": [location["longitude"], location["latitude"]], }, "properties": { "collector": collector, "city": coverage.get("city") or location.get("city"), "country": coverage.get("country") or location.get("country"), "status": "online", "observation_count": coverage.get("observation_count", 0), "prefix_count": coverage.get("prefix_count", 0), "origin_asn_count": coverage.get("origin_asn_count", 0), "peer_asn_count": coverage.get("peer_asn_count", 0), "recent_15m_observation_count": coverage.get("recent_15m_observation_count", 0), "recent_24h_observation_count": coverage.get("recent_24h_observation_count", 0), "recent_7d_observation_count": coverage.get("recent_7d_observation_count", 0), "recent_15m_prefix_count": coverage.get("recent_15m_prefix_count", 0), "recent_24h_prefix_count": coverage.get("recent_24h_prefix_count", 0), "recent_7d_prefix_count": coverage.get("recent_7d_prefix_count", 0), "top_event_types": coverage.get("top_event_types", []), "latest_observed_at": coverage.get("latest_observed_at"), "latest_event_type": coverage.get("latest_event_type"), "baseline_scope": coverage.get( "baseline_scope", { "countries": [location.get("country")] if location.get("country") else [], "cities": [location.get("city")] if location.get("city") else [], }, ), }, } ) return {"type": "FeatureCollection", "features": features} def _incident_estimated_center(valid_regions: List[Dict[str, Any]]) -> Dict[str, float]: x = 0.0 y = 0.0 z = 0.0 for region in valid_regions: lat_rad = math.radians(float(region["latitude"])) lon_rad = math.radians(float(region["longitude"])) x += math.cos(lat_rad) * math.cos(lon_rad) y += math.cos(lat_rad) * math.sin(lon_rad) z += math.sin(lat_rad) total = float(len(valid_regions)) if total <= 0: return {"latitude": 0.0, "longitude": 0.0} x /= total y /= total z /= total hyp = math.sqrt((x * x) + (y * y)) if hyp == 0: return {"latitude": 0.0, "longitude": 0.0} return { "latitude": math.degrees(math.atan2(z, hyp)), "longitude": math.degrees(math.atan2(y, x)), } def _incident_estimated_radius_km(center: Dict[str, float], valid_regions: List[Dict[str, Any]]) -> float: center_coords = (float(center["longitude"]), float(center["latitude"])) distances = [ haversine_distance( center_coords, (float(region["longitude"]), float(region["latitude"])), ) for region in valid_regions ] return round(max(distances) if distances else 0.0, 1) def _normalize_geo_regions(regions: List[Dict[str, Any]]) -> List[Dict[str, Any]]: normalized: list[dict[str, Any]] = [] seen: set[tuple[Any, ...]] = set() for region in regions: if not isinstance(region, dict): continue latitude = region.get("latitude") longitude = region.get("longitude") if not isinstance(latitude, (int, float)) or not isinstance(longitude, (int, float)): continue item = { "collector": region.get("collector"), "country": region.get("country"), "city": region.get("city"), "latitude": float(latitude), "longitude": float(longitude), } key = ( item["collector"], item["country"], item["city"], item["latitude"], item["longitude"], ) if key in seen: continue seen.add(key) normalized.append(item) return normalized def _extract_evidence_geography_hint(evidence: Dict[str, Any]) -> Dict[str, Any] | None: prefix_geo_regions = [] prefix_regions = [] asn_regions = [] evidence_prefix_geography = evidence.get("prefix_geography") or {} prefix_geo_regions.extend( _normalize_geo_regions(evidence_prefix_geography.get("regions") or []) ) prefix_scope = evidence.get("prefix_scope") or {} prefix_regions.extend(_normalize_geo_regions(prefix_scope.get("regions") or [])) for profile_key in ("origin_asn_profile", "new_origin_asn_profile"): profile = evidence.get(profile_key) or {} latitude = profile.get("latitude") longitude = profile.get("longitude") if isinstance(latitude, (int, float)) and isinstance(longitude, (int, float)): asn_regions.append( { "country": profile.get("country"), "city": profile.get("city"), "latitude": float(latitude), "longitude": float(longitude), } ) prefix_geo_regions = _normalize_geo_regions(prefix_geo_regions) prefix_regions = _normalize_geo_regions(prefix_regions) asn_regions = _normalize_geo_regions(asn_regions) if prefix_geo_regions: return {"regions": prefix_geo_regions, "geography_mode": "prefix_geography"} if prefix_regions: return {"regions": prefix_regions, "geography_mode": "prefix_scope"} if asn_regions: return {"regions": asn_regions, "geography_mode": "asn_region"} return None async def build_incident_geography_hints( db: AsyncSession, records: List[BGPIncident], ) -> Dict[str, Dict[str, Any]]: evidence_refs = sorted( { str(ref) for record in records for ref in (record.evidence_refs or []) if ref } ) anomalies = [] if evidence_refs: result = await db.execute( select(BGPAnomaly).where(BGPAnomaly.entity_key.in_(evidence_refs)) ) anomalies = result.scalars().all() anomaly_by_key = { str(anomaly.entity_key): anomaly for anomaly in anomalies if anomaly.entity_key } hints: Dict[str, Dict[str, Any]] = {} for record in records: merged_hint: Dict[str, Any] | None = None priority = {"prefix_geography": 3, "prefix_scope": 2, "asn_region": 1} for ref in record.evidence_refs or []: anomaly = anomaly_by_key.get(str(ref)) if anomaly is None: continue hint = _extract_evidence_geography_hint(anomaly.evidence or {}) if hint is None: continue if merged_hint is None: merged_hint = { "regions": list(hint["regions"]), "geography_mode": hint["geography_mode"], } continue if priority[hint["geography_mode"]] > priority[merged_hint["geography_mode"]]: merged_hint = { "regions": list(hint["regions"]), "geography_mode": hint["geography_mode"], } elif priority[hint["geography_mode"]] == priority[merged_hint["geography_mode"]]: merged_hint["regions"].extend(hint["regions"]) if merged_hint: merged_hint["regions"] = _normalize_geo_regions(merged_hint["regions"]) hints[record.incident_key] = merged_hint return hints def convert_bgp_incidents_to_geojson( records: List[BGPIncident], geography_hints: Optional[Dict[str, Dict[str, Any]]] = None, ) -> Dict[str, Any]: features = [] for record in records: hint = (geography_hints or {}).get(record.incident_key, {}) regions = hint.get("regions") or (record.affected_regions or []) if not regions: continue valid_regions = _normalize_geo_regions(regions) if not valid_regions: continue estimated_center = _incident_estimated_center(valid_regions) estimated_radius_km = _incident_estimated_radius_km(estimated_center, valid_regions) features.append( { "type": "Feature", "geometry": { "type": "Point", "coordinates": [ estimated_center["longitude"], estimated_center["latitude"], ], }, "properties": { "id": record.id, "incident_key": record.incident_key, "incident_type": record.incident_type, "title": record.title, "summary": record.summary, "severity": record.severity, "status": record.status, "confidence": record.confidence, "affected_prefixes": record.affected_prefixes or [], "affected_asns": record.affected_asns or [], "affected_collectors": record.affected_collectors or [], "affected_regions": valid_regions, "estimated_center": estimated_center, "estimated_radius_km": estimated_radius_km, "geography_mode": hint.get("geography_mode") or "collector_centroid", "related_cables": record.related_cables or [], "related_ixps": record.related_ixps or [], "created_at": to_iso8601_utc(record.created_at), "started_at": to_iso8601_utc(record.started_at), }, } ) return {"type": "FeatureCollection", "features": features} # ============== API Endpoints ============== @router.get("/geo/cables") async def get_cables_geojson(db: AsyncSession = Depends(get_db)): """获取海底电缆 GeoJSON 数据 (LineString)""" try: records = await _load_current_collected_data(db, "arcgis_cables") if not records: raise HTTPException( status_code=404, detail="No cable data found. Please run the arcgis_cables collector first.", ) return convert_cable_to_geojson(records) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") @router.get("/geo/landing-points") async def get_landing_points_geojson(db: AsyncSession = Depends(get_db)): try: records_by_source = await _load_current_collected_data_by_sources( db, [ "arcgis_landing_points", "arcgis_cable_landing_relation", "arcgis_cables", ], ) records = records_by_source.get("arcgis_landing_points", []) relation_records = records_by_source.get( "arcgis_cable_landing_relation", [], ) cable_records = records_by_source.get("arcgis_cables", []) city_to_cable_ids_map, cable_id_to_name_map = _build_landing_point_cable_maps( relation_records, cable_records, ) if not records: raise HTTPException( status_code=404, detail="No landing point data found. Please run the arcgis_landing_points collector first.", ) return convert_landing_point_to_geojson(records, city_to_cable_ids_map, cable_id_to_name_map) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") @router.get("/terrain/terrarium/{z}/{x}/{y}.png") async def get_terrarium_tile(z: int, x: int, y: int): """Proxy Terrarium elevation tiles through the backend to avoid browser CORS issues.""" if z < 0 or x < 0 or y < 0: raise HTTPException(status_code=400, detail="Invalid terrain tile coordinates") url = TERRAIN_TILE_URL_TEMPLATE.format(z=z, x=x, y=y) try: async with httpx.AsyncClient( timeout=20.0, follow_redirects=True, ) as client: upstream = await client.get(url) upstream.raise_for_status() except httpx.HTTPStatusError as exc: raise HTTPException( status_code=exc.response.status_code, detail=f"Terrain tile upstream error: {exc.response.status_code}", ) from exc except httpx.HTTPError as exc: raise HTTPException( status_code=502, detail=f"Terrain tile fetch failed: {exc}", ) from exc cache_control = upstream.headers.get("cache-control") or "public, max-age=86400" etag = upstream.headers.get("etag") last_modified = upstream.headers.get("last-modified") headers = { "Cache-Control": cache_control, } if etag: headers["ETag"] = etag if last_modified: headers["Last-Modified"] = last_modified return Response( content=upstream.content, media_type=upstream.headers.get("content-type", "image/png"), headers=headers, ) @router.get("/geo/all") async def get_all_geojson(db: AsyncSession = Depends(get_db)): records_by_source = await _load_current_collected_data_by_sources( db, [ "arcgis_cables", "arcgis_landing_points", "arcgis_cable_landing_relation", ], ) cables_records = records_by_source.get("arcgis_cables", []) points_records = records_by_source.get("arcgis_landing_points", []) relation_records = records_by_source.get("arcgis_cable_landing_relation", []) city_to_cable_ids_map, cable_id_to_name_map = _build_landing_point_cable_maps( relation_records, cables_records, ) cables = ( convert_cable_to_geojson(cables_records) if cables_records else {"type": "FeatureCollection", "features": []} ) points = ( convert_landing_point_to_geojson(points_records, city_to_cable_ids_map, cable_id_to_name_map) if points_records else {"type": "FeatureCollection", "features": []} ) return { "cables": cables, "landing_points": points, "stats": { "cable_count": len(cables.get("features", [])) if cables else 0, "landing_point_count": len(points.get("features", [])) if points else 0, }, } @router.get("/geo/satellites") async def get_satellites_geojson( limit: Optional[int] = Query( None, ge=1, description="Maximum number of satellites to return. Omit for no limit.", ), db: AsyncSession = Depends(get_db), ): """获取卫星 TLE GeoJSON 数据""" records = await _load_current_collected_data( db, "celestrak_tle", exclude_unknown_name=True, limit=limit, ) if not records: return {"type": "FeatureCollection", "features": [], "count": 0} geojson = convert_satellite_to_geojson(list(records)) return { **geojson, "count": len(geojson.get("features", [])), } @router.get("/geo/supercomputers") async def get_supercomputers_geojson( limit: int = 500, db: AsyncSession = Depends(get_db), ): """获取 TOP500 超算中心 GeoJSON 数据""" records = await _load_current_collected_data( db, "top500", exclude_unknown_name=True, limit=limit, ) if not records: return {"type": "FeatureCollection", "features": [], "count": 0} geojson = convert_supercomputer_to_geojson(list(records)) return { **geojson, "count": len(geojson.get("features", [])), } @router.get("/geo/gpu-clusters") async def get_gpu_clusters_geojson( limit: int = 100, db: AsyncSession = Depends(get_db), ): """获取 GPU 集群 GeoJSON 数据""" records = await _load_current_collected_data( db, "epoch_ai_gpu", exclude_unknown_name=True, limit=limit, ) if not records: return {"type": "FeatureCollection", "features": [], "count": 0} geojson = convert_gpu_cluster_to_geojson(list(records)) return { **geojson, "count": len(geojson.get("features", [])), } @router.get("/geo/bgp-anomalies") async def get_bgp_anomalies_geojson( severity: Optional[str] = Query(None), status: Optional[str] = Query("active"), limit: int = Query(200, ge=1, le=1000), db: AsyncSession = Depends(get_db), ): stmt = select(BGPAnomaly).order_by(BGPAnomaly.created_at.desc()).limit(limit) if severity: stmt = stmt.where(BGPAnomaly.severity == severity) if status: stmt = stmt.where(BGPAnomaly.status == status) result = await db.execute(stmt) records = list(result.scalars().all()) geography_hints = await build_anomaly_geography_hints(db, records) geojson = convert_bgp_anomalies_to_geojson(records, geography_hints) return {**geojson, "count": len(geojson.get("features", []))} @router.get("/geo/bgp-incidents") async def get_bgp_incidents_geojson( severity: Optional[str] = Query(None), status: Optional[str] = Query("active"), limit: int = Query(100, ge=1, le=500), db: AsyncSession = Depends(get_db), ): stmt = select(BGPIncident).order_by(BGPIncident.created_at.desc()).limit(limit) if severity: stmt = stmt.where(BGPIncident.severity == severity) if status: stmt = stmt.where(BGPIncident.status == status) result = await db.execute(stmt) records = list(result.scalars().all()) geography_hints = await build_incident_geography_hints(db, records) geojson = convert_bgp_incidents_to_geojson(records, geography_hints) return {**geojson, "count": len(geojson.get("features", []))} @router.get("/geo/bgp-collectors") async def get_bgp_collectors_geojson(db: AsyncSession = Depends(get_db)): coverage = await build_bgp_collector_coverage( db, source_filter=("ris_live_bgp", "bgpstream_bgp"), ) coverage_by_collector = { item["collector"]: item for item in coverage if item.get("collector") } geojson = convert_bgp_collectors_to_geojson(coverage_by_collector) return {**geojson, "count": len(geojson.get("features", []))} @router.get("/all") async def get_all_visualization_data(db: AsyncSession = Depends(get_db)): """获取所有可视化数据的统一端点 Returns GeoJSON FeatureCollections for all data types: - satellites: 卫星 TLE 数据 - cables: 海底电缆 - landing_points: 登陆点 - supercomputers: TOP500 超算 - gpu_clusters: GPU 集群 """ records_by_source = await _load_current_collected_data_by_sources( db, [ "arcgis_cables", "arcgis_landing_points", "celestrak_tle", "top500", "epoch_ai_gpu", ], ) cables_records = records_by_source.get("arcgis_cables", []) points_records = records_by_source.get("arcgis_landing_points", []) satellites_records = _filter_known_records( records_by_source.get("celestrak_tle", []), ) supercomputers_records = _filter_known_records( records_by_source.get("top500", []), ) gpu_records = _filter_known_records( records_by_source.get("epoch_ai_gpu", []), ) cables = ( convert_cable_to_geojson(cables_records) if cables_records else {"type": "FeatureCollection", "features": []} ) landing_points = ( convert_landing_point_to_geojson(points_records) if points_records else {"type": "FeatureCollection", "features": []} ) satellites = ( convert_satellite_to_geojson(satellites_records) if satellites_records else {"type": "FeatureCollection", "features": []} ) supercomputers = ( convert_supercomputer_to_geojson(supercomputers_records) if supercomputers_records else {"type": "FeatureCollection", "features": []} ) gpu_clusters = ( convert_gpu_cluster_to_geojson(gpu_records) if gpu_records else {"type": "FeatureCollection", "features": []} ) return { "generated_at": to_iso8601_utc(datetime.now(UTC)), "version": "1.0", "data": { "satellites": satellites, "cables": cables, "landing_points": landing_points, "supercomputers": supercomputers, "gpu_clusters": gpu_clusters, }, "stats": { "total_features": ( len(satellites.get("features", [])) + len(cables.get("features", [])) + len(landing_points.get("features", [])) + len(supercomputers.get("features", [])) + len(gpu_clusters.get("features", [])) ), "satellites": len(satellites.get("features", [])), "cables": len(cables.get("features", [])), "landing_points": len(landing_points.get("features", [])), "supercomputers": len(supercomputers.get("features", [])), "gpu_clusters": len(gpu_clusters.get("features", [])), }, } # Cache for cable graph _cable_graph: Optional[CableGraph] = None async def get_cable_graph(db: AsyncSession) -> CableGraph: """Get or build cable graph (cached)""" global _cable_graph if _cable_graph is None: cables_records = await _load_current_collected_data(db, "arcgis_cables") points_records = await _load_current_collected_data(db, "arcgis_landing_points") cables_data = convert_cable_to_geojson(cables_records) points_data = convert_landing_point_to_geojson(points_records) _cable_graph = build_graph_from_data(cables_data, points_data) return _cable_graph @router.post("/geo/path") async def find_path( start: List[float], end: List[float], db: AsyncSession = Depends(get_db), ): """Find shortest path between two coordinates via cable network""" if not start or len(start) != 2: raise HTTPException(status_code=400, detail="Start must be [lon, lat]") if not end or len(end) != 2: raise HTTPException(status_code=400, detail="End must be [lon, lat]") graph = await get_cable_graph(db) result = graph.find_shortest_path(start, end) if not result: raise HTTPException( status_code=404, detail="No path found between these points. They may be too far from any landing point.", ) return result