编写自定义传感器#

本页面向要添加新传感器类型的高级用户。它是 传感器管线 的作者视角对照篇：后者描述传感器在运行时如何执行；这里关注你要实现的接口、每个重写必须遵守的 shape 与 dtype 契约，以及框架如何自动把你的传感器与它的选项类配对。如果你只想使用内置传感器，请从 传感器 开始。

基类位于 genesis/engine/sensors/base_sensor.py。几乎所有情况下都应继承 SimpleSensor，只重写自己需要的 hook。只有完全绕过标准管线的传感器才直接继承 Sensor，内置相机就是通过 BaseCameraSensor 这样做的。

最小可运行示例#

一个完整的传感器由三个类组成：面向用户的选项 dataclass、按类共享的元数据容器，以及传感器本身。下面的接近传感器报告所挂连杆到世界原点的距离，并截断到最大量程。它已经足以通过 scene.add_sensor(...) 使用，并从 SimpleSensor 继承了全部误差特性（噪声、偏置、随机游走、延迟、抖动、历史），这些特性由 SimpleSensor 统一施加。

# my_plugin/options.py
from genesis.options.sensors.options import RigidSensorOptionsMixin, SimpleSensorOptions


class MyProximity(
    RigidSensorOptionsMixin["MyProximitySensor"],
    SimpleSensorOptions["MyProximitySensor"],
):
    max_range: float = 1.0  # meters

# my_plugin/sensor.py
from dataclasses import dataclass

import torch

import genesis as gs
from genesis.engine.sensors.base_sensor import (
    RigidSensorMetadataMixin,
    RigidSensorMixin,
    SimpleSensor,
    SimpleSensorMetadata,
)
from genesis.utils.misc import concat_with_tensor, make_tensor_field

from .options import MyProximity


@dataclass
class MyProximityMetadata(RigidSensorMetadataMixin, SimpleSensorMetadata):
    # Per-sensor ranges, concatenated across every instance of this class at build time.
    max_range: torch.Tensor = make_tensor_field((0,))


class MyProximitySensor(
    RigidSensorMixin[MyProximityMetadata],
    SimpleSensor[MyProximity, None, MyProximityMetadata],
):
    def build(self):
        # The manager calls build() once per instance. Append this sensor's parameters
        # to the shared, per-class tensors so one kernel can run over every instance.
        super().build()
        self._shared_metadata.max_range = concat_with_tensor(
            self._shared_metadata.max_range,
            float(self._options.max_range),
            expand=(1,),
        )

    def _get_return_format(self) -> tuple[int, ...]:
        return (1,)  # one scalar per sensor

    @classmethod
    def _get_cache_dtype(cls) -> torch.dtype:
        return gs.tc_float

    @classmethod
    def _update_raw_data(cls, shared_context, shared_metadata, raw_data_T):
        pos = shared_metadata.solver.get_links_pos(shared_metadata.links_idx)  # (B, N, 3)
        dist = pos.norm(dim=-1).clamp(max=shared_metadata.max_range)           # (B, N)
        raw_data_T.copy_(dist.T)  # write in place; buffer is (cols, B), column-major

本页余下部分解释每一块为什么存在，以及更复杂的传感器还会重写哪些额外的 hook。

你需要编写的类#

每个传感器都贡献同样的三样产物，另有两样可选的。

  • 选项类： 一个公共 dataclass，承载每个传感器的全部参数，继承 SimpleSensorOptions（或合适的 mixin）。它以传感器类作为前向引用进行泛型参数化，如 SimpleSensorOptions["MyProximitySensor"]。这是用户唯一需要构造的对象。

  • 元数据类： 按传感器类共享的运行时状态，由场景中该传感器的每个实例共用。SimpleSensor 派生的传感器继承 SimpleSensorMetadata，直接派生自 Sensor 的传感器则继承 SharedSensorMetadata。

  • 传感器类： 具体实现。它继承 SimpleSensor[OptionsT, ContextT, MetadataT]，并重写下面的 hook。

  • NamedTuple 返回类型（可选）： 当传感器返回多个张量时声明一个，并作为第四个类型参数传入。见返回 NamedTuple。

  • SharedSensorContext 子类（可选）： 仅当该传感器与其他传感器类型共享昂贵资源时才声明，并作为第二个类型参数传入。见跨传感器类型共享资源。

