-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapi.py
84 lines (69 loc) · 2.21 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import requests
import json
from dotenv import load_dotenv
import os
import time
load_dotenv()
# Space Track API Integration to fetch TLE data for LEOs
# Space Track credentials
USERNAME = os.getenv('SPACE_TRACK_USER')
PASSWORD = os.getenv('SPACE_TRACK_PASS')
base_url = "https://www.space-track.org"
# Query
QUERY_URL = os.getenv('GP_QUERY_URL')
# Login to Space-Track
def login():
url = f"{base_url}/ajaxauth/login"
payload = {
"identity": USERNAME,
"password": PASSWORD
}
session = requests.Session()
response = session.post(url, data=payload)
if response.status_code == 200:
print("Logged in.")
return session
else:
print(f"Login Error: {response.status_code}, Response: {response.text}")
return
# Perform API request
def get_data(session, url):
try:
response = session.get(url)
if response.status_code == 200:
tle_latest_data = response.json()
return tle_latest_data
elif response.status_code == 429: # API Throttling
print("API OVERLOAD")
time.sleep(60)
return get_data(session, url)
else:
print(f"Failed to retrieve data: {response.status_code}, Response: {response.text}")
return
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
return
# Verify login
# Pass Query and Session to fetch API data
def fetch_data():
session = login()
if not session:
return
url = QUERY_URL
debris_data = get_data(session, url)
if debris_data is None:
return
return debris_data
# Process TLE data containing
# TLE_0 - Contains object Name and Unique NORAD CAT ID
# TLE_1 - Contains object metadata (ID, EPOCH, BStar/Drag)
# TLE_2 - Contains orbital parameters (Inclination, RAAN/Equator Crossing, Eccentricity/Shape of orbit, Perigee/Altitude, Mean Motion/ Orbits per day)
def test_data(raw_data):
tle_data = []
for obj in raw_data:
tle_data.append({
"tle0": obj.get("TLE_LINE0", "Unknown"),
"tle1": obj.get("TLE_LINE1", "Unknown"),
"tle2": obj.get("TLE_LINE2", "Unknown")
})
return tle_data