denshooter
106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
from skyfield.api import load
|
|
import asyncio
|
|
|
|
SATELLITE_GROUPS = {
|
|
"military": "https://celestrak.org/NORAD/elements/gp.php?GROUP=military&FORMAT=tle",
|
|
"stations": "https://celestrak.org/NORAD/elements/gp.php?GROUP=stations&FORMAT=tle",
|
|
"navigation": "https://celestrak.org/NORAD/elements/gp.php?GROUP=gnss&FORMAT=tle",
|
|
}
|
|
|
|
GROUP_LIMITS = {
|
|
"military": 800,
|
|
"stations": 50,
|
|
"navigation": 120,
|
|
}
|
|
|
|
satellites_data: dict = {}
|
|
|
|
|
|
def load_satellites():
|
|
"""
|
|
Blocking I/O: loads TLE data from Celestrak.
|
|
Must be called via asyncio.to_thread to avoid blocking the event loop.
|
|
"""
|
|
global satellites_data
|
|
try:
|
|
ts = load.timescale()
|
|
loaded = {}
|
|
for group, url in SATELLITE_GROUPS.items():
|
|
try:
|
|
filename = f"{group}.tle"
|
|
sats = load.tle_file(url, filename=filename)
|
|
loaded[group] = sats
|
|
print(f"[SPACE] {group}: {len(sats)} satellites loaded.")
|
|
except Exception as e:
|
|
print(f"[SPACE] Failed to load {group}: {e}")
|
|
satellites_data = {**loaded, "ts": ts}
|
|
except Exception as e:
|
|
print(f"[SPACE] Critical load failure: {e}")
|
|
|
|
|
|
def _compute_positions_sync(sats: list, category: str, t, limit: int) -> list:
|
|
"""
|
|
CPU-bound position computation — runs in a thread executor.
|
|
Returns dicts, does NOT mutate shared state.
|
|
"""
|
|
positions = []
|
|
for sat in sats[:limit]:
|
|
try:
|
|
geocentric = sat.at(t)
|
|
subpoint = geocentric.subpoint()
|
|
lat = subpoint.latitude.degrees
|
|
lon = subpoint.longitude.degrees
|
|
alt = subpoint.elevation.m
|
|
|
|
if not (-90 <= lat <= 90) or not (-180 <= lon <= 180):
|
|
continue
|
|
|
|
positions.append({
|
|
"id": sat.model.satnum,
|
|
"name": sat.name,
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"alt": alt,
|
|
"category": category,
|
|
"type": "satellite"
|
|
})
|
|
except Exception:
|
|
pass # skip satellites with bad TLE data
|
|
return positions
|
|
|
|
|
|
async def get_satellite_positions() -> list:
|
|
if "ts" not in satellites_data:
|
|
# First run: load everything in a background thread
|
|
await asyncio.to_thread(load_satellites)
|
|
|
|
if "ts" not in satellites_data:
|
|
print("[SPACE] Satellite data unavailable.")
|
|
return []
|
|
|
|
t = satellites_data["ts"].now()
|
|
all_positions = []
|
|
|
|
for group, limit in GROUP_LIMITS.items():
|
|
if group in satellites_data:
|
|
result = await asyncio.to_thread(
|
|
_compute_positions_sync,
|
|
satellites_data[group],
|
|
group,
|
|
t,
|
|
limit
|
|
)
|
|
all_positions.extend(result)
|
|
|
|
print(f"[SPACE] {len(all_positions)} satellite positions computed.")
|
|
return all_positions
|
|
|
|
|
|
if __name__ == "__main__":
|
|
load_satellites()
|
|
positions = asyncio.run(get_satellite_positions())
|
|
print(f"Total: {len(positions)}")
|
|
from collections import Counter
|
|
for cat, count in Counter(p["category"] for p in positions).items():
|
|
print(f" {cat}: {count}")
|