泛型签名是 Sensor[OptionsT, ContextT, MetadataT, DataT]：先是选项，然后是跨类型共享上下文（没有则为 None），接着是元数据类型，最后是数据类型（DataT 默认为 tuple）。

注册是自动的#

传感器从不需要手动注册。当 Sensor 子类把它的选项类作为第一个类型参数时，Sensor.__init_subclass__ 会在类体执行的那一刻把这对关系记录到 SensorManager.SENSOR_TYPES_MAP 中。之后用户只需构造选项实例并交给 scene.add_sensor(...)，由后者解析出传感器类、实例化并返回传感器。

这样就只剩两种受支持的放置方式：

  • 仓库内（内置传感器）： 选项放在 genesis/options/sensors/*.py，传感器放在 genesis/engine/sensors/*.py。两者都通过各自包的 __init__ 导入，因此在导入时就完成配对注册。

  • 仓库外（第三方插件）： 把选项和传感器放在同一个包的两个同级子模块中（例如 my_plugin/options.py 和 my_plugin/sensor.py）。只要你的代码在构造选项之前导入了选项模块，框架就会在首次调用 scene.add_sensor(...) 时透明地解析出传感器类。

Note

__init_subclass__ 还会强制检查契约：声明了自己选项类的具体传感器也必须声明其元数据类型参数；任何重写 _post_process 的类都必须声明一个中间缓冲区（见投影到不同的返回类型）。两种违规都会在类定义时抛出 TypeError，早于任何场景构建。

必须重写的 hook#

每个具体传感器都必须重写两个 hook，SimpleSensor 又额外增加一个。

  • _get_return_format(self) -> tuple[...]： 实例方法，返回 read() 产物的形状。形状按设计是按实例的，因为选项可以合理地决定它（光线投射器的 pattern、相机的分辨率、接近传感器的探针位置）。单个含 N 个标量的张量返回 (N,)；多张量返回则返回元组的元组，例如 ((3,), (3,), (3,))。

  • _get_cache_dtype(cls) -> torch.dtype： 类方法，返回 read() 产物的 dtype。dtype 是按类统一的，而不是按实例：管理器把一个类的每个实例打包进按 dtype 缓冲区中该类的连续切片，因此所有实例必须共享同一种 dtype。若需要不同 dtype，请使用两个传感器类。

  • _update_raw_data(cls, shared_context, shared_metadata, raw_data_T)： SimpleSensor 的 kernel，在当前步为该类的每个传感器计算真值。这是 SimpleSensor 唯一的抽象产出 hook。

用实例方法定形状、用类方法定 dtype，这种划分是刻意为之。它让管理器无需实例化传感器就能确定每类 dtype，同时仍允许每实例形状依赖选项。

raw_data_T 的形状是 (cols, B)（列主序，批次维 B 在最后），这样当多个传感器类写入同一个中间缓存时，每类的行切片仍是 C 连续的。始终原地填充它：

@classmethod
def _update_raw_data(cls, shared_context, shared_metadata, raw_data_T):
    pos = shared_metadata.solver.get_links_pos(shared_metadata.links_idx)  # (B, N, 3)
    raw_data_T.copy_(pos.reshape(pos.shape[0], -1).T)                      # (3*N, B)

shared_context 是跨类型资源，或者为 None；大多数传感器会忽略它。hook 每步按类调用一次，绝不按实例、也绝不按环境调用，因此要相应地向量化。

选择基类#

基类

使用场景

SimpleSensor[OptionsT, None, MetadataT]

几乎总是选它。标准的逐步管线：原始值、物理误差、变换、硬件误差、后处理、延迟采样。

SimpleSensor[OptionsT, None, MetadataT, DataT]

同样的管线，但 read() 返回 DataT（一个 NamedTuple）的实例，而不是单个张量。IMU 是典型例子。

BaseCameraSensor[OptionsT]

在 read() 时惰性渲染图像的相机类传感器。见相机类传感器。

Sensor[OptionsT, None, MetadataT]

仅当两种标准管线都不适用时。自行实现 _update_shared_cache；如果你的实现完全不触碰时间线环，请设置 uses_ring_pipeline = False。

Mixin 可以组合到基类上：

  • RigidSensorOptionsMixin / KinematicSensorOptionsMixin： 位于选项侧，分别用于挂载到 RigidEntity 或任意 KinematicEntity 的传感器。通过多重继承与 SimpleSensorOptions 组合。

  • RigidSensorMixin / RigidSensorMetadataMixin： 位于传感器和元数据侧，提供带类型的 solver 字段、links_idx 记录，以及 set_pos_offset / set_quat_offset。

按类共享的元数据#

元数据类保存的是按传感器类（而非按实例）且会被 hook 读取的内容：求解器与实体引用、构建时拼接的逐传感器索引张量（links_idx、thresholds、max_range）、逐传感器偏移、滤波器系数，以及用于门控慢路径的预计算标志。

SimpleSensorMetadata 已经提供了误差状态（noise、bias、random_walk、resolution、jitter_ts）以及对应的 has_any_* 标志。继承它，并用 make_tensor_field((shape,)) 声明字段，让它们自动分配：

@dataclass
class MySharedMetadata(RigidSensorMetadataMixin, SimpleSensorMetadata):
    thresholds: torch.Tensor = make_tensor_field((0,))
    custom_offsets: torch.Tensor = make_tensor_field((0, 3))

在传感器的 build() 中用 concat_with_tensor 追加本实例的条目，如上面的最小示例所示。管理器在场景构建时对每个传感器实例调用一次 build()；正是在这里扩充共享张量，才使得单个 kernel 能一次跑遍所有实例。

可选的管线 hook#

SimpleSensor 运行一条固定的逐步管线，每个阶段都有默认实现。只有传感器需要时才重写某个阶段。各阶段按分支、按以下顺序运行：

  • 真值分支： _update_raw_data，然后 _apply_transform(is_measured=False)，然后 _post_process(is_measured=False)。

  • 测量分支： _update_raw_data，然后 _apply_physics_imperfections，然后 _apply_transform(is_measured=True)，然后 _apply_hardware_imperfections，然后 _post_process(is_measured=True)，最后延迟采样。

两个分支各自维护自己的中间空间时间线环，保存变换后、硬件误差前的值，因此有状态的 _apply_transform 滤波器读到的总是未受硬件噪声污染的旧槽位。三个噪声阶段的区别在于扰动在物理上源自何处。

  • _apply_physics_imperfections(cls, shared_metadata, data, timeline)： 仿真器未建模的底层现象的随机波动（真实的漂移、场上的细尺度湍流）。仅作用于测量分支，在 _apply_transform 之前施加，因此它会通过响应模型传播到后续步骤。默认：无操作。

  • _apply_transform(cls, shared_metadata, data, timeline, *, is_measured)： 对传感器元件做坐标变换和/或有状态的响应建模（热质量、RC 时间常数、机械带宽）。两个分支都会调用；坐标部分无条件执行，而对不应出现在真值中的元件特有影响，用 if is_measured: 来门控。原地修改 data；用 timeline.at(1)、timeline.at(2) 读取历史。IMU 用它做本体坐标系对齐旋转；温度网格传感器用它做 RC 滤波。

  • _apply_hardware_imperfections(cls, shared_metadata, measured_slot_0)： 读出电子器件在传感器输出端引入的扰动。SimpleSensor 已在这里实现 noise、bias、random_walk 和 resolution，并由 has_any_* 标志门控，因此全零的类会完全跳过。仅在需要非标准模型时重写，并先调用 super() 处理标准项：

@classmethod
def _apply_hardware_imperfections(cls, shared_metadata, measured_slot_0):
    super()._apply_hardware_imperfections(shared_metadata, measured_slot_0)
    # Signal-dependent noise floor, resampled each step.
    measured_slot_0 += torch.normal(0.0, shared_metadata.signal_noise_coeff) * measured_slot_0.abs()

对于噪声是物理计算固有部分的传感器（单次 kernel pass 必须同时产生理想值和带噪值，因为噪声会影响 kernel 的分支），应改为重写 _update_current_timestep_data。它在一次 pass 中写入真值切片和带噪的测量槽位，管线其余部分仍在其之上运行。

投影到不同的返回类型#

_post_process(cls, shared_metadata, tensor, timeline, *, is_measured) -> torch.Tensor 把管线内部的中间空间投影到面向用户的返回空间。当输出类型或形状与内部表示不同时重写它：例如 ContactSensor 上的 bool 阈值，ContactForceSensor 上的死区与饱和。返回投影后的张量；管理器会把它写入返回空间环并做延迟采样。

@classmethod
def _post_process(cls, shared_metadata, tensor, timeline, *, is_measured):
    return tensor > shared_metadata.thresholds  # float intermediate, bool return

重写 _post_process 要求同时重写 _get_intermediate_format 和/或 _get_intermediate_dtype，否则框架会在类定义时抛出 TypeError。原因是结构性的：中间缓冲区必须是独立的缓冲区，因为时间线环位于中间空间，混用数据空间会破坏任何读取旧槽位的 _apply_transform 滤波器。当投影确实保持 shape 和 dtype 不变时，可把其中之一重写为返回返回空间值的 no-op：这个重写本身就是对两个缓冲区相互独立的显式确认。

  • _get_intermediate_format(self)： 内部缓冲区的形状；默认为 _get_return_format()。

  • _get_intermediate_dtype(cls)： 内部缓冲区的 dtype；默认为 _get_cache_dtype()。ContactSensor 把它重写为 gs.tc_float，因为它的 kernel 是 float，而返回是 bool。

返回 NamedTuple#

多张量返回时，声明一个 NamedTuple 并作为第四个类型参数传入。_get_return_format 随后按字段顺序为每个字段返回一个形状：

class IMUReturnType(NamedTuple):
    lin_acc: torch.Tensor
    ang_vel: torch.Tensor
    mag: torch.Tensor


class IMUSensor(SimpleSensor[IMU, None, IMUSharedMetadata, IMUReturnType]):
    def _get_return_format(self) -> tuple[tuple[int, ...], ...]:
        return ((3,), (3,), (3,))  # shapes match the NamedTuple field order

    @classmethod
    def _get_cache_dtype(cls) -> torch.dtype:
        return gs.tc_float  # one dtype across all fields (class-uniform)

管理器为每个传感器分配一整块连续内存，读取时再切片；read() 重建并返回该 NamedTuple，当场景没有环境时去掉前导批次维。每个字段的形状为 ([n_envs,] *field_shape)。

跨传感器类型共享资源#

元数据和共享上下文都是管理器持有的状态，但它们解决不同的问题，不可混为一谈。元数据是按类型的：它聚合某一个类的逐传感器行，使单个 kernel 能在其上向量化，并随传感器数量增长。上下文是跨类型的：单一资源，在传感器数量上为 O(1)，由多个传感器类读取。

典型的上下文是 RaycasterSensor 和 DepthCameraSensor 共同投射的碰撞 BVH。构建一次并让两者共享，可避免按类型重建相同的树。上下文纯粹是一种优化：无论是否共享，结果都必须完全相同，因此跨传感器的一致性仍是管理器的职责。

SharedSensorContext 是围绕 activate / is_active 构建的抽象基类；子类必须实现每一个生命周期方法。在激活之前读取资源必须抛错，而 update / reset 在未激活时必须为空操作：

from genesis.engine.sensors.base_sensor import SharedSensorContext


class MyBVHContext(SharedSensorContext):
    def __init__(self, sim):
        super().__init__(sim)  # stores the sim, marks the context inactive
        self._bvh = None

    @property
    def bvh(self):
        if not self.is_active:
            raise gs.GenesisException("MyBVHContext queried before activation.")
        return self._bvh

    def activate(self):  # idempotent; the first consumer's build() triggers construction
        if self.is_active:
            return
        self._active = True
        self._bvh = build_bvh(self._sim)

    def update(self):  # once per step, before any consuming sensor reads it
        if self.is_active:
            self._bvh.refresh()

    def reset(self, envs_idx):  # on scene.reset()
        if self.is_active:
            self._bvh.flag_rebuild()

    def destroy(self):  # on teardown
        self._bvh = None

把它声明为 Sensor[...] 的第二个类型参数，在 build() 中激活，并通过产出型 hook 的首个 shared_context 参数读取：

class MySensor(SimpleSensor[MyOptions, MyBVHContext, MySharedMetadata]):
    def build(self):
        super().build()
        self._shared_context.activate()  # idempotent; the first consumer builds the resource

    @classmethod
    def _update_raw_data(cls, shared_context, shared_metadata, raw_data_T):
        raw_data_T.copy_(query(shared_context.bvh, ...))

声明同一上下文类的两种传感器类型共享同一个实例。管理器每步在按类型更新循环之前刷新所有已激活的上下文各一次，因此在 _update_raw_data 内读取的上下文已经是最新的。声明 None 的传感器类型收到 shared_context=None。

相机类传感器#

BaseCameraSensor 是直接继承 Sensor 的类，它把每个内置相机的“读取时惰性渲染”模式固定下来（RasterizerCameraSensor、RaytracerCameraSensor、BatchRendererCameraSensor）。凡是靠渲染场景、而不是靠每步读取物理信号来产生图像的传感器，都可以用它。它为你提供：

  • 读取时惰性渲染 + 逐步缓存： 同一控制步内的多次 read() 共享一次渲染，因此你永远不需要实现 _update_shared_cache。

  • 连杆挂载，配合 pos / lookat / up，把世界空间变换交给你的渲染器逐帧应用。

  • RGB 输出，形状为 ([n_envs,] h, w, 3)，dtype 为 torch.uint8，由 options.res 声明，以 CameraReturnType NamedTuple 返回。

它不参与环形管线（uses_ring_pipeline = False），并在构造时拒绝 delay、jitter 和 history_length，因为这些都依赖它不分配的返回空间环。实现两个 hook：

class MyCameraSensor(BaseCameraSensor[MyCameraOptions]):
    def _apply_camera_transform(self, camera_T: torch.Tensor) -> None:
        # camera_T is a (4, 4) world-space transform. Apply it to your renderer's camera.
        ...

    def _render_current_state(self) -> None:
        # Render into this camera's slot of the per-class image cache. At most once per step.
        ...

完整的示例参见 RasterizerCameraSensor。这里无法使用标准的误差开关；任何误差模型都必须放在 _render_current_state 或重写的 _post_process 中。对于非 RGB 输出（深度、分割、法线），请重写 _get_return_format / _get_cache_dtype 并调整底层存储，或者退回到直接继承 Sensor。

内置传感器重写了什么#

要挑对 hook，就参照最接近的内置传感器。每个都实现了 _get_return_format 和 _get_cache_dtype；下表列出额外的重写。

传感器

_update_raw_data

_apply_transform

_post_process（及中间缓冲区）

JointTorqueSensor

是

—

恒等；返回 (n_dofs,) float

ContactSensor

是

—

bool 阈值；返回 (1,) bool，通过 _get_intermediate_dtype 使中间为 (1,) float

ContactForceSensor

是

—

clamp 与死区；shape 和 dtype 保持不变，用 no-op 的中间重写作为确认

IMUSensor

是

是（本体坐标系对齐）

恒等；返回 NamedTuple

RaycasterSensor / DepthCameraSensor

是

—

恒等

TemperatureGridSensor

是

是（读取 timeline.at(1) 的 RC 滤波）

恒等

任意 *CameraSensor

—

—

恒等；继承 BaseCameraSensor

每个 SimpleSensor 都原样继承 _apply_hardware_imperfections；只有需要非标准误差模型时才重写它。

绘制调试标记#

以 draw_debug=True 创建的传感器会在每次刷新时，在查看器中、以及渲染调试几何的相机中绘制标记。重写 _draw_debug(self, context, envs_idx) 来定义它们。该方法是返回标记，而不是添加标记：用 context.make_debug_* 方法（line、arrow、sphere、spheres、box、mesh、frame、frames、points、pyramid，参数与对应的 scene.draw_debug_* 相同）构建每个标记，并为每个标记返回一个 (env_idx, node, None) 三元组。查看器负责添加它们，并在下次刷新时替换。

envs_idx 列出要绘制的已渲染环境，是 context.rendered_envs_idx 中的索引。env_idx 是标记所属的已渲染环境；对于跨多个环境的标记（例如用第一个环境的数据绘制、在所有环境中显示），它为 None：

def _draw_debug(self, context, envs_idx):
    # Draw the reading of the first rendered environment, if it is not stale
    if 0 not in envs_idx:
        return []
    env_idx = context.rendered_envs_idx[0] if self._manager._sim.n_envs > 0 else None
    if not self.read(env_idx):
        return []
    pos = self._link.get_pos(env_idx, relative=False).reshape((3,))
    return [(None, context.make_debug_sphere(pos=pos, radius=0.02, color=(1.0, 0.0, 0.0, 1.0)), None)]

传感器的数据由一步计算产生，因此当状态在没有执行步进的情况下发生变化（例如通过 entity.set_pos(...) 或 entity.set_qpos(...)）后，数据就不再与场景匹配。类属性 invalidated_by 列出刚体与运动学求解器中会使数据失效、直到下一步才恢复的变更种类（gs.engine.solvers.base_solver.StateChange）。它默认包含所有变更种类，这永远是安全的。数据只依赖部分状态的传感器可以缩小这个范围，例如只依赖连杆位置的传感器可用 frozenset((StateChange.GEOMETRY,))。调试标记展示的就是数据，因此在此类变更触及的环境中，标记会一直隐藏到下一步，而不是在新位姿上显示旧数据。

需要复核的要点#

  • 原地填充 raw_data_T，绝不重新绑定它。 写成 raw_data_T = something_new 会让框架拥有的缓冲区原封不动，并悄悄破坏管线。请通过 raw_data_T.copy_(...)、raw_data_T[...] = ...，或把它作为输出参数的 kernel 来写入。

  • 读取是幂等的。 不要在 read() 内修改状态。状态变更应放在管理器每步调用一次的逐步 hook 中。

  • hook 按类运行一次，而不是按实例或按环境。 在拼接后的每类张量上向量化。

  • shape 按实例，dtype 按类。 _get_return_format 和 _get_intermediate_format 是实例方法，因此选项可以影响形状；_get_cache_dtype 和 _get_intermediate_dtype 是类方法，因为每个实例共享同一种 dtype。

  • 复用代码库中的工具。 genesis.utils.misc 中的 concat_with_tensor、make_tensor_field 和 tensor_to_array 符合每个内置传感器遵循的约定。

另见#

  • 传感器管线：管线在运行时如何执行，以及中间空间与返回空间的完整分离。

  • 传感器：使用内置传感器。