risk_biased/utils/waymo_dataloader.py

from typing import Tuple, List
from cv2 import repeat
from einops import rearrange, repeat
from torch.utils.data import Dataset
from torch.utils.data import DataLoader
import torch
from torch import Tensor
import numpy as np
import pickle
import os

from mmcv import Config


class WaymoDataset(Dataset):
    """
    Dataset loader for custom preprocessed files of Waymo data.
    Args:
        path: path to the dataset directory
        args: global settings
    """

    def __init__(self, cfg: Config, split: str, input_angle: bool = True):
        super(WaymoDataset, self).__init__()
        self.p_exchange_two_first = 1
        if "val" in split.lower():
            path = cfg.val_dataset_path
        elif "test" in split.lower():
            path = cfg.test_dataset_path
        elif "sample" in split.lower():
            path = cfg.sample_dataset_path
        else:
            path = cfg.train_dataset_path
            self.p_exchange_two_first = cfg.p_exchange_two_first

        self.file_list = [
            os.path.join(path, name)
            for name in os.listdir(path)
            if os.path.isfile(os.path.join(path, name))
        ]
        self.normalize = cfg.normalize_angle
        # self.load_dataset(path, 16)
        # self.idx_list = list(self.dataset.keys())
        self.input_angle = input_angle
        self.hist_len = cfg.num_steps
        self.fut_len = cfg.num_steps_future
        self.time_len = self.hist_len + self.fut_len
        self.min_num_obs = cfg.min_num_observation
        self.max_size_lane = cfg.max_size_lane
        self.random_rotation = cfg.random_rotation
        self.random_translation = cfg.random_translation
        self.angle_std = cfg.angle_std
        self.translation_distance_std = cfg.translation_distance_std
        self.max_num_agents = cfg.max_num_agents
        self.max_num_objects = cfg.max_num_objects
        self.state_dim = cfg.state_dim
        self.map_state_dim = cfg.map_state_dim
        self.dt = cfg.dt

        if "val" in os.path.basename(path).lower():
            self.dataset_size_limit = cfg.val_dataset_size_limit
        else:
            self.dataset_size_limit = cfg.train_dataset_size_limit

    def __len__(self):
        if self.dataset_size_limit is not None:
            return min(len(self.file_list), self.dataset_size_limit)
        else:
            return len(self.file_list)

    def __getitem__(
        self, idx: int
    ) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
        """
        Get the item at index idx in the dataset. Normalize the scene and output absolute angle and position.
        Returns:
            trajectories, mask, mask_loss, lanes, mask_lanes, angle, mean_position
        """
        selected_file = self.file_list[idx]
        with open(selected_file, "rb") as handle:
            dataset = pickle.load(handle)
        rel_state_all = dataset["traj"]
        mask_all = dataset["mask_traj"]
        mask_loss = dataset["mask_to_predict"]
        rel_lane_all = dataset["lanes"]
        mask_lane_all = dataset["mask_lanes"]
        mean_pos = dataset["mean_pos"]
        assert (
            (
                rel_state_all[self.hist_len + 5 :, :, :2][mask_all[self.hist_len + 5 :]]
                != 0
            )
            .any(-1)
            .all()
        )
        assert (
            (
                rel_state_all[self.hist_len + 5 :, :, :2][
                    mask_loss[self.hist_len + 5 :]
                ]
                != 0
            )
            .any(-1)
            .all()
        )
        if "lane_states" in dataset.keys():
            lane_states = dataset["lane_states"]
        else:
            lane_states = None
        if np.random.rand() > self.p_exchange_two_first:
            rel_state_all[:, [0, 1]] = rel_state_all[:, [1, 0]]
            mask_all[:, [0, 1]] = mask_all[:, [1, 0]]
            mask_loss[:, [0, 1]] = mask_loss[:, [1, 0]]
        assert (
            (
                rel_state_all[self.hist_len + 5 :, :, :2][mask_all[self.hist_len + 5 :]]
                != 0
            )
            .any(-1)
            .all()
        )
        assert (
            (
                rel_state_all[self.hist_len + 5 :, :, :2][
                    mask_loss[self.hist_len + 5 :]
                ]
                != 0
            )
            .any(-1)
            .all()
        )
        if self.normalize:
            angle = rel_state_all[self.hist_len - 1, 1, 2]

            if self.random_rotation:
                if self.normalize:
                    angle += np.random.normal(0, self.angle_std)
                else:
                    angle += np.random.uniform(-np.pi, np.pi)
            if self.random_translation:
                distance = (
                    np.random.normal([0, 0], self.translation_distance_std, 2)
                    * mask_all[self.hist_len - 1 : self.hist_len, :, None]
                    - rel_state_all[self.hist_len - 1 : self.hist_len, 1:2, :2]
                )
            else:
                distance = -rel_state_all[self.hist_len - 1 : self.hist_len, 1:2, :2]

            rel_state_all[:, :, :2] += distance
            rel_lane_all[:, :, :2] += distance
            mean_pos += distance[0, 0, :]
            rel_state_all = self.scene_rotation(rel_state_all, -angle)
            rel_lane_all = self.scene_rotation(rel_lane_all, -angle)

        else:
            if self.random_translation:
                distance = np.random.normal([0, 0], self.translation_distance_std, 2)
                rel_state_all = (
                    rel_state_all
                    + mask_all[self.hist_len - 1 : self.hist_len, :, None] * distance
                )
                rel_lane_all = (
                    rel_lane_all
                    + mask_all[self.hist_len - 1 : self.hist_len, :, None] * distance
                )

            if self.random_rotation:
                angle = np.random.uniform(0, 2 * np.pi)
                rel_state_all = self.scene_rotation(rel_state_all, angle)
                rel_lane_all = self.scene_rotation(rel_lane_all, angle)
            else:
                angle = 0
        return (
            rel_state_all,
            mask_all,
            mask_loss,
            rel_lane_all,
            mask_lane_all,
            lane_states,
            angle,
            mean_pos,
            idx,
        )

    @staticmethod
    def scene_rotation(coor: np.ndarray, angle: float) -> np.ndarray:
        """
        Rotate all the coordinates with the same angle
        Args:
            coor: array of x, y coordinates
            angle: radiants to rotate the coordinates by
        Returns:
            coor_rotated
        """
        rot_matrix = np.zeros((2, 2))
        c = np.cos(angle)
        s = np.sin(angle)
        rot_matrix[0, 0] = c
        rot_matrix[0, 1] = -s
        rot_matrix[1, 0] = s
        rot_matrix[1, 1] = c
        coor[..., :2] = np.matmul(
            rot_matrix, np.expand_dims(coor[..., :2], axis=-1)
        ).squeeze(-1)
        if coor.shape[-1] > 2:
            coor[..., 2] += angle
        if coor.shape[-1] >= 5:
            coor[..., 3:5] = np.matmul(
                rot_matrix, np.expand_dims(coor[..., 3:5], axis=-1)
            ).squeeze(-1)
        return coor

    def fill_past(self, past, mask_past):
        current_velocity = past[..., 0, 3:5]
        for t in range(1, past.shape[-2]):
            current_velocity = torch.where(
                mask_past[..., t, None], past[..., t, 3:5], current_velocity
            )
            past[..., t, 3:5] = current_velocity
            predicted_position = past[..., t - 1, :2] + current_velocity * self.dt
            past[..., t, :2] = torch.where(
                mask_past[..., t, None], past[..., t, :2], predicted_position
            )
        return past

    def collate_fn(
        self, samples: List
    ) -> Tuple[Tensor, Tensor, Tensor, Tensor, Tensor, Tensor, Tensor]:
        """
        Assemble trajectories into batches with 0-padding.
        Args:
            samples: list of sampled trajectories (list of outputs of __getitem__)
        Returns:
            (starred dimensions have different values from one batch to the next but the ones with the same name are consistent within the batch)
            batch : ((batch_size, num_agents*, num_steps, state_dim),           # past trajectories of all agents in the scene
                     (batch_size, num_agents*, num_steps),                      # mask past False where past trajectories are padding data
                     (batch_size, num_agents*, num_steps_future, state_dim),    # future trajectory
                     (batch_size, num_agents*, num_steps_future),               # mask future False where future trajectories are padding data
                     (batch_size, num_agents*, num_steps_future),               # mask loss False where future trajectories are not to be predicted
                     (batch_size, num_objects*, object_seq_len*, map_state_dim),# map object sequences in the scene
                     (batch_size, num_objects*, object_seq_len*),               # mask map False where map objects are padding data
                     (batch_size, num_agents*, state_dim),                      # position offset of all agents relative to ego at present time
                     (batch_size, num_steps, state_dim),                        # ego past trajectory
                     (batch_size, num_steps_future, state_dim))                 # ego future trajectory
        """
        max_n_vehicle = 50
        max_n_lanes = 0
        for (
            coor,
            mask,
            mask_loss,
            lanes,
            mask_lanes,
            lane_states,
            mean_angle,
            mean_pos,
            idx,
        ) in samples:
            # time_len_coor = self._count_last_obs(coor, hist_len)
            # num_vehicle = np.sum(time_len_coor > self.min_num_obs)
            num_vehicle = coor.shape[1]
            num_lanes = lanes.shape[1]
            max_n_vehicle = max(num_vehicle, max_n_vehicle)
            max_n_lanes = max(num_lanes, max_n_lanes)
        if max_n_vehicle <= 0:
            raise RuntimeError
        data_batch = np.zeros(
            [self.time_len, len(samples), max_n_vehicle, self.state_dim]
        )
        mask_batch = np.zeros([self.time_len, len(samples), max_n_vehicle])
        mask_loss_batch = np.zeros([self.time_len, len(samples), max_n_vehicle])
        lane_batch = np.zeros(
            [self.max_size_lane, len(samples), max_n_lanes, self.map_state_dim]
        )
        mask_lane_batch = np.zeros([self.max_size_lane, len(samples), max_n_lanes])
        mean_angle_batch = np.zeros([len(samples)])
        mean_pos_batch = np.zeros([len(samples), 2])
        tag_list = np.zeros([len(samples)])
        idx_list = [0 for _ in range(len(samples))]

        for sample_ind, (
            coor,
            mask,
            mask_loss,
            lanes,
            mask_lanes,
            lane_states,
            mean_angle,
            mean_pos,
            idx,
        ) in enumerate(samples):
            data_batch[:, sample_ind, : coor.shape[1], :] = coor[: self.time_len, :, :]
            mask_batch[:, sample_ind, : mask.shape[1]] = mask[: self.time_len, :]
            mask_loss_batch[:, sample_ind, : mask.shape[1]] = mask_loss[
                : self.time_len, :
            ]
            lane_batch[: lanes.shape[0], sample_ind, : lanes.shape[1], :2] = lanes
            if lane_states is not None:
                lane_states = repeat(
                    lane_states[:, : self.hist_len],
                    "objects time features -> one objects (time features)",
                    one=1,
                )
                lane_batch[
                    : lanes.shape[0], sample_ind, : lanes.shape[1], 2:
                ] = lane_states
            mask_lane_batch[
                : mask_lanes.shape[0], sample_ind, : mask_lanes.shape[1]
            ] = mask_lanes
            mean_angle_batch[sample_ind] = mean_angle
            mean_pos_batch[sample_ind, :] = mean_pos
            # tag_list[sample_ind] = self.dataset[idx]["tag"]
            idx_list[sample_ind] = idx

        data_batch = torch.from_numpy(data_batch.astype("float32"))
        mask_batch = torch.from_numpy(mask_batch.astype("bool"))
        lane_batch = torch.from_numpy(lane_batch.astype("float32"))
        mask_lane_batch = torch.from_numpy(mask_lane_batch.astype("bool"))
        mean_pos_batch = torch.from_numpy(mean_pos_batch.astype("float32"))
        mask_loss_batch = torch.from_numpy(mask_loss_batch.astype("bool"))

        data_batch = rearrange(
            data_batch, "time batch agents features -> batch agents time features"
        )
        mask_batch = rearrange(mask_batch, "time batch agents -> batch agents time")
        mask_loss_batch = rearrange(
            mask_loss_batch, "time batch agents -> batch agents time"
        )
        lane_batch = rearrange(
            lane_batch,
            "object_seq_len batch objects features-> batch objects object_seq_len features",
        )
        mask_lane_batch = rearrange(
            mask_lane_batch,
            "object_seq_len batch objects -> batch objects object_seq_len",
        )

        # The two first agents are the ones interacting, others are sorted by distance from the first agent
        # Objects are also sorted by distance from the first agent
        # Therefore, the limits in number, max_num_agents and max_num_objects can be seen as adaptative distance limits.

        if not self.input_angle:
            data_batch = torch.cat((data_batch[..., :2], data_batch[..., 3:]), dim=-1)
        traj_past = data_batch[:, : self.max_num_agents, : self.hist_len, :]
        mask_past = mask_batch[:, : self.max_num_agents, : self.hist_len]
        traj_fut = data_batch[
            :, : self.max_num_agents, self.hist_len : self.hist_len + self.fut_len, :
        ]
        mask_fut = mask_batch[
            :, : self.max_num_agents, self.hist_len : self.hist_len + self.fut_len
        ]
        ego_past = data_batch[:, 0:1, : self.hist_len, :]
        ego_fut = data_batch[:, 0:1, self.hist_len : self.hist_len + self.fut_len, :]

        lane_batch = lane_batch[:, : self.max_num_objects]
        mask_lane_batch = mask_lane_batch[:, : self.max_num_objects]

        # Define what to predict (could be from Waymo's label of what to predict or the other agent that interacts with the ego...)
        mask_loss_batch = torch.logical_and(
            mask_loss_batch[
                :, : self.max_num_agents, self.hist_len : self.hist_len + self.fut_len
            ],
            mask_past.any(-1, keepdim=True),
        )
        # Remove all other agents so the model should only predict the first one
        mask_loss_batch[:, 0] = False
        mask_loss_batch[:, 2:] = False

        # Normalize...
        # traj_past = self.fill_past(traj_past, mask_past)
        dynamic_state_size = 5 if self.input_angle else 4
        offset_batch = traj_past[..., -1, :dynamic_state_size].clone()
        traj_past[..., :dynamic_state_size] = traj_past[
            ..., :dynamic_state_size
        ] - offset_batch.unsqueeze(-2)
        traj_fut[..., :dynamic_state_size] = traj_fut[
            ..., :dynamic_state_size
        ] - offset_batch.unsqueeze(-2)

        return (
            traj_past,
            mask_past,
            traj_fut,
            mask_fut,
            mask_loss_batch,
            lane_batch,
            mask_lane_batch,
            offset_batch,
            ego_past,
            ego_fut,
        )


