Implementation:Google deepmind Dm control Viewer Util

Knowledge Sources	Google_deepmind_Dm_control
Domains	Utilities, Visualization, Observer Pattern
Last Updated	2026-02-15 04:00 GMT

Overview

This module provides general-purpose utility classes and functions used throughout the viewer package, including observable collections, time management, value integration, atomic actions, timers, and error logging.

Description

The viewer utility module contains foundational building blocks used pervasively across the viewer application. QuietSet is a set-like container with += and -= operators that silently ignores removal of missing items and discards None values, serving as the observer pattern foundation for event handling throughout the viewer.

TimeMultiplier controls the relative simulation speed from 1/32x to 1x realtime, providing increase() (double) and decrease() (halve) methods. Integrator accumulates values and computes running averages over a configurable refresh period, used for FPS and CPU usage counters. AtomicAction prevents interruption of in-progress actions using a watermark pattern, ensuring that only one camera or manipulation action is active at a time.

ObservableFlag extends QuietSet to provide a toggleable boolean flag that notifies all registered listeners when its state changes, with new listeners receiving the current state immediately upon registration. Timer measures elapsed time between ticks and provides a measure_time() context manager for precise timing. ErrorLogger is a context manager that catches all exceptions, logs them via absl.logging, writes tracebacks to stderr, and notifies listeners while allowing execution to continue. NullErrorLogger is a pass-through replacement that re-raises all exceptions.

Helper functions include is_scalar() for checking if a value can be converted to float, to_iterable() for wrapping single items in a list, and interleave() for interleaving two iterables.

Usage

Use this module's classes when building or extending the viewer. QuietSet is the standard observer mechanism, Timer and Integrator handle performance measurement, AtomicAction ensures input action atomicity, and ErrorLogger provides graceful error handling during environment interaction.

Code Reference

Source Location

• Repository: Google_deepmind_Dm_control
• File: dm_control/viewer/util.py
• Lines: 1-335

Signature

def is_scalar(value):
    """Checks if the supplied value can be converted to a scalar."""

def to_iterable(item):
    """Converts an item or iterable into an iterable."""

def interleave(a, b):
    """Interleaves the contents of two iterables."""

class QuietSet:
    def __init__(self): ...
    def __iadd__(self, items): ...  # += operator
    def __isub__(self, items): ...  # -= operator
    def __len__(self): ...
    def __iter__(self): ...

class TimeMultiplier:
    def __init__(self, initial_time_multiplier): ...
    def get(self): ...
    def set(self, value): ...
    def increase(self): ...
    def decrease(self): ...

class Integrator:
    def __init__(self, refresh_rate=0.5): ...
    @property
    def value(self): ...
    @value.setter
    def value(self, val): ...

class AtomicAction:
    def __init__(self, state_change_callback=None): ...
    def begin(self, watermark): ...
    def end(self, watermark): ...
    @property
    def in_progress(self): ...
    @property
    def watermark(self): ...

class ObservableFlag(QuietSet):
    def __init__(self, initial_value): ...
    def toggle(self): ...
    @property
    def value(self): ...
    @value.setter
    def value(self, val): ...

class Timer:
    def __init__(self): ...
    def tick(self): ...
    def measure_time(self): ...  # context manager
    @property
    def measured_time(self): ...

class ErrorLogger:
    def __init__(self, listeners): ...
    def __enter__(self, *args): ...
    def __exit__(self, exception_type, exception_value, tb): ...
    @property
    def errors_found(self): ...

class NullErrorLogger:
    def __enter__(self, *args): ...
    def __exit__(self, error_type, value, tb): ...
    @property
    def errors_found(self): ...

Import

from dm_control.viewer import util

I/O Contract

Inputs

Name	Type	Required	Description
initial_time_multiplier	float	Yes (TimeMultiplier)	Initial simulation speed (1.0 = realtime, clamped to [1/32, 1])
refresh_rate	float	No	Integrator averaging period in seconds (default 0.5)
state_change_callback	callable	No	AtomicAction callback invoked on state changes
initial_value	bool	Yes (ObservableFlag)	Initial boolean state of the flag
listeners	iterable	Yes (ErrorLogger)	Callables to notify when errors are caught (each accepts a string)

Outputs

Name	Type	Description
QuietSet	iterable	Set-like container supporting += and -= for observer management
TimeMultiplier.get()	float	Current time multiplier value
Integrator.value	float	Running average of integrated values
AtomicAction.in_progress	bool	Whether an action is currently active
AtomicAction.watermark	object	Identifier of the current active action
Timer.tick()	float	Elapsed time since last tick in seconds
Timer.measured_time	float	Time measured by the last tick or measure_time call
ErrorLogger.errors_found	bool	Whether any errors were caught

Usage Examples

QuietSet as Observer

from dm_control.viewer import util

# Create an observable set
listeners = util.QuietSet()

def on_event(data):
    print(f'Received: {data}')

# Add listener
listeners += on_event

# Notify all listeners
for listener in listeners:
    listener('hello')

# Remove listener (silent if not present)
listeners -= on_event

Timer and Integrator

from dm_control.viewer import util

# Measure frame times
timer = util.Timer()
elapsed = timer.tick()

# Use context manager for precise timing
with timer.measure_time():
    # ... do work ...
    pass
print(f'Elapsed: {timer.measured_time}')

# Accumulate FPS measurements
fps_counter = util.Integrator(refresh_rate=1.0)
fps_counter.value = 60.0
fps_counter.value = 58.0
print(f'Average FPS: {fps_counter.value}')

AtomicAction

from dm_control.viewer import util

action = util.AtomicAction()
action.begin('rotate')
print(action.in_progress)   # True
print(action.watermark)     # 'rotate'
action.end('rotate')
print(action.in_progress)   # False