moncenterlib package

Subpackages

moncenterlib.gnss package

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], output_dir: 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.
output_dir (str) – Path to the output directory where the file will be saved.
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, "/some_output_dir", on_start="some_cmd")

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: str) → str

This method is used to start a connection. Importantly!!! 1. You should use a context manager to start a connection. The context manager will allow you to close the connection and write to the file automatically. If this is not done, there is a high risk that the process will not be killed, It will run until the computer is restarted.

2. Every connection have have key ‘process’. Value of this key is the subprocess object. if something goes wrong, you will have access to this object, which started the process of connecting and writing to the file.

When your program finished, all processes will be stopped.

4. It follows from the 3rd point that if you want connections to be constantly active, you need the program to work constantly. For example, you can use an infinite loop or something else.

When you start a connection which is constantly active connection will be stopped and started again.

See code usage examples in the examples folder.

Parameters:: name (str) – Name of the connection.
Returns:: Path to output file.
Return type:: str

Examples

>>> from moncenterlib.gnss.stream2file import Stream2File
>>> import time
>>> with Stream2File() as s2f:
...    s2f.add_connection("TEST",...)
...    s2f.start("TEST")
...    while True:
...        s2f.get_status("TEST")
...        time.sleep(2)

start_all() → list[str]

This method is used to start all connections.

Returns:: List of output files.
Return type:: list[str]

Examples

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

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")

stop_all() → None

This method is used to stop all connections.

Examples

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

moncenterlib package

Subpackages

moncenterlib.stream2file module

Module contents