refactoring

constants.py

@@ -16,7 +16,7 @@
 
 from pathlib import Path
 
-CHECKPOINT_PATH = Path.cwd().joinpath('weights', 'sam_vit_h_4b8939.pth')
+CHECKPOINT_PATH = Path.cwd().joinpath("weights", "sam_vit_h_4b8939.pth")
 
-DEVICE = torch.device('cuda:0' if torch.cuda.is_available() else 'cpu')
+DEVICE = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
 MODEL_TYPE = "vit_h"

main.py

@@ -36,30 +36,32 @@
 
 
 def main():
-    dataset_path = 'dataset/'
-    sequence = '00'
-    image_instances_path = 'pipeline/vfm-labels/sam/00/'
+    dataset_path = "dataset/"
+    sequence = "00"
+    image_instances_path = "pipeline/vfm-labels/sam/00/"
     kitti = KittiDataset(dataset_path, sequence, image_instances_path)
 
-    config = ConfigDTO(**{
-    'dataset': kitti,
-    'start_index': 19,
-    'end_index': 23,
-    'start_image_index_offset': 3,
-    'cam_name': 'cam2',
-    'R': 12,
-    'nb_neighbors': 30,
-    'std_ratio': 5.0,
-    'voxel_size': 0.25
-    })
+    config = ConfigDTO(
+        **{
+            "dataset": kitti,
+            "start_index": 19,
+            "end_index": 23,
+            "start_image_index_offset": 3,
+            "cam_name": "cam2",
+            "R": 12,
+            "nb_neighbors": 30,
+            "std_ratio": 5.0,
+            "voxel_size": 0.25
+        }
+    )
 
     init_pcd = InitMapProcessor().process(config)
     points2instances = InitInstancesMatrixProcessor().process(config, init_pcd)
 
     processors = [
         SelectionNotZeroProcessor(),
         SelectionInCubeProcessor(),
-        StatisticalOutlierProcessor()
+        StatisticalOutlierProcessor(),
     ]
 
     pcd = copy.deepcopy(init_pcd)
@@ -68,7 +70,9 @@ def main():
 
     pcd_for_clustering = copy.deepcopy(pcd)
 
-    pcd, points2instances, trace = VoxelDownProcessor().process(config, pcd, points2instances)
+    pcd, points2instances, trace = VoxelDownProcessor().process(
+        config, pcd, points2instances
+    )
 
     points = np.asarray(pcd.points)
     spatial_distance = cdist(points, points)
@@ -81,9 +85,13 @@ def main():
 
     T = 0.2
     eigenval = 2
-    clusters = normalized_cut(dist, np.array([i for i in range(len(points))], dtype=int), T, eigenval)
+    clusters = normalized_cut(
+        dist, np.array([i for i in range(len(points))], dtype=int), T, eigenval
+    )
 
-    pcd_clustered = color_pcd_by_clusters_and_voxels(pcd_for_clustering, trace, clusters)
+    pcd_clustered = color_pcd_by_clusters_and_voxels(
+        pcd_for_clustering, trace, clusters
+    )
 
     o3d.visualization.draw_geometries([pcd_clustered])
 

pcd_dataset/kitti_dataset.py

@@ -37,50 +37,52 @@ def get_point_cloud(self, index):
 
     def get_lidar_pose(self, index):
         return (
-                np.linalg.inv(self.get_camera_extrinsics('cam0'))
-                @ self.dataset.poses[index]
-                @ self.get_camera_extrinsics('cam0')
+            np.linalg.inv(self.get_camera_extrinsics("cam0"))
+            @ self.dataset.poses[index]
+            @ self.get_camera_extrinsics("cam0")
         )
 
     def get_camera_names(self):
-        return ['cam0', 'cam1', 'cam2', 'cam3']
+        return ["cam0", "cam1", "cam2", "cam3"]
 
     def get_camera_image(self, cam_name, index):
         image, color = None, None
-        if cam_name == 'cam0':
+        if cam_name == "cam0":
             image, color = self.dataset.get_cam0(index), cv2.COLOR_GRAY2BGR
-        elif cam_name == 'cam1':
+        elif cam_name == "cam1":
             image, color = self.dataset.get_cam1(index), cv2.COLOR_GRAY2BGR
-        elif cam_name == 'cam2':
+        elif cam_name == "cam2":
             image, color = self.dataset.get_cam2(index), cv2.COLOR_RGB2BGR
