pythia package#
pythia - pythonic deepstream.
- class pythia.AnnotateFramesBase(pipeline, dst_folder: Path, *args, **kwargs)[source]#
Bases:
Application
,ABC
Base class for creating dataset / annotations.
Construct a Frame annotator.
- Parameters:
pipeline – forwarded to pythia application constructor.
dst_folder – location for the annotations.
args – forwarded to pythia application constructor.
kwargs – forwarded to pythia application constructor.
- __init__(pipeline, dst_folder: Path, *args, **kwargs) None [source]#
Construct a Frame annotator.
- Parameters:
pipeline – forwarded to pythia application constructor.
dst_folder – location for the annotations.
args – forwarded to pythia application constructor.
kwargs – forwarded to pythia application constructor.
- _abc_impl = <_abc_data object>#
- abstract annotator_probe(pad: Pad, info: PadProbeInfo, batch_meta: NvDsBatchMeta) PadProbeReturn [source]#
Implement this to process incoming batch metadata.
- Parameters:
pad – gstreamer pad where the probe was attached.
info – gstreamer pad probe info.
batch_meta – deepstream metadata (batched!).
- static default_probe_target(pipeline: Pipeline) str [source]#
Retreive an element name to attach the probe to.
- Parameters:
pipeline – Pythia Pipeline containing the elements.
- Returns:
- The name of the most downstream element contained in the
pipeline. The precedence order is: analytics, tracker, nvinfer.
- Raises:
LookupError – none of the required deepstream elements was found.
- nvds_frame_meta_parser: Optional[Callable[[pyds.NvDsFrameMeta], Any]]#
- on_message_error(bus: Bus, message: Message) Tuple[str, str] #
Stop application on error.
- Parameters:
self – A stoppable instance.
bus – The application’s pipeline’s bus.
message – The gstreamer error message.
- Returns:
Error and debug string.
- classmethod run(src: SourceUri, model: Union[str, Path, InferenceEngine], dst_folder: str | Path, *args, suffix: str = '.jpg', analytics: Union[Path, Analytics] | None = None, tracker: Union[Path, Tracker] | None = None, **kwargs) None [source]#
Run an annotation application.
- Parameters:
src – Source uri used for frames/video input.
model – Deepstream inference model.
dst_folder – Path to store annotations and frames.
args – Forwarded to class constructor.
suffix – output frames suffix.
analytics – optional analytics for the pipline.
tracker – optional tracker for the pipline.
kwargs – Forwarded to class constructor.
- Raises:
FileExistsError – Non-empty dst_folder
- class pythia.AnnotateFramesBbox(pipeline, dst_folder: Path, *args, **kwargs)[source]#
Bases:
AnnotateFramesBase
Annotate frames with boundingboxes.
Construct a Frame annotator.
- Parameters:
pipeline – forwarded to pythia application constructor.
dst_folder – location for the annotations.
args – forwarded to pythia application constructor.
kwargs – forwarded to pythia application constructor.
- _abc_impl = <_abc_data object>#
- _bus: Optional[Gst.Bus]#
- _registered_probes: Probes#
- annotator_probe(pad: Pad, info: PadProbeInfo, batch_meta: NvDsBatchMeta) PadProbeReturn [source]#
Implement this to process incoming batch metadata.
- Parameters:
pad – gstreamer pad where the probe was attached.
info – gstreamer pad probe info.
batch_meta – deepstream metadata (batched!).
- nvds_frame_meta_parser: Optional[Callable[[pyds.NvDsFrameMeta], Any]]#
- watch_ids: list[int]#
- class pythia.AnnotateFramesMaskRcnn(pipeline, dst_folder: Path, *args, contour_kw: Optional[dict] = None, **kwargs)[source]#
Bases:
AnnotateFramesBase
Annotate frames with maskrcnn.
Run an maskrcnn annotation application.
- Parameters:
pipeline – forwarded to the annotator constructor.
dst_folder – forwarded to the annotator constructor.
args – forwarded to the annotator constructor.
contour_kw – arbitrary dict containing kwargs for ‘cv2.findContours’
kwargs – forwarded to the annotator constructor.
- Raises:
ImportError – opencv missing
See also
https://docs.opencv.org/4.x/d3/dc0/group__imgproc__shape.html#gadf1ad6a0b82947fa1fe3c3d497f260e0
- __init__(pipeline, dst_folder: Path, *args, contour_kw: Optional[dict] = None, **kwargs) None [source]#
Run an maskrcnn annotation application.
- Parameters:
pipeline – forwarded to the annotator constructor.
dst_folder – forwarded to the annotator constructor.
args – forwarded to the annotator constructor.
contour_kw – arbitrary dict containing kwargs for ‘cv2.findContours’
kwargs – forwarded to the annotator constructor.
- Raises:
ImportError – opencv missing
- _abc_impl = <_abc_data object>#
- _bus: Optional[Gst.Bus]#
- _registered_probes: Probes#
- annotator_probe(pad: Pad, info: PadProbeInfo, batch_meta: NvDsBatchMeta) PadProbeReturn [source]#
Implement this to process incoming batch metadata.
- Parameters:
pad – gstreamer pad where the probe was attached.
info – gstreamer pad probe info.
batch_meta – deepstream metadata (batched!).
- generate_mask_polygon(mask: ndarray) List[List[int]] [source]#
Convert 2d numpy array mask into coco-“segmentation”.
- Parameters:
mask – used to convert to polygon as a matrix.
- Returns:
mask in polygon form.
- nvds_frame_meta_parser: Optional[Callable[[pyds.NvDsFrameMeta], Any]]#
- watch_ids: list[int]#
- class pythia.Application(pipeline: BasePipeline)[source]#
Bases:
BaseApplication
Typical pythia application.
Construct an application from a pipeline.
- Parameters:
pipeline – an instantiated Pipeline.
- on_message_eos(bus: Bus, message: Message)#
Stop application on EOS event.
- Parameters:
self – A stoppable instance.
bus – The application’s pipeline’s bus.
message – The gstreamer state change message.
- on_message_error(bus: Bus, message: Message) Tuple[str, str] #
Stop application on error.
- Parameters:
self – A stoppable instance.
bus – The application’s pipeline’s bus.
message – The gstreamer error message.
- Returns:
Error and debug string.
- pythia.objects_per_batch(batch_meta: pyds.NvDsBatchMeta) Iterator[tuple[pyds.NvDsFrameMeta, pyds.NvDsObjectMeta]] [source]#
Yield frame and object metadata from the batch in a buffer.
- Parameters:
batch_meta – deepstream batch metadata from a buffer, as received in a bufer pad probe.
- Yields:
- frame metadata. A single batch might have multiple values when
nvstreammux batches its input frames.
- object metadata. A single frame might have multiple values when
a single or multiple nvinfer generate multiple detections.