-
Notifications
You must be signed in to change notification settings - Fork 0
/
manage.py
186 lines (169 loc) · 7.01 KB
/
manage.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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
import logging
from math import pi
import datetime
from itertools import zip_longest
from typing import Dict
from pathlib import Path
from functools import lru_cache
from collections import namedtuple
import httpx
from app.dbmodels import location, satellite
from app.resources import static_app
def load_initial_satellite_data():
"""
Download satellite database from Jonathan's Space Report and load non-decayed satellites into database
"""
import httpx, csv
from sqlalchemy.sql import select, exists, update
from .dbmodels import satellite, JSR_status_decayed
from .schemas import Satellite
from .database import engine
# r = httpx.get('https://planet4589.org/space/gcat/tsv/cat/satcat.tsv')
with open('satdb.csv', 'r') as f:
sat_reader = csv.DictReader(f)
k = 0
for sat in sat_reader:
sat = empty_string_to_None(sat) # convert empty string to None
k += 1
# if k > 40000: break
with engine.connect() as conn:
satid = int(sat['Satcat']) # NORAD ID
res = conn.execute(select([exists().where(satellite.c.id == satid)])).first()[0]
if (not res) and (sat['Status'] not in JSR_status_decayed):
# Satellite is not in database and is still in orbit, so add it to database
sat_data = Satellite(**{
'id': satid,
'cospar_id': sat['COSPAR_ID'],
'name': sat['Name'],
'decayed': False,
'launch_date': datetime.date.fromisoformat(sat['LaunchDate']) if sat['LaunchDate'] is not None else None,
'launch_year': sat['LaunchYear'],
'mass': sat['Mass'],
'length': sat['Length'],
'diameter': sat['Diameter'],
'span': sat['Span'],
'perigee': sat['Perigee'],
'apogee': sat['Apogee'],
'inclination': sat['Inc']
})
conn.execute(satellite.insert(), sat_data.dict())
msg = f"Added satellite {satid} to database"
logger.info(msg)
print(msg)
elif (res) and (sat['Status'] in JSR_status_decayed):
# Satellite is in database but is now decayed, so mark it decayed
conn.execute(satellite.update().where(satellite.c.id==satid).values(decayed=True))
msg = f"Marked satellite {satid} as decayed in database"
logger.info(msg)
print(msg)
def clean_city_data():
"""
Load world city location data from simplemaps.com, licensed by CC 4.0
https://simplemaps.com/static/data/world-cities/basic/simplemaps_worldcities_basicv1.6.zip
"""
import pathlib, csv, zipfile, io
url = 'https://simplemaps.com/static/data/world-cities/basic/simplemaps_worldcities_basicv1.6.zip'
r = requests.get(url)
z = zipfile.ZipFile(io.BytesIO(r.content))
z.extract('worldcities.csv')
data_dict = {}
fname = pathlib.Path('worldcities.csv')
with open(fname, 'r') as f:
city_reader = csv.reader(f, delimiter=',')
next(city_reader) # skip header row
for city in city_reader:
data_dict[city[0].lower()] = { # Unicode city name
'lat': float(city[2]),
'lon': float(city[3])
}
data_list = [{
'query': key,
'name': key,
'lat': data_dict[key]['lat'],
'lon': data_dict[key]['lat']
} for key in data_dict.keys()
]
# write cleaned output to file
with open('worldcities_cleaned.csv', 'w') as f:
writer = csv.DictWriter(f, fieldnames=['name', 'lat', 'lon'])
writer.writeheader()
writer.writerows(data_list)
def load_initial_city_data():
"""
Load world city location data from simplemaps.com, licensed by CC 4.0
https://simplemaps.com/static/data/world-cities/basic/simplemaps_worldcities_basicv1.6.zip
"""
import pathlib, csv, zipfile, io
from .dbmodels import engine, location
with open('worldcities_cleaned.csv', 'r') as f:
city_reader = csv.DictReader(f)
data = [city for city in city_reader]
with engine.connect() as conn:
conn.execute(location.insert(), data)
def download_orbit_type_data():
"""
Populate orbit type category table from JSR.
Ref: https://www.geeksforgeeks.org/convert-html-table-into-csv-file-in-python/
orbit_type = Table(
'orbit_type', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(75)), # eg. low earth orbit, geostationary orbit
Column('short_name', String(6)), # eg. LEO, GEO
Column('description', Text) # eg. period < 2hr, h > 600 km, incl. < 85deg, ecc ~= .5
)
"""
import csv, json
from bs4 import BeautifulSoup
import requests
from sqlalchemy.sql import select
from .dbmodels import orbit_type
from .database import engine
# download table from JSR website
r = requests.get('https://planet4589.org/space/gcat/web/intro/orbits.html')
soup = BeautifulSoup(r.text, 'html.parser')
data = []
rows = soup.find("table").find_all("tr")[1:]
for row in rows:
# remove trailing </td> and split by <td>
tmp = row.decode_contents().replace('</td>', '').split('<td>')[1:]
items = [item.strip() for item in tmp]
desc_array = []
if items[2]:
desc_array.append("period " + items[2].lower())
if items[3]:
desc_array.append("height " + items[3].lower())
if len(items)==5 and items[4]:
desc_array.append("inclination " + items[4].lower())
if len(items)==6 and items[5]:
desc_array.append("eccentricity " + items[5].lower())
data.append(
{
'short_name': items[0].strip(),
'name': items[1].strip(),
'description' : ", ".join(desc_array)
}
)
with open('orbit_type.tsv', 'w') as f:
writer = csv.DictWriter(f, fieldnames=['short_name', 'name', 'description'], delimiter='\t')
writer.writeheader()
writer.writerows(data)
def import_orbit_type_data():
import csv
from sqlalchemy.sql import select
from app.dbmodels import orbit_type
from app.resources import db
with open('orbit_type.tsv', 'r') as f:
data = csv.DictReader()
# json.dump(data, f)
# with engine.connect() as conn:
# for d in data:
# conn.execute(orbit_type.insert(), d)
# print(f"Inserted Orbit {d['short_name']}")
# def update_satellite_data():
# """
# Update satellite database with dataset from Jonathan's Space Report
# url = https://planet4589.org/space/gcat/tsv/cat/satcat.tsv
# """
# import requests, csv
# from sqlalchemy.sql import select, insert
# from .dbmodels import engine, satellite, JSR_status_decayed