cleanups

Author: haraschax

laika/astro_dog.py

@@ -4,7 +4,7 @@
 from typing import DefaultDict
 from collections.abc import Iterable
 
-from .constants import SECS_IN_DAY, SECS_IN_HR
+from .constants import SECS_IN_DAY
 from .helpers import ConstellationId, get_constellation, get_closest, get_el_az, TimeRangeHolder
 from .ephemeris import Ephemeris, EphemerisType, GLONASSEphemeris, GPSEphemeris, PolyEphemeris, parse_sp3_orbits, parse_rinex_nav_msg_gps, \
   parse_rinex_nav_msg_glonass
@@ -216,20 +216,9 @@ def download_parse_orbit(self, gps_time: GPSTime, skip_before_epoch=None) -> dic
       files = [self.fetch_count(f.result()) for f in futures if f.result()] if futures else []
       ephems = parse_sp3_orbits(files, self.valid_const, skip_before_epoch)
     return ephems
-    #{k: ephems_us.get(k, []) for k in set(list([]) + list(ephems_us.keys()))}
-
-  def download_parse_prediction_orbit(self, gps_time: GPSTime):
-    assert EphemerisType.ULTRA_RAPID_ORBIT in self.valid_ephem_types
-    skip_until_epoch = gps_time - 2 * SECS_IN_HR
-
-    result = [self.fetch_count(download_orbits_gps(t, self.cache_dir, self.valid_ephem_types)) for t in [gps_time - SECS_IN_DAY, gps_time]]
-    return parse_sp3_orbits(result, self.valid_const, skip_until_epoch=skip_until_epoch)
 
   def get_orbit_data(self, time: GPSTime, only_predictions=False):
-    if only_predictions:
-      ephems_sp3 = self.download_parse_prediction_orbit(time)
-    else:
-      ephems_sp3 = self.download_parse_orbit(time)
+    ephems_sp3 = self.download_parse_orbit(time)
     if sum([len(v) for v in ephems_sp3.values()]) < 5:
       raise RuntimeError(f'No orbit data found. For Time {time.as_datetime()} constellations {self.valid_const} valid ephem types {self.valid_ephem_types}')
     self.add_ephem_fetched_time(ephems_sp3, self.orbit_fetched_times)

laika/downloader.py

@@ -27,12 +27,6 @@
 # mirror of sftp://gdc.cddis.eosdis.nasa.gov/gnss/data/hourly
 CDDIS_HOURLY_BASE_URL = os.getenv("CDDIS_HOURLY_BASE_URL", "https://raw.githubusercontent.com/commaai/gnss-data-hourly/master")
 
-# mirror of ftp://ftp.glonass-iac.ru
-GLONAS_IAC_BASE_URL = os.getenv("GLONAS_IAC_BASE_URL", "https://raw.githubusercontent.com/commaai/gnss-data-alt/master")
-
-# no mirror
-IGN_BASE_URL = os.getenv("IGN_BASE_URL", "ftp://igs.ign.fr/pub")
-
 
 class DownloadFailed(Exception):
   pass
@@ -325,36 +319,33 @@ def download_nav(time: GPSTime, cache_dir, constellation: ConstellationId):
 def download_orbits_gps(time, cache_dir, ephem_types):
   url_bases = (
     mirror_url(CDDIS_BASE_URL, '/gnss/products/'),
-    mirror_url(IGN_BASE_URL, '/igs/products/'),
+    mirror_url(CDDIS_BASE_URL, '/glonass/products/'),
   )
 
+  folder_path = "%i/" % time.week
+  filenames = []
+
   if time.week < 2238:
+    assert EphemerisType.FINAL_ORBIT in ephem_types, "Only final orbits are available before 2238"
     compression = '.Z'
