src/Baselines/radarcam-depth/data/datasets.py

import torch
import torch.utils.data
from torch.utils.data import Dataset
import numpy as np
import data.data_utils as data_utils
import random
import os
from PIL import Image
from data.data_utils import load_depth


def random_sample(T):
    '''
    Arg(s):
        T : numpy[float32]
            C x N array
    Returns:
        numpy[float32] : random sample from T
    '''

    index = np.random.randint(0, T.shape[0])
    return T[index, :]


def random_crop(inputs, shape, crop_type=['none']):
    '''
    Apply crop to inputs e.g. images, depth

    Arg(s):
        inputs : list[numpy[float32]]
            list of numpy arrays e.g. images, depth, and validity maps
        shape : list[int]
            shape (height, width) to crop inputs
        crop_type : str
            none, horizontal, vertical, anchored, top, bottom, left, right, center
    Return:
        list[numpy[float32]] : list of cropped inputs
    '''

    n_height, n_width = shape
    _, o_height, o_width = inputs[0].shape

    # Get delta of crop and original height and width

    d_height = o_height - n_height
    d_width = o_width - n_width

    # By default, perform center crop
    y_start = d_height // 2
    x_start = d_width // 2

    # If left alignment, then set starting height to 0
    if 'left' in crop_type:
        x_start = 0

    # If right alignment, then set starting height to right most position
    elif 'right' in crop_type:
        x_start = d_width

    elif 'horizontal' in crop_type:

        # Select from one of the pre-defined anchored locations
        if 'anchored' in crop_type:
            # Create anchor positions
            crop_anchors = [
                0.0, 0.50, 1.0
            ]

            widths = [
                anchor * d_width for anchor in crop_anchors
            ]
            x_start = int(widths[np.random.randint(low=0, high=len(widths))])

        # Randomly select a crop location
        else:
            x_start = np.random.randint(low=0, high=d_width)

    # If top alignment, then set starting height to 0
    if 'top' in crop_type:
        y_start = 0

    # If bottom alignment, then set starting height to lowest position
    elif 'bottom' in crop_type:
        y_start = d_height

    elif 'vertical' in crop_type and np.random.rand() <= 0.30:

        # Select from one of the pre-defined anchored locations
        if 'anchored' in crop_type:
            # Create anchor positions
            crop_anchors = [
                0.0, 0.50, 1.0
            ]

            heights = [
                anchor * d_height for anchor in crop_anchors
            ]
            y_start = int(heights[np.random.randint(low=0, high=len(heights))])

        # Randomly select a crop location
        else:
            y_start = np.random.randint(low=0, high=d_height)

    elif 'center' in crop_type:
        pass

    # Crop each input into (n_height, n_width)
    y_end = y_start + n_height
    x_end = x_start + n_width

    outputs = [
        T[:, y_start:y_end, x_start:x_end] for T in inputs
    ]

    return outputs


