mapper/data/kitti/transform.py

import numpy as np
import torch
from torchvision.transforms import functional as tfn
import torchvision.transforms.functional as tvf

from ..utils import decompose_rotmat
from ..image import pad_image, rectify_image, resize_image
from ...utils.wrappers import Camera
from ..schema import KITTIDataConfiguration


class BEVTransform:
    def __init__(self,
                 cfg: KITTIDataConfiguration, augmentations):
        self.cfg = cfg
        self.augmentations = augmentations

    @staticmethod
    def _compact_labels(msk, cat, iscrowd):
        ids = np.unique(msk)
        if 0 not in ids:
            ids = np.concatenate((np.array([0], dtype=np.int32), ids), axis=0)

        ids_to_compact = np.zeros((ids.max() + 1,), dtype=np.int32)
        ids_to_compact[ids] = np.arange(0, ids.size, dtype=np.int32)

        msk = ids_to_compact[msk]
        cat = cat[ids]
        iscrowd = iscrowd[ids]

        return msk, cat, iscrowd

    def __call__(self, img, bev_msk=None, bev_plabel=None, fv_msk=None, bev_weights_msk=None,
                 bev_cat=None, bev_iscrowd=None, fv_cat=None, fv_iscrowd=None,
                 fv_intrinsics=None, ego_pose=None):
        # Wrap in np.array
        if bev_cat is not None:
            bev_cat = np.array(bev_cat, dtype=np.int32)
        if bev_iscrowd is not None:
            bev_iscrowd = np.array(bev_iscrowd, dtype=np.uint8)

        if ego_pose is not None:
            ego_pose = np.array(ego_pose, dtype=np.float32)

        roll, pitch, yaw = decompose_rotmat(ego_pose[:3, :3])

        # Image transformations
        img = tfn.to_tensor(img)
        # img = [self._normalize_image(rgb) for rgb in img]
        fx = fv_intrinsics[0][0]
        fy = fv_intrinsics[1][1]
        cx = fv_intrinsics[0][2]
        cy = fv_intrinsics[1][2]
        width = img.shape[2]
        height = img.shape[1]

        cam = Camera(torch.tensor(
                        [width, height, fx, fy, cx - 0.5, cy - 0.5])).float()

        if not self.cfg.gravity_align:
            # Turn off gravity alignment
            roll = 0.0
            pitch = 0.0
            img, valid = rectify_image(img, cam, roll, pitch)
        else:
            img, valid = rectify_image(
                img, cam, roll, pitch if self.cfg.rectify_pitch else None
            )
            roll = 0.0
            if self.cfg.rectify_pitch:
                pitch = 0.0

        if self.cfg.target_focal_length is not None:
            # Resize to a canonical focal length
            factor = self.cfg.target_focal_length / cam.f.numpy()
            size = (np.array(img.shape[-2:][::-1]) * factor).astype(int)
            img, _, cam, valid = resize_image(img, size, camera=cam, valid=valid)
            size_out = self.cfg.resize_image
            if size_out is None:
                # Round the edges up such that they are multiple of a factor
                stride = self.cfg.pad_to_multiple
                size_out = (np.ceil((size / stride)) * stride).astype(int)
            # Crop or pad such that both edges are of the given size
            img, valid, cam = pad_image(
                img, size_out, cam, valid, crop_and_center=False
            )
        elif self.cfg.resize_image is not None:
            img, _, cam, valid = resize_image(
                img, self.cfg.resize_image, fn=max, camera=cam, valid=valid
            )
            if self.cfg.pad_to_square:
                # Pad such that both edges are of the given size
                img, valid, cam = pad_image(img, self.cfg.resize_image, cam, valid)

        # Label transformations,
        if bev_msk is not None:
            bev_msk = np.expand_dims(
                np.array(bev_msk, dtype=np.int32, copy=False),
                axis=0
            )
            bev_msk, bev_cat, bev_iscrowd = self._compact_labels(
                bev_msk, bev_cat, bev_iscrowd
            )

            bev_msk = torch.from_numpy(bev_msk)
            bev_cat = torch.from_numpy(bev_cat)

            rotated_mask = torch.rot90(bev_msk, dims=(1, 2))
            cropped_mask = rotated_mask[:, :672, (rotated_mask.size(2) - 672) // 2:-(rotated_mask.size(2) - 672) // 2]

            bev_msk = cropped_mask.squeeze(0)
            seg_masks = bev_cat[bev_msk]

            seg_masks_onehot = seg_masks.clone()
            seg_masks_onehot[seg_masks_onehot == 255] = 0
            seg_masks_onehot = torch.nn.functional.one_hot(
                seg_masks_onehot.to(torch.int64),
                num_classes=self.cfg.num_classes
            )
            seg_masks_onehot[seg_masks == 255] = 0

            seg_masks_onehot = seg_masks_onehot.permute(2, 0, 1)

            seg_masks_down = tvf.resize(seg_masks_onehot, (100, 100))

            seg_masks_down = seg_masks_down.permute(1, 2, 0)

            if self.cfg.class_mapping is not None:
                seg_masks_down = seg_masks_down[:, :, self.cfg.class_mapping]

        img = self.augmentations(img)
        flood_masks = torch.all(seg_masks_down == 0, dim=2).float()


        ret = {
            "image": img,
            "valid": valid,
            "camera": cam,
            "seg_masks": (seg_masks_down).float().contiguous(),
            "flood_masks": flood_masks,
            "roll_pitch_yaw": torch.tensor((roll, pitch, yaw)).float(),
            "confidence_map": flood_masks,
        }
        
        for key, value in ret.items():
            if isinstance(value, np.ndarray):
                ret[key] = torch.from_numpy(value)
    
        return ret