spacecruft
/
satnogsmap
1
0
Fork 0
Code Releases Activity
satnogsmap/satnogs.py

210 lines
8.4 KiB
Python
Raw Blame History

from datetime import datetime , timedelta
import requests
from flask import Flask , render_template,redirect,url_for
import json
from apscheduler.schedulers.background import BackgroundScheduler
import ephem
from satnogs_api_client import fetch_satellites, DB_BASE_URL,fetch_tle_of_observation
from satellite_tle import fetch_tles
scheduler = BackgroundScheduler()
app = Flask(__name__)
modes = {1: 'FM', 2: 'AFSK1k2', 5: 'SSTV', 6: 'CW', 7: 'FMN', 9: 'USB', 15: 'GFSK4k8', 17: 'AHRPT', 18: 'AFSK9k6', 19: 'AM', 20: 'LSB', 21: 'FSK1k2', 22: 'FSK9k6', 26: 'GFSK1k2', 27: 'GFSK2k4', 28: 'GFSK9k6', 29: 'GFSK19k2', 30: 'MSK1k2', 31: 'MSK2k4', 32: 'MSK4k8', 33: 'MSK9k6', 34: 'MSK19k2', 35: 'MSK38k4', 36: 'GMSK1k2', 37: 'GMSK2k4', 38: 'GMSK4k8', 39: 'GMSK9k6', 40: 'PSK31', 41: 'PSK63', 42: 'QPSK31', 43: 'QPSK63', 44: 'APT', 45: 'HRPT', 46: 'FSK4k8', 47: 'BPSK1k2', 48: 'GMSK19k2', 49: 'AFSK', 50: 'BPSK', 51: 'FSK19k2', 52: 'BPSK115k2', 53: 'LRPT', 54: 'BPSK9k6', 55: 'FFSK1k2', 56: 'FSK2k4', 57: 'DSTAR', 58: 'DUV', 59: 'CERTO', 60: 'BPSK400', 61: 'OFDM', 62: 'QPSK38k4'}
Passes = []
Stations = []
TLEs = {}
class Pass:
id = 0
start = None
end = None
ground_station = None
satellite = None
transmitter = None
def __repr__(self):
return "\n<ID>: {}\n<Start>: {}\n<End>: {}\n<Ground Station>: {}\n<Satellite> {}\n<Transmitter>: {}".format(self.id,self.start.strftime('%Y-%m-%dT%H:%M:%S%z'),self.end.strftime('%Y-%m-%dT%H:%M:%S%z'),self.ground_station,json.dumps(self.satellite,indent = 1),json.dumps(self.transmitter,indent=1))
def getActive():
start = (datetime.utcnow() - timedelta(0,0,0,0,20)).strftime('%Y-%m-%dT%H:%M:%S%z')
end = (datetime.utcnow() + timedelta(0,0,0,0,30)).strftime('%Y-%m-%dT%H:%M:%S%z')
passesR = requests.get("https://network.satnogs.org/api/observations/?end="+end+"&format=json&start="+start)
passes = passesR.json()
if passesR.links.has_key("next"):
while passesR.links.has_key("next"):
passesR = requests.get(passesR.links["next"]["url"])
passes += passesR.json()
ground_stations = {}
for x in passes:
if datetime.strptime(x["start"],'%Y-%m-%dT%H:%M:%Sz') > datetime.utcnow() or datetime.strptime(x["end"],'%Y-%m-%dT%H:%M:%Sz') < datetime.utcnow():
passes.remove(x)
else:
if ground_stations.has_key(x["ground_station"]):
ground_stations[x["ground_station"]].append(x)
else:
ground_stations[x["ground_station"]] = []
ground_stations[x["ground_station"]].append(x)
passes = []
for x in ground_stations:
start = datetime.utcnow()
current = {"start":datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S%z')+"z"}
for y in ground_stations[x]:
if datetime.strptime(y["start"],'%Y-%m-%dT%H:%M:%Sz') < datetime.strptime(current["start"],'%Y-%m-%dT%H:%M:%Sz'):
current = y
passes.append(current)
Passes =[]
for x in passes:
temp = Pass()
temp.id = x["id"]
temp.start = datetime.strptime(x["start"],'%Y-%m-%dT%H:%M:%Sz')
temp.end = datetime.strptime(x["end"],'%Y-%m-%dT%H:%M:%Sz')
temp.ground_station = x["ground_station"]
temp.satellite = requests.get("https://db.satnogs.org/api/satellites/"+str(x["norad_cat_id"])) .json()
temp.transmitter = requests.get("https://db.satnogs.org/api/transmitters/"+x["transmitter"]).json()
Passes.append(temp)
return Passes
def GetGroundStations():
stationsR = requests.get("https://network.satnogs.org/api/stations/")
stations = stationsR.json()
while stationsR.links.has_key("next"):
stationsR = requests.get(stationsR.links["next"]["url"])
stations += stationsR.json()
for x in stations:
if x["last_seen"] == None:
stations.remove(x)
continue
if datetime.strptime(x["last_seen"],'%Y-%m-%dT%H:%M:%Sz') < (datetime.utcnow()- timedelta(10,0,0,0)):
stations.remove(x)
for x in stations:
if x["last_seen"] == None:
stations.remove(x)
continue
if datetime.strptime(x["last_seen"],'%Y-%m-%dT%H:%M:%Sz') < (datetime.utcnow()- timedelta(10,0,0,0)):
stations.remove(x)
return stations
@scheduler.scheduled_job('interval',minutes=3)
def updatePasses():
global Passes
print "Updating Passes"
Passes = getActive()
@scheduler.scheduled_job('interval',hours=10)
def updateStations():
global Stations
print "Updating Stations"
Stations = GetGroundStations()
@scheduler.scheduled_job('interval',days=1)
def updateTLE():
print "Updating TLE"
global TlEs
sats = fetch_satellites(url=DB_BASE_URL, max_satellites=None)
satnogs_db_norad_ids = set(sat['norad_cat_id'] for sat in sats if sat['status'] != 're-entered')
# Remove satellites with temporary norad ids
temporary_norad_ids = set(filter(lambda norad_id: norad_id >= 99900, satnogs_db_norad_ids))
satnogs_db_norad_ids = satnogs_db_norad_ids - temporary_norad_ids
# Fetch TLEs for the satellites of interest
tles = fetch_tles(satnogs_db_norad_ids)
TLEs = {}
for norad_id, (source, tle) in tles.items():
TLEs[norad_id] = [str(tle[0]),str(tle[1]),str(tle[2])]
print('\nTLEs for {} of {} requested satellites found ({} satellites with temporary norad ids skipped).'.format(len(tles), len(satnogs_db_norad_ids), len(temporary_norad_ids)))
@app.route('/')
def index():
return 'Satnogs Network Info<br>Updated every 5 min<br><a href="/occuringobservations">IDs of all occuring observations with links to the obs</a><br><a href="/api/activestations">IDs of all stations actively in an observation</a><br><a href="/api/onlinestations">All online stations</a><br><a href="/api/occuringobservations">IDs of all occuring observations</a>'
@app.route("/map_view")
def map_view():
stations = []
for x in Stations:
stations.append({'id':x['id'],'name':x['name'],'lat_lng':[x["lat"],x['lng']]})
return render_template("map.html",stations = stations)
@app.route('/occuringobservations')
def occuring_observations():
obs = []
for x in Passes:
obs.append("<a href='https://network.satnogs.org/observations/"+str(x.id)+"'>"+str(x.id)+"</a>")
return json.dumps(obs)
@app.route('/api/activestations')
def api_active_stations():
stations = []
for x in Passes:
stations.append(x.ground_station)
return json.dumps(stations)
@app.route('/api/onlinestations')
def api_online_stations():
stations = []
for x in Stations:
stations.append(x["id"])
return json.dumps(stations)
@app.route('/api/occuringobservations')
def api_occuring_observations():
obs = []
for x in Passes:
obs.append(x.id)
return json.dumps(obs)
@app.route('/api/occuringsats')
def api_occuring_sats():
obs = {}
for x in Passes:
if x.satellite['norad_cat_id'] not in TLEs.keys():
q = fetch_tle_of_observation(x.id)
TLEs[ x.satellite['norad_cat_id'] ] = [str(x.satellite["name"]),str(q[0]),str(q[1])]
satellite = ephem.readtle(TLEs[x.satellite['norad_cat_id']][0],TLEs[x.satellite['norad_cat_id']][1],TLEs[x.satellite['norad_cat_id']][2])
now = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
satellite.compute(now)
lat = satellite.sublat*57.295779514
long = satellite.sublong*57.295779514
obs[x.satellite['norad_cat_id']] = {"name":x.satellite["name"],"transmitter_name":x.transmitter["description"],"transmitter_downlink":x.transmitter["downlink_low"],"transmitter_mode":modes[x.transmitter["mode_id"]],"lat_lng":[lat,long]}
return json.dumps(obs)
@app.route('/api/satstationpairs')
def api_sat_station_pairs():
pairs = []
for x in Passes:
pairs.append([x.ground_station,x.satellite['norad_cat_id']])
return json.dumps(pairs)
@app.route('/api/getsatloc/<int:norad>')
def get_sat_loc(norad):
satellite = ephem.readtle(TLEs[norad][0],TLEs[norad][1],TLEs[norad][2])
now = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')
satellite.compute(now)
lat = satellite.sublat*57.295779514
long = satellite.sublong*57.295779514
return json.dumps([lat,long])
updatePasses()
updateStations()
updateTLE()
scheduler.start()
app.run(use_reloader=False,host = "0.0.0.0")
View Git Blame Copy Permalink