obci.utils package

Submodules

obci.utils.context module

obci.utils.context.get_dummy_context(name)

obci.utils.context.get_new_context()

obci.utils.debug module

obci.utils.debug.print_asyncio_tasks(title=None)

Print all asyncio tasks for current message loop.

obci.utils.debug.print_threads(title=None)

Print callstacks for all threads.

obci.utils.debug_helper module

obci.utils.debug_helper.get_str_variable_vector(vect)

obci.utils.filesystem module

obci.utils.filesystem.checkpidfile(file)

obci.utils.filesystem.getpidfile(file)

obci.utils.filesystem.is_exe(fpath)

obci.utils.filesystem.removepidfile(file)

obci.utils.filesystem.which(program)

obci.utils.filesystem.which_binary(program)

obci.utils.log_mx_handler module

This module defines logging handler class for Multiplexer peers, a LogMXHandler.

class obci.utils.log_mx_handler.LogMXHandler(conn)

Bases: logging.Handler

The handler pickles log records and sends them to the multiplexer as OBCI_LOG_MESSAGE. The records should be picked up by a log collector mx peer.

close()

emit(record)

Emit a record.

Writes the LogRecord to the queue, preparing it for pickling first.

obci.utils.openbci_logging module

Module defines a single method get_logger that returns logger with set logging level. Change logging.INFO lines to change logging level.

obci.utils.openbci_logging.caller_name(skip=2)

Get a name of a caller in the format module.class.method

skip specifies how many levels of stack to skip while getting caller name. skip=1 means “who calls me”, skip=2 “who calls my caller” etc.

An empty string is returned if skipped levels exceed stack height copied from gist: https://gist.github.com/techtonik/2151727

obci.utils.openbci_logging.console_formatter()

obci.utils.openbci_logging.crash_log_data(exception, callee)

obci.utils.openbci_logging.crash_log_msg(func, args, kwargs, callee=None, exc_info=None, exception=None)

obci.utils.openbci_logging.crash_log_tags(exception, callee)

obci.utils.openbci_logging.file_formatter()

obci.utils.openbci_logging.get_dummy_logger(p_name, p_level='info')

Return logger with p_name as name. And logging level p_level. p_level should be in (starting with the most talkactive): ‘debug’, ‘info’, ‘warning’, ‘error’, ‘critical’.

obci.utils.openbci_logging.get_logger(name, file_level='debug', stream_level='warning', mx_level='warning', sentry_level='error', conn=None, log_dir=None, obci_peer=None)

Return logger with name as name. And logging level p_level. p_level should be in (starting with the most talkactive): ‘debug’, ‘info’, ‘warning’, ‘error’, ‘critical’.

obci.utils.openbci_logging.log_crash(func)

obci.utils.openbci_logging.mx_formatter()

obci.utils.signal_generators module

class obci.utils.signal_generators.AsyncSignalGenerator

Bases: object

Implements async generator for test signal.

Following channels are generated:

1. samples counter
2. time.time() value
3. time.monotonic() value
4. always 0.0
5. always 1.0
6. always -1.0
7. alternating sequence of 0 and 1
8. 100 Hz sinus
9. random.random() generated floats
10. saw signal

obci.utils.signal_generators.MAX_VAL = 10

int – maximal value of saw signal

class obci.utils.signal_generators.SawVerifier(max_val: int = 10)

Bases: object

Raises exception if saw signal is not valid.

verify_next(value)

obci.utils.signal_generators.saw_generator(max_val: int = 10)

Saw signal generator.

Parameters:max_val (int) – maximal value, after which generator drops to 0 and restarts

obci.utils.signal_receiver_peer module

class obci.utils.signal_receiver_peer.SignalReceiver(addresses)

Bases: obci.control.peer.configured_multiplexer_server.ConfiguredMultiplexerServer

handle_message(mxmsg)

obci.utils.streaming_debug module

Use to debug real-life streaming modules like: - amplifier - streamer - filter See signal_streamer_no_filter.py for sample use.

class obci.utils.streaming_debug.Debug(p_sampling, logger, per=1)

Bases: object

By now init externally given logger (python standar logger object) and sampling.

next_sample()

Called after every new sample received. Aftet sel.sampling sample print stats info.

next_sample_timestamp(sample_timestamp)

obci.utils.tagger module

class obci.utils.tagger.Tagger(addresses)

Bases: obci.control.peer.configured_client.ConfiguredClient

send_tag(p_start_timestamp, p_end_timestamp, p_tag_name, p_tag_desc={}, p_tag_channels='')

obci.utils.tagger.get_tagger()

obci.utils.tags_helper module

obci.utils.tags_helper.pack_tag(p_start_timestamp, p_end_timestamp, p_tag_name, p_tag_desc={}, p_tag_channels='')

Return tag with given values. Returned tag is serialised to string. Parameters: - p_start_timestamp - float - p_end_timestamp - float - p_tag_name - string - p_tag_desc - dictionary - p_tag_channels - string like “0 6 7” - numbers of channels

obci.utils.tags_helper.pack_tag_from_tag(tag_dict)

obci.utils.tags_helper.send_tag(conn, p_start_timestamp, p_end_timestamp, p_tag_name, p_tag_desc={}, p_tag_channels='')

For given parameters create tag and send it to mx. Parameters: - p_start_timestamp - float - p_end_timestamp - float - p_tag_name - string - p_tag_desc - dictionary - p_tag_channels - string like “0 6 7” - numbers of channels

obci.utils.tags_helper.send_unpacked_tag(p_tag_dict)

A helper method to send tag in dictionary format.

obci.utils.tags_helper.unpack_tag(p_tag_msg)

For given tag serialised to string, return tag as a dict with fields: - ‘start_timestamp’ - float - ‘end_timestamp’ - float - ‘name’ - string - ‘desc’ - dictionary

obci.utils.zmq module

exception obci.utils.zmq.TimeoutException

Bases: Exception

Raised by recv_multipart_with_timeout when timeout is reached.

obci.utils.zmq.bind_to_urls(socket, urls: typing.List[str]) → typing.List[str]

Bind socket to a given list of URLs. If duplicate URLs are given this function binds only once per URL. A unique list of real bound URLs is returned.

Return type:

List

Parameters:

• socket (zmq.Socket) – ZMQ socket to bind
• urls (list[str]) – list of URLs to bind to

Returns:

list of real bounded URLs

obci.utils.zmq.recv_multipart_with_timeout(socket, timeout: float = 1.0, sleep_interval: float = 0.01) → bytes

This wrapper exists because of a bug in socket.recv_multipart function (zmq.asyncio sockets ignore RCVTIMEO option). For more information see: https://github.com/zeromq/pyzmq/issues/825.

Module contents