class RCNetTrainingDataset(torch.utils.data.Dataset):
    '''
    Dataset for fetching:
        (1) image
        (2) radar point
        (3) ground truth
        (4) bounding boxes for the points
        (5) image crops for summary part of the code

    Arg(s):
        image_paths : list[str]
            paths to images
        radar_paths : list[str]
            paths to radar points
        ground_truth_paths : list[str]
            paths to ground truth depth maps
        crop_width : int
            width of crop centered at the radar point
        total_points_sampled: int
            total number of points sampled from the total radar points available. Repeats the same points multiple times if total points in the frame is less than total sampled points
        sample_probability_of_lidar: int
            randomly sample lidar with this probability and add noise to it instead of using radar points
        min_radar_depth_m: float
            minimum depth accepted for synthetic radar sampling
        max_radar_depth_m: float
            maximum depth accepted for synthetic radar sampling
    '''

    def __init__(self,
                 image_paths,
                 radar_paths,
                 ground_truth_paths,
        patch_size,
        total_points_sampled,
        sample_probability_of_lidar,
        min_radar_depth_m=0.05,
        max_radar_depth_m=11.2):

        self.n_sample = len(image_paths)

        assert self.n_sample == len(ground_truth_paths)
        assert self.n_sample == len(radar_paths)

        self.image_paths = image_paths
        self.radar_paths = radar_paths
        self.ground_truth_paths = ground_truth_paths

        self.patch_size = patch_size
        self.pad_size_x = patch_size[1] // 2
        self.padding = ((0, 0), (0, 0), (self.pad_size_x, self.pad_size_x))

        self.data_format = 'CHW'
        self.total_points_sampled = total_points_sampled
        self.sample_probability_of_lidar = sample_probability_of_lidar
        self.min_radar_depth_m = min_radar_depth_m
        self.max_radar_depth_m = max_radar_depth_m

    def __getitem__(self, index):

        # Load image
        image = data_utils.load_image(
            self.image_paths[index],
            normalize=False,
            data_format=self.data_format)

        height, width = image.shape[1:]
        if height == 720: # ZJU dataset
            image = image[:, 720 // 3: 720 // 4 * 3, :]

        image = np.pad(
            image,
            pad_width=self.padding,
            mode='edge')

        # Load radar points N x 3
        radar_points = np.load(self.radar_paths[index])

        if height == 720:
            radar_points = radar_points[radar_points[:, 1] < 720 // 4 * 3]
            radar_points[:, 1] = radar_points[:, 1] - 720 // 3
            radar_points = radar_points[radar_points[:, 1] >= 0]

        if radar_points.ndim == 1:
            # Only one point (,3), expand to 1 x 3
            radar_points = np.expand_dims(radar_points, axis=0)

        # Store bounding boxes for all radar points
        bounding_boxes_list = []

        # randomly sample radar points to output
        if radar_points.shape[0] <= self.total_points_sampled:
            radar_points = np.repeat(radar_points, 100, axis=0)
        random_idx = np.random.randint(radar_points.shape[0], size=self.total_points_sampled)
        radar_points = radar_points[random_idx, :]

        # Load ground truth depth
        ground_truth = data_utils.load_depth(
            self.ground_truth_paths[index],
            data_format=self.data_format)

        if height == 720:
            ground_truth = ground_truth[:, 720 // 3: 720 // 4 * 3]

        if random.random() < self.sample_probability_of_lidar:
            ground_truth_for_sampling = np.copy(ground_truth)
            ground_truth_for_sampling = ground_truth_for_sampling.squeeze()
            valid_lidar = np.isfinite(ground_truth_for_sampling)
            valid_lidar &= ground_truth_for_sampling >= self.min_radar_depth_m
            valid_lidar &= ground_truth_for_sampling <= self.max_radar_depth_m
            idx_lidar_samples = np.where(valid_lidar)
            n_lidar_samples = len(idx_lidar_samples[0])

            if n_lidar_samples > 0:
                # Keep the fixed point count required by RC-Net. Replacement
                # handles frames with fewer valid GT pixels than requested.
                if n_lidar_samples >= self.total_points_sampled:
                    random_indices = random.sample(
                        range(n_lidar_samples), self.total_points_sampled
                    )
                else:
                    random_indices = np.random.choice(
                        n_lidar_samples,
                        size=self.total_points_sampled,
                        replace=True,
                    )

                points_x = idx_lidar_samples[1][random_indices]
                points_y = idx_lidar_samples[0][random_indices]
                points_z = ground_truth_for_sampling[points_y, points_x]

                noise_for_fake_radar_x = np.random.normal(0, 25, radar_points.shape[0])
                noise_for_fake_radar_z = np.random.uniform(low=0.0, high=0.4, size=radar_points.shape[0])

                fake_radar_points = np.copy(radar_points)
                fake_radar_points[:, 0] = points_x + noise_for_fake_radar_x
                fake_radar_points[:, 0] = np.clip(fake_radar_points[:, 0], 0, ground_truth_for_sampling.shape[1])
                fake_radar_points[:, 2] = points_z + noise_for_fake_radar_z
                # we keep the y as the same it is since it is erroneous

                # convert x and y indices back to int after adding noise
                fake_radar_points[:, 0] = fake_radar_points[:, 0].astype(int)
                fake_radar_points[:, 1] = fake_radar_points[:, 1].astype(int)

                radar_points = np.copy(fake_radar_points)

        # get the shifted radar points after padding
        for radar_point_idx in range(0, radar_points.shape[0]):
            # Set radar point to the center of the patch
            radar_points[radar_point_idx, 0] = radar_points[radar_point_idx, 0] + self.pad_size_x

            bounding_box = [0, 0, 0, 0]
            bounding_box[0] = radar_points[radar_point_idx, 0] - self.pad_size_x
            bounding_box[1] = 0
            bounding_box[2] = radar_points[radar_point_idx, 0] + self.pad_size_x
            bounding_box[3] = self.patch_size[0]
            bounding_boxes_list.append(np.asarray(bounding_box))

        ground_truth = np.pad(
            ground_truth,
            pad_width=self.padding,
            mode='constant',
            constant_values=0)

        ground_truth_crops = []

        # Crop image and ground truth
        for radar_point_idx in range(0, radar_points.shape[0]):
            start_x = int(radar_points[radar_point_idx, 0] - self.pad_size_x)
            end_x = int(radar_points[radar_point_idx, 0] + self.pad_size_x)
            start_y = image.shape[-2] - self.patch_size[0]

            ground_truth_cropped = ground_truth[:, start_y:, start_x:end_x]
            ground_truth_crops.append(ground_truth_cropped)

        image = image[:, start_y:, ...]

        ground_truth = np.asarray(ground_truth_crops)

        # Convert to float32
        image, radar_points, ground_truth = [
            T.astype(np.float32)
            for T in [image, radar_points, ground_truth]
        ]

        bounding_boxes_list = [T.astype(np.float32) for T in bounding_boxes_list]

        bounding_boxes_list = np.stack(bounding_boxes_list, axis=0)

        return image, radar_points, bounding_boxes_list, ground_truth

    def __len__(self):
        return self.n_sample


class RCNetInferenceDataset(torch.utils.data.Dataset):
    '''
    Dataset for fetching:
        (1) image
        (2) radar points
        (3) ground truth (if available)

    Arg(s):
        image_paths : list[str]
            paths to images
        radar_paths : list[str]
            paths to radar points
        ground_truth_paths : list[str]
            paths to ground truth paths
    '''

    def __init__(self, image_paths, radar_paths, ground_truth_paths=None):

        self.n_sample = len(image_paths)

        assert self.n_sample == len(radar_paths)

        self.image_paths = image_paths
        self.radar_paths = radar_paths

        if ground_truth_paths is not None and None not in ground_truth_paths:
            assert self.n_sample == len(ground_truth_paths)
            self.ground_truth_available = True
        else:
            self.ground_truth_available = False

        self.ground_truth_paths = ground_truth_paths

        self.data_format = 'CHW'

    def __getitem__(self, index):

        # Load image
        image = data_utils.load_image(
            self.image_paths[index],
            normalize=False,
            data_format=self.data_format)

        height, width = image.shape[1:]
        if height == 720: # ZJU dataset
            image = image[:, 720 // 3: 720 // 4 * 3, :]

        # Load radar points N x 3
        radar_points = np.load(self.radar_paths[index])

        if height == 720:
            radar_points = radar_points[radar_points[:, 1] < 720 // 4 * 3]
            radar_points[:, 1] = radar_points[:, 1] - 720 // 3
            radar_points = radar_points[radar_points[:, 1] >= 0]

        if radar_points.ndim == 1:
            # Expand to 1 x 3
            radar_points = np.expand_dims(radar_points, axis=0)

        inputs = [image, radar_points]

        if self.ground_truth_available:
            # Load ground truth depth
            ground_truth = data_utils.load_depth(
                self.ground_truth_paths[index],
                data_format=self.data_format)
            if height == 720:
                ground_truth = ground_truth[:, 720 // 3: 720 // 4 * 3]

            inputs.append(ground_truth)

        # Convert to float32
        inputs = [
            T.astype(np.float32)
            for T in inputs
        ]

        return inputs

    def __len__(self):
        return self.n_sample