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

Code before:
    """

    def __init__(self, sensor_options: "SensorOptions", sensor_idx: int, sensor_manager: "SensorManager"):
        self._options: "SensorOptions" = sensor_options
        self._idx: int = sensor_idx
        self._manager: "SensorManager" = sensor_manager
        self._cache_idx: int = -1  # cache_idx is set by the SensorManager during the scene build phase
        self._read_delay_steps: int = 0
        self._cache_buffer_length: int = 0

    # =============================== implementable methods ===============================

    @gs.assert_unbuilt
    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()`.")

Review comment (performance):
I don' think it is the right way to do it. The cache buffer length should be the same for all sensors, and the sensor manager should use `read_delay_step = max(sensor._read_delay_steps for sensor in self._sensors)` to determine how many data mush be stored in cache. It will make everything much easier to implement and faster, but increase the memory footprint to some extent.

Code after:

    def __init__(self, sensor_options: "SensorOptions", sensor_idx: int, sensor_manager: "SensorManager"):
        self._options: "SensorOptions" = sensor_options
        self._idx: int = sensor_idx
        self._manager: "SensorManager" = sensor_manager

        # initialized by SensorManager during build
        self._read_delay_steps: int = 0
        self._shape_indices: list[tuple[int, int]] = []
        self._shared_metadata: dict[str, Any] | None = None
        self._cache: "TensorRingBuffer" | None = None

    # =============================== implementable methods ===============================

    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