pythia.applications.annotation#

Annotation applications.

class pythia.applications.annotation.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>#
_bus: Optional[Gst.Bus]#
static _extract_common(pad, frame, detection, *, extract_analytics: bool = False)[source]#
_registered_probes: Probes#
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.

loop: Optional[RunLoop]#
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

classmethod run_with_probe(*args, probe: SupportedCb | BoundSupportedCb | None = None, **kwargs) → None[source]#

Run an annotator with a custom probe.

Parameters:

  • args – forwarded to run().

  • probe – annotator_probe to use.

  • kwargs – forwarded to run().

watch_ids: list[int]#
class pythia.applications.annotation.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!).

loop: Optional[RunLoop]#
nvds_frame_meta_parser: Optional[Callable[[pyds.NvDsFrameMeta], Any]]#
watch_ids: list[int]#
class pythia.applications.annotation.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

See also

https://docs.opencv.org/4.x/d3/dc0/group__imgproc__shape.html#gadf1ad6a0b82947fa1fe3c3d497f260e0

_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.

See also

https://learnopencv.com/deep-learning-based-object-detection-and-instance-segmentation-using-mask-rcnn-in-opencv-python-c/

loop: Optional[RunLoop]#
nvds_frame_meta_parser: Optional[Callable[[pyds.NvDsFrameMeta], Any]]#
watch_ids: list[int]#
class pythia.applications.annotation._DumpLogger(name, level=0)[source]#

Bases: Logger

Initialize the logger with a name and an optional level.

json(msg, *args, **kwargs)[source]#
pythia.applications.annotation._make_handler(dst: Path | Literal['stdout', 'stderr']) → logging.StreamHandler[source]#
pythia.applications.annotation._make_logger(name: str, dst: Path | Literal['stdout', 'stderr']) → _DumpLogger[source]#