| """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 |
| 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() |
| 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') |
|
|
