Skip to content

column

The telemetry column model: a column's name, split into its parts, together with its metadata.

Every column in a telemetry file is named Category/subgroup/...:attr (for example Bodies/box1/xpos:x). Column is the one place that builds and parses that grammar, so the writer (SignalManager.post) and any reader agree on it, and a name part that would corrupt it fails loudly at construction instead of silently landing in a different position.

MATRIX_ATTRS module-attribute

Python
MATRIX_ATTRS = tuple((str(i)) for i in (range(9)))

The attrs of a flattened 3x3 rotation matrix (xmat, ximat), which posts one column per element.

Column

Bases: BaseModel

One telemetry column: its name parts and its metadata. str(column) is the column name as it appears in the telemetry file.

Returned by: Used by:

category instance-attribute

Python
category: SignalCategory | _Part

Top-level category, e.g. Bodies (see SignalCategory for the built-in ones). Must not be empty or contain / or :.

subgroups class-attribute instance-attribute

Python
subgroups: tuple[_Part, ...] = ()

Folders under the category, e.g. ("box1", "xpos"). Empty parts are dropped; none may contain / or :.

attr class-attribute instance-attribute

Python
attr: _Part | None = Field(default=None)

The component or attribute name, e.g. x. An empty value is treated as unset; it may not contain / or :.

metadata class-attribute instance-attribute

Python
metadata: ColumnMetadata = Field(
    default_factory=ColumnMetadata
)

Metadata persisted for this column in the telemetry file's footer. Validated when the column is built (a plain dict is accepted), so the per-timestep merges that produce it can skip validation.

parse classmethod

Python
parse(
    name: str, metadata: ColumnMetadata | None = None
) -> Column

Splits a column name into its parts; the exact inverse of str(). Raises ValueError for a malformed name (an empty part).

Each distinct name is parsed once per process and then served from a cache, so parsing every column of a large frame repeatedly costs a dictionary lookup per name. The returned Column is shared, which is safe because it is frozen.

Source code in src/mujoco_mojo/utils/column.py
Python
111
112
113
114
115
116
117
118
119
120
121
122
123
@classmethod
def parse(cls, name: str, metadata: ColumnMetadata | None = None) -> Column:
    """
    Splits a column name into its parts; the exact inverse of `str()`. Raises `ValueError` for a malformed name (an empty part).

    Each distinct name is parsed once per process and then served from a cache, so parsing every column of a large frame repeatedly costs a dictionary lookup per name. The returned `Column` is shared, which is safe because it is frozen.
    """
    column = _parse(name)
    if metadata is None:
        return column
    return column.model_copy(
        update={"metadata": ColumnMetadata.validated(metadata)}
    )

covers

Python
covers(other: Column) -> bool

True if other is this column or lies beneath it: the same category, other's subgroups start with this column's subgroups, and this column's attr is unset or equal to other's. Metadata is ignored.

A partial column therefore selects a whole subtree: Column(category="Bodies", subgroups=("box1",)) covers Bodies/box1/xpos:x and Bodies/box1:ke_trans.

Source code in src/mujoco_mojo/utils/column.py
Python
125
126
127
128
129
130
131
132
133
134
135
136
def covers(self, other: Column) -> bool:
    """
    True if `other` is this column or lies beneath it: the same category, `other`'s subgroups start with this column's subgroups, and this column's `attr` is unset or equal to `other`'s. Metadata is ignored.

    A partial column therefore selects a whole subtree: `Column(category="Bodies", subgroups=("box1",))` covers `Bodies/box1/xpos:x` and `Bodies/box1:ke_trans`.
    """
    n = len(self.subgroups)
    return (
        other.category == self.category
        and other.subgroups[:n] == self.subgroups
        and (self.attr is None or self.attr == other.attr)
    )

__contains__

Python
__contains__(other: Self | str) -> bool

child in parent, the same test as parent.covers(child); a name string is parsed first, and a string outside the telemetry grammar is never contained. Equality is untouched, so ==, hashing, and x in [columns] keep meaning exact match.

Source code in src/mujoco_mojo/utils/column.py
Python
144
145
146
147
148
149
150
151
152
def __contains__(self, other: Self | str) -> bool:
    """`child in parent`, the same test as `parent.covers(child)`; a name string is parsed first, and a string outside the telemetry grammar is never contained. Equality is untouched, so `==`, hashing, and `x in [columns]` keep meaning exact match."""
    column = other
    if isinstance(other, str):
        try:
            column = _parse(other)
        except ValueError:
            return False
    return isinstance(column, Column) and self.covers(column)

covering_pattern

Python
covering_pattern(columns: Iterable[Column]) -> str

One anchored regex matching every column name that any of columns covers (see Column.covers), with every name part escaped so it matches literally.

