Implementation:Google deepmind Dm control Observable Base

Knowledge Sources	Google_deepmind_Dm_control
Domains	Robotics Simulation, Reinforcement Learning
Last Updated	2026-02-15 04:00 GMT

Overview

This module defines the abstract base class for observables and three concrete implementations for creating observations from generic callables, named MuJoCo features, and MuJoCo cameras.

Description

The Observable abstract base class is the core abstraction of the dm_control observation system. Every observation in a Composer environment is represented as an Observable subclass. The base class holds configurable properties: update_interval (how often the observation is updated), buffer_size (maximum buffer size for time-series observations), delay (additional simulation steps before returning an observation), aggregator (for reducing buffered observations via min, max, mean, median, or sum), corruptor (for adding noise or other modifications), and enabled (whether the observable is active).

The module provides predefined AGGREGATORS created via _make_aggregator using NumPy reduction functions along axis 0. The observation_callable method wraps the raw observation with optional corruption. The configure method allows setting multiple attributes at once.

Three concrete implementations are provided: Generic wraps an arbitrary callable that accepts a Physics instance, making it the most flexible option. MujocoFeature reads named data fields (e.g., qpos, xpos) from physics.named.data, supporting both string names and callable name generators. MujocoCamera renders images from named MuJoCo cameras with configurable height, width, and depth/RGB modes, and provides an array_spec describing the output shape and dtype.

Usage

Use the Observable classes when defining observations for Composer entities and tasks. Use Generic for custom observation logic, MujocoFeature for standard simulation state observations (joint positions, body positions, etc.), and MujocoCamera for visual observations from camera rendering.

Code Reference

Source Location

• Repository: Google_deepmind_Dm_control
• File: dm_control/composer/observation/observable/base.py
• Lines: 1-309

Signature

AGGREGATORS = {
    'min': ..., 'max': ..., 'mean': ..., 'median': ..., 'sum': ...
}

class Observable(metaclass=abc.ABCMeta):
    def __init__(self, update_interval, buffer_size, delay, aggregator, corruptor): ...
    update_interval: property  # int
    buffer_size: property      # int or None
    delay: property            # int or None
    aggregator: property       # callable or None
    corruptor: property        # callable or None
    enabled: property          # bool
    array_spec: property       # specs.Array or None
    def _callable(self, physics): ...        # abstract
    def observation_callable(self, physics, random_state=None): ...
    def __call__(self, physics, random_state=None): ...
    def configure(self, **kwargs): ...

class Generic(Observable):
    def __init__(self, raw_observation_callable, update_interval=1,
                 buffer_size=None, delay=None, aggregator=None, corruptor=None): ...

class MujocoFeature(Observable):
    def __init__(self, kind, feature_name, update_interval=1,
                 buffer_size=None, delay=None, aggregator=None, corruptor=None): ...

class MujocoCamera(Observable):
    def __init__(self, camera_name, height=240, width=320, update_interval=1,
                 buffer_size=None, delay=None, aggregator=None, corruptor=None, depth=False): ...
    height: property   # int
    width: property    # int
    array_spec: property  # specs.Array

Import

from dm_control.composer.observation.observable import base
# Or specific classes:
from dm_control.composer.observation.observable.base import Observable, Generic, MujocoFeature, MujocoCamera

I/O Contract

Inputs

Name	Type	Required	Description
raw_observation_callable	Callable[[Physics], Any]	Yes (Generic)	A callable accepting a Physics instance and returning the observation value
kind	str	Yes (MujocoFeature)	Field name in MuJoCo's mjData struct (e.g., 'qpos', 'xpos')
feature_name	str or list[str] or Callable	Yes (MujocoFeature)	Name(s) of entities in the MuJoCo XML model
camera_name	str	Yes (MujocoCamera)	Name of the MuJoCo camera to render from
update_interval	int	No	Number of simulation steps between successive updates (default: 1)
buffer_size	int or None	No	Maximum size of the returned buffer
delay	int or None	No	Additional simulation steps before an observation is returned
aggregator	str or Callable or None	No	Reduction operation ('min', 'max', 'mean', 'median', 'sum') or callable
corruptor	Callable or None	No	Function that modifies observations (e.g., adding noise)
depth	bool	No (MujocoCamera)	If True, renders a 1-channel depth image instead of 3-channel RGB

Outputs

Name	Type	Description
observation	np.ndarray	The observation value, potentially corrupted if a corruptor is set
observation_callable	Callable	A zero-argument callable that returns the observation
array_spec	specs.Array or None	Specification of the observation shape and dtype (MujocoCamera provides this directly)

Usage Examples

Basic Usage

from dm_control.composer.observation.observable.base import Generic, MujocoFeature, MujocoCamera

# Generic observable from a lambda
joint_obs = Generic(lambda physics: physics.named.data.qpos['my_joint'])
joint_obs.enabled = True
value = joint_obs(physics)

# MujocoFeature observable for joint positions
qpos_obs = MujocoFeature(kind='qpos', feature_name='hinge_joint')
qpos_obs.enabled = True
value = qpos_obs(physics)

# MujocoCamera observable for RGB images
camera_obs = MujocoCamera(camera_name='front_camera', height=480, width=640)
camera_obs.enabled = True
image = camera_obs(physics)

# Configure multiple properties at once
qpos_obs.configure(enabled=True, update_interval=2, aggregator='mean')

# Using a corruptor to add noise
import numpy as np
def add_noise(obs, random_state):
    return obs + random_state.normal(scale=0.01, size=obs.shape)

noisy_obs = Generic(lambda physics: physics.data.qpos, corruptor=add_noise)