"""Serve Sidereal and a shared, rate-limited CelesTrak orbital-data cache.""" import asyncio import json import html import re import time from datetime import datetime, timezone from pathlib import Path import requests from fastapi import FastAPI, HTTPException, Query from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles import terrain import deep_stars ROOT = Path(__file__).resolve().parent app = FastAPI(docs_url=None, redoc_url=None) cache = json.loads((ROOT / 'public/data/satellites.json').read_text()) starlink_cache = json.loads((ROOT / 'public/data/starlink.json').read_text()) starlink_last_attempt = 0.0 starlink_refreshing = False catalogue_cache = json.loads((ROOT / 'public/data/satellite-catalogue.json').read_text()) last_attempt = 0.0 refreshing = False CACHE_SECONDS = 7200 catalogue_last_attempt = 0.0 catalogue_refreshing = False CATALOGUE_SECONDS = 86400 async def refresh_catalogue(): global catalogue_cache, catalogue_last_attempt, catalogue_refreshing if catalogue_refreshing or time.time() - catalogue_last_attempt < CATALOGUE_SECONDS: return catalogue_refreshing = True catalogue_last_attempt = time.time() try: def fetch(): response = requests.get('https://celestrak.org/satcat/records.php', params={'GROUP': 'visual', 'FORMAT': 'JSON'}, headers={'User-Agent': 'SiderealObservingDesk/0.1'}, timeout=20) response.raise_for_status() records = response.json() if not isinstance(records, list) or not records: raise ValueError('Invalid satellite catalogue') return {str(r['NORAD_CAT_ID']): { 'objectType': r.get('OBJECT_TYPE', 'UNK'), 'owner': r.get('OWNER', ''), 'launchDate': r.get('LAUNCH_DATE', ''), 'internationalId': r.get('OBJECT_ID', '') } for r in records} objects = await asyncio.to_thread(fetch) catalogue_cache = {'fetchedAt': datetime.now(timezone.utc).isoformat(), 'source': 'CelesTrak SATCAT', 'objects': objects} except (requests.RequestException, ValueError, KeyError, TypeError): pass finally: catalogue_refreshing = False async def refresh(): global cache, last_attempt, refreshing if refreshing or time.time() - last_attempt < CACHE_SECONDS: return refreshing = True last_attempt = time.time() try: def fetch(): response = requests.get( 'https://celestrak.org/NORAD/elements/gp.php', params={'GROUP': 'visual', 'FORMAT': 'JSON'}, headers={'User-Agent': 'SiderealObservingDesk/0.1'}, timeout=20) response.raise_for_status() items = response.json() if not isinstance(items, list) or not items or 'NORAD_CAT_ID' not in items[0]: raise ValueError('Invalid orbital-data response') return items elements = await asyncio.to_thread(fetch) cache = {'elements': elements, 'fetchedAt': datetime.now(timezone.utc).isoformat(), 'source': 'CelesTrak visual group'} except (requests.RequestException, ValueError): pass # Keep the timestamped snapshot; the browser marks/hides old elements. finally: refreshing = False async def refresh_starlinks(): global starlink_cache, starlink_last_attempt, starlink_refreshing if starlink_refreshing or time.time() - starlink_last_attempt < CACHE_SECONDS: return starlink_refreshing = True starlink_last_attempt = time.time() try: def fetch(): response = requests.get('https://celestrak.org/NORAD/elements/gp.php', params={'GROUP': 'starlink', 'FORMAT': 'JSON'}, headers={'User-Agent': 'SiderealObservingDesk/0.1'}, timeout=30) response.raise_for_status() items = response.json() if not isinstance(items, list) or not items or len(items) > 50000 or any( not isinstance(e, dict) or 'NORAD_CAT_ID' not in e or 'EPOCH' not in e for e in items): raise ValueError('Invalid Starlink orbital-data response') return items elements = await asyncio.to_thread(fetch) starlink_cache = {'elements': elements, 'fetchedAt': datetime.now(timezone.utc).isoformat(), 'source': 'CelesTrak Starlink group'} except (requests.RequestException, ValueError): pass finally: starlink_refreshing = False @app.get('/api/health') def health(): return {'status': 'ok', 'hardware': 'simulation only'} @app.get('/api/satellites') async def satellites(): try: age = time.time() - datetime.fromisoformat(cache['fetchedAt']).timestamp() except (ValueError, KeyError): age = CACHE_SECONDS + 1 if age > CACHE_SECONDS and not refreshing: asyncio.create_task(refresh()) starlink_age = time.time() - datetime.fromisoformat(starlink_cache['fetchedAt']).timestamp() if starlink_age > CACHE_SECONDS and not starlink_refreshing: asyncio.create_task(refresh_starlinks()) catalogue_age = time.time() - datetime.fromisoformat(catalogue_cache['fetchedAt']).timestamp() if catalogue_age > CATALOGUE_SECONDS and not catalogue_refreshing: asyncio.create_task(refresh_catalogue()) elements = {e['NORAD_CAT_ID']: e for e in [*starlink_cache['elements'], *cache['elements']]} return JSONResponse({**cache, 'source': 'CelesTrak visual + Starlink groups', 'elements': list(elements.values()), 'catalogue': catalogue_cache, 'groups': {'visual': cache['fetchedAt'], 'starlink': starlink_cache['fetchedAt']}, 'cached': max(age, starlink_age) > CACHE_SECONDS}, headers={'Cache-Control': 'public, max-age=300'}) terrain_slots = asyncio.Semaphore(2) terrain_pending = {} @app.get('/api/horizon') async def horizon(lat: float = Query(ge=-90, le=90), lon: float = Query(ge=-180, le=180), height: float = Query(default=1.5, ge=0, le=500)): key = (round(lat, 5), round(lon, 5), round(height, 1)) async def compute(): async with terrain_slots: return await asyncio.to_thread(terrain.horizon, *key) if key not in terrain_pending: if len(terrain_pending) >= 8: raise HTTPException(503, 'Terrain is busy. Try again shortly.') task = asyncio.create_task(compute()) terrain_pending[key] = task def complete(t): terrain_pending.pop(key, None) if not t.cancelled(): t.exception() # Retrieve errors even if every client disconnected. task.add_done_callback(complete) try: result = await asyncio.shield(terrain_pending[key]) return JSONResponse(result, headers={'Cache-Control': 'private, max-age=86400'}) except ValueError as e: raise HTTPException(422, str(e)) from e except (requests.RequestException, OSError, KeyError): raise HTTPException(503, 'Could not load terrain data. Try again.') @app.get('/api/elevation') def elevation(lat: float = Query(ge=-85, le=85), lon: float = Query(ge=-180, le=180), easting: float | None = Query(default=None, ge=2400000, le=2900000), northing: float | None = Query(default=None, ge=1000000, le=1400000)): try: return terrain.elevation(lat, lon, easting, northing) except (requests.RequestException, ValueError, OSError, KeyError): raise HTTPException(503, 'Elevation lookup is unavailable. You can enter it manually.') @app.get('/api/locations/search') def search_locations(q: str = Query(min_length=3, max_length=160)): try: response = requests.get('https://api3.geo.admin.ch/rest/services/ech/SearchServer', params={'searchText': q, 'type': 'locations', 'origins': 'address,gazetteer', 'sr': 2056, 'limit': 6}, headers=terrain.HEADERS, timeout=12) response.raise_for_status() return {'results': [{'name': html.unescape(re.sub('<[^>]*>', '', row['attrs']['label'])), 'lat': row['attrs']['lat'], 'lon': row['attrs']['lon'], 'easting': row['attrs']['y'], 'northing': row['attrs']['x']} for row in response.json()['results']]} except (requests.RequestException, ValueError, KeyError): raise HTTPException(503, 'Address search is unavailable. Use coordinates or your browser location.') star_slots = asyncio.Semaphore(2) star_pending = {} @app.get('/api/stars/overview') async def star_overview(): return FileResponse(ROOT / 'dist/data/gaia-overview.json.gz', media_type='application/json', headers={'Content-Encoding': 'gzip', 'Cache-Control': 'public, max-age=86400'}) @app.get('/api/stars/deep') async def faint_stars(ra: float = Query(ge=0, le=360), dec: float = Query(ge=-90, le=90), radius: float = Query(ge=0.1, le=32), magnitude: float = Query(gt=7.5, le=14)): key = (round(ra, 5), round(dec, 5), round(radius, 3), round(magnitude, 1)) if radius > (32 if magnitude <= 10 else 16 if magnitude <= 12 else 8): raise HTTPException(422, 'Field too wide for this magnitude. Zoom closer.') async def compute(): async with star_slots: return await asyncio.to_thread(deep_stars.cone, *key) if key not in star_pending: if len(star_pending) >= 8: raise HTTPException(503, 'Faint-star catalogue is busy. Try again shortly.') task = asyncio.create_task(compute()) star_pending[key] = task def complete(t): star_pending.pop(key, None) if not t.cancelled(): t.exception() task.add_done_callback(complete) try: result = await asyncio.shield(star_pending[key]) return JSONResponse(result, headers={'Cache-Control': 'public, max-age=86400'}) except (requests.RequestException, ValueError, OSError, KeyError): raise HTTPException(503, 'Could not load Gaia stars. Try again; the bright-star map remains available.') app.mount('/', StaticFiles(directory=ROOT / 'dist', html=True), name='frontend')