es_sfgtools.pride_tools.gnss_product_operations module

es_sfgtools.pride_tools.gnss_product_operations.download(source: RemoteResourceFTP, dest: Path) → Path

Downloads a file from a remote FTP server to a local destination. :param source: An object containing the FTP server details, directory, and file name. :type source: RemoteResource :param dest: The local path where the file will be saved. :type dest: Path

Returns:

The local path where the file has been saved.

Return type:

Path

Raises:

ftplib.all_errors – If any FTP-related error occurs during the download process.
FileNotFoundError – If the downloaded file is empty (0 bytes).

Example

>>> source = WuhanIGS.get_rinex_3_nav(datetime.date(2021,1,1))
>>> updated_source = update_source(source)
>>> dest_dir = Path("dest/to/data")
>>> dest = dest_dir/updated_source.file_name
>>> download(source, dest)
"Downloading ftp://igs.gnsswhu.cn/pub/gps/data/daily/2021/001/21p/BRDC20210010000.rnx.gz" to dest/to/data/BRDC20210010000.rnx.gz"
>>> dest.exists()
True

es_sfgtools.pride_tools.gnss_product_operations.get_daily_rinex_url(date: date) → Dict[str, Dict[str, RemoteResourceFTP]]

This function returns the ‘RemoteResource’ for the IGS rinex observation file for a given date. url config docs at https://igs.org/products-access/#gnss-broadcast-ephemeris

Parameters:: date (datetime.date) – The date for which the rinex observation file is required.
Returns:: A dictionary containing the urls for the rinex 2 and 3 observation files.
Return type:: Dict[str,Dict[str,RemoteResource]]

Examples

>>> date = datetime.date(2021,1,1)
>>> urls = get_daily_rinex_url(date)
>>> str(urls["rinex_2"]["wuhan_gps"])
"ftp://igs.gnsswhu.cn/pub/gps/data/daily/21/001/21n/"
>>> str(urls["rinex_2"]["wuhan_glonass"])
"ftp://igs.gnsswhu.cn/pub/gps/data/daily/21/001/21g/"

Note

Until the remote resorces are updated with ‘update_source’, the file_name attribute will be None.

es_sfgtools.pride_tools.gnss_product_operations.get_gnss_common_products_urls(date: date) → Dict[str, Dict[str, Path]]

Retrieve GNSS common products for a given date. This function fetches various GNSS products (sp3, clk, bias, obx, erp) from different sources (WuhanIGS and CLSIGS) for the specified date. :param date: The date for which to retrieve the GNSS products. :type date: datetime.date

Returns:

A dictionary containing the GNSS products categorized by product type and source. Example structure:

{
    "sp3": {
        "wuhan": Path to WuhanIGS sp3 product,
        "cligs": Path to CLSIGS sp3 product,
    },
    "clk": {
        "wuhan": Path to WuhanIGS clk product,
        "cligs": Path to CLSIGS clk product,
    },
    "bias": {
        "wuhan": Path to WuhanIGS bias product,
        "cligs": Path to CLSIGS bias product,
    },
    "obx": {
        "wuhan": Path to WuhanIGS obx product,
        "cligs": Path to CLSIGS obx product,
    },
    "erp": {
        "wuhan": Path to WuhanIGS erp product,
        "cligs": Path to CLSIGS erp product,
    },
}

Return type:

Dict[str, Dict[str, Path]]

Note

Until the remote resources are updated with update_source, the file_name attribute will be None.

es_sfgtools.pride_tools.gnss_product_operations.get_gnss_products(rinex_path: Path, pride_dir: Path, override: bool = False, source: Literal['all', 'wuhan', 'cligs'] = 'all', date: date | datetime | None = None, override_config: bool = True) → Path | None

Generates or retrieves GNSS products for a given RINEX file or date and returns a pride config file path that catalogs the products.

Parameters:

rinex_path (Path) – The path to the RINEX file.
pride_dir (Path) – The directory where the PRIDE products are stored.
override (bool) – If True, the function will attempt to download the products even if they already exist.
source (Literal["all", "wuhan", "cligs"]) – The source from which to download the products. Defaults to “all”.
date (Optional[datetime.date | datetime.datetime]) – The date for which to retrieve the products. If provided, it will be used to determine the year and day of year (DOY) for the products. If rinex_path is provided, this will be ignored.
override_config (bool) – If True, the function will attempt to re-download the products even if a config file already exists.

Returns:

The path to the config file that catalogs the products, or None if the products could not be retrieved or generated.

Return type:

Path | None

es_sfgtools.pride_tools.gnss_product_operations.get_nav_file(rinex_path: Path, override: bool = False) → Path

Attempts to build a navigation file for a given RINEX file by downloading the necessary files from the IGS FTP server.

Parameters:

rinex_path (Path) – The path to the RINEX file.
override (bool) – If True, the function will attempt to download the navigation file even if it already exists.
mode (Literal['process','test']) – The mode in which the function is running. Test mode attempt downloads from all resources

Returns:

The path to the navigation file.

Return type:

brdm_path (Path)

Raises:

Warning – If the navigation file cannot be built or located.

Examples

>>> rinex_path = Path("data/NCB11750.23o")
>>> nav_path = get_nav_file(rinex_path)
Attempting to build nav file for data/NCB11750.23o
>>> nav_path
Path("data/BRDC00IGS_R_20231750000_01D_MN.rnx.gz")

es_sfgtools.pride_tools.gnss_product_operations.merge_broadcast_files(brdn: Path, brdg: Path, output_folder: Path) → Path

Merge GPS and GLONASS broadcast ephemerides into a single BRDM file. Functionality inspired by https://github.com/PrideLab/PRIDE-PPPAR/blob/master/scripts/merge2brdm.py

Parameters:

brdn (Path) – Path to the GPS broadcast ephemerides file.
brdg (Path) – Path to the GLONASS broadcast ephemerides file.
output_folder (Path) – Path to the output folder where the merged file will be saved.

Returns:

Path to the merged BRDM file.

Return type:

Path

Raises:

FileNotFoundError – If either brdn or brdg file does not exist.
Exception – If an unexpected error occurs while reading or writing the files.

Examples

>>> brdn = Path("data/brdc1500.21n")
>>> brdg = Path("data/brdc1500.21g")
>>> output_folder = Path("data")
>>> merged_brdm = merge_broadcast_files(brdn, brdg, output_folder)
>>> merged_brdm
Path("data/brdm1500.21p")

es_sfgtools.pride_tools.gnss_product_operations.uncompress_file(file_path: Path, dest_dir: Path | None) → Path

Decompresses a file using zlib and returns the path of the decompressed file. :param file: The path of the compressed file. :type file: Path

Returns:: The path of the decompressed file.
Return type:: Path
Raises:: FileNotFoundError – If the file does not exist.

Examples

>>> file = Path("data/brdc1500.21n.gz")
>>> uncompress_file(file)
Path("data/brdc1500.21n")

es_sfgtools.pride_tools.gnss_product_operations.update_source(source: RemoteResourceFTP) → RemoteResourceFTP

Get the contents of the directory on a remote FTP server and return the first file that matches the sorted remote query. :param source: An object containing the FTP server details, directory to list, and the remote query for matching files. :type source: RemoteResource

Returns:: The updated source object with the file_name attribute set to the first matching file in the sorted order, or None if no match is found.
Return type:: RemoteResource
Raises:: ftplib.all_errors – Any FTP-related errors encountered during the connection, login, or directory listing process.

Example

>>> remote_resource = WuhanIGS.get_rinex_3_nav(datetime.date(2021,1,1))
>>> remote_resource.file_name
None
>>> updated_remote_resource = update_source(remote_resource)
>>> updated_remote_resource.file_name
"BRDC20210010000.rnx.gz"