moncenterlib.gnss.tools4rnx module

A module for manipulating RINEX files. - Converting raw satellite receiver data into a universal RINEX format; - Soon (e.g. merge, cut rinex files).

The module has the following classes: - RtkLibConvbin; - Soon.

Learn more about the specific class.

class moncenterlib.gnss.tools4rnx.RtkLibConvbin(workers: int = 1, logger: bool | Logger | None = None)

Bases: object

This class is based on the RTKLib software package. Convert RTCM, receiver raw data log and RINEX file to RINEX and SBAS/LEX message file. SBAS message file complies with RTKLIB SBAS/LEX message format. See more about RTKLIB here: https://rtklib.com/ This class can convert one or more files. See code usage examples in the examples folder.

__init__(workers: int = 1, logger: bool | Logger | None = None)

Parameters:

workers (int) – The number of parallel processes. It reduces the processing time for multiple files. Defaults to 1.
logger (bool | Logger, optional) – if the logger is None, a logger will be created inside the default class. If the logger is False, then no information will be output. If you pass an instance of your logger, the information output will be implemented according to your logger. Defaults to None.

get_default_config() → dict

Return variable __default_config. Default config isn’t editable. In the future, you will manually configure this config and send it to the start method. See documentation RTKLIB, how to configuration.

Returns:: default config for convbin of RTKLib
Return type:: dict

get_full_status() → dict[str, dict[str, list]]

Returns the full lines of processing status information.

Returns:: The dictionary contains 2 keys. stderr, stdout. The stderr key stores the list lines of error information. The stdout key stores the list lines of information.
Return type:: dict[str, dict[str, list]]

Examples

>>> t4r = RtkLibConvib()
>>> t4r.start(list_files, "/some_output_dir", config, True)
>>> print(t4r.get_full_status())
{
    "/home/file1": {
        "stdout": ["Status: Running", "bla bla"],
        "stderr": ["Error: File not found", "bum bum"]
    }
    "/home/file2": {
        "stdout": ["ok"],
        "stderr": []
    }
}

get_last_status() → dict[str, dict]

Returns the last line of processing status information.

Returns:: The dictionary contains 3 keys. stderr, stdout, isStop. The stderr key stores the last line of error information. The stdout key stores the last line of information. The isStop key stores a boolean value. True if the process has stopped, False otherwise.
Return type:: dict[str, str | bool]

Examples

>>> t4r = RtkLibConvib()
>>> t4r.start(list_files, "/some_output_dir", config, True)
>>> While True:
>>>     status = t4r.get_last_status()
>>>     if status["isStop"]:
>>>         break
>>>     time.sleep(1)
>>>     print(status)
{
    "isStop": True,
    "stdout": "some info",
    "stderr": ""
}

merge_rinex(input_dir: str, output_dir: str, output_filename: str, config: dict, wait_process: bool = True)

Allows you to combine several RINEX files into one common file.

Parameters:

input_dir (str) – Path to the directory. There should definitely be an asterisk at the end of the path. For example /input_dir/*
output_dir (str) – Path to the output directory.
output_filename (str) – Name of the output file.
config (dict) – Dictionary with configuration. You can get the configuration by calling the get_default_config() method.
wait_process (bool) – If wait_process is set to False, the file merging process is performed in a separate thread. The method becomes non-blocking. If wait_process is True, then the process becomes blocking. The file merging process is waiting to be completed.

Examples

>>> t4r = RtkLibConvib()
>>> config = t4r.get_default_config()
>>> t4r.merge_rinex("/input_dir/*", "/output_dir", "name_file", config, True)
>>> print(t4r.output_files)

scan_dir(input_dir: str, recursion: bool = False) → list[str]

This method scans the directory and makes a list of files for further work of the class. The method can also recursively search for files.

Parameters:

input_dir (str) – Path to the directory.
recursion (bool, optional) – Recursively search for files. Defaults to False.

Raises:

ValueError – Path to dir is strange.

Returns:

List of files.

Return type:

list[str]

Examples

>>> t4r = RtkLibConvib()
>>> res = t4r.scan_dir("/some_path_to_dir", True)
>>> res
["file_1", "file_2"]

start(input_raw: str | list, output: str, config: dict, recursion: bool = False)

The method starts the process of preserving files in the RINEX format.This method is non-blocking. You can use an endless loop to wait for the process to complete. See the usage example.

Parameters:

input_raw (str | list) – The input method can accept a path of up to one file. The path to the directory where several files are stored. As well as a list of files generated by the scan_dirs method.
output (str) – The path to the directory where the files will be saved.
config (dict) – Dictionary with configuration. You can get the configuration by calling the get_default_config() method.
recursion (bool, optional) – When you put path to directory, you can use recursively search for files. Defaults to False.

Raises:

ValueError – Path to file or dir is strange.
TypeError – The type of the ‘input_raw’ variable should be ‘str’ or ‘list’.
ValueError – Path to output dir is strange.

Examples

>>> t4r = RtkLibConvib()
>>> list_files = t4r.scan_dir("/some_path_to_dir", True)
>>> config = t4r.get_default_config()
>>> t4r.start(list_files, "/some_output_dir", config, True)
>>> while True:
...     time.sleep(1)
...     status = t4r.get_last_status()
...     print(status)
...     states = [s['isStop'] for s in status.values()]
...     if all(states):
...         break
>>> print(t4r.output_files)
>>> # check output files
>>> from moncenterlib.tools import files_check
>>> return_files = {}
>>> for type_file, files in t4r.output_files.items():
...     return_files[type_file] = files_check(files)
>>> print(return_files)