Skip to content

Save Detections

CSV Sink

supervision.detection.tools.csv_sink.CSVSink

A utility class for saving detection data to a CSV file. This class is designed to efficiently serialize detection objects into a CSV format, allowing for the inclusion of bounding box coordinates and additional attributes like confidence, class_id, and tracker_id.

Tip

CSVSink allows passing custom data alongside detection fields, providing flexibility for logging various types of information. When a NumPy array, list, or tuple value in custom_data (or detections.data) has the same length as the detection count, each element is written to the corresponding detection row; any other value is broadcast to all rows.

Parameters:

Name Type Description Default

file_name

str

The name of the CSV file where the detections will be stored. Defaults to 'output.csv'.

 'output.csv'
Example 
>>> import supervision as sv
>>> import numpy as np
>>> import tempfile
>>> import os
>>> # Create synthetic detections
>>> detections = sv.Detections(
...     xyxy=np.array([[10, 20, 30, 40], [50, 60, 70, 80]]),
...     confidence=np.array([0.9, 0.8]),
...     class_id=np.array([0, 1])
... )
>>> # Use temporary file
>>> temp_file = tempfile.NamedTemporaryFile(
...     mode='w', suffix='.csv', delete=False
... )
>>> temp_file.close()
>>> csv_sink = sv.CSVSink(temp_file.name)
>>> with csv_sink as sink:
...     sink.append(detections, custom_data={'frame': 0})
>>> os.unlink(temp_file.name)  # Clean up
Source code in src/supervision/detection/tools/csv_sink.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class CSVSink:
    """
    A utility class for saving detection data to a CSV file. This class is designed to
    efficiently serialize detection objects into a CSV format, allowing for the
    inclusion of bounding box coordinates and additional attributes like `confidence`,
    `class_id`, and `tracker_id`.

    !!! tip

        CSVSink allows passing custom data alongside detection fields, providing
        flexibility for logging various types of information.
        When a NumPy array, list, or tuple value in custom_data (or
        detections.data) has the same length as the detection count, each
        element is written to the corresponding detection row; any other value
        is broadcast to all rows.

    Args:
        file_name: The name of the CSV file where the detections will be stored.
            Defaults to 'output.csv'.

    Example:
        ```pycon
        >>> import supervision as sv
        >>> import numpy as np
        >>> import tempfile
        >>> import os
        >>> # Create synthetic detections
        >>> detections = sv.Detections(
        ...     xyxy=np.array([[10, 20, 30, 40], [50, 60, 70, 80]]),
        ...     confidence=np.array([0.9, 0.8]),
        ...     class_id=np.array([0, 1])
        ... )
        >>> # Use temporary file
        >>> temp_file = tempfile.NamedTemporaryFile(
        ...     mode='w', suffix='.csv', delete=False
        ... )
        >>> temp_file.close()
        >>> csv_sink = sv.CSVSink(temp_file.name)
        >>> with csv_sink as sink:
        ...     sink.append(detections, custom_data={'frame': 0})
        >>> os.unlink(temp_file.name)  # Clean up

        ```
    """

    def __init__(self, file_name: str = "output.csv") -> None:
        """
        Initialize the CSVSink instance.

        Args:
            file_name: The name of the CSV file.
        """
        self.file_name = file_name
        self.file: io.TextIOWrapper | None = None
        self.writer: WriterProtocol | None = None
        self.header_written = False
        self.field_names: list[str] = []
        self.deferred_field_names: list[str] = []

    def __enter__(self) -> CSVSink:
        self.open()
        return self

    def __exit__(
        self,
        exc_type: type | None,
        exc_val: Exception | None,
        exc_tb: Any | None,
    ) -> None:
        self.close()

    def open(self) -> None:
        """
        Open the CSV file for writing.
        """
        parent_directory = os.path.dirname(self.file_name)
        if parent_directory and not os.path.exists(parent_directory):
            os.makedirs(parent_directory)

        self.file = open(self.file_name, "w", newline="")
        self.writer = csv.writer(self.file)
        self.header_written = False
        self.field_names = []
        self.deferred_field_names = []

    def close(self) -> None:
        """
        Close the CSV file.

        When every appended batch was empty no header has been written yet, so
        the schema remembered from the first such batch is emitted here. This
        keeps a run that never detected anything readable as an empty table
        rather than as a zero-byte file.
        """
        if self.file is None or self.file.closed:
            return

        # close() also runs from __exit__, so a writer raising here would both leak
        # the file handle and replace the with-body's own exception. Closing in a
        # finally keeps the handle released and lets the original error surface.
        try:
            if not self.header_written and self.deferred_field_names and self.writer:
                self.field_names = self.deferred_field_names
                self.writer.writerow(self.field_names)
                self.header_written = True
        finally:
            self.file.close()

    @staticmethod
    def _slice_value(value: Any, i: int, n: int) -> Any:
        """
        Return the i-th element when the value stores per-detection data.

        Dispatch rules:
            - np.ndarray with ndim == 0: return as-is for broadcasting
            - np.ndarray with len equal to n: return value[i]
            - list or tuple with len equal to n: return value[i]
            - any other type: return as-is for broadcasting

        Args:
            value: Custom-data field value.
            i: Zero-based detection index.
            n: Total number of detections.

        Returns:
            Element at position i if value is a per-detection sequence,
            otherwise value unchanged.
        """
        if isinstance(value, np.ndarray):
            return value[i] if value.ndim > 0 and len(value) == n else value
        if isinstance(value, (list, tuple)) and len(value) == n:
            return value[i]
        return value

    @staticmethod
    def parse_detection_data(
        detections: Detections, custom_data: dict[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        """
        Convert detections and optional custom data into per-detection rows.

        Builds one dictionary per detection containing bounding box coordinates,
        detection attributes, and any values from ``detections.data`` or
        ``custom_data``. NumPy array, list, and tuple values in
        ``custom_data`` with length equal to ``len(detections.xyxy)`` are
        sliced one element per row; all other values are broadcast to every row.

        Args:
            detections: Detection data to serialize into row dictionaries.
            custom_data: Optional extra fields to include in each row.

        Returns:
            A list of dictionaries, one per detection, containing ``xyxy``
            coordinates, ``class_id``, ``confidence``, ``tracker_id``, and any
            values from ``detections.data`` or ``custom_data``.
        """
        parsed_rows = []
        n = len(detections.xyxy)
        for i in range(n):
            row = {
                "x_min": detections.xyxy[i][0],
                "y_min": detections.xyxy[i][1],
                "x_max": detections.xyxy[i][2],
                "y_max": detections.xyxy[i][3],
                "class_id": ""
                if detections.class_id is None
                else str(detections.class_id[i]),
                "confidence": ""
                if detections.confidence is None
                else str(detections.confidence[i]),
                "tracker_id": ""
                if detections.tracker_id is None
                else str(detections.tracker_id[i]),
            }

            if hasattr(detections, "data"):
                for key, value in detections.data.items():
                    row[key] = CSVSink._slice_value(value, i, n)

            if custom_data:
                for key, value in custom_data.items():
                    row[key] = CSVSink._slice_value(value, i, n)

            parsed_rows.append(row)
        return parsed_rows

    def append(
        self, detections: Detections, custom_data: dict[str, Any] | None = None
    ) -> None:
        """
        Append detection data to the CSV file.

        The CSV header is fixed by the first batch that actually contains
        detections; batches with no detections write nothing and leave the
        header undecided, so an empty first frame does not strip the columns
        of the frames that follow. While no populated batch has appeared, the
        schema of the first empty batch is the one ``close()`` falls back to.

        Args:
            detections: The detection data.
            custom_data: Custom data to include. Scalars, dictionaries, and
                other non-sequence values are broadcast to every detection in
                this batch. NumPy arrays, lists, and tuples with length equal
                to ``len(detections)`` are sliced per detection; other lists
                and tuples are broadcast unchanged.
        """
        if not self.writer:
            raise Exception(
                f"Cannot append to CSV: The file '{self.file_name}' is not open."
            )
        field_names = CSVSink.parse_field_names(detections, custom_data)

        # An empty batch produces no rows, so letting it fix the header would
        # pin the file to a schema no detection ever contributed to and drop
        # every extra column of the batches that follow. Remember the first
        # such schema for close() instead, in case no batch ever carries
        # detections; later empty batches must not redefine it, mirroring the
        # first-populated-batch rule for the header itself.
        if len(detections) == 0:
            if not self.header_written and not self.deferred_field_names:
                self.deferred_field_names = field_names
            return

        if not self.header_written:
            self.field_names = field_names
            self.writer.writerow(field_names)
            self.header_written = True

        if field_names != self.field_names:
            logger.warning(
                "Field names do not match the header. Expected: %s, given: %s",
                self.field_names,
                field_names,
            )

        parsed_rows = CSVSink.parse_detection_data(detections, custom_data)
        for row in parsed_rows:
            self.writer.writerow(
                [row.get(field_name, "") for field_name in self.field_names]
            )

    @staticmethod
    def parse_field_names(
        detections: Detections, custom_data: dict[str, Any] | None = None
    ) -> list[str]:
        custom_keys = set(custom_data.keys()) if custom_data else set()
        dynamic_header = sorted(
            custom_keys | set(getattr(detections, "data", {}).keys())
        )
        return BASE_HEADER + dynamic_header

Methods:

__init__(file_name: str = 'output.csv') -> None

Initialize the CSVSink instance.

Parameters:

Name Type Description Default
file_name
str

The name of the CSV file.

 'output.csv'
Source code in src/supervision/detection/tools/csv_sink.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
def __init__(self, file_name: str = "output.csv") -> None:
    """
    Initialize the CSVSink instance.

    Args:
        file_name: The name of the CSV file.
    """
    self.file_name = file_name
    self.file: io.TextIOWrapper | None = None
    self.writer: WriterProtocol | None = None
    self.header_written = False
    self.field_names: list[str] = []
    self.deferred_field_names: list[str] = []

append(detections: Detections, custom_data: dict[str, Any] | None = None) -> None

Append detection data to the CSV file.

The CSV header is fixed by the first batch that actually contains detections; batches with no detections write nothing and leave the header undecided, so an empty first frame does not strip the columns of the frames that follow. While no populated batch has appeared, the schema of the first empty batch is the one close() falls back to.

Parameters:

Name Type Description Default
detections
Detections

The detection data.

 required
custom_data
dict[str, Any] | None

Custom data to include. Scalars, dictionaries, and other non-sequence values are broadcast to every detection in this batch. NumPy arrays, lists, and tuples with length equal to len(detections) are sliced per detection; other lists and tuples are broadcast unchanged.

 None
Source code in src/supervision/detection/tools/csv_sink.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def append(
    self, detections: Detections, custom_data: dict[str, Any] | None = None
) -> None:
    """
    Append detection data to the CSV file.

    The CSV header is fixed by the first batch that actually contains
    detections; batches with no detections write nothing and leave the
    header undecided, so an empty first frame does not strip the columns
    of the frames that follow. While no populated batch has appeared, the
    schema of the first empty batch is the one ``close()`` falls back to.

    Args:
        detections: The detection data.
        custom_data: Custom data to include. Scalars, dictionaries, and
            other non-sequence values are broadcast to every detection in
            this batch. NumPy arrays, lists, and tuples with length equal
            to ``len(detections)`` are sliced per detection; other lists
            and tuples are broadcast unchanged.
    """
    if not self.writer:
        raise Exception(
            f"Cannot append to CSV: The file '{self.file_name}' is not open."
        )
    field_names = CSVSink.parse_field_names(detections, custom_data)

    # An empty batch produces no rows, so letting it fix the header would
    # pin the file to a schema no detection ever contributed to and drop
    # every extra column of the batches that follow. Remember the first
    # such schema for close() instead, in case no batch ever carries
    # detections; later empty batches must not redefine it, mirroring the
    # first-populated-batch rule for the header itself.
    if len(detections) == 0:
        if not self.header_written and not self.deferred_field_names:
            self.deferred_field_names = field_names
        return

    if not self.header_written:
        self.field_names = field_names
        self.writer.writerow(field_names)
        self.header_written = True

    if field_names != self.field_names:
        logger.warning(
            "Field names do not match the header. Expected: %s, given: %s",
            self.field_names,
            field_names,
        )

    parsed_rows = CSVSink.parse_detection_data(detections, custom_data)
    for row in parsed_rows:
        self.writer.writerow(
            [row.get(field_name, "") for field_name in self.field_names]
        )

close() -> None

Close the CSV file.

When every appended batch was empty no header has been written yet, so the schema remembered from the first such batch is emitted here. This keeps a run that never detected anything readable as an empty table rather than as a zero-byte file.

Source code in src/supervision/detection/tools/csv_sink.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def close(self) -> None:
    """
    Close the CSV file.

    When every appended batch was empty no header has been written yet, so
    the schema remembered from the first such batch is emitted here. This
    keeps a run that never detected anything readable as an empty table
    rather than as a zero-byte file.
    """
    if self.file is None or self.file.closed:
        return

    # close() also runs from __exit__, so a writer raising here would both leak
    # the file handle and replace the with-body's own exception. Closing in a
    # finally keeps the handle released and lets the original error surface.
    try:
        if not self.header_written and self.deferred_field_names and self.writer:
            self.field_names = self.deferred_field_names
            self.writer.writerow(self.field_names)
            self.header_written = True
    finally:
        self.file.close()

open() -> None

Open the CSV file for writing.

Source code in src/supervision/detection/tools/csv_sink.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
def open(self) -> None:
    """
    Open the CSV file for writing.
    """
    parent_directory = os.path.dirname(self.file_name)
    if parent_directory and not os.path.exists(parent_directory):
        os.makedirs(parent_directory)

    self.file = open(self.file_name, "w", newline="")
    self.writer = csv.writer(self.file)
    self.header_written = False
    self.field_names = []
    self.deferred_field_names = []

parse_detection_data(detections: Detections, custom_data: dict[str, Any] | None = None) -> list[dict[str, Any]] staticmethod

Convert detections and optional custom data into per-detection rows.

Builds one dictionary per detection containing bounding box coordinates, detection attributes, and any values from detections.data or custom_data. NumPy array, list, and tuple values in custom_data with length equal to len(detections.xyxy) are sliced one element per row; all other values are broadcast to every row.

Parameters:

Name Type Description Default
detections
Detections

Detection data to serialize into row dictionaries.

 required
custom_data
dict[str, Any] | None

Optional extra fields to include in each row.

 None

Returns:

Type Description
list[dict[str, Any]]

A list of dictionaries, one per detection, containing xyxy
list[dict[str, Any]]

coordinates, class_id, confidence, tracker_id, and any
list[dict[str, Any]]

values from detections.data or custom_data.
Source code in src/supervision/detection/tools/csv_sink.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@staticmethod
def parse_detection_data(
    detections: Detections, custom_data: dict[str, Any] | None = None
) -> list[dict[str, Any]]:
    """
    Convert detections and optional custom data into per-detection rows.

    Builds one dictionary per detection containing bounding box coordinates,
    detection attributes, and any values from ``detections.data`` or
    ``custom_data``. NumPy array, list, and tuple values in
    ``custom_data`` with length equal to ``len(detections.xyxy)`` are
    sliced one element per row; all other values are broadcast to every row.

    Args:
        detections: Detection data to serialize into row dictionaries.
        custom_data: Optional extra fields to include in each row.

    Returns:
        A list of dictionaries, one per detection, containing ``xyxy``
        coordinates, ``class_id``, ``confidence``, ``tracker_id``, and any
        values from ``detections.data`` or ``custom_data``.
    """
    parsed_rows = []
    n = len(detections.xyxy)
    for i in range(n):
        row = {
            "x_min": detections.xyxy[i][0],
            "y_min": detections.xyxy[i][1],
            "x_max": detections.xyxy[i][2],
            "y_max": detections.xyxy[i][3],
            "class_id": ""
            if detections.class_id is None
            else str(detections.class_id[i]),
            "confidence": ""
            if detections.confidence is None
            else str(detections.confidence[i]),
            "tracker_id": ""
            if detections.tracker_id is None
            else str(detections.tracker_id[i]),
        }

        if hasattr(detections, "data"):
            for key, value in detections.data.items():
                row[key] = CSVSink._slice_value(value, i, n)

        if custom_data:
            for key, value in custom_data.items():
                row[key] = CSVSink._slice_value(value, i, n)

        parsed_rows.append(row)
    return parsed_rows

JSON Sink

supervision.detection.tools.json_sink.JSONSink

A utility class for saving detection data to a JSON file. This class is designed to efficiently serialize detection objects into a JSON format, allowing for the inclusion of bounding box coordinates and additional attributes like confidence, class_id, and tracker_id.

Tip

JSONSink allows passing custom data alongside detection fields, providing flexibility for logging various types of information. When a NumPy array, list, or tuple value in custom_data (or detections.data) has the same length as the detection count, each element is written to the corresponding detection row; any other value is broadcast to all rows. NumPy scalars (e.g. np.int64, np.float32) are serialized as JSON numbers; NumPy arrays are serialized as JSON arrays.

Parameters:

Name Type Description Default

file_name

str

The name of the JSON file where the detections will be stored. Defaults to 'output.json'.

 'output.json'
Example 
import supervision as sv
from rfdetr import RFDETRMedium

model = RFDETRMedium()
json_sink = sv.JSONSink(<RESULT_JSON_FILE_PATH>)
frames_generator = sv.get_video_frames_generator("<SOURCE_VIDEO_PATH>")

with json_sink as sink:
    for frame in frames_generator:
        detections = model.predict(frame[:, :, ::-1])
        sink.append(detections, custom_data={"<CUSTOM_LABEL>":"<CUSTOM_DATA>"})
Source code in src/supervision/detection/tools/json_sink.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
class JSONSink:
    """
    A utility class for saving detection data to a JSON file. This class is designed to
    efficiently serialize detection objects into a JSON format, allowing for the
    inclusion of bounding box coordinates and additional attributes like `confidence`,
    `class_id`, and `tracker_id`.

    !!! tip

        JSONSink allows passing custom data alongside detection fields, providing
        flexibility for logging various types of information.
        When a NumPy array, list, or tuple value in custom_data (or
        detections.data) has the same length as the detection count, each
        element is written to the corresponding detection row; any other value
        is broadcast to all rows.
        NumPy scalars (e.g. ``np.int64``, ``np.float32``) are serialized as
        JSON numbers; NumPy arrays are serialized as JSON arrays.

    Args:
        file_name: The name of the JSON file where the detections will be stored.
            Defaults to 'output.json'.

    Example:
        ```python
        import supervision as sv
        from rfdetr import RFDETRMedium

        model = RFDETRMedium()
        json_sink = sv.JSONSink(<RESULT_JSON_FILE_PATH>)
        frames_generator = sv.get_video_frames_generator("<SOURCE_VIDEO_PATH>")

        with json_sink as sink:
            for frame in frames_generator:
                detections = model.predict(frame[:, :, ::-1])
                sink.append(detections, custom_data={"<CUSTOM_LABEL>":"<CUSTOM_DATA>"})
        ```
    """

    def __init__(self, file_name: str = "output.json") -> None:
        """
        Initialize the JSONSink instance.

        Args:
            file_name: The name of the JSON file.
        """
        self.file_name = file_name
        self.file: io.TextIOWrapper | None = None
        self.data: list[dict[str, Any]] = []

    def __enter__(self) -> JSONSink:
        self.open()
        return self

    def __exit__(
        self,
        exc_type: type | None,
        exc_val: Exception | None,
        exc_tb: Any | None,
    ) -> None:
        self.write_and_close()

    def open(self) -> None:
        """
        Open the JSON file for writing.
        """
        parent_directory = os.path.dirname(self.file_name)
        if parent_directory and not os.path.exists(parent_directory):
            os.makedirs(parent_directory)

        self.file = open(self.file_name, "w")
        self.data = []

    @staticmethod
    def _json_default(value: Any) -> Any:
        """Return a JSON-serializable equivalent of a NumPy scalar or array.

        Called as the ``default`` hook by :func:`json.dump`. Converts
        :class:`numpy.generic` scalars via ``.item()`` and
        :class:`numpy.ndarray` instances via ``.tolist()``.

        Args:
            value: Object the standard JSON encoder could not serialize.

        Returns:
            A Python scalar or nested list equivalent of ``value``.

        Raises:
            TypeError: If ``value`` is neither a NumPy scalar nor an ndarray.
        """
        if isinstance(value, np.generic):
            return value.item()
        if isinstance(value, np.ndarray):
            return value.tolist()
        raise TypeError(
            f"Object of type {type(value).__name__} is not JSON serializable"
        )

    def write_and_close(self) -> None:
        """
        Write and close the JSON file.
        """
        if self.file:
            try:
                json.dump(
                    self.data, self.file, indent=4, default=JSONSink._json_default
                )
            finally:
                self.file.close()

    @staticmethod
    def _slice_value(value: Any, i: int, n: int) -> Any:
        """
        Return the i-th element when the value stores per-detection data.

        Dispatch rules:
            - np.ndarray with ndim == 0: return as-is for broadcasting
            - np.ndarray with len equal to n: return value[i]
            - list or tuple with len equal to n: return value[i]
            - any other type: return as-is for broadcasting

        Args:
            value: Custom-data field value.
            i: Zero-based detection index.
            n: Total number of detections.

        Returns:
            Element at position i if value is a per-detection sequence,
            otherwise value unchanged.
        """
        if isinstance(value, np.ndarray):
            return value[i] if value.ndim > 0 and len(value) == n else value
        if isinstance(value, (list, tuple)) and len(value) == n:
            return value[i]
        return value

    @staticmethod
    def parse_detection_data(
        detections: Detections, custom_data: dict[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        """
        Convert detections and optional custom data into per-detection rows.

        Builds one dictionary per detection containing bounding box coordinates,
        detection attributes, and any values from ``detections.data`` or
        ``custom_data``. NumPy array, list, and tuple values in
        ``custom_data`` with length equal to ``len(detections.xyxy)`` are
        sliced one element per row; all other values are broadcast to every row.

        Args:
            detections: Detection data to serialize into row dictionaries.
            custom_data: Optional extra fields to include in each row.

        Returns:
            A list of dictionaries, one per detection, containing ``xyxy``
            coordinates, ``class_id``, ``confidence``, ``tracker_id``, and any
            values from ``detections.data`` or ``custom_data``.
        """
        parsed_rows = []
        n = len(detections.xyxy)
        for i in range(n):
            row = {
                "x_min": float(detections.xyxy[i][0]),
                "y_min": float(detections.xyxy[i][1]),
                "x_max": float(detections.xyxy[i][2]),
                "y_max": float(detections.xyxy[i][3]),
                "class_id": ""
                if detections.class_id is None
                else int(detections.class_id[i]),
                "confidence": ""
                if detections.confidence is None
                else float(detections.confidence[i]),
                "tracker_id": ""
                if detections.tracker_id is None
                else int(detections.tracker_id[i]),
            }

            if hasattr(detections, "data"):
                for key, value in detections.data.items():
                    row[key] = JSONSink._slice_value(value, i, n)

            if custom_data:
                for key, value in custom_data.items():
                    row[key] = JSONSink._slice_value(value, i, n)

            parsed_rows.append(row)
        return parsed_rows

    def append(
        self, detections: Detections, custom_data: dict[str, Any] | None = None
    ) -> None:
        """
        Append detection data to the JSON file.

        Args:
            detections: The detection data.
            custom_data: Custom data to include. Scalars, dictionaries, and
                other non-sequence values are broadcast to every detection in
                this batch. NumPy arrays, lists, and tuples with length equal
                to ``len(detections)`` are sliced per detection; other lists
                and tuples are broadcast unchanged.
        """
        parsed_rows = JSONSink.parse_detection_data(detections, custom_data)
        self.data.extend(parsed_rows)

Methods:

__init__(file_name: str = 'output.json') -> None

Initialize the JSONSink instance.

Parameters:

Name Type Description Default
file_name
str

The name of the JSON file.

 'output.json'
Source code in src/supervision/detection/tools/json_sink.py 
51
52
53
54
55
56
57
58
59
60
def __init__(self, file_name: str = "output.json") -> None:
    """
    Initialize the JSONSink instance.

    Args:
        file_name: The name of the JSON file.
    """
    self.file_name = file_name
    self.file: io.TextIOWrapper | None = None
    self.data: list[dict[str, Any]] = []

append(detections: Detections, custom_data: dict[str, Any] | None = None) -> None

Append detection data to the JSON file.

Parameters:

Name Type Description Default
detections
Detections

The detection data.

 required
custom_data
dict[str, Any] | None

Custom data to include. Scalars, dictionaries, and other non-sequence values are broadcast to every detection in this batch. NumPy arrays, lists, and tuples with length equal to len(detections) are sliced per detection; other lists and tuples are broadcast unchanged.

 None
Source code in src/supervision/detection/tools/json_sink.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def append(
    self, detections: Detections, custom_data: dict[str, Any] | None = None
) -> None:
    """
    Append detection data to the JSON file.

    Args:
        detections: The detection data.
        custom_data: Custom data to include. Scalars, dictionaries, and
            other non-sequence values are broadcast to every detection in
            this batch. NumPy arrays, lists, and tuples with length equal
            to ``len(detections)`` are sliced per detection; other lists
            and tuples are broadcast unchanged.
    """
    parsed_rows = JSONSink.parse_detection_data(detections, custom_data)
    self.data.extend(parsed_rows)

open() -> None

Open the JSON file for writing.

Source code in src/supervision/detection/tools/json_sink.py 
74
75
76
77
78
79
80
81
82
83
def open(self) -> None:
    """
    Open the JSON file for writing.
    """
    parent_directory = os.path.dirname(self.file_name)
    if parent_directory and not os.path.exists(parent_directory):
        os.makedirs(parent_directory)

    self.file = open(self.file_name, "w")
    self.data = []

parse_detection_data(detections: Detections, custom_data: dict[str, Any] | None = None) -> list[dict[str, Any]] staticmethod

Convert detections and optional custom data into per-detection rows.

Builds one dictionary per detection containing bounding box coordinates, detection attributes, and any values from detections.data or custom_data. NumPy array, list, and tuple values in custom_data with length equal to len(detections.xyxy) are sliced one element per row; all other values are broadcast to every row.

Parameters:

Name Type Description Default
detections
Detections

Detection data to serialize into row dictionaries.

 required
custom_data
dict[str, Any] | None

Optional extra fields to include in each row.

 None

Returns:

Type Description
list[dict[str, Any]]

A list of dictionaries, one per detection, containing xyxy
list[dict[str, Any]]

coordinates, class_id, confidence, tracker_id, and any
list[dict[str, Any]]

values from detections.data or custom_data.
Source code in src/supervision/detection/tools/json_sink.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@staticmethod
def parse_detection_data(
    detections: Detections, custom_data: dict[str, Any] | None = None
) -> list[dict[str, Any]]:
    """
    Convert detections and optional custom data into per-detection rows.

    Builds one dictionary per detection containing bounding box coordinates,
    detection attributes, and any values from ``detections.data`` or
    ``custom_data``. NumPy array, list, and tuple values in
    ``custom_data`` with length equal to ``len(detections.xyxy)`` are
    sliced one element per row; all other values are broadcast to every row.

    Args:
        detections: Detection data to serialize into row dictionaries.
        custom_data: Optional extra fields to include in each row.

    Returns:
        A list of dictionaries, one per detection, containing ``xyxy``
        coordinates, ``class_id``, ``confidence``, ``tracker_id``, and any
        values from ``detections.data`` or ``custom_data``.
    """
    parsed_rows = []
    n = len(detections.xyxy)
    for i in range(n):
        row = {
            "x_min": float(detections.xyxy[i][0]),
            "y_min": float(detections.xyxy[i][1]),
            "x_max": float(detections.xyxy[i][2]),
            "y_max": float(detections.xyxy[i][3]),
            "class_id": ""
            if detections.class_id is None
            else int(detections.class_id[i]),
            "confidence": ""
            if detections.confidence is None
            else float(detections.confidence[i]),
            "tracker_id": ""
            if detections.tracker_id is None
            else int(detections.tracker_id[i]),
        }

        if hasattr(detections, "data"):
            for key, value in detections.data.items():
                row[key] = JSONSink._slice_value(value, i, n)

        if custom_data:
            for key, value in custom_data.items():
                row[key] = JSONSink._slice_value(value, i, n)

        parsed_rows.append(row)
    return parsed_rows

write_and_close() -> None

Write and close the JSON file.

Source code in src/supervision/detection/tools/json_sink.py 
110
111
112
113
114
115
116
117
118
119
120
def write_and_close(self) -> None:
    """
    Write and close the JSON file.
    """
    if self.file:
        try:
            json.dump(
                self.data, self.file, indent=4, default=JSONSink._json_default
            )
        finally:
            self.file.close()

Comments