str or path and format

Author: Matt Sokoloff

labelbox/data/serialization/coco/annotation.py

@@ -6,7 +6,7 @@
 import numpy as np
 
 
-def rle_decoding(rle_arr : List[int], w : int, h: int) -> np.ndarray:
+def rle_decoding(rle_arr: List[int], w: int, h: int) -> np.ndarray:
     indices = []
     for idx, cnt in zip(rle_arr[0::2], rle_arr[1::2]):
         indices.extend(list(range(idx - 1,

labelbox/data/serialization/coco/converter.py

@@ -1,4 +1,4 @@
-from typing import Dict, Any
+from typing import Dict, Any, Union
 from pathlib import Path
 import os
 
@@ -7,16 +7,20 @@
 from labelbox.data.serialization.coco.panoptic_dataset import CocoPanopticDataset
 
 
-def create_path_if_not_exists(path: Path, ignore_existing_data=False):
+def create_path_if_not_exists(path: Union[Path, str],
+                              ignore_existing_data=False):
+    path = Path(path)
     if not path.exists():
         path.mkdir(parents=True, exist_ok=True)
     elif not ignore_existing_data and os.listdir(path):
         raise ValueError(
             f"Directory `{path}`` must be empty. Or set `ignore_existing_data=True`"
         )
+    return path
 
 
-def validate_path(path: Path, name: str):
+def validate_path(path: Union[Path, str], name: str):
+    path = Path(path)
     if not path.exists():
         raise ValueError(f"{name} `{path}` must exist")
 
@@ -29,10 +33,12 @@ class COCOConverter:
     Subclasses are currently ignored.
     To use subclasses, manually flatten them before using the converter.
     """
+
     @staticmethod
     def serialize_instances(labels: LabelCollection,
-                            image_root: Path,
-                            ignore_existing_data=False, max_workers = 8) -> Dict[str, Any]:
+                            image_root: Union[Path, str],
+                            ignore_existing_data=False,
+                            max_workers=8) -> Dict[str, Any]:
         """
         Convert a Labelbox LabelCollection into an mscoco dataset.
         This function will only convert masks, polygons, and rectangles.
@@ -48,16 +54,18 @@ def serialize_instances(labels: LabelCollection,
         Returns:
             A dictionary containing labels in the coco object format.
         """
-        create_path_if_not_exists(image_root, ignore_existing_data)
+        image_root = create_path_if_not_exists(image_root, ignore_existing_data)
         return CocoInstanceDataset.from_common(labels=labels,
-                                               image_root=image_root, max_workers = max_workers).dict()
+                                               image_root=image_root,
+                                               max_workers=max_workers).dict()
 
     @staticmethod
     def serialize_panoptic(labels: LabelCollection,
-                           image_root: Path,
-                           mask_root: Path,
+                           image_root: Union[Path, str],
+                           mask_root: Union[Path, str],
                            all_stuff: bool = False,
-                           ignore_existing_data=False, max_workers = 8) -> Dict[str, Any]:
+                           ignore_existing_data=False,
+                           max_workers=8) -> Dict[str, Any]:
         """
         Convert a Labelbox LabelCollection into an mscoco dataset.
         This function will only convert masks, polygons, and rectangles.
@@ -76,16 +84,18 @@ def serialize_panoptic(labels: LabelCollection,
         Returns:
             A dictionary containing labels in the coco panoptic format.
         """
-        create_path_if_not_exists(image_root, ignore_existing_data)
-        create_path_if_not_exists(mask_root, ignore_existing_data)
+        image_root = create_path_if_not_exists(image_root, ignore_existing_data)
+        mask_root = create_path_if_not_exists(mask_root, ignore_existing_data)
         return CocoPanopticDataset.from_common(labels=labels,
                                                image_root=image_root,
                                                mask_root=mask_root,
-                                               all_stuff=all_stuff , max_workers = max_workers).dict()
+                                               all_stuff=all_stuff,
+                                               max_workers=max_workers).dict()
 
     @staticmethod
-    def deserialize_panoptic(json_data: Dict[str, Any], image_root: Path,
-                             mask_root: Path) -> LabelGenerator:
+    def deserialize_panoptic(json_data: Dict[str, Any], image_root: Union[Path,
+                                                                          str],
+                             mask_root: Union[Path, str]) -> LabelGenerator:
         """
         Convert coco panoptic data into the labelbox format (as a LabelGenerator).
 
@@ -96,8 +106,8 @@ def deserialize_panoptic(json_data: Dict[str, Any], image_root: Path,
         Returns:
             LabelGenerator
         """
-        validate_path(image_root, 'image_root')
-        validate_path(mask_root, 'mask_root')
+        image_root = validate_path(image_root, 'image_root')
+        mask_root = validate_path(mask_root, 'mask_root')
         objs = CocoPanopticDataset(**json_data)
         gen = objs.to_common(image_root, mask_root)
         return LabelGenerator(data=gen)
@@ -114,7 +124,7 @@ def deserialize_instances(json_data: Dict[str, Any],
         Returns:
             LabelGenerator
         """
-        validate_path(image_root, 'image_root')
+        image_root = validate_path(image_root, 'image_root')
         objs = CocoInstanceDataset(**json_data)
         gen = objs.to_common(image_root)
         return LabelGenerator(data=gen)

labelbox/data/serialization/coco/image.py

@@ -26,18 +26,15 @@ def get_image_id(label: Label, idx: int) -> int:
     return idx
 
 
-def get_image(label: Label, image_root : Path, image_id: str) -> CocoImage:
+def get_image(label: Label, image_root: Path, image_id: str) -> CocoImage:
     path = Path(image_root, f"{image_id}.jpg")
     if not path.exists():
         im = Image.fromarray(label.data.value)
         im.save(path)
         w, h = im.size
     else:
         w, h = imagesize.get(str(path))
-    return CocoImage(id=image_id,
-                     width=w,
-                     height=h,
-                     file_name=Path(path.name))
+    return CocoImage(id=image_id, width=w, height=h, file_name=Path(path.name))
 
 
 def id_to_rgb(id: int) -> Tuple[int, int, int]:

labelbox/data/serialization/coco/instance_dataset.py

@@ -15,8 +15,9 @@
 from .image import CocoImage, get_image, get_image_id
 
 
-def mask_to_coco_object_annotation(annotation: ObjectAnnotation, annot_idx : int,
-                                   image_id : int, category_id : int) -> COCOObjectAnnotation:
+def mask_to_coco_object_annotation(annotation: ObjectAnnotation, annot_idx: int,
+                                   image_id: int,
+                                   category_id: int) -> COCOObjectAnnotation:
     # This is going to fill any holes into the multipolygon
     # If you need to support holes use the panoptic data format
     shapely = annotation.value.shapely.simplify(1).buffer(0)
@@ -38,8 +39,9 @@ def mask_to_coco_object_annotation(annotation: ObjectAnnotation, annot_idx : int
         iscrowd=0)
 
 
-def vector_to_coco_object_annotation(annotation: ObjectAnnotation, annot_idx : int,
-                                     image_id: int, category_id: int) -> COCOObjectAnnotation:
+def vector_to_coco_object_annotation(annotation: ObjectAnnotation,
+                                     annot_idx: int, image_id: int,
+                                     category_id: int) -> COCOObjectAnnotation:
     shapely = annotation.value.shapely
     xmin, ymin, xmax, ymax = shapely.bounds
     segmentation = []
@@ -62,15 +64,17 @@ def vector_to_coco_object_annotation(annotation: ObjectAnnotation, annot_idx : i
                                 iscrowd=0)
 
 
-def rle_to_common(class_annotations : COCOObjectAnnotation, class_name : str) -> ObjectAnnotation:
+def rle_to_common(class_annotations: COCOObjectAnnotation,
+                  class_name: str) -> ObjectAnnotation:
     mask = rle_decoding(class_annotations.segmentation.counts,
                         *class_annotations.segmentation.size[::-1])
     return ObjectAnnotation(name=class_name,
                             value=Mask(mask=MaskData.from_2D_arr(mask),
                                        color=[1, 1, 1]))
 
 
-def segmentations_to_common(class_annotations : COCOObjectAnnotation, class_name: str) -> List[ObjectAnnotation]:
+def segmentations_to_common(class_annotations: COCOObjectAnnotation,
+                            class_name: str) -> List[ObjectAnnotation]:
     # Technically it is polygons. But the key in coco is called segmentations..
     annotations = []
     for points in class_annotations.segmentation:
@@ -83,10 +87,12 @@ def segmentations_to_common(class_annotations : COCOObjectAnnotation, class_name
     return annotations
 
 
-def process_label(label: Label,
-                  idx : int,
-                  image_root :str,
-                  max_annotations_per_image=10000) -> Tuple[np.ndarray, List[COCOObjectAnnotation], Dict[str, str]]:
+def process_label(
+    label: Label,
+    idx: int,
+    image_root: str,
+    max_annotations_per_image=10000
+) -> Tuple[np.ndarray, List[COCOObjectAnnotation], Dict[str, str]]:
     annot_idx = idx * max_annotations_per_image
     image_id = get_image_id(label, idx)
     image = get_image(label, image_root, image_id)
@@ -119,7 +125,10 @@ class CocoInstanceDataset(BaseModel):
     categories: List[Categories]
 
     @classmethod
-    def from_common(cls, labels: LabelCollection, image_root : Path, max_workers = 8):
+    def from_common(cls,
+                    labels: LabelCollection,
+                    image_root: Path,
+                    max_workers=8):
         all_coco_annotations = []
         categories = {}
         images = []

labelbox/data/serialization/coco/panoptic_dataset.py

@@ -1,4 +1,3 @@
-
 from concurrent.futures import ProcessPoolExecutor, as_completed
 from typing import Dict, Any, List, Union
 from pathlib import Path
@@ -104,8 +103,12 @@ class CocoPanopticDataset(BaseModel):
     categories: List[Categories]
 
     @classmethod
-    def from_common(cls, labels: LabelCollection, image_root, mask_root,
-                    all_stuff, max_workers = 8):
+    def from_common(cls,
+                    labels: LabelCollection,
+                    image_root,
+                    mask_root,
+                    all_stuff,
+                    max_workers=8):
         all_coco_annotations = []
         coco_categories = {}
         coco_things = {}
@@ -167,8 +170,7 @@ def to_common(self, image_root: Path, mask_root: Path):
                 raise ValueError(
                     f"COCO masks must be stored as png files and their extension must be `.png`. Found {annotation.file_name}"
                 )
-            mask = MaskData(
-                file_path=Path(mask_root, annotation.file_name))
+            mask = MaskData(file_path=Path(mask_root, annotation.file_name))
 
             for segmentation in annotation.segments_info:
                 category = category_lookup[segmentation.category_id]