Example Python Script
An example Python script for reference only.
"""
Example ODS API Client
Usage, to get ODS Data:
python example_ods_api_client.py --get_ods_data
To upload a TLE file:
python example_ods_api_client.py --tle_file_path <path to TLE file>
To upload a directory of satellite avoidance logs:
python example_ods_api_client.py --tba_dir_path <log directory>
"""
# --------------------------------------------------------------------------------------------------
# NOTE: For demo purposes, loads client credentials from an .env file and persists an
# access token back to the .env file to be used for subsequent requests.
#
# For production, use a secure credential store like your OS keychain or secrets manager instead.
#
# Use the ODS API responsibly. Avoid unnecessary or repeated requests, rotate tokens when needed
# and respect rate limits.
# --------------------------------------------------------------------------------------------------
import argparse
import os
import time
from io import BufferedReader
from typing import Callable, Dict, Optional
import requests
from dotenv import load_dotenv, set_key
from loguru import logger
API_BASE = "https://ods.nrao.edu"
TOKEN_URL = f"{API_BASE}/token"
ODS_DATA_URL = f"{API_BASE}/ods_data"
SAT_AVOIDANCE_LOG_URL = f"{API_BASE}/sat_avoidance_logs"
TLE_DATA_URL = f"{API_BASE}/tle_data"
MAX_REQUEST_RETRIES = 3
RETRY_STATUS_CODES = [403, 408, 429, 500, 502, 503, 504]
# Load credentials from env file
ENV_FILE = ".ods.env"
load_dotenv(dotenv_path=ENV_FILE, verbose=True)
class ODSAPIClient:
"""Example Python ODS API Client for calling the NRAO ODS API"""
def __init__(self, client_id: str, client_secret: str):
self.client_id = client_id
self.client_secret = client_secret
self.token = os.environ.get("ODS_ACCESS_TOKEN")
def refresh_token(self) -> None:
"""Fetch a new access token from the API and persist to be used for
all subsequent requests. Only re-generate tokens as necessary, not per-request.
In production, use a secure storage mechanism for client credentials.
"""
resp: requests.Response = requests.post(
TOKEN_URL,
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
},
timeout=5,
)
if resp.status_code == 200:
new_token = resp.json()["token"]
set_key(ENV_FILE, "ODS_ACCESS_TOKEN", new_token)
self.token = new_token
else:
logger.info("Failed to refresh token")
resp.raise_for_status()
def ods_request_with_retry(
self,
method: Callable,
url: str,
files: Optional[Dict[str, BufferedReader]] = None,
):
"""
Make requests to the ODS API with exponential backoff
Reuse the existing token and rotate once the ODS API indicates it is no longer valid.
"""
retries: int = 0
response: Optional[requests.Response] = None
while retries < MAX_REQUEST_RETRIES:
if not self.token:
self.refresh_token()
response = method(
url,
headers={"Authorization": f"Bearer {self.token}"},
files=files,
timeout=5,
)
if response.status_code in RETRY_STATUS_CODES:
if response.status_code == 403:
# Reactively refresh token if we get a 403
logger.info("Refreshing token")
self.refresh_token()
delay: float = backoff_delay(2, retries)
logger.info("Retrying in {} seconds...", delay)
time.sleep(delay)
retries += 1
else:
response.raise_for_status()
return response
if response is not None:
response.raise_for_status()
return response
def get_ods_data(self) -> None:
"""Example request for getting ODS Data
Client with appropriate scope required.
"""
logger.info("Making request to get ODS Data")
api_response = self.ods_request_with_retry(requests.get, ODS_DATA_URL)
logger.info("Retrieved ODS data: ")
logger.info(api_response.json()["ods_data"])
def post_satellite_avoidance_logs(self, filepath: str) -> None:
"""Example function for uploading a directory of satellite avoidance logs to ODS
Client with appropriate scope required.
"""
if not os.path.isdir(filepath):
raise Exception(f"Expecting directory: {filepath}")
with os.scandir(filepath) as entries:
entries = [
entry
for entry in entries
if entry.name.endswith(".csv") or entry.name.endswith(".txt")
]
sorted_entries = sorted(entries, key=lambda entry: entry.name)
logger.info("Preparing to upload {} satellite avoidance logs...", len(entries))
for entry in sorted_entries:
with open(entry, "rb") as f:
files = {"file": f}
api_response = self.ods_request_with_retry(
requests.post, SAT_AVOIDANCE_LOG_URL, files
)
logger.info(
"Uploaded {} to ODS: {}", entry.name, api_response.json()
)
logger.info("Sleeping for ten seconds")
time.sleep(10)
def post_tle_file(self, filepath: str) -> None:
"""Example function for uploading a TLE file to ODS
Client with appropriate scope required.
"""
logger.info("Uploading {} to ODS".format(filepath))
with open(filepath, "rb") as f:
files = {"tle_file": f}
api_response = self.ods_request_with_retry(
requests.post, TLE_DATA_URL, files
)
logger.info("Uploaded TLE file to ODS: {}", api_response.json())
def backoff_delay(backoff_factor: int, attempts: int) -> float:
"""Add exponential backoff delay"""
return backoff_factor * (2 ** (attempts - 1))
def main(
get_ods_data: Optional[bool] = False,
tba_dir_path: Optional[str] = None,
tle_file_path: Optional[str] = None,
) -> None:
"""
Make requests to the ODS API to do things like getting ODS Data, uploading satellite avoidance logs,
or uploading TLE files.
"""
# Contact an NRAO ODS Admin to generate client credentials
# with applicable permissions. Store securely.
ods_client_id = os.getenv("ODS_CLIENT_ID")
ods_client_secret = os.getenv("ODS_CLIENT_SECRET")
api_client = ODSAPIClient(ods_client_id, ods_client_secret)
if get_ods_data:
api_client.get_ods_data()
if tba_dir_path:
api_client.post_satellite_avoidance_logs(tba_dir_path)
if tle_file_path:
api_client.post_tle_file(tle_file_path)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="A simple Python script to demonstrate talking to the ODS API."
)
parser.add_argument(
"--get_ods_data", action="store_true", help="Flag to get ODS Data"
)
parser.add_argument(
"--tba_dir_path",
type=str,
help="Optional path for a file of satellite avoidance logs to be uploaded to ODS",
)
parser.add_argument(
"--tle_file_path",
type=str,
help="Optional path to a TLE file to be uploaded to ODS",
)
args = parser.parse_args()
main(args.get_ods_data, args.tle_file_path, args.tba_dir_path)