diff --git a/app.arc b/app.arc index d0b5ab829..828684a3a 100644 --- a/app.arc +++ b/app.arc @@ -43,6 +43,12 @@ client_credentials client_id **String PointInTimeRecovery true +acrossapi_tle + satname *String + epoch **String + tle1 String + tle2 String + sessions _idx *String _ttl TTL diff --git a/python/across_api/base/api.py b/python/across_api/base/api.py index c9e25b5b1..81414a9c0 100644 --- a/python/across_api/base/api.py +++ b/python/across_api/base/api.py @@ -2,8 +2,13 @@ # Administrator of the National Aeronautics and Space Administration. # All Rights Reserved. -from fastapi import FastAPI +from datetime import datetime +from typing import Annotated, Optional +from astropy.time import Time # type: ignore +from fastapi import Depends, FastAPI, Query + +# FastAPI app definition app = FastAPI( title="ACROSS API", summary="Astrophysics Cross-Observatory Science Support (ACROSS).", @@ -13,3 +18,19 @@ }, root_path="/labs/api/v1", ) + + +# Globally defined Depends definitions +async def epoch( + epoch: Annotated[ + datetime, + Query( + title="Epoch", + description="Epoch in UTC or ISO format.", + ), + ], +) -> Optional[Time]: + return Time(epoch) + + +EpochDep = Annotated[datetime, Depends(epoch)] diff --git a/python/across_api/base/schema.py b/python/across_api/base/schema.py index 637794ad3..c38302000 100644 --- a/python/across_api/base/schema.py +++ b/python/across_api/base/schema.py @@ -2,7 +2,27 @@ # Administrator of the National Aeronautics and Space Administration. # All Rights Reserved. -from pydantic import BaseModel, ConfigDict + +from datetime import datetime +from typing import Any, List, Optional + +import astropy.units as u # type: ignore +from arc import tables # type: ignore +from astropy.time import Time # type: ignore +from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, computed_field +from typing_extensions import Annotated + +# Define a Pydantic type for astropy Time objects, which will be serialized as +# a naive UTC datetime object, or a string in ISO format for JSON. If Time is +# in list form, then it will be serialized as a list of UTC naive datetime +# objects. +AstropyTime = Annotated[ + Time, + PlainSerializer( + lambda x: x.utc.datetime, + return_type=datetime, + ), +] class BaseSchema(BaseModel): @@ -13,4 +33,105 @@ class BaseSchema(BaseModel): Subclasses can inherit from this class and override the `from_attributes` method to define their own schema logic. """ - model_config = ConfigDict(from_attributes=True) + model_config = ConfigDict(from_attributes=True, arbitrary_types_allowed=True) + + +class TLEGetSchema(BaseSchema): + epoch: AstropyTime + + +class TLEEntry(BaseSchema): + """ + Represents a single TLE entry in the TLE database. + + Parameters + ---------- + satname : str + The name of the satellite from the Satellite Catalog. + tle1 : str + The first line of the TLE. + tle2 : str + The second line of the TLE. + + Attributes + ---------- + epoch + """ + + __tablename__ = "acrossapi_tle" + satname: str # Partition Key + tle1: str = Field(min_length=69, max_length=69) + tle2: str = Field(min_length=69, max_length=69) + + @computed_field # type: ignore[misc] + @property + def epoch(self) -> AstropyTime: + """ + Calculate the Epoch of the TLE file. See + https://celestrak.org/columns/v04n03/#FAQ04 for more information on + how the year / epoch encoding works. + + Returns + ------- + The calculated epoch of the TLE. + """ + # Extract epoch from TLE + tleepoch = self.tle1.split()[3] + + # Convert 2 number year into 4 number year. + tleyear = int(tleepoch[0:2]) + if tleyear < 57: + year = 2000 + tleyear + else: + year = 1900 + tleyear + + # Convert day of year into float + day_of_year = float(tleepoch[2:]) + + # Return Time epoch + return Time(f"{year}-01-01", scale="utc") + (day_of_year - 1) * u.day + + @classmethod + def find_tles_between_epochs( + cls, satname: str, start_epoch: Time, end_epoch: Time + ) -> List[Any]: + """ + Find TLE entries between two epochs in the TLE database for a given + satellite TLE name. + + Arguments + --------- + satname + The common name for the spacecraft based on the Satellite Catalog. + start_epoch + The start time over which to search for TLE entries. + end_epoch + The end time over which to search for TLE entries. + + Returns + ------- + A list of TLEEntry objects between the specified epochs. + """ + table = tables.table(cls.__tablename__) + + # Query the table for TLEs between the two epochs + response = table.query( + KeyConditionExpression="satname = :satname AND epoch BETWEEN :start_epoch AND :end_epoch", + ExpressionAttributeValues={ + ":satname": satname, + ":start_epoch": str(start_epoch.utc.datetime), + ":end_epoch": str(end_epoch.utc.datetime), + }, + ) + + # Convert the response into a list of TLEEntry objects and return them + return [cls(**item) for item in response["Items"]] + + def write(self) -> None: + """Write the TLE entry to the database.""" + table = tables.table(self.__tablename__) + table.put_item(Item=self.model_dump(mode="json")) + + +class TLESchema(BaseSchema): + tle: Optional[TLEEntry] diff --git a/python/across_api/base/tle.py b/python/across_api/base/tle.py new file mode 100644 index 000000000..8c6a6c28d --- /dev/null +++ b/python/across_api/base/tle.py @@ -0,0 +1,385 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. + + +import logging +import os +from typing import List, Optional + +import requests +from astropy.time import Time # type: ignore +from astropy.units import Quantity # type: ignore +from requests import HTTPError +from spacetrack import SpaceTrackClient # type: ignore + +from .common import ACROSSAPIBase +from .schema import TLEEntry, TLEGetSchema, TLESchema + + +class TLEBase(ACROSSAPIBase): + """ + Class for retrieving and updating spacecraft TLEs in the TLE database. If + the TLEs are not found in the database, they are retrieved from + space-track.org based on the name of the spacecraft given by `tle_name`. + Other backup methods for fetching the TLE include using either the supplied + `tle_url`, or from the URL specified in the `tle_concat` attribute (in the + concatenated TLE format). TLEs fetched are then written to the database for + future use. + + Parameters + ---------- + epoch + Epoch of TLE to retrieve + + Attributes + ---------- + tles + List of TLEs currently loaded + tle + TLE entry for given epoch + offset + Offset between TLE epoch and requested epoch in days + tle_name + Name of the spacecraft as it appears in the Spacecraft Catalog. + tle_url + URL to retrieve the TLE from. + tle_concat + URL to retrieve the TLE from in concatenated format. + tle_bad + If the TLE is this many days old, it is considered outdated, and a new + TLE will be retrieved. + tle_min_epoch + Minimum epoch for which TLEs are available, typically this will + correspond to a date after the launch of the spacecraft. + + Methods + ------- + get + Get TLEs for given epoch + tle_out_of_date + Check if the given TLE is out of date + read_tle_web + Read TLE from dedicated weblink + read_tle_concat + Read TLEs in the concatenated format + read_tle_db + Read the best TLE for a given epoch from the local database of TLEs + write_db + Write a TLE to the database + write_db_all_tles + Write all loaded TLEs to database + """ + + _schema = TLESchema + _get_schema = TLEGetSchema + + # Configuration parameters + tle_concat: Optional[str] + tle_url: Optional[str] + tle_bad: Quantity + tle_name: str + tle_norad_id: int + tle_min_epoch: Time + # Arguments + epoch: Time + # Attributes + tles: List[TLEEntry] = [] + # Return values + error: Optional[str] + + def __init__(self, epoch: Time): + """ + Initialize a TLE object with the given epoch. + + Arguments + --------- + epoch + The epoch of the TLE object. + """ + self.epoch = epoch + self.tles = [] + if self.validate_get(): + self.get() + + def read_tle_db(self) -> bool: + """ + Read the best TLE for a given epoch from the local database of TLEs + + Returns + ------- + Did it work? + """ + # Read TLEs from the database for a given `tle_name` and epoch within + # the allowed range + self.tles = TLEEntry.find_tles_between_epochs( + self.tle_name, + self.epoch - self.tle_bad, + self.epoch + self.tle_bad, + ) + + return True + + def read_tle_web(self) -> bool: + """ + Read TLE from dedicated weblink. + + This method downloads the TLE (Two-Line Elements) from a dedicated + weblink. It retrieves the TLE data, parses it, and stores the valid TLE + entries in a list. Often websites (e.g. Celestrak) will have multiple + TLEs for a given satellite, so this method will only store the TLEs + that match the given satellite name, as stored in the `tle_name` + attribute. + + Returns + ------- + True if the TLE was successfully read and stored, False otherwise. + """ + # Check if the URL is set + if self.tle_url is None: + return False + + # Download TLE from internet + r = requests.get(self.tle_url) + try: + # Check for HTTP errors + r.raise_for_status() + except HTTPError as e: + logging.exception(e) + return False + + # Read valid TLEs into a list + tlefile = r.text.splitlines() + tles = [ + TLEEntry( + satname=tlefile[i].strip(), + tle1=tlefile[i + 1].strip(), + tle2=tlefile[i + 2].strip(), + ) + for i in range(0, len(tlefile), 3) + if self.tle_name in tlefile[i] + ] + + # Append them to the list of stored TLEs + self.tles.extend(tles) + + # Check if a good TLE for the current epoch was found + if self.tle_out_of_date is False: + return True + + return False + + def read_tle_space_track(self) -> bool: + """ + Read TLE from Space-Track.org. + + This method downloads the TLE (Two-Line Elements) from Space-Track.org. + It retrieves the TLE data, parses it, and stores the valid TLE entries + in a list. Often websites (e.g. Celestrak) will have multiple TLEs for + a given satellite, so this method will only store the TLEs that match + the given satellite name, as stored in the `tle_name` attribute. + + Returns + ------- + True if the TLE was successfully read and stored, False otherwise. + """ + # Check if the URL is set + if self.tle_url is None: + return False + + # Build space-track.org query + epoch_start = self.epoch - self.tle_bad + epoch_stop = self.epoch + self.tle_bad + + # Log into space-track.org + st = SpaceTrackClient( + identity=os.environ.get("SPACE_TRACK_USER"), + password=os.environ.get("SPACE_TRACK_PASS"), + ) + + # Fetch the TLEs between the requested epochs + tletext = st.tle( + norad_cat_id=self.tle_norad_id, + orderby="epoch desc", + limit=22, + format="tle", + epoch=f">{epoch_start.datetime},<{epoch_stop.datetime}", + ) + # Check if we got a return + if tletext == "": + return False + + # Split the TLEs into individual lines + tletext = tletext.splitlines() + + # Parse the results into a list of TLEEntry objects + tles = [ + TLEEntry( + satname=self.tle_name, + tle1=tletext[i].strip(), + tle2=tletext[i + 1].strip(), + ) + for i in range(0, len(tletext), 2) + ] + + # Append them to the list of stored TLEs + self.tles.extend(tles) + + # Check if a good TLE for the current epoch was found + if self.tle_out_of_date is False: + return True + + return False + + def read_tle_concat(self) -> bool: + """ + Read TLEs in the CONCAT MISSION_TLE_ARCHIVE.tle format. This format is + used by the CONCAT to store TLEs for various missions. The format + consists of a concatenation of all available TLEs, without the name + header. + + Returns + ------- + True if TLEs were successfully read, False otherwise. + """ + # Check if the URL is set + if self.tle_concat is None: + return False + + # Download TLEs from internet + r = requests.get(self.tle_concat) + try: + # Check for HTTP errors + r.raise_for_status() + except HTTPError as e: + logging.exception(e) + return False + + # Parse the file into a list of TLEEntry objects + tlefile = r.text.splitlines() + tles = [ + TLEEntry( + satname=self.tle_name, + tle1=tlefile[i].strip(), + tle2=tlefile[i + 1].strip(), + ) + for i in range(0, len(tlefile), 2) + ] + + # Append that list to the list of TLEs + self.tles.extend(tles) + + # Check if a good TLE for the requested epoch was found + if self.tle_out_of_date is False: + return True + return False + + @property + def tle(self) -> Optional[TLEEntry]: + """ + Return the best TLE out of the TLEs currently loaded for a the given + epoch. + + Returns + ------- + Best TLE for the given epoch, or None if no TLEs are loaded. + """ + if self.epoch is not None and len(self.tles) > 0: + return min( + self.tles, key=lambda x: abs((x.epoch - self.epoch).to_value("s")) + ) + return None + + @property + def tle_out_of_date(self) -> Optional[bool]: + """ + Is this TLE outside of the allowed range? + + Returns + ------- + True if the epoch of the loaded TLE is more the `tle_bad` days off. + Returns None if no TLE is loaded. + """ + # Check if we have a TLE + if self.tle is None: + return None + + # Calculate the number of days between the TLE epoch and the requested + # epoch. If this is greater than the allowed number of days (given by + # `tle_bad`), then return True + if abs(self.epoch - self.tle.epoch) > self.tle_bad: + return True + return False + + def get(self) -> bool: + """ + Find in the best TLE for a given epoch. This method will first try to + read the TLE from the local database. If that fails, it will try to + read the TLE from the internet (with support for two different TLE + formats). If that fails, it will return False, indicating that no TLE + was found. + + Parameters + ---------- + epoch + Epoch for which you want to retrieve a TLE. + + Returns + ------- + True if a TLE was found, False otherwise. + """ + + # Check if the requested arguments are valid + if self.validate_get() is False: + return False + + # Check that the epoch is within the allowed range. If set + # to before `tle_min_epoch`, then set it to `tle_min_epoch`. If set to + # a value in the future, then set it to the current time to give the most + # up to date TLE. + if self.epoch < self.tle_min_epoch: + self.epoch = self.tle_min_epoch + elif self.epoch > Time.now().utc: + self.epoch = Time.now().utc + + # Fetch TLE from the TLE database + if self.read_tle_db() is True: + if self.tle is not None: + return True + + # Next try querying space-track.org for the TLE. This will only work if + # the environment variables SPACE_TRACK_USER and + # SPACE_TRACK_PASS are set, and valid. + if self.read_tle_space_track() is True: + # Write the TLE to the database for next time + if self.tle is not None: + self.tle.write() + return True + + # Next try try reading the TLE given in the concatenated format at the + # URL given by `tle_concat`. Concatenated format should have every TLE + # for the satellite since launch in a single file, so it's safe to + # query this for any date within the mission lifetime. For an example + # of this format (for the NuSTAR mission), see here: + # https://nustarsoc.caltech.edu/NuSTAR_Public/NuSTAROperationSite/NuSTAR.tle + if self.tle_concat is not None: + if self.read_tle_concat() is True: + # Write the TLE to the database for next time + if self.tle is not None: + self.tle.write() + return True + + # Finally try reading from the web at the URL given by `tle_url`. Note + # that URL based TLEs are usually only valid for the current epoch, so + # we will only use this if the epoch being requested is within + # `tle_bad` days of the current epoch. + if self.tle_url is not None: + if self.epoch > Time.now().utc - self.tle_bad: + if self.read_tle_web() is True: + # Write the TLE to the database for next time + if self.tle is not None: + self.tle.write() + return True + + # If we did not find any valid TLEs, then return False + return False diff --git a/python/across_api/burstcube/__init__.py b/python/across_api/burstcube/__init__.py new file mode 100644 index 000000000..ff1d22748 --- /dev/null +++ b/python/across_api/burstcube/__init__.py @@ -0,0 +1,3 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. diff --git a/python/across_api/burstcube/api.py b/python/across_api/burstcube/api.py new file mode 100644 index 000000000..33348fa02 --- /dev/null +++ b/python/across_api/burstcube/api.py @@ -0,0 +1,17 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. + +from ..base.api import EpochDep, app +from ..base.schema import TLESchema +from .tle import BurstCubeTLE + + +@app.get("/burstcube/tle") +async def burstcube_tle( + epoch: EpochDep, +) -> TLESchema: + """ + Returns the best TLE for BurstCube for a given epoch. + """ + return BurstCubeTLE(epoch=epoch).schema diff --git a/python/across_api/burstcube/tle.py b/python/across_api/burstcube/tle.py new file mode 100644 index 000000000..e1ecea25d --- /dev/null +++ b/python/across_api/burstcube/tle.py @@ -0,0 +1,21 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. + +import astropy.units as u # type: ignore +from astropy.time import Time # type: ignore +from cachetools import TTLCache, cached + +from ..base.tle import TLEBase + + +@cached(cache=TTLCache(maxsize=128, ttl=3600)) +class BurstCubeTLE(TLEBase): + # Configuration options for BurstCubeTLE + # FIXME: These values are placeholders until BurstCube is launched + tle_name = "ISS (ZARYA)" + tle_norad_id = 25544 + tle_url = "https://celestrak.com/NORAD/elements/stations.txt" + tle_concat = None + tle_bad = 4 * u.day + tle_min_epoch = Time("2023-12-18", scale="utc") diff --git a/python/across_api/swift/__init__.py b/python/across_api/swift/__init__.py new file mode 100644 index 000000000..ff1d22748 --- /dev/null +++ b/python/across_api/swift/__init__.py @@ -0,0 +1,3 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. diff --git a/python/across_api/swift/api.py b/python/across_api/swift/api.py new file mode 100644 index 000000000..26e1e78cb --- /dev/null +++ b/python/across_api/swift/api.py @@ -0,0 +1,17 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. + +from ..base.api import EpochDep, app +from ..base.schema import TLESchema +from .tle import SwiftTLE + + +@app.get("/swift/tle") +async def swift_tle( + epoch: EpochDep, +) -> TLESchema: + """ + Returns the best TLE for Swift for a given epoch. + """ + return SwiftTLE(epoch=epoch).schema diff --git a/python/across_api/swift/tle.py b/python/across_api/swift/tle.py new file mode 100644 index 000000000..045948318 --- /dev/null +++ b/python/across_api/swift/tle.py @@ -0,0 +1,20 @@ +# Copyright © 2023 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. + +import astropy.units as u # type: ignore +from astropy.time import Time # type: ignore +from cachetools import TTLCache, cached + +from ..base.tle import TLEBase + + +@cached(cache=TTLCache(maxsize=128, ttl=3600)) +class SwiftTLE(TLEBase): + # Configuration options for SwiftTLE + tle_name = "SWIFT" + tle_norad_id = 28485 + tle_url = "https://celestrak.org/NORAD/elements/gp.php?INTDES=2004-047" + tle_concat = "https://www.swift.ac.uk/about/status_files/tle" + tle_bad = 4 * u.day + tle_min_epoch = Time("2004-11-20 23:00:00", scale="utc") diff --git a/python/lambda.py b/python/lambda.py index 78ba2f94b..156b08a24 100644 --- a/python/lambda.py +++ b/python/lambda.py @@ -5,7 +5,10 @@ from mangum import Mangum from across_api.base.api import app -from across_api.across.api import * # noqa F401 +import across_api.across.api # noqa F401 +import across_api.burstcube.api # noqa F401 +import across_api.swift.api # noqa F401 + from env import feature if feature("LABS"): diff --git a/python/requirements.in b/python/requirements.in index a1d417e23..bf1389e8a 100644 --- a/python/requirements.in +++ b/python/requirements.in @@ -1,4 +1,8 @@ +architect-functions astropy +cachetools +email-validator fastapi mangum requests +spacetrack diff --git a/python/requirements.txt b/python/requirements.txt index d2c347ef7..b49f6d44f 100644 --- a/python/requirements.txt +++ b/python/requirements.txt @@ -2,28 +2,58 @@ # This file is autogenerated by pip-compile with Python 3.11 # by the following command: # -# pip-compile python/requirements.in +# pip-compile # annotated-types==0.6.0 # via pydantic anyio==3.7.1 # via # fastapi + # httpx # starlette +architect-functions==1.0.0 + # via -r requirements.in astropy==6.0.0 # via -r requirements.in -astropy-iers-data==0.2023.12.18.0.30.18 +astropy-iers-data # via astropy +attrs==23.2.0 + # via rush +cachetools==5.3.2 + # via -r requirements.in certifi==2023.11.17 - # via requests + # via + # httpcore + # httpx + # requests +cffi==1.16.0 + # via cryptography charset-normalizer==3.3.2 # via requests +cryptography==41.0.7 + # via architect-functions +dnspython==2.4.2 + # via email-validator +ecdsa==0.18.0 + # via python-jose +email-validator==2.1.0.post1 + # via -r requirements.in fastapi==0.105.0 # via -r requirements.in +h11==0.14.0 + # via httpcore +httpcore==1.0.2 + # via httpx +httpx==0.26.0 + # via spacetrack idna==3.6 # via # anyio + # email-validator + # httpx # requests +logbook==1.7.0.post0 + # via spacetrack mangum==0.17.0 # via -r requirements.in numpy==1.26.2 @@ -32,18 +62,41 @@ numpy==1.26.2 # pyerfa packaging==23.2 # via astropy +pyasn1==0.5.1 + # via + # python-jose + # rsa +pycparser==2.21 + # via cffi pydantic==2.5.2 # via fastapi pydantic-core==2.14.5 # via pydantic pyerfa==2.0.1.1 # via astropy +python-jose==3.3.0 + # via architect-functions pyyaml==6.0.1 # via astropy +represent==2.0 + # via spacetrack requests==2.31.0 # via -r requirements.in +rsa==4.9 + # via python-jose +rush==2021.4.0 + # via spacetrack +simplejson==3.19.2 + # via architect-functions +six==1.16.0 + # via ecdsa sniffio==1.3.0 - # via anyio + # via + # anyio + # httpx + # spacetrack +spacetrack==1.2.0 + # via -r requirements.in starlette==0.27.0 # via fastapi typing-extensions==4.9.0 @@ -52,5 +105,5 @@ typing-extensions==4.9.0 # mangum # pydantic # pydantic-core -urllib3==2.1.0 +urllib3==2.0.7 # via requests diff --git a/requirements.txt b/requirements.txt index 3c402076c..7f00831ee 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,7 @@ black +boto3 mypy pytest types-requests +types-cachetools -r python/requirements.txt