-        elif cam_name == 'cam3':
+        elif cam_name == "cam3":
             image, color = self.dataset.get_cam3(index), cv2.COLOR_RGB2BGR
         return cv2.cvtColor(np.array(image), color)
 
     def get_image_instances(self, cam_name, index):
-        masks_path = Path.cwd().joinpath(self.image_instances_path, cam_name, '{}.npz'.format(str(index).zfill(6)))
-        return np.load(masks_path, allow_pickle=True)['masks']
+        masks_path = Path.cwd().joinpath(
+            self.image_instances_path, cam_name, "{}.npz".format(str(index).zfill(6))
+        )
+        return np.load(masks_path, allow_pickle=True)["masks"]
 
     def get_camera_intrinsics(self, cam_name):
-        if cam_name == 'cam0':
+        if cam_name == "cam0":
             return self.dataset.calib.K_cam0
-        elif cam_name == 'cam1':
+        elif cam_name == "cam1":
             return self.dataset.calib.K_cam1
-        elif cam_name == 'cam2':
+        elif cam_name == "cam2":
             return self.dataset.calib.K_cam2
-        elif cam_name == 'cam3':
+        elif cam_name == "cam3":
             return self.dataset.calib.K_cam3
         else:
             return None
 
     def get_camera_extrinsics(self, cam_name):
-        if cam_name == 'cam0':
+        if cam_name == "cam0":
             return self.dataset.calib.T_cam0_velo
-        elif cam_name == 'cam1':
+        elif cam_name == "cam1":
             return self.dataset.calib.T_cam1_velo
-        elif cam_name == 'cam2':
+        elif cam_name == "cam2":
             return self.dataset.calib.T_cam2_velo
-        elif cam_name == 'cam3':
+        elif cam_name == "cam3":
             return self.dataset.calib.T_cam3_velo
         else:
             return None

services/normalized_cut_service.py

@@ -33,14 +33,15 @@ def ncut_cost(W, D, cut):
         current n-cut, whose cost needs to be calculated
     """
     cost = cut_cost(W, cut)
-    assoc_a = D.todense()[cut].sum()  # Anastasiia: this also can be optimized in the future
+    # Anastasiia: this also can be optimized in the future
+    assoc_a = D.todense()[cut].sum()
     assoc_b = D.todense()[~cut].sum()
     return (cost / assoc_a) + (cost / assoc_b)
 
 
 def get_min_ncut(ev, d, w, num_cuts):
     """Construction of a minimal graph cut based on a normalized similarity criterion
-    
+
     Parameters
     ----------
     ev : eigenvector