class WaymoDataloaders:
    def __init__(self, cfg: Config) -> None:
        self.cfg = cfg

    def sample_dataloader(self) -> DataLoader:
        """Setup and return sample DataLoader

        Returns:
            DataLoader: sample DataLoader
        """
        dataset = WaymoDataset(self.cfg, "sample")
        sample_loader = DataLoader(
            dataset=dataset,
            batch_size=self.cfg.batch_size,
            shuffle=False,
            num_workers=self.cfg.num_workers,
            collate_fn=dataset.collate_fn,
            drop_last=True,
        )
        return sample_loader

    def val_dataloader(
        self, drop_last=True, shuffle=False, input_angle=True
    ) -> DataLoader:
        """Setup and return validation DataLoader

        Returns:
            DataLoader: validation DataLoader
        """
        dataset = WaymoDataset(self.cfg, "val", input_angle)
        val_loader = DataLoader(
            dataset=dataset,
            batch_size=self.cfg.batch_size,
            shuffle=shuffle,
            num_workers=self.cfg.num_workers,
            collate_fn=dataset.collate_fn,
            drop_last=drop_last,
        )
        torch.cuda.empty_cache()
        return val_loader

    def train_dataloader(
        self, drop_last=True, shuffle=True, input_angle=True
    ) -> DataLoader:
        """Setup and return training DataLoader

        Returns:
            DataLoader: training DataLoader
        """
        dataset = WaymoDataset(self.cfg, "train", input_angle)
        train_loader = DataLoader(
            dataset=dataset,
            batch_size=self.cfg.batch_size,
            shuffle=shuffle,
            num_workers=self.cfg.num_workers,
            collate_fn=dataset.collate_fn,
            drop_last=drop_last,
        )
        torch.cuda.empty_cache()
        return train_loader

    def test_dataloader(self) -> DataLoader:
        """Setup and return test DataLoader

        Returns:
            DataLoader: test DataLoader
        """
        raise NotImplementedError("The waymo dataloader cannot load test samples yet.")

    @staticmethod
    def unnormalize_trajectory(
        input: torch.Tensor, offset: torch.Tensor
    ) -> torch.Tensor:
        """Unnormalize trajectory by adding offset to input

        Args:
            input : (..., (n_sample), num_steps_future, state_dim) tensor of future
            trajectory y
            offset : (..., state_dim) tensor of offset to add to y

        Returns:
            Unnormalized trajectory that has the same size as input
        """
        assert offset.ndim == 3
        batch_size, num_agents = offset.shape[:2]
        offset_state_dim = offset.shape[-1]
        assert offset_state_dim <= input.shape[-1]
        assert input.shape[0] == batch_size
        assert input.shape[1] == num_agents
        input_copy = input.clone()

        input_copy[..., :offset_state_dim] = input_copy[
            ..., :offset_state_dim
        ] + offset[..., : input.shape[-1]].reshape(
            [batch_size, num_agents, *([1] * (input.ndim - 3)), offset_state_dim]
        )
        return input_copy