moncenterlib.stream2file module

class moncenterlib.stream2file.Stream2File(logger: bool | Logger | None = None)

Bases: object

This class is used to convert a stream to a file. You can choose type of connections: serial, tcpcli and NTRIP. You can manage connections. Add and remove connections, start and stop connections. Also you can get status of connections.

This class use binary file of RTKLib library https://rtklib.com/. The name of file is str2str. This bin file was modified for this class.

See example code in example folder.

__init__(logger: bool | Logger | None = None) → None

Parameters:: 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.

add_connection(name: str, param: dict[str, str], on_start: str = '') → None

This method is used to add a connection. Data of connection stored in variable ‘connections’.

The parameters that you can use are listed below in the example.

Parameters:

name (str) – Name of the connection.
param (dict[str, str]) – Dictionary of parameters.
on_start (str, optional) – Command that will be send to host when the connection starts.

Raises:

ValueError – Path ‘output_dir’ to dir is strange.

Examples

>>> # example parameters
>>> {"type": "serial", "port": "COM1", "brate": "9600", "bsize": "8", "parity": "None", "stopb": "1", "fctr": "None"}
>>> {"type": "tcpcli", "addr": "1.2.3.4", "port": "123"}
>>> {"type": "ntrip", "user": "anonymous", "passwd": "anonymous", "addr": "1.2.3.4", "port": "123", "mntpnt": "AAAA"}
>>> # example code
>>> from moncenterlib.gnss.stream2file import Stream2File
>>> s2f = Stream2File()
>>> your_param = {"type": "..."}
>>> s2f.add_connection("test_connect", your_param, on_start="some_cmd")

create_output_file_name(name_conn: str, output_dir: str) → str

Create path to the output file. Format is ‘nameConnection_YYYYMMDD_HHMMSS.log’

Parameters:

name_conn (str) – Name of the connection.
output_dir (str) – Path to a directory where file will be save.

Returns:

Return path to the output file. Format is ‘nameConnection_YYYYMMDD_HHMMSS.log’

Return type:

str

get_connection_names() → list[str]

Get name of connections

Returns:: List name of connections
Return type:: list[str]

get_status(name: str) → dict[str, str]

This method is used to get the status of a connection. This method returns a dict. Where keys are the names of the metric, and values are the metric. If the metric doesn’t exist, the value will be empty str.

Parameters:

name (str) – Name of the connection.

Raises:

FileNotFoundError – File to get status doesn’t exist.

Returns:

The keys of the dict are:

time (str): Time when the data was collected.
byte (str): Number of bytes received.
bps (str): Bit per second.
connect (str): Status of connection.

Return type:

dict[str, str]

Examples

>>> from moncenterlib.gnss.stream2file import Stream2File
>>> s2f = Stream2File()
>>> s2f.add_connection("TEST", ...)
>>> s2f.start("TEST")
>>> s2f.get_status("TEST")

remove_connection(name: str) → None

This method is used to remove a connection. Data of connection remove from variable ‘connections’. When you remove a connection, the process will be stopped.

Parameters:: name (str) – Name of the connection.

Examples

>>> from moncenterlib.gnss.stream2file import Stream2File
>>> s2f = Stream2File()
>>> s2f.remove_connection("TEST")

start(name_con: str, name_file: str, timeout_time: int = 50000, reconnect_interval: int = 60000) → None

This method is used to start a connection. This method is blocking. It uses an infinite loop internally. Use the threading module to solve this problem or other tools. The connection can be stopped using the ‘stop’ method. See example codes how to use module threading with this method.

Parameters:

name_con (str) – Name of the connection.
name_file (str) – Path to the output file.
timeout_time (int) – Timeout time (ms).
reconnect_interval (int) – Reconnect interval (ms).

Returns:

Path to output file.

Return type:

str

Examples

>>> from moncenterlib.gnss.stream2file import Stream2File
>>> s2f = Stream2File()
>>> s2f.add_connection("TEST",...)
>>> s2f.start("TEST", "path_to_output_file")

stop(name: str) → None

This method is used to stop a connection. After you can start a connection again.

Parameters:: name (str) – Name of the connection.

Examples

>>> from moncenterlib.gnss.stream2file import Stream2File
>>> s2f = Stream2File()
>>> s2f.stop("TEST")