@@ -77,7 +78,7 @@ def get_min_ncut(ev, d, w, num_cuts):
 
 def normalized_cut(w, labels, T=0.01, eigenvalues_count=2):
     """Implementation of the GraphCut algorithm for segmentation labels based on a matrix of distances W between them
-    
+
     Parameters
     ----------
     w : matrix
@@ -100,18 +101,24 @@ def normalized_cut(w, labels, T=0.01, eigenvalues_count=2):
 
         A = D2 * (D - W) * D2
 
-        eigvals, eigvecs = sparse.linalg.eigsh(A, eigenvalues_count, sigma=0, which='LM')
+        eigvals, eigvecs = sparse.linalg.eigsh(
+            A, eigenvalues_count, sigma=0, which='LM'
+        )
 
         index2 = np.argsort(eigvals)[1]
 
         ev = eigvecs[:, index2]
-        
+
         mask, mcut = get_min_ncut(ev, D, w, 10)
         print("mcut = {}".format(mcut))
 
         if mcut < T:
-            labels1 = normalized_cut(w[mask][:, mask], labels[mask], T, eigenvalues_count)
-            labels2 = normalized_cut(w[~mask][:, ~mask], labels[~mask], T, eigenvalues_count)
+            labels1 = normalized_cut(
+                w[mask][:, mask], labels[mask], T, eigenvalues_count
+            )
+            labels2 = normalized_cut(
+                w[~mask][:, ~mask], labels[~mask], T, eigenvalues_count
+            )
             return labels1 + labels2
         else:
             return [labels]

services/preprocessing/common/config.py

@@ -39,43 +39,62 @@ def image_offset_less_than_start_index(instance, attribute, value):
 
 @attr.s
 class ConfigDTO:
-   """Config is used to set cloud preprocessing parameters before segmentation
-
-   Attributes
-   ----------
-   dataset : AbstractDataset
-      an instance of a dataset of clouds and images (e.g. KITTI)
-   start_index : int
-      number of the first cloud in the processed sequence
-   end_index : int
-      number of the last cloud in the processed sequence
-   start_image_index_offset : int
-      offset of the first number of images of the covering sequence relative to the start cloud
-   cam_name : str
-      name of the camera whose images will be used
-   R : int
-      side of a cube within which points are considered
-   nb_neighbors : int
-      number of neighbors taken into account in the statistical outlier removal function
-   std_ratio : float
-      the threshold level in the statistical outlier removal function
-   voxel_size : float
-      voxel size to downsample into
-   """
-
-   dataset: AbstractDataset = attr.ib()
-
-   start_index: int = attr.ib(default=5, validator=[attr.validators.instance_of(int), is_positive])
-   end_index: int = attr.ib(default=10, validator=[attr.validators.instance_of(int), end_greater_than_start])
-   start_image_index_offset: int = attr.ib(
-      default=3, validator=[attr.validators.instance_of(int), is_positive, image_offset_less_than_start_index]
-   )
-
-   cam_name: str = attr.ib(default='cam2', validator=[attr.validators.instance_of(str), is_valid_cam_name])
-
-   R: int = attr.ib(default=12, validator=[attr.validators.instance_of(int), is_positive])
-
-   nb_neighbors: int = attr.ib(default=25, validator=[attr.validators.instance_of(int), is_positive])
-   std_ratio: float = attr.ib(default=5.0, validator=[attr.validators.instance_of(float), is_positive])
-
-   voxel_size: float = attr.ib(default=0.3, validator=[attr.validators.instance_of(float), is_positive])
+    """Config is used to set cloud preprocessing parameters before segmentation
+
+    Attributes
+    ----------
+    dataset : AbstractDataset
+       an instance of a dataset of clouds and images (e.g. KITTI)
+    start_index : int
+       number of the first cloud in the processed sequence
+    end_index : int
+       number of the last cloud in the processed sequence
+    start_image_index_offset : int
+       offset of the first number of images of the covering sequence relative to the start cloud
+    cam_name : str
+       name of the camera whose images will be used
+    R : int
+       side of a cube within which points are considered
+    nb_neighbors : int
+       number of neighbors taken into account in the statistical outlier removal function
+    std_ratio : float
+       the threshold level in the statistical outlier removal function
+    voxel_size : float
+       voxel size to downsample into
+    """
+
+    dataset: AbstractDataset = attr.ib()
+
+    start_index: int = attr.ib(
+       default=5, validator=[attr.validators.instance_of(int), is_positive]
+    )
+    end_index: int = attr.ib(
+       default=10, validator=[attr.validators.instance_of(int), end_greater_than_start]
+    )
+    start_image_index_offset: int = attr.ib(
+       default=3, 
+       validator=[
+          attr.validators.instance_of(int), 
+          is_positive, 
+          image_offset_less_than_start_index
+       ],
+    )
+
+    cam_name: str = attr.ib(
+       default='cam2', validator=[attr.validators.instance_of(str), is_valid_cam_name]
+    )
+
+    R: int = attr.ib(
+       default=12, validator=[attr.validators.instance_of(int), is_positive]
+    )
+
+    nb_neighbors: int = attr.ib(
+       default=25, validator=[attr.validators.instance_of(int), is_positive]
+    )
+    std_ratio: float = attr.ib(
+       default=5.0, validator=[attr.validators.instance_of(float), is_positive]
+    )
+
+    voxel_size: float = attr.ib(
+       default=0.3, validator=[attr.validators.instance_of(float), is_positive]
+    )

services/preprocessing/in_cube.py

@@ -28,12 +28,13 @@ def process(self, config, pcd, points2instances):
         The cube starts from the lidar location at the moment of recording the config.start_index cloud.
         """
 
-        T_first_cam = (
-            config.dataset.get_lidar_pose(config.start_index) 
-            @ np.linalg.inv(config.dataset.get_camera_extrinsics(config.cam_name))
+        T_first_cam = config.dataset.get_lidar_pose(config.start_index) @ np.linalg.inv(
+            config.dataset.get_camera_extrinsics(config.cam_name)
         )
 
-        close_point_indices = self.get_close_point_indices_cube(pcd, T_first_cam, config.R)
+        close_point_indices = self.get_close_point_indices_cube(
+            pcd, T_first_cam, config.R
+        )
 
         pcd_in_cube = get_subpcd(pcd, close_point_indices)
         points2instances_in_cube = points2instances[close_point_indices]