python_plugin_template ====================== .. py:module:: python_plugin_template Attributes ---------- .. autoapisummary:: python_plugin_template.CURR_DIR python_plugin_template.APP_BASE_PATH python_plugin_template.MANIFEST_JSON_PATH python_plugin_template.PROCESS_MLA_RESOURCES python_plugin_template.SIMAAI_META_STR python_plugin_template.PLUGIN_CPU_TYPE python_plugin_template.current_dir python_plugin_template.manifest_config python_plugin_template.manifest_path python_plugin_template.logger Classes ------- .. autoapisummary:: python_plugin_template.MetadataStruct python_plugin_template.LogLevel python_plugin_template.Logger python_plugin_template.MetaStruct python_plugin_template.SimaaiPythonBuffer python_plugin_template.ValueType python_plugin_template.AggregatorTemplate Functions --------- .. autoapisummary:: python_plugin_template.get_monotonic_timestamp Module Contents --------------- .. py:data:: CURR_DIR .. py:data:: APP_BASE_PATH .. py:data:: MANIFEST_JSON_PATH .. py:data:: PROCESS_MLA_RESOURCES .. py:class:: MetadataStruct Structure base class .. py:data:: SIMAAI_META_STR :value: 'GstSimaMeta' .. py:data:: PLUGIN_CPU_TYPE :value: 'APU' .. py:class:: 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 - value lookup: >>> Color(1) - name lookup: >>> Color['RED'] Enumerations can be iterated over, and know how many members they have: >>> len(Color) 3 >>> list(Color) [, , ] Methods can be added to enumerations, and members can have their own attributes -- see the documentation for details. .. py:attribute:: EMERG :value: 0 .. py:attribute:: ALERT :value: 1 .. py:attribute:: CRIT :value: 2 .. py:attribute:: ERR :value: 3 .. py:attribute:: WARNING :value: 4 .. py:attribute:: NOTICE :value: 5 .. py:attribute:: INFO :value: 6 .. py:attribute:: DEBUG :value: 7 .. py:class:: Logger(log_file='/var/log/simaai.log', enable_console=False) .. py:attribute:: enable_console :value: False .. py:attribute:: logger .. py:attribute:: level :value: 6 .. py:attribute:: log_methods .. py:method:: set_level(level: LogLevel) Set log level using LogLevel enum. .. py:method:: log(level: LogLevel, message: str, *args, **kwargs) Log a message at the given level. .. py:method:: log_exception(exc_type, exc_value, exc_traceback) Log uncaught exceptions. .. py:data:: current_dir .. py:data:: manifest_config :value: None .. py:data:: manifest_path .. py:data:: logger .. py:class:: MetaStruct(buffer_name, stream_id, timestamp, frame_id) .. py:attribute:: buffer_name .. py:attribute:: stream_id .. py:attribute:: timestamp .. py:attribute:: frame_id .. py:class:: SimaaiPythonBuffer(metadata: MetaStruct, map: gi.repository.Gst.MapInfo, reshape_tensors: Optional[List[numpy.ndarray]] = None) .. py:attribute:: metadata .. py:attribute:: data .. py:attribute:: size .. py:attribute:: reshape_tensors :value: None .. py:attribute:: array :value: [] .. py:class:: 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 - value lookup: >>> Color(1) - name lookup: >>> Color['RED'] Enumerations can be iterated over, and know how many members they have: >>> len(Color) 3 >>> list(Color) [, , ] Methods can be added to enumerations, and members can have their own attributes -- see the documentation for details. .. py:attribute:: INT64 .. py:attribute:: UINT64 .. py:attribute:: STRING .. py:attribute:: DOUBLE .. py:function:: get_monotonic_timestamp() .. py:class:: 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 .. py:attribute:: transmit .. py:attribute:: silent .. py:attribute:: config .. py:attribute:: plugin_name .. py:attribute:: dynamic_pads :value: [] .. py:attribute:: src_caps_set :value: False .. py:attribute:: timestamp :value: 0 .. py:attribute:: frame_id :value: 0 .. py:attribute:: is_pcie :value: False .. py:attribute:: in_pcie_buf_id :value: 0 .. py:attribute:: stream_id :value: 'unknown-stream' .. py:attribute:: buffer_name :value: 'default' .. py:attribute:: buffer_id :value: 0 .. py:attribute:: plugin_id :value: 'python-agg-template' .. py:attribute:: t0 :value: None .. py:attribute:: t1 :value: None .. py:attribute:: manifest_json :value: None .. py:attribute:: next_plugin_is_metaparser :value: False .. py:attribute:: mpk_path :value: None .. py:attribute:: model_output_shapes :value: [] .. py:attribute:: grandparent_config :value: None .. py:attribute:: detess_dequant_pad_name :value: '' .. py:attribute:: pcie_buffer_id :value: 0 .. py:attribute:: metadata_add_failed :value: False .. py:attribute:: metadata_len :value: 0 .. py:method:: register_metadata() -> None .. py:method:: request_new_pad(templ, direction=None, name=None) Handle dynamic pad requests. Pads are created when a new input stream is added dynamically. .. py:method:: do_start() Handle start even for the aggregator. .. py:method:: finish_buffer(buffer) Finalizes and pushes the buffer downstream. .. py:method:: do_set_property(property_id, value) .. py:method:: do_get_property(property_id) .. py:method:: extract_metadata(buffer: gi.repository.Gst.Buffer) -> None Input: buffer Gst.Buffer: Input buffer from which metadata will be extracted. .. py:method:: pack_metadata() Pack metadata into bytes .. py:method:: 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 .. py:method:: 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 .. py:method:: get_gobject_value(value, value_type: ValueType) -> gi.repository.GObject.Value .. py:method:: do_aggregate(timeout) Called when buffers are queued on all sinkpads. Calls the run() function defined by the user .. py:method:: run(input_buffers: List[gi.repository.Gst.Buffer], output_buffer: gi.repository.Gst.Buffer) -> None :abstractmethod: 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.