utils/find_passes: Remove unused passid

This ID is unused, arbitrary and prevents this code to be parallized.
merge-requests/51/head
Fabian P. Schmidt 2019-11-02 20:07:37 +00:00 committed by cgbsat
parent 4f654502aa
commit 3577778eb8
7 changed files with 237 additions and 217 deletions

View File

@ -0,0 +1,2 @@
from .tle import Twolineelement
from .satellite import Satellite

View File

@ -0,0 +1,122 @@
import ephem
import math
from datetime import datetime, timedelta
def overlap(satpass, scheduledpasses, wait_time_seconds):
"""Check if this pass overlaps with already scheduled passes"""
# No overlap
overlap = False
# Add wait time
tr = satpass['tr']
ts = satpass['ts'] + timedelta(seconds=wait_time_seconds)
# Loop over scheduled passes
for scheduledpass in scheduledpasses:
# Test pass falls within scheduled pass
if tr >= scheduledpass['tr'] and ts < scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds):
overlap = True
# Scheduled pass falls within test pass
elif scheduledpass['tr'] >= tr and scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds) < ts:
overlap = True
# Pass start falls within pass
elif tr >= scheduledpass['tr'] and tr < scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds):
overlap = True
# Pass end falls within end
elif ts >= scheduledpass['tr'] and ts < scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds):
overlap = True
if overlap:
break
return overlap
def find_passes(satellite,
observer,
tmin,
tmax,
minimum_altitude,
min_pass_duration):
passes = []
# Set start time
observer.date = ephem.date(tmin)
# Load TLE
try:
sat_ephem = ephem.readtle(str(satellite.tle0), str(satellite.tle1), str(satellite.tle2))
except (ValueError, AttributeError):
return []
# Loop over passes
keep_digging = True
while keep_digging:
sat_ephem.compute(observer)
try:
tr, azr, tt, altt, ts, azs = observer.next_pass(sat_ephem)
except ValueError:
break # there will be sats in our list that fall below horizon, skip
except TypeError:
break # if there happens to be a non-EarthSatellite object in the list
except Exception:
break
if tr is None:
break
# using the angles module convert the sexagesimal degree into
# something more easily read by a human
try:
elevation = format(math.degrees(altt), '.0f')
azimuth_r = format(math.degrees(azr), '.0f')
azimuth_s = format(math.degrees(azs), '.0f')
except TypeError:
break
pass_duration = ts.datetime() - tr.datetime()
# show only if >= configured horizon and till tmax,
# and not directly overhead (tr < ts see issue 199)
if tr < ephem.date(tmax):
if (float(elevation) >= minimum_altitude and tr < ts and
pass_duration > timedelta(minutes=min_pass_duration)):
valid = True
# invalidate passes that start too soon
if tr < ephem.Date(datetime.now() + timedelta(minutes=5)):
valid = False
# get pass information
satpass = {
'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,
'mode': satellite.mode,
'scheduled': False
}
passes.append(satpass)
observer.date = ephem.Date(ts).datetime() + timedelta(minutes=1)
else:
keep_digging = False
return passes

View File

@ -0,0 +1,20 @@
class Satellite:
"""Satellite class"""
def __init__(self, tle, transmitter, success_rate, good_count, data_count, mode):
"""Define a satellite"""
self.tle0 = tle.tle0
self.tle1 = tle.tle1
self.tle2 = tle.tle2
self.id = tle.id
self.name = tle.name.strip()
self.transmitter = transmitter
self.success_rate = success_rate
self.good_count = good_count
self.data_count = data_count
self.mode = mode
def __repr__(self):
return "%s %s %d %d %d %s %s" % (self.id, self.transmitter, self.success_rate, self.good_count,
self.data_count, self.mode, self.name)

View File

@ -0,0 +1,50 @@
import logging
import random
from .pass_predictor import overlap
def ordered_scheduler(passes, scheduledpasses, wait_time_seconds):
"""Loop through a list of ordered passes and schedule each next one that fits"""
# Loop over passes
for satpass in passes:
# Schedule if there is no overlap with already scheduled passes
if not overlap(satpass, scheduledpasses, wait_time_seconds):
scheduledpasses.append(satpass)
return scheduledpasses
def random_scheduler(passes, scheduledpasses, wait_time_seconds):
"""Schedule passes based on random ordering"""
# Shuffle passes
random.shuffle(passes)
return ordered_scheduler(passes, scheduledpasses, wait_time_seconds)
def report_efficiency(scheduledpasses, passes):
if scheduledpasses:
# Loop over passes
start = False
for satpass in scheduledpasses:
if not start:
dt = satpass['ts'] - satpass['tr']
tmin = satpass['tr']
tmax = satpass['ts']
start = True
else:
dt += satpass['ts'] - satpass['tr']
if satpass['tr'] < tmin:
tmin = satpass['tr']
if satpass['ts'] > tmax:
tmax = satpass['ts']
# Total time covered
dttot = tmax - tmin
logging.info("%d passes selected out of %d, %.0f s out of %.0f s at %.3f%% efficiency" %
(len(scheduledpasses), len(passes), dt.total_seconds(), dttot.total_seconds(),
100 * dt.total_seconds() / dttot.total_seconds()))
else:
logging.info("No appropriate passes found for scheduling.")

