Writing a custom sensor#
This page is for advanced users adding a new sensor type. It is the author’s counterpart to the sensor pipeline, which describes how sensors execute at runtime; here the focus is the interface you implement, the shape and dtype contracts each override must honor, and how the framework pairs your sensor with its options automatically. If you only want to use the built-in sensors, start with Sensors instead.
The base classes live in genesis/engine/sensors/base_sensor.py. In almost every case you derive from SimpleSensor and override only the hooks you need. Deriving directly from Sensor is reserved for sensors that bypass the standard pipeline entirely, and the built-in cameras do exactly that through BaseCameraSensor.
Minimal working example#
A complete sensor is three classes: a user-facing options dataclass, a per-class metadata container, and the sensor itself. The following proximity sensor reports the distance from an attached link to the world origin, clamped to a maximum range. It is enough to be usable through scene.add_sensor(...), and every imperfection feature (noise, bias, random walk, delay, jitter, history) is inherited from SimpleSensor and applied uniformly.
# 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
The rest of this page explains why each piece exists and which additional hooks the more elaborate sensors override.
The classes you write#
Every sensor contributes the same three artifacts, plus two optional ones.
Options class: a public dataclass carrying every per-sensor parameter, inheriting
SimpleSensorOptions(or the appropriate mixin). It is generic-parameterized with the sensor class as a forward reference,SimpleSensorOptions["MyProximitySensor"]. This is the only object the user constructs.Metadata class: the per-sensor-class runtime state shared by every instance of the sensor in a scene. It inherits
SimpleSensorMetadataforSimpleSensor-derived sensors, orSharedSensorMetadatafor sensors deriving fromSensordirectly.Sensor class: the implementation. It inherits
SimpleSensor[OptionsT, ContextT, MetadataT]and overrides the hooks below.A
NamedTuplereturn type (optional): declare one when the sensor returns several tensors, and pass it as the fourth type parameter. See Returning a NamedTuple.A
SharedSensorContextsubclass (optional): declare one only when this sensor shares an expensive resource with other sensor types, and pass it as the second type parameter. See Sharing a resource across sensor types.
The generic signature is Sensor[OptionsT, ContextT, MetadataT, DataT]: options first, then the cross-type shared context (None when there is none), then the metadata type, then the data type (DataT defaults to tuple).
Registration is automatic#
Sensors are never registered by hand. When a Sensor subclass names its options class as the first type parameter, Sensor.__init_subclass__ records the pairing in SensorManager.SENSOR_TYPES_MAP the moment the class body runs. The user then only ever constructs the options instance and hands it to scene.add_sensor(...), which resolves the sensor class, instantiates it, and returns the sensor.
That leaves two supported placements:
In-tree (built-in sensors): options in
genesis/options/sensors/*.py, sensor ingenesis/engine/sensors/*.py. Both are imported through their package__init__, so the pairing is registered at import.Out-of-tree (third-party plugins): put your options and sensor in sibling submodules of one package (for example
my_plugin/options.pyandmy_plugin/sensor.py). As long as your code imports the options module before constructing the options, the framework resolves the sensor class transparently on the firstscene.add_sensor(...)call.
Note
__init_subclass__ also enforces the contract: a concrete sensor that declares its own options class must also declare its metadata type parameter, and any class that overrides _post_process must declare an intermediate buffer (see Projecting to a different return type). Both violations raise TypeError at class-definition time, before any scene is built.
Required overrides#
Two overrides are required of every concrete sensor, and SimpleSensor adds a third.
_get_return_format(self) -> tuple[...]: an instance method returning the shape of whatread()produces. Shape is per-instance by design, because options may legitimately determine it (a raycaster’s pattern, a camera’s resolution, a proximity sensor’s probe positions). Return(N,)for a single tensor ofNscalars, or a tuple of tuples such as((3,), (3,), (3,))for a multi-tensor return._get_cache_dtype(cls) -> torch.dtype: a classmethod returning the dtype of whatread()produces. Dtype is class-uniform, not per-instance: the manager packs every instance of a class into one contiguous per-class slice of a per-dtype buffer, so all instances must share one dtype. If you need different dtypes, use two sensor classes._update_raw_data(cls, shared_context, shared_metadata, raw_data_T): theSimpleSensorkernel that computes the ground-truth value for every sensor of this class at the current step. This isSimpleSensor’s single abstract producing hook.
The split between an instance method for shape and a classmethod for dtype is deliberate. It lets the manager resolve the per-class dtype without instantiating a sensor, while still letting the per-instance shape depend on options.
raw_data_T has shape (cols, B) (column-major, with the batch dimension B last) so that per-class row slices are C-contiguous when several sensor classes write into the same intermediate cache. Always populate it in place:
@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 is the cross-type resource, or None; most sensors ignore it. Hooks are called once per class per step, never per instance and never per environment, so vectorize accordingly.
Choosing a base class#
Base |
When to use |
|---|---|
|
Almost always. The standard per-step pipeline: raw, physics imperfections, transform, hardware imperfections, post-process, delay sampling. |
|
Same pipeline, but |
|
Camera-style sensors that render an image lazily on |
|
Only when neither standard pipeline fits. You then implement |
Mixins compose onto the base:
RigidSensorOptionsMixin/KinematicSensorOptionsMixin: on the options side, for sensors attached to aRigidEntityor anyKinematicEntityrespectively. Combine withSimpleSensorOptionsthrough multiple inheritance.RigidSensorMixin/RigidSensorMetadataMixin: on the sensor and metadata side, giving you a typedsolverfield, thelinks_idxbookkeeping, andset_pos_offset/set_quat_offset.
Per-class metadata#
The metadata class holds anything that is per-sensor-class rather than per-instance and is read by your hooks: solver and entity references, per-sensor index tensors concatenated at build time (links_idx, thresholds, max_range), per-sensor offsets, filter coefficients, and precomputed flags that gate slow paths.
SimpleSensorMetadata already provides the imperfection state (noise, bias, random_walk, resolution, jitter_ts) and the matching has_any_* flags. Subclass it and declare your fields with make_tensor_field((shape,)) so they are auto-allocated:
@dataclass
class MySharedMetadata(RigidSensorMetadataMixin, SimpleSensorMetadata):
thresholds: torch.Tensor = make_tensor_field((0,))
custom_offsets: torch.Tensor = make_tensor_field((0, 3))
In your sensor’s build(), append this instance’s entries with concat_with_tensor, as in the minimal example above. The manager calls build() once per sensor instance at scene-build time; growing the shared tensors there is what lets a single kernel run over every instance at once.
Optional pipeline hooks#
SimpleSensor runs a fixed per-step pipeline and gives every stage a default. Override a stage only when your sensor needs it. The stages run per branch, in this order:
Ground-truth branch:
_update_raw_data, then_apply_transform(is_measured=False), then_post_process(is_measured=False).Measured branch:
_update_raw_data, then_apply_physics_imperfections, then_apply_transform(is_measured=True), then_apply_hardware_imperfections, then_post_process(is_measured=True), then delay sampling.
Both branches keep their own intermediate-space timeline ring holding post-transform, pre-hardware-imperfection values, so a stateful _apply_transform filter always reads previous slots that are clean of hardware noise. The distinction between the three noise stages is where the perturbation physically originates.
_apply_physics_imperfections(cls, shared_metadata, data, timeline): random fluctuation of the underlying phenomenon that the simulator does not model (genuine drift, fine-scale turbulence on the field). Measured-only, applied before_apply_transform, so it propagates through the response model on later steps. Default: no-op._apply_transform(cls, shared_metadata, data, timeline, *, is_measured): a coordinate transform and/or a stateful response model of the sensor element (thermal mass, RC time constant, mechanical bandwidth). Called on both branches; the coordinate part runs unconditionally, and you gate an element-specific effect that must not appear in ground truth onif is_measured:. Mutatedatain place; read history withtimeline.at(1),timeline.at(2). The IMU uses this for its body-frame alignment rotation; the temperature-grid sensor uses it for an RC filter._apply_hardware_imperfections(cls, shared_metadata, measured_slot_0): the perturbations the readout electronics introduce at the sensor output.SimpleSensoralready implementsnoise,bias,random_walk, andresolutionhere, gated by thehas_any_*flags so an all-zero class pays nothing. Override only for a non-standard model, and callsuper()first for the standard terms:
@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()
For a sensor whose noise is intrinsic to the physics computation (a single kernel pass must produce both the ideal and the noised value because the noise shapes the kernel’s branches), override _update_current_timestep_data instead. It writes the ground-truth slice and the noised measured slot in one pass, and the rest of the pipeline still runs on top.
Projecting to a different return type#
_post_process(cls, shared_metadata, tensor, timeline, *, is_measured) -> torch.Tensor projects from the pipeline-internal intermediate space into the user-facing return space. Override it when the output type or shape differs from the internal representation: a bool threshold on ContactSensor, a deadband and saturation on ContactForceSensor. Return the projected tensor; the manager writes it into the return-space ring and delay-samples it.
@classmethod
def _post_process(cls, shared_metadata, tensor, timeline, *, is_measured):
return tensor > shared_metadata.thresholds # float intermediate, bool return
Overriding _post_process requires also overriding _get_intermediate_format and/or _get_intermediate_dtype; the framework raises TypeError at class-definition time otherwise. The reason is structural: the intermediate buffer must be a distinct buffer, because the timeline ring lives in intermediate space and mixing data spaces would corrupt any _apply_transform filter that reads previous slots. When the projection genuinely preserves shape and dtype, override one of them as a no-op returning the return-space value: the override is the explicit acknowledgment that the buffers are distinct.
_get_intermediate_format(self): shape of the internal buffer; defaults to_get_return_format()._get_intermediate_dtype(cls): dtype of the internal buffer; defaults to_get_cache_dtype().ContactSensoroverrides it togs.tc_floatbecause its kernel is float but its return is bool.
Returning a NamedTuple#
For a multi-tensor return, declare a NamedTuple and pass it as the fourth type parameter. _get_return_format then returns one shape per field, in field order:
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)
The manager allocates a single contiguous slab per sensor and slices it on read; read() reconstructs and returns the NamedTuple, with the leading batch dimension dropped when the scene has no environments. Each field has shape ([n_envs,] *field_shape).
Camera-style sensors#
BaseCameraSensor is a Sensor-direct subclass that codifies the lazy-render-on-read pattern of every built-in camera (RasterizerCameraSensor, RaytracerCameraSensor, BatchRendererCameraSensor). Use it for any sensor that produces an image by rendering the scene rather than by reading physics signals each step. It gives you:
Lazy render-on-read with per-step caching: multiple
read()calls in one step share a single render, so you never implement_update_shared_cache.Link attachment with
pos/lookat/up, handing you the world-space transform to apply to your renderer each frame.An RGB output of shape
([n_envs,] h, w, 3)and dtypetorch.uint8, declared fromoptions.res, returned as aCameraReturnTypeNamedTuple.
It opts out of the ring pipeline (uses_ring_pipeline = False) and rejects delay, jitter, and history_length at construction, since those depend on the return-space ring it does not allocate. You implement two hooks:
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.
...
See RasterizerCameraSensor for a complete worked example. The standard imperfection knobs are unavailable here; any imperfection model must live inside _render_current_state or a _post_process override. For non-RGB output (depth, segmentation, normals), override _get_return_format / _get_cache_dtype and adapt the backing store, or drop down to a bare Sensor subclass.
What the built-in sensors override#
To pick the right hooks, mirror the closest built-in sensor. Every one implements _get_return_format and _get_cache_dtype; the table shows the additional overrides.
Sensor |
|
|
|
|---|---|---|---|
|
yes |
— |
identity; return |
|
yes |
— |
bool threshold; return |
|
yes |
— |
clamp and deadband; shape and dtype preserved, no-op intermediate override as acknowledgment |
|
yes |
yes (body-frame alignment) |
identity; |
|
yes |
— |
identity |
|
yes |
yes (RC filter reading |
identity |
Any |
— |
— |
identity; derives from |
_apply_hardware_imperfections is inherited unchanged by every SimpleSensor; override it only for a non-standard imperfection model.
Things to double-check#
Populate
raw_data_Tin place; never rebind it. Assigningraw_data_T = something_newleaves the framework-owned buffer untouched and silently breaks the pipeline. Write viaraw_data_T.copy_(...),raw_data_T[...] = ..., or a kernel that takes it as an output argument.Reads are idempotent. Do not mutate state inside
read(). State changes belong in the per-step hooks the manager calls once per step.Hooks run once per class, not per instance or per environment. Vectorize over the concatenated per-class tensors.
Shape is per-instance; dtype is class-uniform.
_get_return_formatand_get_intermediate_formatare instance methods so options can affect shape;_get_cache_dtypeand_get_intermediate_dtypeare classmethods because every instance shares one dtype.Reuse the codebase utilities.
concat_with_tensor,make_tensor_field, andtensor_to_arrayfromgenesis.utils.miscmatch the conventions every built-in sensor follows.
See also#
Sensor pipeline: how the pipeline executes at runtime, and the intermediate-versus-return separation in full.
Sensors: using the built-in sensors.