neclib.recorders.necstdb_writer

Aliases

TextLike	None
Writer	None
get_logger	Get logger instance which prints operation logs to console.

parse_str_size(data)

Parameters:

data (Union[TextLike, List[TextLike]]) –

Return type:

Tuple[str, int]

str_to_bytes(data)

Parameters:

data (Union[TextLike, List[TextLike]]) –

Return type:

Union[bytes, List[bytes]]

class NECSTDBWriter(*args, **kwargs)

Bases: Writer

Dump data to NECSTDB.

Variables:

• db (necstdb.necstdb.necstdb or None) – Database instance this writer is currently dumping data to.

• tables (Dict[str, necstdb.necstdb.table]) – NECSTDB Table to which this writer is dumping data.

Examples

>>> writer = neclib.recorders.NECSTDBWriter()
>>> writer.start_recording("path/to/database/directory.necstdb")
>>> chunk = [
...     {"key": "timestamp", "type": "double", "value": 1664195057.022712},
...     {"key": "pressure", "type": "float32", "value": 850.5},
... ]
>>> writer.append("/observatory/pressure", chunk)
>>> writer.stop_recording()

LivelinessDuration: float = 15.0

If a table isn’t updated for this duration (in sec), it will be closed.

WarningQueueSize: int = 1000

Warn if number of data waiting for being dumped is greater than this.

DTypeConverters: Dict[str, Callable[[Any], Tuple[Any, str, int]]] = {'bool': <function NECSTDBWriter.<lambda>>, 'byte': <function NECSTDBWriter.<lambda>>, 'char': <function NECSTDBWriter.<lambda>>, 'double': <function NECSTDBWriter.<lambda>>, 'float': <function NECSTDBWriter.<lambda>>, 'float32': <function NECSTDBWriter.<lambda>>, 'float64': <function NECSTDBWriter.<lambda>>, 'int16': <function NECSTDBWriter.<lambda>>, 'int32': <function NECSTDBWriter.<lambda>>, 'int64': <function NECSTDBWriter.<lambda>>, 'int8': <function NECSTDBWriter.<lambda>>, 'string': <function NECSTDBWriter.<lambda>>, 'uint16': <function NECSTDBWriter.<lambda>>, 'uint32': <function NECSTDBWriter.<lambda>>, 'uint64': <function NECSTDBWriter.<lambda>>, 'uint8': <function NECSTDBWriter.<lambda>>}

Converter from readable type name to C data structure.

start_recording(record_dir)

Parameters:

record_dir (Path) –

Return type:

None

append(topic=None, chunk=None, *args, **kwargs)

Append a chunk of data.

Returns:

Whether the data is handled by this writer.

Return type:

handled

Parameters:

• topic (Optional[str]) –

• chunk (Optional[List[Dict[str, Any]]]) –

Examples

>>> chunk = [
        {"key": "timestamp", "type": "double", "value": 1664195057.022712},
        {"key": "reading", "type": "int32", "value": 5},
    ]
>>> writer.append("/meter_reading", chunk)

stop_recording()

Return type:

None

add_table(topic, metadata)

Parameters:

• topic (str) –

• metadata (Dict[str, Any]) –

Return type:

None

remove_table(topic)

Parameters:

topic (str) –

Return type:

None