spacecruft
/
satnogs-auto-scheduler
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fix code for pep8 via yapf 

Signed-off-by: Pierros Papadeas <pierros@papadeas.gr>
merge-requests/18/head
Pierros Papadeas 2019-05-10 10:33:54 +03:00
parent 289bad05cb
commit fee88c5fc1
No known key found for this signature in database
GPG Key ID: 8DB97129D9982991
3 changed files with 101 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
schedule_single_station.py
View File

@ -55,8 +55,8 @@ class satellite:
self.data_count = data_count
def __repr__(self):
return "%s %s %d %d %d %s" % (self.id, self.transmitter, self.success_rate,
self.good_count, self.data_count, self.name)
return "%s %s %d %d %d %s" % (self.id, self.transmitter, self.success_rate, self.good_count,
self.data_count, self.name)
def _log_level_string_to_int(log_level_string):
@ -77,27 +77,49 @@ def main():
parser = argparse.ArgumentParser(
description="Automatically schedule observations on a SatNOGS station.")
parser.add_argument("-s", "--station", help="Ground station ID", type=int)
parser.add_argument("-t", "--starttime", help="Start time (YYYY-MM-DD HH:MM:SS) [default: now]",
parser.add_argument("-t",
"--starttime",
help="Start time (YYYY-MM-DD HH:MM:SS) [default: now]",
default=datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S"))
parser.add_argument("-d", "--duration", help="Duration to schedule [hours; default 1.0]",
type=float, default=1)
parser.add_argument("-m", "--min-horizon", help="Minimum horizon [default 0]", type=float,
parser.add_argument("-d",
"--duration",
help="Duration to schedule [hours; default 1.0]",
type=float,
default=1)
parser.add_argument("-m",
"--min-horizon",
help="Minimum horizon [default 0]",
type=float,
default=0.)
parser.add_argument("-f", "--no-search-transmitters",
parser.add_argument("-f",
"--no-search-transmitters",
help="Do not search good transmitters [default searching]",
dest='search_transmitters', action='store_false')
dest='search_transmitters',
action='store_false')
parser.set_defaults(search_transmitters=True)
parser.add_argument("-w", "--wait",
parser.add_argument("-w",
"--wait",
help="Wait time between consecutive observations (for setup and slewing)" +
" [seconds; default: 0.0]", type=float, default=0)
parser.add_argument("-n", "--dryrun", help="Dry run (do not schedule passes)",
" [seconds; default: 0.0]",
type=float,
default=0)
parser.add_argument("-n",
"--dryrun",
help="Dry run (do not schedule passes)",
action="store_true")
parser.add_argument("-P", "--priorities", help="File with transmitter priorities. Should have" +
parser.add_argument("-P",
"--priorities",
help="File with transmitter priorities. Should have" +
"columns of the form |NORAD priority UUID| like |43017 0.9" +
" KgazZMKEa74VnquqXLwAvD|. Priority is fractional, one transmitter " +
"per line.", default=None)
parser.add_argument("-l", "--log-level", default="INFO", dest="log_level",
type=_log_level_string_to_int, nargs="?",
"per line.",
default=None)
parser.add_argument("-l",
"--log-level",
default="INFO",
dest="log_level",
type=_log_level_string_to_int,
nargs="?",
help="Set the logging output level. {0}".format(_LOG_LEVEL_STRINGS))
args = parser.parse_args()
@ -172,9 +194,11 @@ def main():
# Get NORAD IDs
norad_cat_ids = sorted(
set([transmitter["norad_cat_id"] for transmitter in transmitters.values()
if transmitter["norad_cat_id"] < settings.MAX_NORAD_CAT_ID and
transmitter["norad_cat_id"] in alive_norad_cat_ids]))
set([
transmitter["norad_cat_id"] for transmitter in transmitters.values()
if transmitter["norad_cat_id"] < settings.MAX_NORAD_CAT_ID
and transmitter["norad_cat_id"] in alive_norad_cat_ids
]))
# Store transmitters
fp = open(os.path.join(cache_dir, "transmitters_%d.txt" % ground_station_id), "w")
@ -189,12 +213,10 @@ def main():
if transmitters[uuid]["norad_cat_id"] not in alive_norad_cat_ids:
continue
fp.write("%05d %s %d %d %d\n" %
(transmitters[uuid]["norad_cat_id"],
uuid,
transmitter["stats"]["success_rate"],
transmitter["stats"]["good_count"],
transmitter["stats"]["total_count"]))
fp.write(
"%05d %s %d %d %d\n" %
(transmitters[uuid]["norad_cat_id"], uuid, transmitter["stats"]["success_rate"],
transmitter["stats"]["good_count"], transmitter["stats"]["total_count"]))
logging.info("Transmitter success rates received!")
fp.close()
@ -218,8 +240,9 @@ def main():
# Read tles
with open(os.path.join(cache_dir, "tles_%d.txt" % ground_station_id), "r") as f:
lines = f.readlines()
tles = [twolineelement(lines[i], lines[i + 1], lines[i + 2])
for i in range(0, len(lines), 3)]
tles = [
twolineelement(lines[i], lines[i + 1], lines[i + 2]) for i in range(0, len(lines), 3)
]
# Read transmitters
satellites = []
@ -231,12 +254,7 @@ def main():
item[0]), item[1], float(item[2]) / 100.0, int(item[3]), int(item[4])
for tle in tles:
if tle.id == norad_cat_id:
satellites.append(satellite(
tle,
uuid,
success_rate,
good_count,
data_count))
satellites.append(satellite(tle, uuid, success_rate, good_count, data_count))
# Find passes
passes = find_passes(satellites, observer, tmin, tmax, minimum_altitude)
@ -273,8 +291,7 @@ def main():
* satpass['success_rate'] \
* float(satpass['good_count']) / max_good_count
else:
satpass['priority'] = (
float(satpass['altt']) / 90.0) * satpass['success_rate']
satpass['priority'] = (float(satpass['altt']) / 90.0) * satpass['success_rate']
normalpasses.append(satpass)
# Priority scheduler
@ -302,15 +319,10 @@ def main():
for satpass in sorted(scheduledpasses, key=lambda satpass: satpass['tr']):
logging.info(
"%3d | %3.d | %05d | %s | %s | %3.0f | %4.6f | %s | %s" %
(ground_station_id,
satpass['scheduled'],
int(satpass['id']),
satpass['tr'].strftime("%Y-%m-%dT%H:%M:%S"),
satpass['ts'].strftime("%Y-%m-%dT%H:%M:%S"),
float(satpass['altt']) if satpass['altt'] else 0.,
satpass['priority'],
satpass['uuid'],
satpass['name'].rstrip()))
(ground_station_id, satpass['scheduled'], int(
satpass['id']), satpass['tr'].strftime("%Y-%m-%dT%H:%M:%S"),
satpass['ts'].strftime("%Y-%m-%dT%H:%M:%S"), float(satpass['altt']) if satpass['altt']
else 0., satpass['priority'], satpass['uuid'], satpass['name'].rstrip()))
if not satpass['scheduled']:
schedule_needed = True
@ -320,8 +332,7 @@ def main():
session = requests.session()
login = session.get(loginUrl) # Get login page for CSFR token
login_html = lxml.html.fromstring(login.text)
login_hidden_inputs = login_html.xpath(
r'//form//input[@type="hidden"]') # Get CSFR token
login_hidden_inputs = login_html.xpath(r'//form//input[@type="hidden"]') # Get CSFR token
form = {x.attrib["name"]: x.attrib["value"] for x in login_hidden_inputs}
form["login"] = settings.NETWORK_USERNAME
form["password"] = settings.NETWORK_PASSWORD
@ -329,8 +340,10 @@ def main():
# Login
result = session.post(loginUrl,
data=form,
headers={'referer': loginUrl,
'user-agent': 'satnogs-auto-scheduler/0.0.1'})
headers={
'referer': loginUrl,
'user-agent': 'satnogs-auto-scheduler/0.0.1'
})
if result.url.endswith("/accounts/login/"):
logging.info("Authentication failed")
else:
@ -342,18 +355,11 @@ def main():
logging.info('Checking and scheduling passes as needed.')
for satpass in tqdm(scheduledpasses_sorted):
if not satpass['scheduled']:
logging.debug(
"Scheduling %05d %s %s %3.0f %4.3f %s %s" %
(int(satpass['id']),
satpass['tr'].strftime("%Y-%m-%dT%H:%M:%S"),
satpass['ts'].strftime("%Y-%m-%dT%H:%M:%S"),
float(satpass['altt']),
satpass['priority'],
satpass['uuid'],
satpass['name'].rstrip()))
schedule_observation(session,
int(satpass['id']),
satpass['uuid'],
logging.debug("Scheduling %05d %s %s %3.0f %4.3f %s %s" %
(int(satpass['id']), satpass['tr'].strftime("%Y-%m-%dT%H:%M:%S"),
satpass['ts'].strftime("%Y-%m-%dT%H:%M:%S"), float(satpass['altt']),
satpass['priority'], satpass['uuid'], satpass['name'].rstrip()))
schedule_observation(session, int(satpass['id']), satpass['uuid'],
ground_station_id,
satpass['tr'].strftime("%Y-%m-%d %H:%M:%S") + ".000",
satpass['ts'].strftime("%Y-%m-%d %H:%M:%S") + ".000")

4
setup.cfg
View File

@ -33,3 +33,7 @@ install_requires =
max-complexity = 25
max-line-length = 100
ignore = F403,W504
[yapf]
based_on_style = pep8
column_limit = 100

84
utils.py
View File

@ -67,8 +67,7 @@ def get_active_transmitter_info(fmin, fmax):
for o in r.json():
if o["downlink_low"]:
if o["status"] == "active" and o["downlink_low"] > fmin and o["downlink_low"] <= fmax:
transmitter = {"norad_cat_id": o["norad_cat_id"],
"uuid": o["uuid"]}
transmitter = {"norad_cat_id": o["norad_cat_id"], "uuid": o["uuid"]}
transmitters.append(transmitter)
logging.info("Transmitters filtered based on ground station capability.")
return transmitters
@ -92,8 +91,7 @@ def get_scheduled_passes_from_network(ground_station, tmin, tmax):
while True:
if start:
r = client.get('{}/api/observations/?ground_station={:d}'.format(
settings.NETWORK_BASE_URL,
ground_station))
settings.NETWORK_BASE_URL, ground_station))
start = False
else:
nextpage = r.links.get("next")
@ -103,21 +101,14 @@ def get_scheduled_passes_from_network(ground_station, tmin, tmax):
for o in r.json():
satpass = {
"id": o['norad_cat_id'],
"tr": datetime.strptime(
o['start'].replace(
"Z",
""),
"%Y-%m-%dT%H:%M:%S"),
"ts": datetime.strptime(
o['end'].replace(
"Z",
""),
"%Y-%m-%dT%H:%M:%S"),
"tr": datetime.strptime(o['start'].replace("Z", ""), "%Y-%m-%dT%H:%M:%S"),
"ts": datetime.strptime(o['end'].replace("Z", ""), "%Y-%m-%dT%H:%M:%S"),
"scheduled": True,
"altt": o['max_altitude'],
"priority": 1,
"uuid": o['transmitter'],
"name": ''}
"name": ''
}
if satpass['ts'] > tmin and satpass['tr'] < tmax:
scheduledpasses.append(satpass)
@ -195,8 +186,7 @@ def efficiency(passes):
# Total time covered
dttot = tmax - tmin
return dt.total_seconds(), dttot.total_seconds(
), dt.total_seconds() / dttot.total_seconds()
return dt.total_seconds(), dttot.total_seconds(), dt.total_seconds() / dttot.total_seconds()
def find_passes(satellites, observer, tmin, tmax, minimum_altitude):
@ -210,9 +200,7 @@ def find_passes(satellites, observer, tmin, tmax, minimum_altitude):
# Load TLE
try:
sat_ephem = ephem.readtle(str(satellite.tle0),
str(satellite.tle1),
str(satellite.tle2))
sat_ephem = ephem.readtle(str(satellite.tle0), str(satellite.tle1), str(satellite.tle2))
except (ValueError, AttributeError):
continue
@ -246,30 +234,30 @@ def find_passes(satellites, observer, tmin, tmax, minimum_altitude):
if tr < ephem.date(tmax):
if (float(elevation) >= minimum_altitude and tr < ts):
valid = True
if tr < ephem.Date(datetime.now() +
timedelta(minutes=5)):
if tr < ephem.Date(datetime.now() + timedelta(minutes=5)):
valid = False
satpass = {'passid': passid,
'mytime': str(observer.date),
'name': str(satellite.name),
'id': str(satellite.id),
'tle1': str(satellite.tle1),
'tle2': str(satellite.tle2),
'tr': tr.datetime(), # Rise time
'azr': azimuth_r, # Rise Azimuth
'tt': tt.datetime(), # Max altitude time
'altt': elevation, # Max altitude
'ts': ts.datetime(), # Set time
'azs': azimuth_s, # Set azimuth
'valid': valid,
'uuid': satellite.transmitter,
'success_rate': satellite.success_rate,
'good_count': satellite.good_count,
'data_count': satellite.data_count,
'scheduled': False}
satpass = {
'passid': passid,
'mytime': str(observer.date),
'name': str(satellite.name),
'id': str(satellite.id),
'tle1': str(satellite.tle1),
'tle2': str(satellite.tle2),
'tr': tr.datetime(), # Rise time
'azr': azimuth_r, # Rise Azimuth
'tt': tt.datetime(), # Max altitude time
'altt': elevation, # Max altitude
'ts': ts.datetime(), # Set time
'azs': azimuth_s, # Set azimuth
'valid': valid,
'uuid': satellite.transmitter,
'success_rate': satellite.success_rate,
'good_count': satellite.good_count,
'data_count': satellite.data_count,
'scheduled': False
}
passes.append(satpass)
observer.date = ephem.Date(
ts).datetime() + timedelta(minutes=1)
observer.date = ephem.Date(ts).datetime() + timedelta(minutes=1)
else:
keep_digging = False
@ -283,9 +271,7 @@ def get_groundstation_info(ground_station_id):
# Loop
found = False
r = client.get("{}/api/stations/?id={:d}".format(
settings.NETWORK_BASE_URL,
ground_station_id))
r = client.get("{}/api/stations/?id={:d}".format(settings.NETWORK_BASE_URL, ground_station_id))
for o in r.json():
if o['id'] == ground_station_id:
if o['status'] == 'Online' or o['status'] == 'Testing':
@ -312,13 +298,7 @@ def get_last_update(fname):
return None
def schedule_observation(
session,
norad_cat_id,
uuid,
ground_station_id,
starttime,
endtime):
def schedule_observation(session, norad_cat_id, uuid, ground_station_id, starttime, endtime):
obsURL = '{}/observations/new/'.format(settings.NETWORK_BASE_URL) # Observation URL
# Get the observation/new/ page to get the CSFR token