src/models/data/radym.py

# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import os
import zipfile
from pathlib import Path
from typing import Any, List, Optional

import cv2
import numpy as np
import OpenEXR
import torch
import torch.utils.data
from decord import VideoReader
from lru import LRU

from src.models.data.base import BaseDataset
from src.models.data.datafield import DataField

class Radym(BaseDataset):
    MAX_ZIP_DESCRIPTORS = 10
    MAX_MP4_READERS = 2

    def __init__(
        self, root_path, filter_list_path: Optional[str] = None, num_views: int = -1, depth_folder: str = "depth", 
        custom_folders: Optional[List[str]] = None, custom_fields: Optional[List[str]] = None, start_view_idx: int = 0,
        depth_path: Optional[List[str]] = None, camera_path: Optional[List[str]] = None,
    ):
        # For multi-view datasets, root_path is the path to camera idx 0.
        self.root_path = root_path

        # filter_list_path is a text file containing the list of mp4 files to load.
        # Each line in the file should contain the name of the mp4 file with or without the extension.
        if filter_list_path is None:
            self.filter_set = None
        else:
            self.filter_list_path = filter_list_path if os.path.isabs(filter_list_path) else os.path.join(root_path, filter_list_path)
            with open(self.filter_list_path, "r") as f:
                self.filter_set = [line.strip() for line in f.readlines()]
            self.filter_set = set([x.split(".")[0] for x in self.filter_set])
        self.n_views = num_views

        # Recursively grab all mp4 files in subfolders with name 'rgb'.
        self.mp4_file_paths = []
        for rgb_root in Path(root_path).rglob("rgb"):
            if not rgb_root.is_dir():
                continue
            print(rgb_root)
            for mp4_file in rgb_root.glob("*.mp4"):
                if self.filter_set is None or mp4_file.stem in self.filter_set:
                    self.mp4_file_paths.append(mp4_file)
        self.mp4_file_paths = sorted(self.mp4_file_paths)
        # Process-dependent LRU cache for file handles of the tar files.
        self.worker_id = None
        self.zip_descriptors = LRU(
            self.MAX_ZIP_DESCRIPTORS, callback=self._evict_zip_handle
        )
        # self.mp4_readers = LRU(self.MAX_MP4_READERS, callback=self._evict_mp4_reader)
        self.depth_folder = depth_folder
        self.custom_folders = custom_folders
        self.custom_fields = custom_fields

        # If view index doesn't start at 0
        self.start_view_idx = start_view_idx

        # Store some annotations in different folders
        self.depth_path = Path(depth_path) if depth_path is not None else depth_path
        self.camera_path = Path(camera_path) if camera_path is not None else camera_path

    @staticmethod
    def _evict_zip_handle(_, zip_handle):
        zip_handle.close()

    @staticmethod
    def _evict_mp4_reader(_, mp4_reader: VideoReader):
        # This is no-op, just a placeholder.
        del mp4_reader

    def _check_worker_id(self):
        # Protect handle boundary:
        worker_info = torch.utils.data.get_worker_info()
        if worker_info is not None:
            if self.worker_id is not None:
                assert self.worker_id == worker_info.id, "Worker id mismatch"
            else:
                self.worker_id = worker_info.id

    def _get_zip_handle(self, idx, attr, view_idx, attr_path = None):
        self._check_worker_id()

        if self.n_views != -1:
            dict_key = f"{idx}_{view_idx}_{attr}"
        else:
            dict_key = f"{idx}_{attr}"
        if dict_key in self.zip_descriptors:
            return self.zip_descriptors[dict_key]

        rgb_path = self.mp4_file_paths[idx]
        root_path, zip_name = rgb_path.parent.parent, rgb_path.stem + ".zip"
        if self.n_views != -1:
            root_path = root_path.parent
            if attr_path is not None:
                root_path = attr_path
            root_path = root_path / str(view_idx)
        else:
            if attr_path is not None:
                root_path = attr_path
        file_path_in = root_path / attr / zip_name
        zip_handle = zipfile.ZipFile(file_path_in, "r")
        self.zip_descriptors[dict_key] = zip_handle
        return zip_handle

    def _get_mp4_reader(self, idx, attr, view_idx):
        # self._check_worker_id()
 
        # if self.n_views != -1:
        #     dict_key = f"{idx}_{view_idx}_{attr}"
        # else:
        #     dict_key = f"{idx}_{attr}"
        # if dict_key in self.mp4_readers:
        #     return self.mp4_readers[dict_key]
        
        rgb_path = self.mp4_file_paths[idx]
        if self.n_views != -1:
            root_path, mp4_name = rgb_path.parent.parent.parent, rgb_path.name
        else:
            root_path, mp4_name = rgb_path.parent.parent, rgb_path.name
        if self.n_views != -1:
            root_path = root_path / str(view_idx)

        mp4_reader = VideoReader(str(root_path / attr / mp4_name), num_threads=4)
        # self.mp4_readers[dict_key] = mp4_reader
        return mp4_reader

    def available_data_fields(self) -> list[DataField]:
        return [
            DataField.IMAGE_RGB,
            DataField.CAMERA_C2W_TRANSFORM,
            DataField.CAMERA_INTRINSICS,
            DataField.METRIC_DEPTH,
            DataField.DYNAMIC_INSTANCE_MASK,
            DataField.BACKWARD_FLOW,
            DataField.OBJECT_BBOX,
            DataField.CAPTION,
            DataField.LATENT_RGB,
        ]

    def num_videos(self) -> int:
        return len(self.mp4_file_paths)

    def num_views(self, video_idx: int) -> int:
        return 1 if self.n_views == -1 else self.n_views

    def num_frames(self, video_idx: int, view_idx: int = None) -> int:
        if view_idx is None:
            view_idx = self.start_view_idx
        return len(self._get_mp4_reader(video_idx, "rgb", view_idx))

    def _read_data(
        self,
        video_idx: int,
        frame_idxs: List[int],
        view_idxs: List[int],
        data_fields: List[DataField],
    ):
        frame_indices = np.asarray(frame_idxs).astype(np.int64)
        rgb_path = self.mp4_file_paths[video_idx]
        data_base_path, data_key = rgb_path.parent.parent, rgb_path.stem
        if self.camera_path is not None:
            camera_path = self.camera_path
        else:
            camera_path = data_base_path
        if self.n_views != -1:
            # Currently support only load at most one camera.
            assert len(set(view_idxs)) == 1, "Currently support only one view"
            view_idx = view_idxs[0]
            data_base_path = data_base_path.parent / str(view_idx)
            if self.camera_path is None:
                camera_path = data_base_path
            else:
                camera_path = camera_path / str(view_idx)
        else:
            view_idx = self.start_view_idx
        output_dict: dict[str | DataField, Any] = {"__key__": data_key}

        for data_field in data_fields:
            if data_field == DataField.IMAGE_RGB:
                rgb_reader = self._get_mp4_reader(video_idx, "rgb", view_idx)
                rgb_read = rgb_reader.get_batch(frame_indices)
                try:
                    rgb_np = rgb_read.asnumpy()
                except AttributeError:
                    rgb_np = rgb_read.numpy()
                rgb_np = rgb_np.astype(np.float32) / 255.0
                rgb_torch = torch.from_numpy(rgb_np).moveaxis(-1, 1).contiguous()
                output_dict[data_field] = rgb_torch

                rgb_reader.seek(0)  # set video reader point back to 0 to clean up cache
                del rgb_reader
            
            elif data_field == DataField.LATENT_RGB:
                # Load rgb latent
                rgb_latents_path = data_base_path / "latent" / f"{data_key}.pkl"
                rgb_latents_data = torch.load(rgb_latents_path, map_location='cpu', weights_only=False)
                if isinstance(rgb_latents_data, np.ndarray):
                    rgb_latents_data = torch.from_numpy(rgb_latents_data)
                # Remove batch index
                rgb_latents_data = rgb_latents_data[0]
                output_dict[data_field] = rgb_latents_data
                # NOTE: Assuming all latents are loaded (no subsampling with frame_indices)

            elif data_field == DataField.CAMERA_C2W_TRANSFORM:
                c2w_data = np.load(camera_path / "pose" / f"{data_key}.npz")
                f_idx = np.searchsorted(c2w_data["inds"], frame_indices)
                assert np.all(
                    c2w_data["inds"][f_idx] == frame_indices
                ), "Pose not found"
                c2w_np = c2w_data["data"][f_idx].astype(np.float32)
                c2w_torch = torch.from_numpy(c2w_np).contiguous()
                output_dict[data_field] = c2w_torch

            elif data_field == DataField.CAMERA_INTRINSICS:
                intrinsics_data = np.load(
                    camera_path / "intrinsics" / f"{data_key}.npz"
                )
                f_idx = np.searchsorted(intrinsics_data["inds"], frame_indices)
                assert np.all(
                    intrinsics_data["inds"][f_idx] == frame_indices
                ), "Intrinsics not found"
                intrinsics_np = intrinsics_data["data"][f_idx].astype(np.float32)
                intrinsics_torch = torch.from_numpy(intrinsics_np).contiguous()
                output_dict[data_field] = intrinsics_torch

            elif data_field == DataField.METRIC_DEPTH:
                depth_zip_handle = self._get_zip_handle(video_idx, self.depth_folder, view_idx, self.depth_path)
                depth_np = []
                for frame_idx in frame_indices:
                    frame_name = f"{frame_idx:05d}.exr"
                    with depth_zip_handle.open(frame_name, "r") as f:
                        exr_file = OpenEXR.InputFile(f)
                        exr_dw = exr_file.header()["dataWindow"]
                        depth_np.append(
                            np.frombuffer(exr_file.channel("Z"), np.float16).reshape(
                                exr_dw.max.y - exr_dw.min.y + 1,
                                exr_dw.max.x - exr_dw.min.x + 1,
                            )
                        )
                depth_np = np.stack(depth_np, axis=0).astype(np.float32)
                depth_torch = torch.from_numpy(depth_np).contiguous()
                output_dict[data_field] = depth_torch

            elif data_field == DataField.OBJECT_BBOX:
                bbox_zip_handle = self._get_zip_handle(
                    video_idx, "object_info", view_idx
                )
                bbox_list = []
                for frame_idx in frame_indices:
                    frame_name = f"{frame_idx:05d}.json"
                    with bbox_zip_handle.open(frame_name, "r") as f:
                        bbox_data = json.load(f)
                        bbox_list.append(bbox_data)
                output_dict[data_field] = bbox_list

            elif data_field == DataField.DYNAMIC_INSTANCE_MASK:
                mask_zip_handle = self._get_zip_handle(video_idx, "mask", view_idx)
                mask_np = []
                for frame_idx in frame_indices:
                    frame_name = f"{frame_idx:05d}.png"
                    with mask_zip_handle.open(frame_name, "r") as f:
                        mask_np.append(
                            cv2.imdecode(
                                np.frombuffer(f.read(), np.uint8), cv2.IMREAD_UNCHANGED
                            )
                        )
                mask_np = np.stack(mask_np, axis=0).astype(np.uint8)
                mask_torch = torch.from_numpy(mask_np).contiguous()
                output_dict[data_field] = mask_torch

            elif data_field == DataField.BACKWARD_FLOW:
                flow_zip_handle = self._get_zip_handle(video_idx, "flow", view_idx)
                flow_np = []
                for frame_idx in frame_indices:
                    frame_name = f"{frame_idx:05d}.exr"
                    with flow_zip_handle.open(frame_name, "r") as f:
                        exr_file = OpenEXR.InputFile(f)
                        exr_dw = exr_file.header()["dataWindow"]
                        height, width = (
                            exr_dw.max.y - exr_dw.min.y + 1,
                            exr_dw.max.x - exr_dw.min.x + 1,
                        )
                        flow_np.append(
                            np.stack(
                                [
                                    np.frombuffer(
                                        exr_file.channel(f"{channel}"), np.float16
                                    )
                                    for channel in ["U", "V"]
                                ],
                                axis=-1,
                            ).reshape(height, width, 2)
                        )
                flow_np = np.stack(flow_np, axis=0).astype(np.float32)
                flow_torch = torch.from_numpy(flow_np).contiguous()
                output_dict[data_field] = flow_torch

            elif data_field == DataField.CAPTION:
                caption_path = data_base_path / "caption" / f"{data_key}.txt"
                with open(caption_path, "r") as f:
                    caption = f.read()
                output_dict[data_field] = caption
            elif data_field == "custom":
                if self.custom_folders is not None:
                    output_dict[data_field] = {}
                    assert len(self.custom_folders) == len(self.custom_fields), "Custom folders and types must have the same length"
                    for custom_folder, custom_fields in zip(self.custom_folders, self.custom_fields):
                        if custom_fields == "ftheta_intrinsic":
                            intrinsics_data = np.load(
                                data_base_path / custom_folder / f"{data_key}.npz"
                            )
                            f_idx = np.searchsorted(intrinsics_data["inds"], frame_indices)
                            assert np.all(
                                intrinsics_data["inds"][f_idx] == frame_indices
                            ), "Intrinsics not found"
                            intrinsics_np = intrinsics_data["data"][f_idx].astype(np.float32)
                            intrinsics_torch = torch.from_numpy(intrinsics_np).contiguous()
                            output_dict[data_field][custom_fields] = intrinsics_torch
                        elif custom_fields in ["hdmap"]:
                            mp4_reader = self._get_mp4_reader(video_idx, custom_folder, view_idx)
                            mp4_read = mp4_reader.get_batch(frame_indices)
                            try:
                                mp4_np = mp4_read.asnumpy()
                            except AttributeError:
                                mp4_np = mp4_read.numpy()
                            mp4_np = mp4_np.astype(np.float32) / 255.0
                            mp4_torch = torch.from_numpy(mp4_np).moveaxis(-1, 1).contiguous()
                            output_dict[data_field][custom_fields] = mp4_torch

                            mp4_reader.seek(0)  # set video reader point back to 0 to clean up cache
                            del mp4_reader

                        else:
                            raise NotImplementedError(f"Can't handle custom data field {data_field}")
            else:
                raise NotImplementedError(f"Can't handle data field {data_field}")

        return output_dict