563 lines
19 KiB
Python
563 lines
19 KiB
Python
"""UE scene state built from visualization aggregate output."""
|
|
|
|
import asyncio
|
|
import json
|
|
import zlib
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING, Any, Dict, List
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.time import to_iso8601_utc
|
|
from app.models.bgp_anomaly import BGPAnomaly
|
|
from app.models.bgp_incident import BGPIncident
|
|
|
|
if TYPE_CHECKING:
|
|
from app.models.collected_data import CollectedData
|
|
|
|
DEFAULT_LAYER_ORDER = (
|
|
"satellites",
|
|
"supercomputers",
|
|
"gpu_clusters",
|
|
"submarine_cables",
|
|
"landing_points",
|
|
"bgp_anomalies",
|
|
"bgp_incidents",
|
|
"bgp_collectors",
|
|
"alerts",
|
|
)
|
|
|
|
FULL_RESYNC_REASONS = {"initial_connect", "manual_resync", "profile_changed"}
|
|
UE_SCENE_LAYER_CHUNK_ITEM_LIMITS = {
|
|
"satellites": 500,
|
|
"supercomputers": 200,
|
|
"gpu_clusters": 100,
|
|
"submarine_cables": 25,
|
|
"landing_points": 250,
|
|
"bgp_anomalies": 100,
|
|
"bgp_incidents": 100,
|
|
"bgp_collectors": 100,
|
|
"alerts": 100,
|
|
}
|
|
|
|
|
|
def _default_display_profile() -> Dict[str, Any]:
|
|
return {
|
|
"profile_id": "polarized-wall-a",
|
|
"stereo_mode": "polarized",
|
|
"screen_width_m": 3.0,
|
|
"screen_height_m": 2.0,
|
|
"target_refresh_hz": 120,
|
|
}
|
|
|
|
|
|
def _default_camera_state() -> Dict[str, Any]:
|
|
return {
|
|
"mode": "auto_cruise",
|
|
"path_id": "global_overview",
|
|
"fov": 42.0,
|
|
}
|
|
|
|
|
|
def _stable_json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def _entity_type_for_layer(layer_name: str) -> str:
|
|
return {
|
|
"satellites": "satellite",
|
|
"supercomputers": "supercomputer",
|
|
"gpu_clusters": "gpu_cluster",
|
|
"submarine_cables": "submarine_cable",
|
|
"landing_points": "landing_point",
|
|
"bgp_anomalies": "bgp_anomaly",
|
|
"bgp_incidents": "bgp_incident",
|
|
"bgp_collectors": "bgp_collector",
|
|
"alerts": "alert",
|
|
}[layer_name]
|
|
|
|
|
|
def _default_visual_for_layer(layer_name: str) -> Dict[str, Any]:
|
|
visuals = {
|
|
"satellites": {"style": "satellite_marker", "size": 0.7, "color": "#9BDBFF"},
|
|
"supercomputers": {"style": "supercomputer_marker", "size": 1.2, "color": "#FF6B6B"},
|
|
"gpu_clusters": {"style": "pulse_marker", "size": 1.0, "color": "#FF8C42"},
|
|
"submarine_cables": {"style": "cable_arc", "width": 2.0, "color": "#4ECDC4"},
|
|
"landing_points": {"style": "landing_point_marker", "size": 0.8, "color": "#45B7D1"},
|
|
"bgp_anomalies": {"style": "anomaly_marker", "size": 1.0, "color": "#FFB703"},
|
|
"bgp_incidents": {"style": "incident_marker", "size": 1.2, "color": "#E63946"},
|
|
"bgp_collectors": {"style": "collector_marker", "size": 0.9, "color": "#7B9ACC"},
|
|
"alerts": {"style": "alert_marker", "size": 1.0, "color": "#FFD166"},
|
|
}
|
|
return visuals[layer_name]
|
|
|
|
|
|
def _empty_scene_layers() -> Dict[str, Dict[str, Any]]:
|
|
return {
|
|
layer_name: {"revision": 0, "items": {}}
|
|
for layer_name in DEFAULT_LAYER_ORDER
|
|
}
|
|
|
|
|
|
def _empty_changes() -> Dict[str, Dict[str, List[Any]]]:
|
|
return {
|
|
layer_name: {"added": [], "updated": [], "removed": []}
|
|
for layer_name in DEFAULT_LAYER_ORDER
|
|
}
|
|
|
|
|
|
def _point_geo(coordinates: List[Any]) -> Dict[str, Any]:
|
|
lng = coordinates[0] if len(coordinates) > 0 else None
|
|
lat = coordinates[1] if len(coordinates) > 1 else None
|
|
alt = coordinates[2] if len(coordinates) > 2 else 0.0
|
|
return {"lat": lat, "lng": lng, "alt": alt}
|
|
|
|
|
|
def _path_geo(geometry_type: str, coordinates: Any) -> Dict[str, Any]:
|
|
if geometry_type == "LineString":
|
|
return {
|
|
"path": [
|
|
{"lat": point[1], "lng": point[0], "alt": point[2] if len(point) > 2 else 0.0}
|
|
for point in coordinates
|
|
if isinstance(point, list) and len(point) >= 2
|
|
]
|
|
}
|
|
|
|
if geometry_type == "MultiLineString":
|
|
return {
|
|
"segments": [
|
|
[
|
|
{"lat": point[1], "lng": point[0], "alt": point[2] if len(point) > 2 else 0.0}
|
|
for point in line
|
|
if isinstance(point, list) and len(point) >= 2
|
|
]
|
|
for line in coordinates
|
|
if isinstance(line, list)
|
|
]
|
|
}
|
|
|
|
return {}
|
|
|
|
|
|
def _item_identifier(layer_name: str, feature: Dict[str, Any], index: int) -> str:
|
|
properties = feature.get("properties") or {}
|
|
feature_id = feature.get("id") or properties.get("id") or properties.get("source_id")
|
|
return f"{layer_name}:{feature_id or index}"
|
|
|
|
|
|
def _item_revision(item: Dict[str, Any]) -> int:
|
|
return zlib.crc32(_stable_json(item).encode("utf-8")) & 0xFFFFFFFF
|
|
|
|
|
|
def _layer_revision(items: Dict[str, Dict[str, Any]]) -> int:
|
|
if not items:
|
|
return 0
|
|
joined = "|".join(
|
|
f"{item_id}:{items[item_id]['revision']}"
|
|
for item_id in sorted(items)
|
|
)
|
|
return zlib.crc32(joined.encode("utf-8")) & 0xFFFFFFFF
|
|
|
|
|
|
def _serialize_feature(layer_name: str, feature: Dict[str, Any], index: int) -> Dict[str, Any]:
|
|
properties = dict(feature.get("properties") or {})
|
|
geometry = feature.get("geometry") or {}
|
|
geometry_type = geometry.get("type", "")
|
|
coordinates = geometry.get("coordinates") or []
|
|
item_id = _item_identifier(layer_name, feature, index)
|
|
|
|
geo: Dict[str, Any] = {}
|
|
if geometry_type == "Point":
|
|
geo = _point_geo(coordinates)
|
|
elif geometry_type in {"LineString", "MultiLineString"}:
|
|
geo = _path_geo(geometry_type, coordinates)
|
|
|
|
title = (
|
|
properties.get("name")
|
|
or properties.get("Name")
|
|
or properties.get("title")
|
|
or item_id
|
|
)
|
|
subtitle_parts = [
|
|
properties.get("city"),
|
|
properties.get("country"),
|
|
properties.get("region"),
|
|
]
|
|
item = {
|
|
"id": item_id,
|
|
"entity_type": _entity_type_for_layer(layer_name),
|
|
"geo": geo,
|
|
"visual": {
|
|
**_default_visual_for_layer(layer_name),
|
|
**({"color": properties["color"]} if properties.get("color") else {}),
|
|
},
|
|
"metrics": properties,
|
|
"labels": {
|
|
"title": title,
|
|
"subtitle": ", ".join([part for part in subtitle_parts if part]),
|
|
},
|
|
"status": {
|
|
"health": properties.get("status", "normal"),
|
|
"alert_level": properties.get("severity", "none"),
|
|
},
|
|
}
|
|
item["revision"] = _item_revision(item)
|
|
return item
|
|
|
|
|
|
async def build_visualization_scene_state(db: AsyncSession) -> Dict[str, Any]:
|
|
from app.api.v1.visualization import (
|
|
_build_landing_point_cable_maps,
|
|
_filter_known_records,
|
|
_load_current_collected_data_by_sources,
|
|
build_anomaly_geography_hints,
|
|
build_incident_geography_hints,
|
|
convert_bgp_anomalies_to_geojson,
|
|
convert_bgp_collectors_to_geojson,
|
|
convert_bgp_incidents_to_geojson,
|
|
convert_cable_to_geojson,
|
|
convert_gpu_cluster_to_geojson,
|
|
convert_landing_point_to_geojson,
|
|
convert_satellite_to_geojson,
|
|
convert_supercomputer_to_geojson,
|
|
)
|
|
from app.services.bgp_collectors import build_bgp_collector_coverage
|
|
|
|
records_by_source = await _load_current_collected_data_by_sources(
|
|
db,
|
|
[
|
|
"arcgis_cables",
|
|
"arcgis_landing_points",
|
|
"arcgis_cable_landing_relation",
|
|
"celestrak_tle",
|
|
"top500",
|
|
"epoch_ai_gpu",
|
|
],
|
|
)
|
|
|
|
cables_records = records_by_source.get("arcgis_cables", [])
|
|
landing_point_records = records_by_source.get("arcgis_landing_points", [])
|
|
relation_records = records_by_source.get("arcgis_cable_landing_relation", [])
|
|
satellites_records = _filter_known_records(records_by_source.get("celestrak_tle", []))
|
|
supercomputer_records = _filter_known_records(records_by_source.get("top500", []))
|
|
gpu_records = _filter_known_records(records_by_source.get("epoch_ai_gpu", []))
|
|
bgp_anomalies_result = await db.execute(
|
|
select(BGPAnomaly)
|
|
.where(BGPAnomaly.status == "active")
|
|
.order_by(BGPAnomaly.created_at.desc())
|
|
.limit(200)
|
|
)
|
|
bgp_anomalies = list(bgp_anomalies_result.scalars().all())
|
|
bgp_anomaly_geography_hints = await build_anomaly_geography_hints(db, bgp_anomalies)
|
|
bgp_incidents_result = await db.execute(
|
|
select(BGPIncident)
|
|
.where(BGPIncident.status == "active")
|
|
.order_by(BGPIncident.created_at.desc())
|
|
.limit(100)
|
|
)
|
|
bgp_incidents = list(bgp_incidents_result.scalars().all())
|
|
bgp_incident_geography_hints = await build_incident_geography_hints(db, bgp_incidents)
|
|
bgp_collector_coverage = await build_bgp_collector_coverage(
|
|
db,
|
|
source_filter=("ris_live_bgp", "bgpstream_bgp"),
|
|
)
|
|
bgp_coverage_by_collector = {
|
|
item["collector"]: item
|
|
for item in bgp_collector_coverage
|
|
if item.get("collector")
|
|
}
|
|
|
|
city_to_cable_ids_map, cable_id_to_name_map = _build_landing_point_cable_maps(
|
|
relation_records,
|
|
cables_records,
|
|
)
|
|
|
|
aggregate = {
|
|
"satellites": convert_satellite_to_geojson(satellites_records),
|
|
"supercomputers": convert_supercomputer_to_geojson(supercomputer_records),
|
|
"gpu_clusters": convert_gpu_cluster_to_geojson(gpu_records),
|
|
"submarine_cables": convert_cable_to_geojson(cables_records),
|
|
"landing_points": convert_landing_point_to_geojson(
|
|
landing_point_records,
|
|
city_to_cable_ids_map,
|
|
cable_id_to_name_map,
|
|
),
|
|
"bgp_anomalies": convert_bgp_anomalies_to_geojson(
|
|
bgp_anomalies,
|
|
bgp_anomaly_geography_hints,
|
|
),
|
|
"bgp_incidents": convert_bgp_incidents_to_geojson(
|
|
bgp_incidents,
|
|
bgp_incident_geography_hints,
|
|
),
|
|
"bgp_collectors": convert_bgp_collectors_to_geojson(bgp_coverage_by_collector),
|
|
"alerts": {"type": "FeatureCollection", "features": []},
|
|
}
|
|
|
|
layers = _empty_scene_layers()
|
|
total_records = 0
|
|
for layer_name in DEFAULT_LAYER_ORDER:
|
|
feature_collection = aggregate.get(layer_name) or {"features": []}
|
|
items = {
|
|
item["id"]: item
|
|
for index, feature in enumerate(feature_collection.get("features", []), start=1)
|
|
for item in [_serialize_feature(layer_name, feature, index)]
|
|
}
|
|
layers[layer_name]["items"] = items
|
|
layers[layer_name]["revision"] = _layer_revision(items)
|
|
total_records += len(items)
|
|
|
|
timestamp = to_iso8601_utc(datetime.now(UTC))
|
|
state_hash = zlib.crc32(
|
|
_stable_json(
|
|
{
|
|
layer_name: {
|
|
item_id: item["revision"]
|
|
for item_id, item in layers[layer_name]["items"].items()
|
|
}
|
|
for layer_name in DEFAULT_LAYER_ORDER
|
|
}
|
|
).encode("utf-8")
|
|
) & 0xFFFFFFFF
|
|
|
|
return {
|
|
"generated_at": timestamp,
|
|
"state_hash": state_hash,
|
|
"total_records": total_records,
|
|
"layers": layers,
|
|
}
|
|
|
|
|
|
def _changes_exist(changes: Dict[str, Dict[str, List[Any]]]) -> bool:
|
|
return any(
|
|
layer_changes["added"] or layer_changes["updated"] or layer_changes["removed"]
|
|
for layer_changes in changes.values()
|
|
)
|
|
|
|
|
|
def build_incremental_changes(
|
|
previous_state: Dict[str, Any],
|
|
current_state: Dict[str, Any],
|
|
) -> Dict[str, Dict[str, List[Any]]]:
|
|
changes = _empty_changes()
|
|
|
|
for layer_name in DEFAULT_LAYER_ORDER:
|
|
previous_items = previous_state["layers"][layer_name]["items"]
|
|
current_items = current_state["layers"][layer_name]["items"]
|
|
previous_ids = set(previous_items)
|
|
current_ids = set(current_items)
|
|
|
|
for added_id in sorted(current_ids - previous_ids):
|
|
changes[layer_name]["added"].append(current_items[added_id])
|
|
|
|
for removed_id in sorted(previous_ids - current_ids):
|
|
changes[layer_name]["removed"].append(removed_id)
|
|
|
|
for common_id in sorted(previous_ids & current_ids):
|
|
if _stable_json(previous_items[common_id]) != _stable_json(current_items[common_id]):
|
|
changes[layer_name]["updated"].append(current_items[common_id])
|
|
|
|
return changes
|
|
|
|
|
|
def _full_payload_from_state(state: Dict[str, Any], sequence: int) -> Dict[str, Any]:
|
|
return {
|
|
"update_type": "full",
|
|
"sequence": sequence,
|
|
"cluster_time": state["generated_at"],
|
|
"display_profile": _default_display_profile(),
|
|
"camera_state": _default_camera_state(),
|
|
"payload": {
|
|
"meta": {
|
|
"generated_at": state["generated_at"],
|
|
"total_records": state["total_records"],
|
|
"state_hash": state["state_hash"],
|
|
},
|
|
"layers": {
|
|
layer_name: {
|
|
"revision": layer_data["revision"],
|
|
"items": [
|
|
layer_data["items"][item_id]
|
|
for item_id in sorted(layer_data["items"])
|
|
],
|
|
}
|
|
for layer_name, layer_data in state["layers"].items()
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _incremental_payload(
|
|
*,
|
|
current_state: Dict[str, Any],
|
|
sequence: int,
|
|
base_sequence: int,
|
|
changes: Dict[str, Dict[str, List[Any]]],
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"update_type": "incremental",
|
|
"sequence": sequence,
|
|
"base_sequence": base_sequence,
|
|
"cluster_time": current_state["generated_at"],
|
|
"changes": changes,
|
|
"meta": {
|
|
"generated_at": current_state["generated_at"],
|
|
"total_records": current_state["total_records"],
|
|
"state_hash": current_state["state_hash"],
|
|
},
|
|
}
|
|
|
|
|
|
def _noop_incremental_payload(state: Dict[str, Any], sequence: int) -> Dict[str, Any]:
|
|
return _incremental_payload(
|
|
current_state=state,
|
|
sequence=sequence,
|
|
base_sequence=sequence,
|
|
changes=_empty_changes(),
|
|
)
|
|
|
|
|
|
def expand_scene_payload_for_transport(scene_payload: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Split oversized scene payloads into smaller transport-safe chunks."""
|
|
update_type = scene_payload.get("update_type")
|
|
if update_type != "full":
|
|
return [scene_payload]
|
|
|
|
payload = scene_payload.get("payload") or {}
|
|
layers = payload.get("layers") or {}
|
|
if not layers:
|
|
return [scene_payload]
|
|
|
|
chunks: List[Dict[str, Any]] = []
|
|
for layer_name in DEFAULT_LAYER_ORDER:
|
|
layer_data = layers.get(layer_name) or {}
|
|
items = list(layer_data.get("items") or [])
|
|
if not items:
|
|
continue
|
|
|
|
chunk_size = UE_SCENE_LAYER_CHUNK_ITEM_LIMITS.get(layer_name, 100)
|
|
total_layer_chunks = max(1, (len(items) + chunk_size - 1) // chunk_size)
|
|
for chunk_index, offset in enumerate(range(0, len(items), chunk_size), start=1):
|
|
chunk_items = items[offset : offset + chunk_size]
|
|
chunks.append(
|
|
{
|
|
**scene_payload,
|
|
"payload": {
|
|
"meta": {
|
|
**(payload.get("meta") or {}),
|
|
"chunked": True,
|
|
"layer_name": layer_name,
|
|
"layer_chunk_index": chunk_index,
|
|
"layer_chunk_count": total_layer_chunks,
|
|
},
|
|
"layers": {
|
|
layer_name: {
|
|
"revision": layer_data.get("revision", 0),
|
|
"items": chunk_items,
|
|
}
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
if not chunks:
|
|
return [scene_payload]
|
|
|
|
total_chunks = len(chunks)
|
|
for transport_index, chunk in enumerate(chunks, start=1):
|
|
chunk_payload = chunk.setdefault("payload", {})
|
|
chunk_meta = chunk_payload.setdefault("meta", {})
|
|
chunk_meta["transport_chunk_index"] = transport_index
|
|
chunk_meta["transport_chunk_count"] = total_chunks
|
|
|
|
return chunks
|
|
|
|
|
|
class UeSceneStateStore:
|
|
"""Maintain cached ue_scene state and a bounded incremental replay history."""
|
|
|
|
def __init__(self, history_limit: int = 20) -> None:
|
|
self.history_limit = history_limit
|
|
self.sequence = 0
|
|
self.state: Dict[str, Any] | None = None
|
|
self.history: List[Dict[str, Any]] = []
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def _refresh_locked(self, db: AsyncSession) -> List[Dict[str, Any]]:
|
|
current_state = await build_visualization_scene_state(db)
|
|
|
|
if self.state is None:
|
|
self.sequence = 1
|
|
self.state = current_state
|
|
return [_full_payload_from_state(self.state, self.sequence)]
|
|
|
|
if current_state["state_hash"] == self.state["state_hash"]:
|
|
self.state = current_state
|
|
return []
|
|
|
|
previous_sequence = self.sequence
|
|
previous_state = self.state
|
|
self.sequence += 1
|
|
self.state = current_state
|
|
changes = build_incremental_changes(previous_state, current_state)
|
|
incremental = _incremental_payload(
|
|
current_state=current_state,
|
|
sequence=self.sequence,
|
|
base_sequence=previous_sequence,
|
|
changes=changes,
|
|
)
|
|
self.history.append(incremental)
|
|
if len(self.history) > self.history_limit:
|
|
self.history = self.history[-self.history_limit :]
|
|
return [incremental]
|
|
|
|
async def get_broadcast_payloads(self, db: AsyncSession) -> List[Dict[str, Any]]:
|
|
async with self.lock:
|
|
payloads = await self._refresh_locked(db)
|
|
return [payload for payload in payloads if payload["update_type"] == "full" or _changes_exist(payload.get("changes", {}))]
|
|
|
|
async def get_sync_payloads(
|
|
self,
|
|
db: AsyncSession,
|
|
*,
|
|
last_sequence: int | None,
|
|
reason: str | None,
|
|
) -> List[Dict[str, Any]]:
|
|
async with self.lock:
|
|
await self._refresh_locked(db)
|
|
if self.state is None:
|
|
return []
|
|
|
|
normalized_reason = (reason or "").strip()
|
|
if last_sequence is None or normalized_reason in FULL_RESYNC_REASONS:
|
|
return [_full_payload_from_state(self.state, self.sequence)]
|
|
|
|
if last_sequence == self.sequence:
|
|
return [_noop_incremental_payload(self.state, self.sequence)]
|
|
|
|
if last_sequence > self.sequence:
|
|
return [_full_payload_from_state(self.state, self.sequence)]
|
|
|
|
replay_payloads = [
|
|
payload
|
|
for payload in self.history
|
|
if payload["base_sequence"] >= last_sequence
|
|
and payload["sequence"] > last_sequence
|
|
]
|
|
if replay_payloads:
|
|
expected_base = last_sequence
|
|
ordered_payloads: List[Dict[str, Any]] = []
|
|
for payload in replay_payloads:
|
|
if payload["base_sequence"] != expected_base:
|
|
return [_full_payload_from_state(self.state, self.sequence)]
|
|
ordered_payloads.append(payload)
|
|
expected_base = payload["sequence"]
|
|
if ordered_payloads and ordered_payloads[-1]["sequence"] == self.sequence:
|
|
return ordered_payloads
|
|
|
|
return [_full_payload_from_state(self.state, self.sequence)]
|
|
|
|
|
|
ue_scene_state_store = UeSceneStateStore()
|