de32552159
- Add data_sources.yaml for configurable data source URLs - Add data_sources.py to load config with database override support - Add arcgis_landing_points and arcgis_cable_landing_relation collectors - Change visualization API to query arcgis_landing_points - Add /api/v1/datasources/configs/all endpoint - Update Earth to fetch from API instead of static files - Fix scheduler collector ID mappings
245 lines
8.5 KiB
Python
245 lines
8.5 KiB
Python
"""Visualization API - GeoJSON endpoints for 3D Earth display"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
from app.db.session import get_db
|
|
from app.models.collected_data import CollectedData
|
|
from app.services.cable_graph import build_graph_from_data, CableGraph
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
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,
|
|
"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": record.value,
|
|
"length_km": record.value,
|
|
"SHAPE__Length": 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]) -> Dict[str, Any]:
|
|
"""Convert landing point records to GeoJSON FeatureCollection"""
|
|
features = []
|
|
|
|
for record in records:
|
|
try:
|
|
lat = float(record.latitude) if record.latitude else None
|
|
lon = float(record.longitude) if record.longitude else None
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
if lat is None or lon is None:
|
|
continue
|
|
|
|
metadata = record.extra_data or {}
|
|
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
"geometry": {"type": "Point", "coordinates": [lon, lat]},
|
|
"properties": {
|
|
"id": record.id,
|
|
"source_id": record.source_id,
|
|
"name": record.name,
|
|
"country": record.country,
|
|
"city": record.city,
|
|
"is_tbd": metadata.get("is_tbd", False),
|
|
},
|
|
}
|
|
)
|
|
|
|
return {"type": "FeatureCollection", "features": features}
|
|
|
|
|
|
@router.get("/geo/cables")
|
|
async def get_cables_geojson(db: AsyncSession = Depends(get_db)):
|
|
"""获取海底电缆 GeoJSON 数据 (LineString)"""
|
|
try:
|
|
stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
|
|
result = await db.execute(stmt)
|
|
records = result.scalars().all()
|
|
|
|
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)):
|
|
"""获取登陆点 GeoJSON 数据 (Point)"""
|
|
try:
|
|
stmt = select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
|
|
result = await db.execute(stmt)
|
|
records = result.scalars().all()
|
|
|
|
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)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")
|
|
|
|
|
|
@router.get("/geo/all")
|
|
async def get_all_geojson(db: AsyncSession = Depends(get_db)):
|
|
"""获取所有可视化数据 (电缆 + 登陆点)"""
|
|
cables_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
|
|
cables_result = await db.execute(cables_stmt)
|
|
cables_records = cables_result.scalars().all()
|
|
|
|
points_stmt = select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
|
|
points_result = await db.execute(points_stmt)
|
|
points_records = points_result.scalars().all()
|
|
|
|
cables = (
|
|
convert_cable_to_geojson(cables_records)
|
|
if cables_records
|
|
else {"type": "FeatureCollection", "features": []}
|
|
)
|
|
points = (
|
|
convert_landing_point_to_geojson(points_records)
|
|
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,
|
|
},
|
|
}
|
|
|
|
|
|
# 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_stmt = select(CollectedData).where(CollectedData.source == "arcgis_cables")
|
|
cables_result = await db.execute(cables_stmt)
|
|
cables_records = list(cables_result.scalars().all())
|
|
|
|
points_stmt = select(CollectedData).where(CollectedData.source == "arcgis_landing_points")
|
|
points_result = await db.execute(points_stmt)
|
|
points_records = list(points_result.scalars().all())
|
|
|
|
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
|