It lets a frame select by Column inside polars' regex engine instead of parsing each column name in Python, which matters for frames with tens of thousands of columns.

Source code in src/mujoco_mojo/utils/column.py
Python
175
176
177
178
179
180
181
def covering_pattern(columns: Iterable[Column]) -> str:
    """
    One anchored regex matching every column name that any of `columns` covers (see `Column.covers`), with every name part escaped so it matches literally.

    It lets a frame select by `Column` inside polars' regex engine instead of parsing each column name in Python, which matters for frames with tens of thousands of columns.
    """
    return "^(?:" + "|".join(c._pattern_body() for c in columns) + ")$"

fan_out

Python
fan_out(
    category: str,
    subgroups: tuple[str, ...],
    attrs: Iterable[str],
    metadata: ColumnMetadata | None = None,
) -> tuple[Column, ...]

One Column per attr, all sharing the same category, subgroups, and metadata (e.g. the x/y/z/m columns of a vector). Build these once and reuse them: SignalManager.post is called every timestep.

Source code in src/mujoco_mojo/utils/column.py
Python
184
185
186
187
188
189
190
191
192
193
194
195
def fan_out(
    category: str,
    subgroups: tuple[str, ...],
    attrs: Iterable[str],
    metadata: ColumnMetadata | None = None,
) -> tuple[Column, ...]:
    """One `Column` per attr, all sharing the same category, subgroups, and metadata (e.g. the `x`/`y`/`z`/`m` columns of a vector). Build these once and reuse them: `SignalManager.post` is called every timestep."""
    meta = metadata if metadata is not None else ColumnMetadata()
    return tuple(
        Column(category=category, subgroups=subgroups, attr=attr, metadata=meta)
        for attr in attrs
    )

pose_column_names cached

Python
pose_column_names(
    source: Column | str,
    pos_channel: str = "xpos",
    quat_channel: str = "quat",
) -> tuple[tuple[str, ...], tuple[str, ...]]

The telemetry column names holding an object's pose: (x, y, z) position names and (w, x, y, z) quaternion names, in that order.

source names the object, e.g. Sites/A or Column(category="Sites", subgroups=("A",)), and gives Sites/A/xpos:x through Sites/A/quat:z for the default channels. Cached per source, so repeated lookups cost a dictionary hit. Raises ValueError if source names a single column (has an attr) instead of an object.

Source code in src/mujoco_mojo/utils/column.py
Python
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
@lru_cache(maxsize=4096)
def pose_column_names(
    source: Column | str, pos_channel: str = "xpos", quat_channel: str = "quat"
) -> tuple[tuple[str, ...], tuple[str, ...]]:
    """
    The telemetry column names holding an object's pose: `(x, y, z)` position names and `(w, x, y, z)` quaternion names, in that order.

    `source` names the object, e.g. `Sites/A` or `Column(category="Sites", subgroups=("A",))`, and gives `Sites/A/xpos:x` through `Sites/A/quat:z` for the default channels. Cached per source, so repeated lookups cost a dictionary hit. Raises `ValueError` if `source` names a single column (has an attr) instead of an object.
    """
    column = Column.parse(source) if isinstance(source, str) else source
    if column.attr is not None:
        raise ValueError(
            f"{str(column)!r} names a single column; pass the object it belongs to (for example {'/'.join((column.category, *column.subgroups))!r})"
        )
    pos = fan_out(column.category, (*column.subgroups, pos_channel), "xyz")
    quat = fan_out(column.category, (*column.subgroups, quat_channel), "wxyz")
    return tuple(str(c) for c in pos), tuple(str(c) for c in quat)

pose_missing_error

Python
pose_missing_error(
    source: Column | str,
    available: Iterable[str],
    pos_channel: str = "xpos",
    quat_channel: str = "quat",
) -> ValueError | None

A ValueError naming every pose column of source that is not in available, or None if the pose is complete.

Source code in src/mujoco_mojo/utils/column.py
Python
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def pose_missing_error(
    source: Column | str,
    available: Iterable[str],
    pos_channel: str = "xpos",
    quat_channel: str = "quat",
) -> ValueError | None:
    """A `ValueError` naming every pose column of `source` that is not in `available`, or `None` if the pose is complete."""
    pos_names, quat_names = pose_column_names(source, pos_channel, quat_channel)
    have = (
        available if isinstance(available, (set, frozenset, dict)) else set(available)
    )
    missing = [n for n in (*pos_names, *quat_names) if n not in have]
    if not missing:
        return None
    return ValueError(
        f"No pose for {str(source)!r}: missing columns {', '.join(missing)}. "
        f"Were they requested (for example `request(channels=[{pos_channel!r}, {quat_channel!r}])`)?"
    )