xiaoxue_admin
/
supervision
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

LazyDataset Loading - accept List[str] of images 

* Unfinished
* Untested
* Committing to show Piotr a bit earlier
This commit is contained in:
LinasKo 2024-07-04 14:21:45 +03:00
parent c973647cb8
commit 4b9b70fbb2
7 changed files with 261 additions and 146 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
docs/changelog.md
View File

@ -1,3 +1,15 @@
### 0.22.0 <small>Jul X, 2024</small>
- Detections dataset now accepts a list of image paths, lazy-loading images when needed. This enables use cases when you have a large dataset and don't want to load all images into memory at once.
!!! failure "Deprecated"
Constructing `DetectionDataset` with parameter `images` as `Dict[str, np.ndarray]` is deprecated and will be removed in `supervision-0.26.0`. Please pass a list of paths `List[str]` instead.
!!! failure "Deprecated"
The `DetectionDataset.images` property is deprecated and will be removed in `supervision-0.26.0`. Please loop over images with `for path, image, annotation in dataset:`, as that does not require loading all images into memory.
### 0.21.0 <small>Jun 5, 2024</small>
- Added [#500](https://github.com/roboflow/supervision/pull/500): [`sv.Detections.with_nmm`](https://supervision.roboflow.com/develop/detection/core/#supervision.detection.core.Detections.with_nmm) to perform non-maximum merging on the current set of object detections.
@ -164,7 +176,7 @@ with csv_sink:
- Added [#819](https://github.com/roboflow/supervision/pull/819): [`sv.JSONSink`](/0.19.0/detection/tools/save_detections/#supervision.detection.tools.csv_sink.JSONSink) allowing for the straightforward saving of image, video, or stream inference results in a `.json` file.
```python
````python
```python
import supervision as sv
@ -179,7 +191,7 @@ with json_sink:
result = model(frame)[0]
detections = sv.Detections.from_ultralytics(result)
json_sink.append(detections, custom_data={<CUSTOM_LABEL>:<CUSTOM_DATA>})
```
````
- Added [#847](https://github.com/roboflow/supervision/pull/847): [`sv.mask_iou_batch`](/0.19.0/detection/utils/#supervision.detection.utils.mask_iou_batch) allowing to compute Intersection over Union (IoU) of two sets of masks.
@ -285,7 +297,6 @@ ColorPalette(colors=[Color(r=68, g=1, b=84), Color(r=59, g=82, b=139), ...])
`sv.ColorPalette.default()` is deprecated and will be removed in `supervision-0.22.0`. Use `sv.ColorPalette.DEFAULT` instead.
- Changed [#769](https://github.com/roboflow/supervision/pull/769): [`sv.ColorPalette.DEFAULT`](/0.18.0/draw/color/#colorpalette) value, giving users a more extensive set of annotation colors.
- Changed [#677](https://github.com/roboflow/supervision/pull/677): `sv.Detections.from_roboflow` to [`sv.Detections.from_inference`](/0.18.0/detection/core/#supervision.detection.core.Detections.from_inference) streamlining its functionality to be compatible with both the both [inference](https://github.com/roboflow/inference) pip package and the Robloflow [hosted API](https://docs.roboflow.com/deploy/hosted-api).
@ -294,7 +305,6 @@ ColorPalette(colors=[Color(r=68, g=1, b=84), Color(r=59, g=82, b=139), ...])
`Detections.from_roboflow()` is deprecated and will be removed in `supervision-0.22.0`. Use `Detections.from_inference` instead.
- Fixed [#735](https://github.com/roboflow/supervision/pull/735): [`sv.LineZone`](/0.18.0/detection/tools/line_zone/#linezone) functionality to accurately update the counter when an object crosses a line from any direction, including from the side. This enhancement enables more precise tracking and analytics, such as calculating individual in/out counts for each lane on the road.
### 0.17.0 <small>December 06, 2023</small>
@ -387,13 +397,12 @@ ColorPalette(colors=[Color(r=68, g=1, b=84), Color(r=59, g=82, b=139), ...])
- Fixed [#477](https://github.com/roboflow/supervision/pull/477): Poetry env definition allowing proper local installation.
- Fixed [#430](https://github.com/roboflow/supervision/pull/430): [`sv.ByteTrack`](/0.16.0/trackers/#supervision.tracker.byte_tracker.core.ByteTrack) to return `np.array([], dtype=int)` when `svDetections` is empty.
- Fixed [#430](https://github.com/roboflow/supervision/pull/430): [`sv.ByteTrack`](/0.16.0/trackers/#supervision.tracker.byte_tracker.core.ByteTrack) to return `np.array([], dtype=int)` when `svDetections` is empty.
!!! failure "Deprecated"
`sv.Detections.from_yolov8` and `sv.Classifications.from_yolov8` as those are now replaced by [`sv.Detections.from_ultralytics`](/0.16.0/detection/core/#supervision.detection.core.Detections.from_ultralytics) and [`sv.Classifications.from_ultralytics`](/0.16.0/classification/core/#supervision.classification.core.Classifications.from_ultralytics).
### 0.15.0 <small>October 5, 2023</small>
- Added [#170](https://github.com/roboflow/supervision/pull/170): [`sv.BoundingBoxAnnotator`](/0.15.0/annotators/#supervision.annotators.core.BoundingBoxAnnotator) allowing to annotate images and videos with bounding boxes.

2
docs/deprecated.md
View File

@ -17,3 +17,5 @@ These features are phased out due to better alternatives or potential issues in
- The `track_buffer`, `track_thresh`, and `match_thresh` parameters in [`ByterTrack`](trackers.md/#supervision.tracker.byte_tracker.core.ByteTrack) are deprecated and will be removed in `supervision-0.23.0`. Use `lost_track_buffer,` `track_activation_threshold`, and `minimum_matching_threshold` instead.
- The `triggering_position ` parameter in [`sv.PolygonZone`](detection/tools/polygon_zone.md/#supervision.detection.tools.polygon_zone.PolygonZone) is deprecated and will be removed in `supervision-0.23.0`. Use `triggering_anchors ` instead.
- The `frame_resolution_wh ` parameter in [`sv.PolygonZone`](detection/tools/polygon_zone.md/#supervision.detection.tools.polygon_zone.PolygonZone) is deprecated and will be removed in `supervision-0.24.0`.
- Constructing `DetectionDataset` with parameter `images` as `Dict[str, np.ndarray]` is deprecated and will be removed in `supervision-0.26.0`. Please pass a list of paths `List[str]` instead.
- The `DetectionDataset.images` property is deprecated and will be removed in `supervision-0.26.0`. Please loop over images with `for path, image, annotation in dataset:`, as that does not require loading all images into memory.

303
supervision/dataset/core.py
View File

@ -3,8 +3,9 @@ from __future__ import annotations
import os
from abc import ABC, abstractmethod
from dataclasses import dataclass
from itertools import chain
from pathlib import Path
from typing import Dict, Iterator, List, Optional, Tuple
from typing import Dict, Iterator, List, Optional, Tuple, Union
import cv2
import numpy as np
@ -27,13 +28,13 @@ from supervision.dataset.utils import (
build_class_index_mapping,
map_detections_class_id,
merge_class_lists,
save_dataset_images,
train_test_split,
)
from supervision.detection.core import Detections
from supervision.utils.internal import deprecated, warn_deprecated
from supervision.utils.iterables import find_duplicates
@dataclass
class BaseDataset(ABC):
@abstractmethod
def __len__(self) -> int:
@ -46,21 +47,68 @@ class BaseDataset(ABC):
pass
@dataclass
class DetectionDataset(BaseDataset):
"""
Dataclass containing information about object detection dataset.
Attributes:
classes (List[str]): List containing dataset class names.
images (Dict[str, np.ndarray]): Dictionary mapping image name to image.
images (Union[List[str], Dict[str, np.ndarray]]):
Accepts a list of image paths, or dictionaries of loaded cv2 images
with paths as keys. If you pass a list of paths, the dataset will
lazily load images on demand, which is much more memory-efficient.
annotations (Dict[str, Detections]): Dictionary mapping
image name to annotations.
image path to detections.
"""
classes: List[str]
images: Dict[str, np.ndarray]
annotations: Dict[str, Detections]
def __init__(
self,
classes: List[str],
images: Union[List[str], Dict[str, np.ndarray]],
annotations: Dict[str, Detections],
) -> None:
self.classes = classes
self.annotations = annotations
self._image_paths_as_unique_keys = dict.fromkeys(images)
self.image_paths = list(self._image_paths_as_unique_keys)
self._images_in_memory: Dict[str, np.ndarray] = {}
if isinstance(images, dict):
self._images_in_memory = images
warn_deprecated(
"Passing a `Dict[str, np.ndarray]` into `DetectionDataset` is deprecated and "
"will be removed in `supervision-0.26.0`. Use a list of paths `List[str]` "
"instead."
)
# TODO: when supervision-0.26.0 is released, and images: Dict[str, np.ndarray] is
# no longer supported, also simplify the rest of the code. E.g. list(images)
# is no longer needed, and merge can be simplified.
@property
@deprecated(
"`DetectionDataset.images` property is deprecated and will be removed in "
"`supervision-0.26.0`. Iterate with `for path, image, annotation in dataset:` instead."
)
def images(self) -> Dict[str, np.ndarray]:
"""
Load all images to memory and return them as a dictionary.
Warning: only use this when you need all images at once.
It is much more memory-efficient to initialize dataset with
image paths and use `for image in dataset:`.
"""
if self._images_in_memory:
return self._images_in_memory
images = {image_path: cv2.imread(image_path) for image_path in self.image_paths}
return images
def _get_image(self, image_path: str) -> np.ndarray:
"""Assumes that image is in dataset"""
if self._images_in_memory:
return self._images_in_memory[image_path]
return cv2.imread(image_path)
def __len__(self) -> int:
"""
@ -69,7 +117,13 @@ class DetectionDataset(BaseDataset):
Returns:
int: The number of images.
"""
return len(self.images)
return len(self._images_in_memory) or len(self.image_paths)
def __getitem__(self, i: int) -> Tuple[str, np.ndarray, Detections]:
image_path = self.image_paths[i]
image = self._get_image(image_path)
annotation = self.annotations[image_path]
return image_path, image, annotation
def __iter__(self) -> Iterator[Tuple[str, np.ndarray, Detections]]:
"""
@ -80,21 +134,30 @@ class DetectionDataset(BaseDataset):
An iterator that yields tuples containing the image name,
the image data, and its corresponding annotation.
"""
for image_name, image in self.images.items():
yield image_name, image, self.annotations.get(image_name, None)
for i in range(len(self)):
image_path, image, annotation = self[i]
yield image_path, image, annotation
def __eq__(self, other):
def __eq__(self, other) -> bool:
if not isinstance(other, DetectionDataset):
return False
if set(self.classes) != set(other.classes):
return False
for key in self.images:
if not np.array_equal(self.images[key], other.images[key]):
return False
if not self.annotations[key] == other.annotations[key]:
return False
if self.image_paths != other.image_paths:
return False
try:
for self_values, other_values in zip(self, other):
_, image_self, annotation_self = self_values
_, image_other, annotation_other = other_values
if not np.array_equal(image_self, image_other):
return False
if not annotation_self == annotation_other:
return False
except KeyError:
return False
return True
@ -127,26 +190,131 @@ class DetectionDataset(BaseDataset):
```
"""
image_names = list(self.images.keys())
train_names, test_names = train_test_split(
data=image_names,
train_paths, test_paths = train_test_split(
data=self.image_paths,
train_ratio=split_ratio,
random_state=random_state,
shuffle=shuffle,
)
train_input: Union[List[str], Dict[str, np.ndarray]]
test_input: Union[List[str], Dict[str, np.ndarray]]
if self._images_in_memory:
train_input = {path: self._images_in_memory[path] for path in train_paths}
test_input = {path: self._images_in_memory[path] for path in test_paths}
else:
train_input = train_paths
test_input = test_paths
train_annotations = {path: self.annotations[path] for path in train_paths}
test_annotations = {path: self.annotations[path] for path in test_paths}
train_dataset = DetectionDataset(
classes=self.classes,
images={name: self.images[name] for name in train_names},
annotations={name: self.annotations[name] for name in train_names},
images=train_input,
annotations=train_annotations,
)
test_dataset = DetectionDataset(
classes=self.classes,
images={name: self.images[name] for name in test_names},
annotations={name: self.annotations[name] for name in test_names},
images=test_input,
annotations=test_annotations,
)
return train_dataset, test_dataset
@classmethod
def merge(cls, dataset_list: List[DetectionDataset]) -> DetectionDataset:
"""
Merge a list of `DetectionDataset` objects into a single
`DetectionDataset` object.
This method takes a list of `DetectionDataset` objects and combines
their respective fields (`classes`, `images`,
`annotations`) into a single `DetectionDataset` object.
Args:
dataset_list (List[DetectionDataset]): A list of `DetectionDataset`
objects to merge.
Returns:
(DetectionDataset): A single `DetectionDataset` object containing
the merged data from the input list.
Examples:
```python
import supervision as sv
ds_1 = sv.DetectionDataset(...)
len(ds_1)
# 100
ds_1.classes
# ['dog', 'person']
ds_2 = sv.DetectionDataset(...)
len(ds_2)
# 200
ds_2.classes
# ['cat']
ds_merged = sv.DetectionDataset.merge([ds_1, ds_2])
len(ds_merged)
# 300
ds_merged.classes
# ['cat', 'dog', 'person']
```
"""
def is_in_memory(dataset: DetectionDataset) -> bool:
return len(dataset._images_in_memory) > 0 or len(dataset.image_paths) == 0
def is_lazy(dataset: DetectionDataset) -> bool:
return not is_in_memory(dataset) or len(dataset._images_in_memory) == 0
all_in_memory = all([is_in_memory(dataset) for dataset in dataset_list])
all_lazy = all([is_lazy(dataset) for dataset in dataset_list])
if not all_in_memory and not all_lazy:
raise ValueError(
"Merging lazy and in-memory DetectionDatasets is not supported."
)
images_in_memory = {}
for dataset in dataset_list:
images_in_memory.update(dataset._images_in_memory)
image_paths: List[str] = []
if not images_in_memory:
image_paths = list(
chain.from_iterable(dataset.image_paths for dataset in dataset_list)
)
image_paths_unique = list(dict.fromkeys(image_paths))
if len(image_paths) != len(image_paths_unique):
duplicates = find_duplicates(image_paths)
raise ValueError(
f"Image paths {duplicates} are not unique across datasets."
)
image_paths = image_paths_unique
classes = merge_class_lists(
class_lists=[dataset.classes for dataset in dataset_list]
)
annotations = {}
for dataset in dataset_list:
annotations.update(dataset.annotations)
for dataset in dataset_list:
class_index_mapping = build_class_index_mapping(
source_classes=dataset.classes, target_classes=classes
)
for image_path in dataset.image_paths:
annotations[image_path] = map_detections_class_id(
source_to_target_mapping=class_index_mapping,
detections=annotations[image_path],
)
return cls(
classes=classes,
images=images_in_memory or image_paths,
annotations=annotations,
)
def as_pascal_voc(
self,
images_directory_path: Optional[str] = None,
@ -179,23 +347,19 @@ class DetectionDataset(BaseDataset):
in the range [0, 1). Argument is used only for segmentation datasets.
"""
if images_directory_path:
save_dataset_images(
images_directory_path=images_directory_path, images=self.images
self._save_images(
images_directory_path=images_directory_path,
)
if annotations_directory_path:
Path(annotations_directory_path).mkdir(parents=True, exist_ok=True)
for image_path, image in self.images.items():
detections = self.annotations[image_path]
if annotations_directory_path:
for image_path, image, annotations in self.images.items():
annotation_name = Path(image_path).stem
annotations_path = os.path.join(
annotations_directory_path, f"{annotation_name}.xml"
)
image_name = Path(image_path).name
pascal_voc_xml = detections_to_pascal_voc(
detections=detections,
detections=annotations,
classes=self.classes,
filename=image_name,
image_shape=image.shape,
@ -358,9 +522,7 @@ class DetectionDataset(BaseDataset):
Argument is used only for segmentation datasets.
"""
if images_directory_path is not None:
save_dataset_images(
images_directory_path=images_directory_path, images=self.images
)
self._save_images(images_directory_path=images_directory_path)
if annotations_directory_path is not None:
save_yolo_annotations(
annotations_directory_path=annotations_directory_path,
@ -482,70 +644,11 @@ class DetectionDataset(BaseDataset):
approximation_percentage=approximation_percentage,
)
@classmethod
def merge(cls, dataset_list: List[DetectionDataset]) -> DetectionDataset:
"""
Merge a list of `DetectionDataset` objects into a single
`DetectionDataset` object.
This method takes a list of `DetectionDataset` objects and combines
their respective fields (`classes`, `images`,
`annotations`) into a single `DetectionDataset` object.
Args:
dataset_list (List[DetectionDataset]): A list of `DetectionDataset`
objects to merge.
Returns:
(DetectionDataset): A single `DetectionDataset` object containing
the merged data from the input list.
Examples:
```python
import supervision as sv
ds_1 = sv.DetectionDataset(...)
len(ds_1)
# 100
ds_1.classes
# ['dog', 'person']
ds_2 = sv.DetectionDataset(...)
len(ds_2)
# 200
ds_2.classes
# ['cat']
ds_merged = sv.DetectionDataset.merge([ds_1, ds_2])
len(ds_merged)
# 300
ds_merged.classes
# ['cat', 'dog', 'person']
```
"""
merged_images, merged_annotations = {}, {}
class_lists = [dataset.classes for dataset in dataset_list]
merged_classes = merge_class_lists(class_lists=class_lists)
for dataset in dataset_list:
class_index_mapping = build_class_index_mapping(
source_classes=dataset.classes, target_classes=merged_classes
)
for image_name, image, detections in dataset:
if image_name in merged_annotations:
raise ValueError(
f"Image name {image_name} is not unique across datasets."
)
merged_images[image_name] = image
merged_annotations[image_name] = map_detections_class_id(
source_to_target_mapping=class_index_mapping,
detections=detections,
)
return cls(
classes=merged_classes, images=merged_images, annotations=merged_annotations
)
def _save_images(self, images_directory_path: str) -> None:
Path(images_directory_path).mkdir(parents=True, exist_ok=True)
for image_path, image, _ in self:
final_path = os.path.join(images_directory_path, image_path)
cv2.imwrite(final_path, image)
@dataclass

30
supervision/dataset/formats/pascal_voc.py
View File

@ -138,7 +138,7 @@ def load_pascal_voc_annotations(
images_directory_path: str,
annotations_directory_path: str,
force_masks: bool = False,
) -> Tuple[List[str], Dict[str, np.ndarray], Dict[str, Detections]]:
) -> Tuple[List[str], List[str], Dict[str, Detections]]:
"""
Loads PASCAL VOC XML annotations and returns the image name,
a Detections instance, and a list of class names.
@ -151,44 +151,40 @@ def load_pascal_voc_annotations(
annotations, regardless of whether they are present.
Returns:
Tuple[List[str], Dict[str, np.ndarray], Dict[str, Detections]]: A tuple
containing a list of class names,
a dictionary with image names as keys and
images as values, and a dictionary with image names as
keys and corresponding Detections instances as values.
Tuple[List[str], List[str], Dict[str, Detections]]: A tuple with a list
of class names, a list of paths to images, and a dictionary with image
paths as keys and corresponding Detections instances as values.
"""
image_paths = list_files_with_extensions(
directory=images_directory_path, extensions=["jpg", "jpeg", "png"]
)
image_paths = [
str(path)
for path in list_files_with_extensions(
directory=images_directory_path, extensions=["jpg", "jpeg", "png"]
)
]
classes = []
images = {}
classes: List[str] = []
annotations = {}
for image_path in image_paths:
image_name = Path(image_path).stem
image_path = str(image_path)
image = cv2.imread(image_path)
annotation_path = os.path.join(annotations_directory_path, f"{image_name}.xml")
if not os.path.exists(annotation_path):
images[image_path] = image
annotations[image_path] = Detections.empty()
continue
tree = parse(annotation_path)
root = tree.getroot()
image = cv2.imread(image_path)
resolution_wh = (image.shape[1], image.shape[0])
annotation, classes = detections_from_xml_obj(
root, classes, resolution_wh, force_masks
)
images[image_path] = image
annotations[image_path] = annotation
return classes, images, annotations
return classes, image_paths, annotations
def detections_from_xml_obj(

15
supervision/dataset/utils.py
View File

@ -1,10 +1,7 @@
import copy
import os
import random
from pathlib import Path
from typing import Dict, List, Optional, Tuple, TypeVar, Union
import cv2
import numpy as np
import numpy.typing as npt
@ -59,6 +56,7 @@ def merge_class_lists(class_lists: List[List[str]]) -> List[str]:
def build_class_index_mapping(
source_classes: List[str], target_classes: List[str]
) -> Dict[int, int]:
"""Returns the index map of source classes -> target classes."""
index_mapping = {}
for i, class_name in enumerate(source_classes):
@ -93,17 +91,6 @@ def map_detections_class_id(
return detections_copy
def save_dataset_images(
images_directory_path: str, images: Dict[str, np.ndarray]
) -> None:
Path(images_directory_path).mkdir(parents=True, exist_ok=True)
for image_path, image in images.items():
image_name = Path(image_path).name
target_image_path = os.path.join(images_directory_path, image_name)
cv2.imwrite(target_image_path, image)
def train_test_split(
data: List[T],
train_ratio: float = 0.8,

22
supervision/utils/internal.py
View File

@ -31,6 +31,16 @@ else:
warnings.simplefilter("always", SupervisionWarnings)
def warn_deprecated(message: str):
"""
Issue a warning that a function is deprecated.
Args:
message (str): The message to display when the function is called.
"""
warnings.warn(message, category=SupervisionWarnings, stacklevel=2)
def deprecated_parameter(
old_parameter: str,
new_parameter: str,
@ -82,15 +92,13 @@ def deprecated_parameter(
else:
function_name = func.__name__
warnings.warn(
warn_deprecated(
message=warning_message.format(
function_name=function_name,
old_parameter=old_parameter,
new_parameter=new_parameter,
**message_kwargs,
),
category=SupervisionWarnings,
stacklevel=2,
)
)
kwargs[new_parameter] = map_function(kwargs.pop(old_parameter))
@ -106,11 +114,7 @@ def deprecated(reason: str):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
warnings.warn(
f"{func.__name__} is deprecated: {reason}",
category=SupervisionWarnings,
stacklevel=2,
)
warn_deprecated(f"{func.__name__} is deprecated: {reason}")
return func(*args, **kwargs)
return wrapper

14
supervision/utils/iterables.py
View File

@ -68,3 +68,17 @@ def fill(sequence: List[V], desired_size: int, content: V) -> List[V]:
missing_size = max(0, desired_size - len(sequence))
sequence.extend([content] * missing_size)
return sequence
def find_duplicates(sequence: List) -> List:
"""
Find all duplicate elements in the input sequence.
"""
seen = set()
duplicates = set()
for element in sequence:
if element in seen:
duplicates.add(element)
else:
seen.add(element)
return list(duplicates)