Coverage for src/pythia/iterators.py: 32%
58 statements
« prev ^ index » next coverage.py v6.4.4, created at 2022-10-07 19:27 +0000
1"""Shortcuts to iterate over `pyds.GList` -from metadata."""
2from __future__ import annotations
4from typing import Iterator
5from typing import Tuple
6from typing import Type
8import numpy as np
9import pyds
11from pythia.types import PydsClass
12from pythia.utils.ds import FrameAnalytics
13from pythia.utils.ds import ObjectAnalytics
14from pythia.utils.ds import SemanticMasks
15from pythia.utils.ds import SupportedUserMeta
16from pythia.utils.maskrcnn import extract_maskrcnn_mask
19def glist_iter(
20 container_list: pyds.GList,
21 klass: Type[PydsClass],
22) -> Iterator:
23 """Iterate over `pyds.GList` objects.
25 A Pythonic way for iterating over deepstream metadata.
27 Example:
29 When extracting object metadata from frame_metadata, in c++:
31 .. code-block:: cpp
33 for (
34 NvDsMetaList * l_obj = frame_meta->obj_meta_list;
35 l_obj != NULL;
36 l_obj = l_obj->next
37 ) {
38 NvDsObjectMeta *object_meta = (NvDsObjectMeta *) (l_obj->data);
39 do_something_with(object_meta);
40 }
42 And its equivalent in python:
44 .. code-block:: python
46 for object_meta in glist_iter(
47 frame_metadata.obj_meta_list,
48 pyds.NvDsObjectMeta
49 ):
50 do_something_with(object_meta)
52 Args:
53 container_list: The object to iterate over. Its `data` attribute
54 (PyCapsule) is used to instantiate objects.
55 klass: This class is instantiated via its `cast` constructor,
56 using `container_list.data` as its only argument.
58 Yields:
59 The Object instantiated when calling the `cast` function.
61 """
62 while container_list is not None:
63 try:
64 meta = klass.cast(container_list.data)
65 except StopIteration:
66 break
67 yield meta
68 try:
69 container_list = container_list.next
70 except StopIteration:
71 break
74def _iter_user_meta(
75 container_list: pyds.GList,
76 kind: Type[SupportedUserMeta],
77) -> Iterator:
78 for meta in glist_iter(container_list, pyds.NvDsUserMeta):
79 if kind.condition(meta):
80 try:
81 yield kind.klass.cast(meta.user_meta_data)
82 except StopIteration:
83 break
86def frames_per_batch(
87 batch_meta: pyds.NvDsBatchMeta,
88) -> Iterator[pyds.NvDsFrameMeta]:
89 """Yield frame metadata from the batch in a buffer.
91 Args:
92 batch_meta: deepstream batch metadata from a buffer, as received
93 in a bufer pad probe.
95 Yields:
96 frame metadata. A single batch might have multiple values when
97 `nvstreammux` batches its input frames.
99 """
100 for frame_meta in glist_iter(
101 batch_meta.frame_meta_list, pyds.NvDsFrameMeta
102 ):
103 yield frame_meta
106def objects_per_frame(
107 frame_meta: pyds.NvDsFrameMeta,
108) -> Iterator[pyds.NvDsObjectMeta]:
109 """Yield object metadata from a single frame in a batch.
111 Args:
112 frame_meta: single frame metadata
114 Yields:
115 object metadata. A single frame might have multiple values when
116 a single or multiple `nvinfer` generate multiple detections.
118 """
119 yield from glist_iter(frame_meta.obj_meta_list, pyds.NvDsObjectMeta)
122def classification_per_object(
123 object_metadata: pyds.NvDsObjectMeta,
124) -> Iterator[pyds.NvDsClassifierMeta]:
125 """Yield classification metadata from a single object in a frame.
127 Args:
128 object_metadata: deepstream detection metadata from `nvinfer`.
130 Yields:
131 classification metadata. A single object might have multiple
132 values.
134 """
135 yield from glist_iter(
136 object_metadata.classifier_meta_list, pyds.NvDsClassifierMeta
137 )
140def analytics_per_obj(
141 object_metadata: pyds.NvDsObjectMeta,
142) -> Iterator[pyds.NvDsAnalyticsObjInfo]:
143 """Yield analytics metadata from a single object in a frame.
145 Args:
146 object_metadata: deepstream detection metadata from `nvinfer`.
148 Yields:
149 analytics metadata. A single object might have multiple values.
151 """
152 yield from _iter_user_meta(
153 object_metadata.obj_user_meta_list, kind=ObjectAnalytics
154 )
157def labels_per_classification(
158 classifier_meta: pyds.NvDsClassifierMeta,
159) -> Iterator[pyds.NvDsLabelInfo]:
160 """Yield label metadata from a single classification in an object.
162 Args:
163 classifier_meta: deepstream classification metadata from
164 `nvinfer`.
166 Yields:
167 label metadata. A single classification might have multiple
168 values.
170 """
171 yield from glist_iter(
172 classifier_meta.label_info_list,
173 pyds.NvDsLabelInfo,
174 )
177def labels_per_obj(
178 object_metadata: pyds.NvDsObjectMeta,
179) -> Iterator[pyds.NvDsLabelInfo]:
180 """Yield label metadata from a single object in a frame.
182 Args:
183 object_metadata: deepstream detection metadata from `nvinfer`.
185 Yields:
186 label metadata. A single object might have multiple labels.
188 """
189 for classifier_meta in classification_per_object(object_metadata):
190 yield from labels_per_classification(classifier_meta)
193def analytics_per_frame(
194 frame_meta: pyds.NvDsFrameMeta,
195) -> Iterator[pyds.NvDsAnalyticsFrameMeta]:
196 """Yield frame-level analytics metadata.
198 Args:
199 frame_meta: deepstream frame-level metadata from `nvinfer`.
201 Yields:
202 analytics metadata. A single frame might have multiple values.
204 """
205 yield from _iter_user_meta(
206 frame_meta.frame_user_meta_list,
207 kind=FrameAnalytics,
208 )
211def semantic_masks_per_frame(
212 frame_meta: pyds.NvDsFrameMeta,
213) -> Iterator[np.ndarray]:
214 """Yield frame-level semantic segmentation metadata.
216 Args:
217 frame_meta: deepstream frame-level metadata from `nvinfer`.
219 Yields:
220 semantic segmentation bidimensional matrix. A single frame will
221 probably have several values.
223 """
224 for segmeta in _iter_user_meta(
225 frame_meta.frame_user_meta_list,
226 kind=SemanticMasks,
227 ):
228 masks_ = pyds.get_segmentation_masks(segmeta)
229 masks = np.array(masks_, copy=True, order="C")
230 yield masks
233def objects_per_batch(
234 batch_meta: pyds.NvDsBatchMeta,
235) -> Iterator[tuple[pyds.NvDsFrameMeta, pyds.NvDsObjectMeta]]:
236 """Yield frame and object metadata from the batch in a buffer.
238 Args:
239 batch_meta: deepstream batch metadata from a buffer, as received
240 in a bufer pad probe.
242 Yields:
243 frame metadata. A single batch might have multiple values when
244 `nvstreammux` batches its input frames.
245 object metadata. A single frame might have multiple values when
246 a single or multiple `nvinfer` generate multiple detections.
248 """
249 for frame_meta in frames_per_batch(batch_meta):
250 for obj_meta in objects_per_frame(frame_meta):
251 yield frame_meta, obj_meta
254def instance_masks_per_batch(
255 batch_meta: pyds.NvDsBatchMeta,
256) -> Iterator[Tuple[pyds.NvDsFrameMeta, pyds.NvDsObjectMeta, np.ndarray]]:
257 """Yield frame, object, and mask metadata from the batch in a buffer.
259 Args:
260 batch_meta: deepstream batch metadata from a buffer, as received
261 in a bufer pad probe.
263 Yields:
264 * frame metadata. A single batch might have multiple values when
265 `nvstreammux` batches its input frames.
266 * object metadata. A single frame might have multiple values when
267 a single or multiple `nvinfer` generate multiple detections.
268 * instance segmentation. bidimensional matrix.
270 """
271 for frame_meta, obj_meta in objects_per_batch(batch_meta):
272 yield frame_meta, obj_meta, extract_maskrcnn_mask(obj_meta)