-    ephem_strs = {
-      EphemerisType.FINAL_ORBIT: ['igs{wwww}{dow}.sp3'.format(wwww=time.week, dow=time.dow)],
-      EphemerisType.RAPID_ORBIT: ['igr{wwww}{dow}.sp3'.format(wwww=time.week, dow=time.dow)],
-      EphemerisType.ULTRA_RAPID_ORBIT: ['igu{wwww}{dow}_{hh}.sp3'.format(wwww=time.week, dow=time.dow, hh=hour) for hour in ['18', '12', '06', '00']]
-    }
+    filenames.extend(['igs{wwww}{dow}.sp3'.format(wwww=time.week, dow=time.dow)])
   else:
     # TODO deal with version number
     compression = '.gz'
     ephem_strs =  {
-      EphemerisType.FINAL_ORBIT: ['IGS0OPSFIN_{yyyy}{doy:03d}0000_01D_15M_ORB.SP3'.format(yyyy=time.year, doy=time.doy)],
-      EphemerisType.RAPID_ORBIT: ['IGS0OPSRAP_{yyyy}{doy:03d}0000_01D_15M_ORB.SP3'.format(yyyy=time.year, doy=time.doy)],
-      EphemerisType.ULTRA_RAPID_ORBIT: ['IGS0OPSULT_{yyyy}{doy:03d}{hh}00_02D_15M_ORB.SP3'.format(yyyy=time.year, doy=time.doy, hh=hour) \
+      EphemerisType.FINAL_ORBIT: ['COD0OPSFIN_{yyyy}{doy:03d}0000_01D_05M_ORB.SP3'.format(yyyy=time.year, doy=time.doy)],
+      EphemerisType.RAPID_ORBIT: ['COD0OPSRAP_{yyyy}{doy:03d}0000_01D_05M_ORB.SP3'.format(yyyy=time.year, doy=time.doy)],
+      EphemerisType.ULTRA_RAPID_ORBIT: ['COD0OPSULT_{yyyy}{doy:03d}{hh}00_02D_05M_ORB.SP3'.format(yyyy=time.year, doy=time.doy, hh=hour) \
         for hour in ['18', '12', '06', '00']],
     }
 
-  folder_path = "%i/" % time.week
-  filenames = []
-
-  # Download filenames in order of quality. Final -> Rapid -> Ultra-Rapid(newest first)
-  if EphemerisType.FINAL_ORBIT in ephem_types and GPSTime.from_datetime(datetime.utcnow()) - time > 3 * SECS_IN_WEEK:
-    filenames.extend(ephem_strs[EphemerisType.FINAL_ORBIT])
-  if EphemerisType.RAPID_ORBIT in ephem_types and GPSTime.from_datetime(datetime.utcnow()) - time > 3 * SECS_IN_DAY:
-    filenames.extend(ephem_strs[EphemerisType.RAPID_ORBIT])
-  if EphemerisType.ULTRA_RAPID_ORBIT in ephem_types:
-    filenames.extend(ephem_strs[EphemerisType.ULTRA_RAPID_ORBIT])
+    # Download filenames in order of quality. Final -> Rapid -> Ultra-Rapid(newest first)
+    if EphemerisType.FINAL_ORBIT in ephem_types and GPSTime.from_datetime(datetime.utcnow()) - time > 3 * SECS_IN_WEEK:
+      filenames.extend(ephem_strs[EphemerisType.FINAL_ORBIT])
+    if EphemerisType.RAPID_ORBIT in ephem_types and GPSTime.from_datetime(datetime.utcnow()) - time > 3 * SECS_IN_DAY:
+      filenames.extend(ephem_strs[EphemerisType.RAPID_ORBIT])
+    if EphemerisType.ULTRA_RAPID_ORBIT in ephem_types:
+      filenames.extend(ephem_strs[EphemerisType.ULTRA_RAPID_ORBIT])
 
   folder_file_names = [(folder_path, filename) for filename in filenames]
   ret = download_and_cache_file_return_first_success(url_bases, folder_file_names, cache_dir+'cddis_products/', compression=compression)
@@ -388,7 +379,6 @@ def download_dcb(time, cache_dir):
   folder_paths = []
   url_bases = (
     mirror_url(CDDIS_BASE_URL, '/gnss/products/bias/'),
-    mirror_url(IGN_BASE_URL, '/igs/products/mgex/dcb/'),
   )
   # seem to be a lot of data missing, so try many days
   for time_step in [time - i * SECS_IN_DAY for i in range(14)]:

tests/test_ephemerides.py

@@ -33,6 +33,7 @@ def test_nav_vs_orbit_old(self):
     dog_nav = AstroDog(valid_ephem_types=EphemerisType.NAV)
     for gps_time in gps_times:
       for svId in svIds:
+        print(svId, gps_time)
         sat_info_nav = dog_nav.get_sat_info(svId, gps_time)
         assert sat_info_nav is not None
         sat_info_orbit = dog_orbit.get_sat_info(svId, gps_time)

tests/test_prediction_orbits.py

@@ -10,20 +10,19 @@
 class TestPredictionOrbits(unittest.TestCase):
 
   def test_gps(self):
-    available_date = GPSTime.from_datetime(datetime(2020, 5, 1, 12))
+    available_date = GPSTime.from_datetime(datetime.now())
     dog = AstroDog(valid_const=(ConstellationId.GPS,), valid_ephem_types=EphemerisType.ULTRA_RAPID_ORBIT)
-    dog.get_orbit_data(available_date, only_predictions=True)
+    dog.get_orbit_data(available_date)
     self.assertGreater(len(dog.orbits.keys()), 0)
     self.assertTrue(available_date in dog.orbit_fetched_times)
 
-  def test_gps_and_glonass_2022(self):
-    # Test GPS and GLONASS separately from the first date that GLONASS Ultra-Rapid prediction orbits were available
-    available_date = GPSTime.from_datetime(datetime(2022, 1, 29, 11, 31))
+  def test_glonass(self):
+    available_date = GPSTime.from_datetime(datetime.now())
     for t in range(0, 24, 3):
       check_date = available_date + t * SECS_IN_HR
-      for const in [ConstellationId.GPS, ConstellationId.GLONASS]:
+      for const in [ConstellationId.GLONASS]:
         dog = AstroDog(valid_const=(const,), valid_ephem_types=EphemerisType.ULTRA_RAPID_ORBIT)
-        dog.get_orbit_data(check_date, only_predictions=True)
+        dog.get_orbit_data(check_date)
         self.assertGreater(len(dog.orbits.keys()), 0)
         self.assertTrue(check_date in dog.orbit_fetched_times)