Repository: Genesis-Embodied-AI/Genesis
File: genesis/sensors/base_sensor.py (Python)

Code before:
    def build(self):
        """
        This method is called by the SensorManager during the scene build phase to initialize the sensor.
        This is where any shared metadata should be initialized.
        """
        delay_steps_float = self._options.read_delay / self._manager._sim.dt
        self._read_delay_steps = round(delay_steps_float)
        if not np.isclose(delay_steps_float, self._read_delay_steps):
            gs.logger.warn(
                f"Read delay should be a multiple of the simulation time step. Got {self._options.read_delay} and "
                f"{self._manager._sim.dt}. Actual read delay will be {1/self._read_delay_steps}."
            )
        self._cache_buffer_length = self._read_delay_steps + 1

    def _get_return_format(self) -> dict[str, tuple[int, int]] | tuple[int, int] | None:
        """
        Data format of the read() return value.
        dict: a dictionary with string keys and tensor values will be returned.
              The tuple (start_idx, end_idx) will be used to get a slice of the values from the cache.
        tuple: the slice of the cache will be returned.
        None: the entire tensor (B, cache_length, cache_shape) will be returned.
        """
        raise NotImplementedError("Sensors must implement `return_format()`.")

    @gs.assert_built
    def _update_shared_gt_cache(self):
        """
        Update the shared sensor ground truth cache for all sensors of this class.
        Shared information is stored in SensorManager
        """
        raise NotImplementedError("Sensors must implement `update_shared_gt_cache()`.")

    @gs.assert_built
    def _update_shared_cache(self):
        """
        Update the shared sensor cache for all sensors of this class using information stored in shared metadata.
        """
        raise NotImplementedError("Sensors must implement `update_shared_cache()`.")

    def _get_cache_length(self) -> int:
        """
        The length (first dimension of cache shape) of the cache for this sensor instance.
        The sum of cache_length for all sensors of this type determines the length of the cache.
        """
        raise NotImplementedError("Sensors must implement `cache_length()`.")

    @classmethod
    def _get_cache_size(cls) -> int:
        """
        The length (second dimension of cache shape) of the cache for this sensor type.
        """

Review comment (suggestion):
`gt` is not a great acronym because it usually means "greater than". What about 'true' or 'raw' ? Or just 'ground_truth'

Code after:

    def build(self):
        """
        This method is called by the SensorManager during the scene build phase to initialize the sensor.
        This is where any shared metadata should be initialized.
        """
        raise NotImplementedError("Sensors must implement `build()`.")

    def _get_return_format(self) -> dict[str, tuple[int, ...]] | tuple[int, ...]:
        """
        Data format of the read() return value.

        Returns
        -------
        return_format : dict | tuple
            - If tuple, the final shape of the read() return value.
                e.g. (2, 3) means read() will return a tensor of shape (2, 3).
            - If dict a dictionary with string keys and tensor values will be returned.
                e.g. {"pos": (3,), "quat": (4,)} returns a dict of tensors [0:3] and [3:7] from the cache.
        """
        raise NotImplementedError("Sensors must implement `return_format()`.")

    def _get_cache_length(self) -> int:
        """
        The length of the cache for this sensor instance, e.g. number of points for a Lidar point cloud.
        """
        raise NotImplementedError("Sensors must implement `cache_length()`.")

    @classmethod
    def _update_shared_ground_truth_cache(
        cls, shared_metadata: dict[str, Any], shared_ground_truth_cache: torch.Tensor
    ):
        """
        Update the shared sensor ground truth cache for all sensors of this class using metadata in SensorManager.
        """
        raise NotImplementedError("Sensors must implement `update_shared_ground_truth_cache()`.")

    @classmethod
    def _update_shared_cache(
        cls, shared_metadata: dict[str, Any], shared_ground_truth_cache: torch.Tensor, shared_cache: "TensorRingBuffer"
    ):
        """
        Update the shared sensor cache for all sensors of this class using metadata in SensorManager.
        """
        raise NotImplementedError("Sensors must implement `update_shared_cache()`.")

    @classmethod
    def _get_cache_dtype(cls) -> torch.dtype:
        """
        The dtype of the cache for this sensor.
        """