python_plugin_template

Attributes

`CURR_DIR`
`APP_BASE_PATH`
`MANIFEST_JSON_PATH`
`PROCESS_MLA_RESOURCES`
`SIMAAI_META_STR`
`PLUGIN_CPU_TYPE`
`current_dir`
`manifest_config`
`manifest_path`
`logger`

Classes

`MetadataStruct`	Structure base class
`LogLevel`	Create a collection of name/value pairs.
`Logger`
`MetaStruct`
`SimaaiPythonBuffer`
`ValueType`	Create a collection of name/value pairs.
`AggregatorTemplate`	A Python based gstreamer plugin template. Enables the user to:

Functions

get_monotonic_timestamp()

Module Contents

python_plugin_template.CURR_DIR

python_plugin_template.APP_BASE_PATH

python_plugin_template.MANIFEST_JSON_PATH

python_plugin_template.PROCESS_MLA_RESOURCES

class python_plugin_template.MetadataStruct: Structure base class

python_plugin_template.SIMAAI_META_STR = 'GstSimaMeta'

python_plugin_template.PLUGIN_CPU_TYPE = 'APU'

class python_plugin_template.LogLevel(*args, **kwds)

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

attribute access:
```
>>> Color.RED
<Color.RED: 1>
```
value lookup:
```
>>> Color(1)
<Color.RED: 1>
```
name lookup:
```
>>> Color['RED']
<Color.RED: 1>
```

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

EMERG = 0

ALERT = 1

CRIT = 2

ERR = 3

WARNING = 4

NOTICE = 5

INFO = 6

DEBUG = 7

class python_plugin_template.Logger(log_file='/var/log/simaai.log', enable_console=False)

enable_console = False

logger

level = 6

log_methods

set_level(level: LogLevel): Set log level using LogLevel enum.

log(level: LogLevel, message: str, *args, **kwargs): Log a message at the given level.

log_exception(exc_type, exc_value, exc_traceback): Log uncaught exceptions.

python_plugin_template.current_dir

python_plugin_template.manifest_config = None

python_plugin_template.manifest_path

python_plugin_template.logger

class python_plugin_template.MetaStruct(buffer_name, stream_id, timestamp, frame_id)

buffer_name

stream_id

timestamp

frame_id

class python_plugin_template.SimaaiPythonBuffer(metadata: MetaStruct, map: gi.repository.Gst.MapInfo, reshape_tensors: List[numpy.ndarray] | None = None)

metadata

data

size

reshape_tensors = None

array = []

class python_plugin_template.ValueType(*args, **kwds)

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

attribute access:
```
>>> Color.RED
<Color.RED: 1>
```
value lookup:
```
>>> Color(1)
<Color.RED: 1>
```
name lookup:
```
>>> Color['RED']
<Color.RED: 1>
```

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

INT64

UINT64

STRING

DOUBLE

python_plugin_template.get_monotonic_timestamp()

class python_plugin_template.AggregatorTemplate(plugin_name, out_size, next_metaparser=False)

A Python based gstreamer plugin template. Enables the user to: - Accept incoming buffers from dynamic pads - Define any custom plugin runtime logic User has to only override the run() function

transmit

silent

config

plugin_name

dynamic_pads = []

src_caps_set = False

timestamp = 0

frame_id = 0

is_pcie = False

in_pcie_buf_id = 0

stream_id = 'unknown-stream'

buffer_name = 'default'

buffer_id = 0

plugin_id = 'python-agg-template'

t0 = None

t1 = None

manifest_json = None

next_plugin_is_metaparser = False

mpk_path = None

model_output_shapes = []

grandparent_config = None

detess_dequant_pad_name = ''

pcie_buffer_id = 0

metadata_add_failed = False

metadata_len = 0

register_metadata() → None

request_new_pad(templ, direction=None, name=None): Handle dynamic pad requests. Pads are created when a new input stream is added dynamically.

do_start(): Handle start even for the aggregator.

finish_buffer(buffer): Finalizes and pushes the buffer downstream.

do_set_property(property_id, value)

do_get_property(property_id)

extract_metadata(buffer: gi.repository.Gst.Buffer) → None: Input: buffer Gst.Buffer: Input buffer from which metadata will be extracted.

pack_metadata(): Pack metadata into bytes

insert_metadata_as_header(buffer: gi.repository.Gst.Buffer) → bool: Fallback method to insert metadata as buffer header when GstMeta fails. Returns: True if successful, False if failed

insert_metadata(buffer: gi.repository.Gst.Buffer) → bool: Insert metadata into buffer using GstMeta and verify the attachment. Returns: True if successful, False if failed

get_gobject_value(value, value_type: ValueType) → gi.repository.GObject.Value

do_aggregate(timeout): Called when buffers are queued on all sinkpads. Calls the run() function defined by the user

abstractmethod run(input_buffers: List[gi.repository.Gst.Buffer], output_buffer: gi.repository.Gst.Buffer) → None

Input: input_buffers: List[Gst.Buffer] List of input buffers, source from each pad. output_buffer: Gst.Buffer Output buffer that needs to be overwritten.

Implement your logic within this function. Process the input buffers, and modify the output buffer.