add_sv_states from precise

Author: betaBison

gnss_lib_py/__init__.py

@@ -19,8 +19,8 @@
 from gnss_lib_py.utils.coordinates import *
 from gnss_lib_py.utils.ephemeris_downloader import *
 from gnss_lib_py.utils.filters import *
-from gnss_lib_py.utils.sv_models import *
 from gnss_lib_py.utils.gnss_models import *
+from gnss_lib_py.utils.sv_models import *
 from gnss_lib_py.utils.time_conversions import *
 from gnss_lib_py.utils.visualizations import *
 

gnss_lib_py/utils/sv_models.py

@@ -22,43 +22,100 @@
 from gnss_lib_py.utils.ephemeris_downloader import DEFAULT_EPHEM_PATH, load_ephemeris
 
 
-def add_sv_states(measurements, source = 'precise', file_paths = None,
+def add_sv_states(navdata, source = 'precise', file_paths = None,
                   download_directory = DEFAULT_EPHEM_PATH,
-                  constellations = None, delta_t_dec = -2,
                   verbose = True):
     """Add SV states to measurements using SP3 and CLK or Rinex files.
+
+    If source is 'precise' then will use SP3 and CLK files.
+
+    Parameters
+    ----------
+    navdata : gnss_lib_py.parsers.navdata.NavData
+        Instance of the NavData class that must include rows for
+        ``gps_millis``, ``gnss_id``, ``sv_id``, and ``raw_pr_m``.
+    source : string
+        The method used to compute SV states. If 'precise', then will
+        use SP3 and CLK precise files.
+    file_paths : list, string or path-like
+        Paths to existing ephemeris files if they exist.
+    download_directory : string or path-like
+        Directory where ephemeris files are downloaded if necessary.
+    verbose : bool
+        Prints extra debugging statements if true.
+
+    Returns
+    -------
+    navdata_w_sv_states : gnss_lib_py.parsers.navdata.NavData
+        Updated NavData class with satellite information computed.
+
     """
     if source == 'precise':
-        measurements_w_sv_states = add_sv_states_precise(measurements,
-                                                         sp3_clk_paths = file_paths,
-                                                         download_directory = download_directory,
-                                                         constellations = constellations,
-                                                         delta_t_dec = delta_t_dec,
-                                                         verbose = verbose)
+        navdata_w_sv_states = add_sv_states_precise(navdata,
+                                file_paths = file_paths,
+                                download_directory = download_directory,
+                                verbose = verbose)
     else:
         raise RuntimeError('Only Precise SV state estimation supported')
-    return measurements_w_sv_states
-
+    return navdata_w_sv_states
 
 
-def add_sv_states_precise(measurements, sp3_clk_paths = None,
+def add_sv_states_precise(navdata, file_paths = None,
                           download_directory = DEFAULT_EPHEM_PATH,
-                          constellations=None, delta_t_dec=-2,
                           verbose=True):
     """Add SV states to measurements using SP3 and CLK files.
 
-    Given received measurements, add SV states for measurements corresponding
-    to received time and SV ID using SP3 and CLK files. If receiver
-    position is given, that position is used to calculate the difference
-    between signal transmission and reception to find the SV states
-    corresponding to the time at which the signal was transmitted.
-
     Parameters
     ----------
-    """
-    return measurements
+    navdata : gnss_lib_py.parsers.navdata.NavData
+        Instance of the NavData class that must include rows for
+        ``gps_millis``, ``gnss_id``, ``sv_id``, and ``raw_pr_m``.
+    source : string
+        The method used to compute SV states. If 'precise', then will
+        use SP3 and CLK precise files.
+    file_paths : list, string or path-like
+        Paths to existing SP3 or CLK files if they exist.
+    download_directory : string or path-like
+        Directory where ephemeris files are downloaded if necessary.
+    verbose : bool
+        Prints extra debugging statements if true.
 
+    Returns
+    -------
+    navdata_w_sv_states : gnss_lib_py.parsers.navdata.NavData
+        Updated NavData class with satellite information computed.
 
+    """
+
+    # get unique gps_millis and constellations for ephemeris loader
+    unique_gps_millis = np.unique(navdata["gps_millis"])
+    constellations = np.unique(navdata["gnss_id"])
+
+    # load sp3 files
+    sp3_paths = load_ephemeris("sp3", gps_millis = unique_gps_millis,
+                               constellations=constellations,
+                               file_paths = file_paths,
+                               download_directory=download_directory,
+                               verbose=verbose
+                              )
+    sp3 = Sp3(sp3_paths)
+
+    # load clk files
+    clk_paths = load_ephemeris("clk", gps_millis = unique_gps_millis,
+                               constellations=constellations,
+                               file_paths = file_paths,
+                               download_directory=download_directory,
+                               verbose=verbose
+                              )
+    clk = Clk(clk_paths)
+
+    # add SV states using sp3 and clk
+    navdata_w_sv_states = single_gnss_from_precise_eph(navdata,
+                                                       sp3,
+                                                       clk,
+                                                       verbose=verbose)
+
+    return navdata_w_sv_states
 
 def add_sv_states_rinex(measurements, ephemeris_path= DEFAULT_EPHEM_PATH,
                   constellations=['gps'], delta_t_dec = -2):
@@ -709,9 +766,6 @@ def _find_delxyz_range(sv_posvel, rx_ecef):
     true_range = np.linalg.norm(del_pos, axis=0)
     return del_pos, true_range
 
-
-
-
 def single_gnss_from_precise_eph(navdata, sp3_parsed_file,
                                  clk_parsed_file, inplace=False,
                                  verbose = False):
@@ -862,39 +916,3 @@ def single_gnss_from_precise_eph(navdata, sp3_parsed_file,
     if inplace:
         return None
     return new_navdata
-
-def add_sv_states_sp3_and_clk(navdata, sp3_path, clk_path,
-                                inplace=False, verbose = False):
-    """Compute satellite information using .sp3 and .clk for multiple GNSS
-
-    Parameters
-    ----------
-    navdata : gnss_lib_py.parsers.navdata.NavData
-        Instance of the NavData class that depicts android derived dataset
-    sp3_path : path
-        File path for .sp3 file to extract precise ephemerides
-    clk_path : path
-        File path for .clk file to extract precise ephemerides
-    inplace : bool
-        If true, adds satellite positions and clock bias to the input
-        navdata object, otherwise returns a new NavData object with the
-        satellite rows added.
-    verbose : bool
-        Flag for whether to print intermediate steps useful
-        for debugging/reviewing (the default is False)
-
-    Returns
-    -------
-    navdata : gnss_lib_py.parsers.navdata.NavData
-        Updated NavData class with satellite information computed using
-        precise ephemerides from .sp3 and .clk files
-    """
-    sp3_parsed_gnss = Sp3(sp3_path)
-    clk_parsed_gnss = Clk(clk_path)
-    precise_navdata = single_gnss_from_precise_eph(navdata,
-                                                   sp3_parsed_gnss,
-                                                   clk_parsed_gnss,
-                                                   inplace = inplace,
-                                                   verbose = verbose)
-
-    return precise_navdata

tests/utils/test_sv_models.py

@@ -6,6 +6,7 @@
 __date__ = "21 Mar 2023"
 
 import os
+from datetime import datetime, timezone
 
 import pytest
 import numpy as np
@@ -16,6 +17,7 @@
 from gnss_lib_py.parsers.sp3 import Sp3
 from gnss_lib_py.parsers.navdata import NavData
 import gnss_lib_py.utils.sv_models as sv_models
+import gnss_lib_py.utils.time_conversions as tc
 from gnss_lib_py.parsers.android import AndroidDerived2021
 # pylint: disable=protected-access
 
@@ -452,22 +454,6 @@ def test_add_visible_svs_for_trajectory(android_gps_l1, ephemeris_path,
                                                 measure_frame_sv[row],
                                                 decimal=error_tol_dec['brd_eph'])
 
-@pytest.fixture(name="root_path")
-def fixture_root_path():
-    """Location of measurements for unit test
-
-    Returns
-    -------
-    root_path : string
-        Folder location containing measurements
-    """
-    root_path = os.path.dirname(
-                os.path.dirname(
-                os.path.dirname(
-                os.path.realpath(__file__))))
-    root_path = os.path.join(root_path, 'data/unit_test/')
-    return root_path
-
 @pytest.fixture(name="sp3_path")
 def fixture_sp3_path(root_path):
     """Filepath of valid .sp3 measurements
@@ -873,7 +859,7 @@ def test_gpscheck_sp3_eph(navdata_gpsl1, sp3data, clkdata, ephemeris_path):
 
 
 def test_compute_concat_precise_eph(navdata, sp3_path, clk_path):
-    """Tests that add_sv_states_sp3_and_clk does not fail for multi-GNSS
+    """Tests that single_gnss_from_precise_eph does not fail for multi-GNSS
 
     Notes
     -----
@@ -895,8 +881,11 @@ def test_compute_concat_precise_eph(navdata, sp3_path, clk_path):
     navdata_merged = NavData()
     navdata_merged = navdata.where('gnss_id',gnss_consts)
 
-    navdata_prcs_merged = sv_models.add_sv_states_sp3_and_clk(navdata, sp3_path,
-                                            clk_path,  verbose = True)
+    sp3 = Sp3(sp3_path)
+    clk = Clk(clk_path)
+
+    navdata_prcs_merged = sv_models.single_gnss_from_precise_eph(navdata,
+                                            sp3, clk, verbose=True)
 
     navdata_prcs_merged = navdata_prcs_merged.where("gnss_id",gnss_consts)
 
@@ -993,3 +982,78 @@ def test_compute_gps_brdcst_eph(navdata_gpsl1, navdata, navdata_glonassg1,
     pd.testing.assert_frame_equal(navdata_gpsl1_df.sort_index(axis=1),
                                   navdata_gpsl1_eph_df.sort_index(axis=1),
                                   check_dtype=False, check_names=True)
+
+@pytest.fixture(name="all_ephem_paths")
+def fixture_all_ephem_paths(root_path):
+    """Location of ephemeris files for unit test
+
+    Parameters
+    ----------
+    root_path : string
+        Location where ephemeris files are stored/to be downloaded.
+
+    Returns
+    -------
+    all_ephem_paths : string
+        Location of all unit test ephemeris files.
+
+    """
+
+    all_ephem_paths = []
+
+    ephem_dirs = [
+                       os.path.join(root_path,"rinex","nav"),
+                       os.path.join(root_path,"sp3"),
+                       os.path.join(root_path,"clk"),
+                      ]
+    for ephem_dir in ephem_dirs:
+        all_ephem_paths += [os.path.join(ephem_dir,file) \
+                            for file in os.listdir(ephem_dir)]
+
+    return all_ephem_paths
+
+def test_add_sv_states_precise(all_ephem_paths):
+    """Test adding SV states
+
+    Parameters
+    ----------
+    all_ephem_paths : string
+        Location of all unit test ephemeris files.
+
+    """
+
+    # create minimal test inputs
+    navdata = NavData()
+
+    datetimes = np.array([
+                          datetime(2020, 5, 17, 11, tzinfo=timezone.utc),
+                          datetime(2021, 4, 28, 12, tzinfo=timezone.utc),
+                          datetime(2023, 3, 14, 11, tzinfo=timezone.utc),
+                         ])
+    gps_millis = tc.datetime_to_gps_millis(datetimes)
+
+    navdata["gps_millis"] = gps_millis
+    navdata["gnss_id"] = np.array(["gps","glonass","qzss"])
+    navdata["sv_id"] = np.array([2,1,1])
+    navdata["raw_pr_m"] = np.array([26E3]*3)
+
+    sv_states = sv_models.add_sv_states(navdata,"precise",
+                                        file_paths=all_ephem_paths,
+                                        verbose=True)
+
+    assert not np.isnan(sv_states[["x_sv_m","y_sv_m","z_sv_m"]]).any()
+
+    # test minimal inputs
+
+def test_add_sv_states_fails():
+    """Test options which should fail.
+
+    """
+
+    navdata = NavData()
+
+    with pytest.raises(RuntimeError) as excinfo:
+        sv_models.add_sv_states(navdata, source="broadcast")
+
+    with pytest.raises(RuntimeError) as excinfo:
+        sv_models.add_sv_states(navdata, source="rinex")