Cobra Extras

The top-level pntos.cobra.extras module should only directly contain plugin imports. These are all the plugins that can be directly imported from pntos.cobra.extras.

Plugins in this submodule are separated from the core components for one or more of the following reasons:

1. They may be more advanced than core plugins 1. They may be somewhat experimental 1. They may serve a more niche purpose 1. They may not conform to the same typing and error-checking standards as core plugins.

class pntos.cobra.extras.LcmLogTransportPluginWithProfiling(identifier, config_group=LcmLogTransportConfig.group)

Bases: LcmLogTransportPlugin

A transport plugin which process LCM messages from a log with profiling enabled.

To use, simply swap out the LcmLogTransportPlugin in your app for this one, using the same LcmLogTransportConfig.

NOTE: This plugin uses a simple profiling method that only profiles the thread used to read the data from the LCM log. Thus, if this plugin is used with a controller plugin that utilizes a multi-threaded or multi-processed concurrency model, this plugin has very limited usefulness.

read_log(): Process messages from LCM log, with profiling enabled.

class pntos.cobra.extras.AdvancedPreprocessorPlugin(identifier)

Bases: PreprocessorPlugin

A preprocessor plugin that provides an advanced set of preprocessors.

The preprocessors this plugin provides are:

ZeroVelocity2dGenerator - Generates 2-D zero-velocity measurements based on the assumption that a ground vehicle does not slide laterally or lift off of the ground.

__init__(identifier)

Constructor.

Parameters:: identifier (str) – The plugin identifier used to set this plugin’s pntos.api.CommonPlugin.identifier field.

init_plugin(plugin_resources_location=None, mediator=None)

The first plugin method called; initializes the plugin.

A function that will be called by pntOS once and only once when it first initializes the plugin before any other functions on the plugin are called. Here the plugin may do dynamic runtime initialization of its members, and is given the full path to the location of a data folder specific to the plugin, in case it needs to acquire additional files. An instance of pntos.api.Mediator will be passed which the plugin should save off for later use. Whenever the plugin needs to make a request of pntOS, it should use one of the fields in the pntos.api.Mediator instance received by the plugin in this function call.

Note

Implementation note: This inversion of control allows the controller to implement the pntos.api.Mediator class, and abstracts away the return communication channel from the plugin to the rest of the system. Thus, the plugin need only implement pntos.api.Mediator by simply saving a copy of the functions that the controller passes into it. Then, when the plugin later needs to make requests of the system, it may call a function in its copy of pntos.api.Mediator, without needing any knowledge of how the controller implemented pntos.api.Mediator. This allows controllers to implement arbitrary concurrency models, including single-threaded, multi-threaded, multi-process, and distributed computing.

Parameters:

plugin_resources_location (str | None, optional) – Specifies the location of the plugin’s resources. The location is determined by the controller plugin, and therefore is controller implementation specific. Plugin implementers wishing to provide a resource to their plugin should consult the documentation of the controller to determine which location scheme will be passed into this function.
mediator (Mediator | None, optional) – None-able if the plugin type being initialized is a pntos.api.ControllerPlugin. Non-controller plugins may assume that the mediator parameter is not None.

shutdown_plugin()

A function that will be called by pntOS when it is done using the plugin.

Here the plugin should release any resources it has acquired. When this function call returns pntOS may only call the destructor function (it will not call any other functions of this plugin). The plugin may not call any function on any other plugin, mediator, or use any resource that was given to it by pntOS after it returns from this function.

new_preprocessor(preprocessor_index, config_group=None)

Get a newly created pntos.api.Preprocessor.

Parameters:: preprocessor_index (int) – Since the pntos.api.PreprocessorPlugin can create a different preprocessor for each element in preprocessor_identifiers, the preprocessor_index parameter is used to select which kind of preprocessor to create a new instance of. The pntos.api.PreprocessorPlugin.preprocessor_identifiers field contains identifying strings for the kinds of preprocessors.

Example

For example, if the plugin can create 45 different preprocessors, the identifier of the last preprocessor that can be created is found in preprocessor_identifiers[44]. An instance of this preprocessor can be created by calling new_preprocessor(44, ...). Note that 0 <= preprocessor_index < length of preprocessor_identifiers.

config_group (str | None, optional): Indicates which (if any) parameter group in the
registry may be used to obtain additional configuration values to generate the new preprocessor. If the preprocessor requires no outside configuration, config_group may be None.

Returns:: A newly created pntos.api.Preprocessor. Returns None if preprocessor_index is greater than or equal to the length of pntos.api.PreprocessorPlugin.preprocessor_identifiers or if config_group is invalid.
Return type:: Preprocessor

Cobra Extras Config

pntos.cobra.extras.config contains config for features in pntos.cobra.extras, much like pntos.cobra.config contains config for features in pntos.cobra.

class pntos.cobra.extras.config.ZeroVelocity2dGeneratorConfig(*, group, channels=None, regex=False, trigger_dt_sec=0.0, lateral_vel_sigma, vertical_vel_sigma, output_channel)

Bases: PreprocessorConfig

Config for ZeroVelocity2dGenerator preprocessor.

Parameters:

channels – Trigger channels for generating zero-velocity measurement. Zero-velocity measurements will be generated when a message is received on the given channel, provided trigger_dt_sec has also been surpassed. If None, will generate zero-velocity measurements for all incoming channels.
trigger_dt_sec – Trigger time for generating zero-velocity measurement. Zero-velocity measurements will be generated whenever this delta-time has surpassed since last measurement was generated. If 0.0, will generate zero-velocity measurement whenever a message is received on a channel specified in channels.
lateral_vel_sigma – 1-sigma [m/s] to use for lateral zero-velocity measurement.
vertical_vel_sigma – 1-sigma [m/s] to use for vertical zero-velocity measurement.
output_channel – Channel on which to output the zero-velocity measurements.

identifier = 'zero_velocity2d_generator'

A string that specifies which preprocessor this config should be used in.

This field will be matched against the preprocessor_identifiers field on the preprocessor plugin.

channels = None

Specifier for which channels/chains should include this preprocessor.

If regex field is True, the strings in the channels will be interpreted as regex patterns.

Cobra Extras Internal

pntos.cobra.extras.internal contains features used internally by the plugins in pntos.cobra.extras.

class pntos.cobra.extras.internal.ZeroVelocity2dGenerator(mediator, channels, dt, lat_sigma, vert_sigma, output_channel)

Bases: Preprocessor

A preprocessor that generates 2-D zero-velocity measurements.

These measurements are based on the assumption that a ground vehicle does not slide laterally or lift vertically off the ground.

__init__(mediator, channels, dt, lat_sigma, vert_sigma, output_channel): Cobra 2d zero-velocity generator Preprocessor.

process_pntos_message(message)

Process a message.

Parameters:: message (Message) – A message to be processed.
Returns:: A list of pntos.api.Message s. Usually this will be a single message, a modified version of message. It could be None if message is rejected or dropped. The preprocessor could also accumulate several messages, returning None for each one then returning an array with multiple processed messages.
Return type:: list[Message] | None