View File

@ -0,0 +1,17 @@
class Twolineelement:
"""TLE class"""
def __init__(self, tle0, tle1, tle2):
"""Define a TLE"""
self.tle0 = tle0
self.tle1 = tle1
self.tle2 = tle2
if tle0[:2] == "0 ":
self.name = tle0[2:]
else:
self.name = tle0
if tle1.split(" ")[1] == "":
self.id = int(tle1.split(" ")[2][:4])
else:
self.id = int(tle1.split(" ")[1][:5])

View File

@ -8,10 +8,19 @@ import os
import lxml.html
import argparse
import logging
from utils import get_active_transmitter_info, get_transmitter_stats, \
get_groundstation_info, get_scheduled_passes_from_network, ordered_scheduler, \
report_efficiency, find_passes, schedule_observation, read_priorities_transmitters, \
get_satellite_info, update_needed, get_priority_passes
from utils import get_active_transmitter_info, \
get_transmitter_stats, \
get_groundstation_info, \
get_scheduled_passes_from_network, \
schedule_observation, \
read_priorities_transmitters, \
get_satellite_info, \
update_needed, \
get_priority_passes
from auto_scheduler import Twolineelement, Satellite
from auto_scheduler.pass_predictor import find_passes
from auto_scheduler.schedulers import ordered_scheduler, \
report_efficiency
import settings
from tqdm import tqdm
import sys
@ -19,47 +28,6 @@ import sys
_LOG_LEVEL_STRINGS = ['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG']
class twolineelement:
"""TLE class"""
def __init__(self, tle0, tle1, tle2):
"""Define a TLE"""
self.tle0 = tle0
self.tle1 = tle1
self.tle2 = tle2
if tle0[:2] == "0 ":
self.name = tle0[2:]
else:
self.name = tle0
if tle1.split(" ")[1] == "":
self.id = int(tle1.split(" ")[2][:4])
else:
self.id = int(tle1.split(" ")[1][:5])
class satellite:
"""Satellite class"""
def __init__(self, tle, transmitter, success_rate, good_count, data_count, mode):
"""Define a satellite"""
self.tle0 = tle.tle0
self.tle1 = tle.tle1
self.tle2 = tle.tle2
self.id = tle.id
self.name = tle.name.strip()
self.transmitter = transmitter
self.success_rate = success_rate
self.good_count = good_count
self.data_count = data_count
self.mode = mode
def __repr__(self):
return "%s %s %d %d %d %s %s" % (self.id, self.transmitter, self.success_rate, self.good_count,
self.data_count, self.mode, self.name)
def _log_level_string_to_int(log_level_string):
if log_level_string not in _LOG_LEVEL_STRINGS:
message = 'invalid choice: {0} (choose from {1})'.format(log_level_string,
@ -290,7 +258,7 @@ def main():
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)
Twolineelement(lines[i], lines[i + 1], lines[i + 2]) for i in range(0, len(lines), 3)
]
# Read transmitters
@ -303,10 +271,19 @@ def main():
item[0]), item[1], float(item[2]) / 100.0, int(item[3]), int(item[4]), item[5]
for tle in tles:
if tle.id == norad_cat_id:
satellites.append(satellite(tle, uuid, success_rate, good_count, data_count, mode))
satellites.append(Satellite(tle, uuid, success_rate, good_count, data_count, mode))
# Find passes
passes = find_passes(satellites, observer, tmin, tmax, min_culmination, min_pass_duration)
passes = []
logging.info('Finding all passes for %s satellites:' % len(satellites))
# Loop over satellites
for satellite in tqdm(satellites):
passes.extend(find_passes(satellite,
observer,
tmin,
tmax,
min_culmination,
min_pass_duration))
priorities, favorite_transmitters = read_priorities_transmitters(priority_filename)

170
utils.py
View File

@ -1,12 +1,8 @@
import requests
import logging
import math
import random
from datetime import datetime, timedelta
import ephem
from datetime import datetime
import lxml
import settings
from tqdm import tqdm
import os
import sys
@ -141,170 +137,6 @@ def get_scheduled_passes_from_network(ground_station, tmin, tmax):
return scheduledpasses
def overlap(satpass, scheduledpasses, wait_time_seconds):
"""Check if this pass overlaps with already scheduled passes"""
# No overlap
overlap = False
# Add wait time
tr = satpass['tr']
ts = satpass['ts'] + timedelta(seconds=wait_time_seconds)
# Loop over scheduled passes
for scheduledpass in scheduledpasses:
# Test pass falls within scheduled pass
if tr >= scheduledpass['tr'] and ts < scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds):
overlap = True
# Scheduled pass falls within test pass
elif scheduledpass['tr'] >= tr and scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds) < ts:
overlap = True
# Pass start falls within pass
elif tr >= scheduledpass['tr'] and tr < scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds):
overlap = True
# Pass end falls within end
elif ts >= scheduledpass['tr'] and ts < scheduledpass['ts'] + timedelta(
seconds=wait_time_seconds):
overlap = True
if overlap:
break
return overlap
def ordered_scheduler(passes, scheduledpasses, wait_time_seconds):
"""Loop through a list of ordered passes and schedule each next one that fits"""
# Loop over passes
for satpass in passes:
# Schedule if there is no overlap with already scheduled passes
if not overlap(satpass, scheduledpasses, wait_time_seconds):
scheduledpasses.append(satpass)
return scheduledpasses
def random_scheduler(passes, scheduledpasses, wait_time_seconds):
"""Schedule passes based on random ordering"""
# Shuffle passes
random.shuffle(passes)
return ordered_scheduler(passes, scheduledpasses, wait_time_seconds)
def report_efficiency(scheduledpasses, passes):
if scheduledpasses:
# Loop over passes
start = False
for satpass in scheduledpasses:
if not start:
dt = satpass['ts'] - satpass['tr']
tmin = satpass['tr']
tmax = satpass['ts']
start = True
else:
dt += satpass['ts'] - satpass['tr']
if satpass['tr'] < tmin:
tmin = satpass['tr']
if satpass['ts'] > tmax:
tmax = satpass['ts']
# Total time covered
dttot = tmax - tmin
logging.info("%d passes selected out of %d, %.0f s out of %.0f s at %.3f%% efficiency" %
(len(scheduledpasses), len(passes), dt.total_seconds(), dttot.total_seconds(),
100 * dt.total_seconds() / dttot.total_seconds()))
else:
logging.info("No appropriate passes found for scheduling.")
def find_passes(satellites, observer, tmin, tmax, minimum_altitude, min_pass_duration):
# Loop over satellites
passes = []
passid = 0
logging.info('Finding all passes for %s satellites:' % len(satellites))
for satellite in tqdm(satellites):
# Set start time
observer.date = ephem.date(tmin)
# Load TLE
try:
sat_ephem = ephem.readtle(str(satellite.tle0), str(satellite.tle1), str(satellite.tle2))
except (ValueError, AttributeError):
continue
# Loop over passes
keep_digging = True
while keep_digging:
sat_ephem.compute(observer)
try:
tr, azr, tt, altt, ts, azs = observer.next_pass(sat_ephem)
except ValueError:
break # there will be sats in our list that fall below horizon, skip
except TypeError:
break # if there happens to be a non-EarthSatellite object in the list
except Exception:
break
if tr is None:
break
# using the angles module convert the sexagesimal degree into
# something more easily read by a human
try:
elevation = format(math.degrees(altt), '.0f')
azimuth_r = format(math.degrees(azr), '.0f')
azimuth_s = format(math.degrees(azs), '.0f')
except TypeError:
break
passid += 1
pass_duration = ts.datetime() - tr.datetime()
# show only if >= configured horizon and till tmax,
# and not directly overhead (tr < ts see issue 199)
if tr < ephem.date(tmax):
if (float(elevation) >= minimum_altitude and tr < ts and
pass_duration > timedelta(minutes=min_pass_duration)):
valid = True
# invalidate passes that start too soon
if tr < ephem.Date(datetime.now() + timedelta(minutes=5)):
valid = False
# get pass information
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,
'mode': satellite.mode,
'scheduled': False
}
passes.append(satpass)
observer.date = ephem.Date(ts).datetime() + timedelta(minutes=1)
else:
keep_digging = False
return passes
def get_priority_passes(passes, priorities, favorite_transmitters, only_priority, min_priority):
priority = []
normal = []