unienv_interface.space.space_utils.batch_utils¶
Space(backend: ComputeBackend[ArrayAPIArray, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT], shape: Optional[Sequence[int]] = None, device: Optional[_SpaceBDeviceT] = None, dtype: Optional[_SpaceBDTypeT] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: ABC, Generic[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Abstract description of a valid data domain.
Spaces carry backend, device, shape, and dtype metadata and define the operations needed by the rest of UniEnv: validation, sampling, empty value creation, serialization-friendly representation, and backend/device conversion for both the space definition and its data.
In-place changes to a space (reassigning one of its attributes, or e.g.
DictSpace.__setitem__) increment the global Space._mutation_epoch.
Caches keyed on a space's identity (e.g. in ChainedTransformation)
include this counter, so mutating a space can never make them stale.
BoxSpace is immutable after construction and is not tracked.
Metadata¶
Every space carries an optional, read-only metadata mapping
(constructor keyword metadata=) describing what its data means, e.g.
BoxSpace(..., shape=(480, 640, 3), metadata={"kind": "image", "format": "rgb"}).
Values must be JSON-serializable; they are validated and deep-copied at
construction and exposed as a :class:SpaceMetadata (empty when absent).
Nested spaces carry their own metadata, so every camera of a DictSpace
observation can be described separately. Use :meth:with_metadata to get
a copy of a space with other metadata.
Metadata is a description, not part of the data domain: __eq__,
is_subspace and is_subspaceeq ignore it (use
space_utils.metadata_utils.spaces_equal_with_metadata to compare it
too). It is kept by to(), the batching utilities, serialization and the
transformations that keep the meaning of the data, and dropped where the
meaning is lost (e.g. flatten_space).
The keys below form a shared vocabulary understood by UniEnv utilities
(e.g. the video wrappers and unienv_data.storages.dict_storage_kwargs_from_space).
None of them is required or enforced at construction, and any other keys
(and kinds) are allowed. "Per element" lists have one entry per element of
the space's last axis:
kind(str): what the data is. Recommended:"image","depth","segmentation","normal","pointcloud","proprio","action","pose","force","tactile","teleop","keypoints","mask","intrinsics","extrinsics","privileged","task"(goals / targets),"state".format(str): pixel format of images,"rgb","bgr","rgba","bgra"or"gray".layout(str): axis order of the trailing image dims,"HWC","CHW"or"HW"(single channel without a channel axis); leading dims are batch dims.units(str, or per element list of str): physical units, e.g."m","mm","rad","rad/s","m/s","m/s^2","N","N*m","A","V","degC","px","s";"normalized"for unitless normalized data.frame(str): reference frame, e.g."world","base","camera_cv"(OpenCV: x right, y down, z forward),"camera_gl","link_local","wrist".rotation(str): rotation representation inside a pose / action vector,"quat_wxyz","quat_xyzw","euler_<axes>"with transforms3d axes (sstatic / extrinsic,rrotating / intrinsic, e.g."euler_sxyz","euler_rxyz"),"rot6d","matrix"or"axis_angle".control(str, actions): what is commanded, e.g."joint_position","joint_velocity","joint_torque","eef_pose","eef_velocity","gripper".mode(str, actions):"absolute","delta"or"velocity".normalized(bool): the values are normalized.range([low, high], scalars or per element lists): the physical range normalized values map to, when known.names(per element list of str): e.g. joint names.source(str): name of the camera / sensor producing the data.fps(positive number): capture rate of the source.label(str): human readable name.
A vector packing several quantities (e.g. position 3 + quaternion 4 +
gripper 1) is described per element with names and / or a units
list, while rotation names the representation of its rotation part.
space_utils.metadata_utils has builders for these keys
(image_metadata, action_metadata, ...), annotate to attach
metadata to nested spaces and validate_metadata to check it.
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
abstractmethod
¶
to(backend: Optional[ComputeBackend] = None, device: Optional[Union[_SpaceBDeviceT, Any]] = None) -> Union[Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT], Space]
Return an equivalent space on another backend and/or device.
abstractmethod
¶
sample(rng: _SpaceBDRNGT, **kwargs) -> Tuple[_SpaceBDRNGT, SpaceDataT]
Draw one valid value from the space and return the advanced RNG.
abstractmethod
¶
create_empty() -> SpaceDataT
Create an empty data structure for this space.
abstractmethod
¶
is_bounded(manner: Literal['both', 'below', 'above'] = 'both') -> bool
Return boolean specifying if this space is bounded in the specified manner.
abstractmethod
¶
contains(x: Any) -> bool
Return boolean specifying if x is a valid member of this space.
is_subspaceeq(other: Space) -> bool
Return whether this space is a non-strict subspace of other (self ⊆ other).
A space a is a non-strict subspace of b (a ⊆ b) when every
valid sample of a is also a valid member of b; equality is
allowed, i.e. a == b implies a.is_subspaceeq(b).
The primary use case is a controller that declares its REQUIRED
observation space and checks
required.is_subspaceeq(env_observation_space): for DictSpace
this must hold even when the environment space exposes EXTRA keys
beyond the required ones (unlike DictSpace.contains, which demands
exact key equality). Controller-required-space checks should typically
use is_subspaceeq rather than the strict is_subspace because a
controller's required space may exactly equal the env space.
Comparison policy (structural only):
- The two spaces must share the same backend type.
- Dtypes must be strictly equal; no implicit cast widening is performed.
deviceis intentionally ignored — two spaces on different devices may still be in a subspace relation.
Cross-type comparisons (e.g. BoxSpace vs DictSpace) return
False rather than raising. Subclasses override this method to
provide concrete structural containment checks; the base implementation
raises NotImplementedError to mirror the abstract-method style of
this class.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
clip(x: Any) -> SpaceDataT
Project x onto this space and return the projected value.
clip is defined on the TARGET (subset) space. Contract: if
self.is_subspaceeq(source) holds, then for every x with
source.contains(x)::
self.contains(self.clip(x))
unless no meaningful projection exists, in which case clip raises
ValueError instead of inventing data (e.g. a required DictSpace
key is missing, or a string / dynamic axis / graph is shorter or
smaller than this space's minimum; each space documents its cases).
Additionally:
clipis idempotent on members:self.clip(y)equalsywheneverself.contains(y).- The input is never mutated; the result is new data, or
x(or parts of it) itself where nothing needs to change. - The result follows this space's backend and dtype conventions (e.g.
a
BoxSpacecasts to its dtype).
Inputs that are not members of any superspace (wrong structure, wrong shape, ...) are not supported and may raise.
The base implementation raises NotImplementedError to mirror the
abstract-method style of this class; subclasses override it.
abstractmethod
¶
get_repr(abbreviate: bool = False, include_backend: bool = True, include_device: bool = True, include_dtype: bool = True) -> str
Return a string representation of the space.
abstractmethod
¶
data_to(data: SpaceDataT, backend: Optional[ComputeBackend] = None, device: Optional[Union[_SpaceBDeviceT, Any]] = None) -> Union[SpaceDataT, Any]
Convert space-compatible data to another backend and/or device.
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
SpaceMetadata(data: Optional[Mapping[str, Any]] = None)
Bases: Mapping
Immutable, JSON-serializable str-keyed mapping describing a space.
Created from any mapping of JSON data (validated and deep-copied); nested
mappings are frozen into SpaceMetadata and lists into tuples. Compares
equal to plain mappings with the same content (lists and tuples are
interchangeable). to_dict() returns a plain, mutable JSON copy.
BinarySpace(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], shape: Sequence[int], dtype: Optional[BDtypeType] = None, device: Optional[BDeviceType] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[BArrayType, BDeviceType, BDtypeType, BRNGType]
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
clip(x: BArrayType) -> BArrayType
Return x (cast to this space's boolean dtype if needed); see Space.clip.
A binary space is only a subspace of an identical binary space, so any
member of a superset is already a member: there is nothing to project.
Raises ValueError if x does not have this space's shape.
get_repr(abbreviate=False, include_backend=True, include_device=True, include_dtype=True)
is_subspaceeq(other: Any) -> bool
Return whether this binary space is a non-strict subspace of other (⊆).
True iff other is a BinarySpace on the same backend with equal
shape and dtype. Since a binary space contains exactly the boolean
tensors of its shape, two such spaces are in a subspace relation iff
they are structurally identical (modulo device).
BoxSpace(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], low: SupportsFloat | BArrayType, high: SupportsFloat | BArrayType, dtype: BDtypeType, device: Optional[BDeviceType] = None, shape: Optional[Sequence[int]] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[BArrayType, BDeviceType, BDtypeType, BRNGType]
Continuous or integer hyper-rectangle defined by elementwise bounds.
A box is immutable after construction (_low/_high are never
reassigned), so values derived from the bounds are computed lazily and
cached on the instance; caches are dropped when pickling.
Create a box with broadcastable low and high bounds.
metadata is an optional JSON-serializable description of the data (see Space).
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
to(backend: Optional[ComputeBackend] = None, device: Optional[Union[BDeviceType, Any]] = None) -> Union[BoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType], BoxSpace]
Return an equivalent box on another backend and/or device.
sample(rng: BRNGType) -> Tuple[BRNGType, BArrayType]
Generates a single random sample inside the Box.
In creating a sample of the box, each coordinate is sampled (independently) from a distribution that is chosen according to the form of the interval:
- :math:
[a, b]: uniform distribution - :math:
[a, \infty): shifted exponential distribution - :math:
(-\infty, b]: shifted negative exponential distribution - :math:
(-\infty, \infty): normal distribution
Returns:
| Type | Description |
|---|---|
Tuple[BRNGType, BArrayType]
|
A sampled value from the Box |
create_empty() -> BArrayType
Allocate an uninitialized array with the box shape and dtype.
clip(x: BArrayType) -> BArrayType
Clamp x elementwise to the bounds of this box (see Space.clip).
x is cast to the box dtype first (members of a superset box are
losslessly castable), so the result always has the box dtype; ±inf
bounds leave the corresponding coordinates untouched. Never raises
for data of the box shape. Returns a new array.
get_repr(abbreviate: bool = False, include_backend: bool = True, include_device: bool = True, include_dtype: bool = True) -> str
is_subspaceeq(other: Any) -> bool
Return whether this box is a non-strict subspace of other (⊆).
True iff other is a BoxSpace on the same backend, with equal
shape and dtype, and other's bounds contain self's bounds
elementwise (other.low <= self.low and other.high >= self.high).
Infinite bounds are handled correctly via direct comparison since
±inf compares as expected against finite values and itself.
device is ignored.
data_to(data: BArrayType, backend: Optional[ComputeBackend] = None, device: Optional[Union[BDeviceType, Any]] = None) -> Union[BArrayType, Any]
Convert data to another backend.
DynamicBoxSpace(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], low: Union[int, float, BArrayType], high: Union[int, float, BArrayType], shape_low: Sequence[int], shape_high: Sequence[int], dtype: BDtypeType, device: Optional[BDeviceType] = None, fill_value: Union[int, float] = 0, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[BArrayType, BDeviceType, BDtypeType, BRNGType]
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
property
¶
low: BArrayType
Broadcasted lower bounds at the space's broadcast shape.
For a dynamic box there is no single fixed shape — only a range
[shape_low, shape_high] per axis. The stored _low/_high
arrays are broadcastable to self._broadcast_shape (shape_low
with size 1 on every dynamic axis, i.e. axes where
shape_low != shape_high). This property returns the bounds
broadcast to that _broadcast_shape, mirroring BoxSpace.low
(which broadcasts to self.shape). Comparisons against data of any
in-range shape broadcast correctly because dynamic axes carry size 1
here. Use get_low(shape) / get_high(shape) to materialize the
bounds at a specific concrete shape.
property
¶
high: BArrayType
Broadcasted upper bounds at the space's broadcast shape.
See low for the rationale behind the chosen shape.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
staticmethod
¶
pad_array_on_axis(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], data: BArrayType, axis: int, target_size: int, fill_value: Union[int, float] = 0) -> BArrayType
staticmethod
¶
get_array_axis_length(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], data: BArrayType, axis: int, fill_value: Union[int, float] = 0) -> int
staticmethod
¶
unpad_array_on_axis(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], data: BArrayType, axis: int, fill_value: Union[int, float] = 0) -> BArrayType
to(backend: Optional[ComputeBackend] = None, device: Optional[Union[BDeviceType, Any]] = None) -> Union[DynamicBoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType], DynamicBoxSpace]
sample(rng: BRNGType) -> Tuple[BRNGType, BArrayType]
Generates a single random sample inside the Box.
In creating a sample of the box, each coordinate is sampled (independently) from a distribution that is chosen according to the form of the interval:
- :math:
[a, b]: uniform distribution - :math:
[a, \infty): shifted exponential distribution - :math:
(-\infty, b]: shifted negative exponential distribution - :math:
(-\infty, \infty): normal distribution
Returns:
| Type | Description |
|---|---|
Tuple[BRNGType, BArrayType]
|
A sampled value from the Box |
pad_data(data: BArrayType, start_axis: Optional[int] = None, end_axis: Optional[int] = None) -> BArrayType
Pad the data to the maximum shape of this space.
unpad_data(data: BArrayType, start_axis: Optional[int] = None, end_axis: Optional[int] = None) -> BArrayType
Unpad the data to the minimum shape of this space.
shape_contains(shape: Sequence[int]) -> bool
Check if the shape is within the bounds of this space.
clip(x: BArrayType) -> BArrayType
Project x onto this dynamic box (see Space.clip).
Axes longer than shape_high are trimmed to their first
shape_high[i] entries, then the values are clamped to the bounds
and cast to the space dtype. Like contains, fill_value gets no
special treatment (padding entries are clamped like any other value).
Raises ValueError if x has a different number of dimensions or
an axis shorter than shape_low (padding would invent data).
get_repr(abbreviate: bool = False, include_backend: bool = True, include_device: bool = True, include_dtype: bool = True) -> str
is_subspaceeq(other: Any) -> bool
Return whether this dynamic box is a non-strict subspace of other (⊆).
True iff other is a DynamicBoxSpace on the same backend with
equal dtype, self's shape range is contained within other's
(other.shape_low <= self.shape_low and
self.shape_high <= other.shape_high per dimension, inf-aware via
plain integer comparison), and the value bounds of self are
contained within other's. The value-bounds check broadcasts both
spaces' bounds to self's broadcast shape (shape_low with size
1 on its dynamic axes).
This covers every stored bound: other's bounds broadcast to it
because an axis that is static in other is static with the same
size in self, and neither space's bounds vary along a dynamic axis.
data_to(data: BArrayType, backend: Optional[ComputeBackend] = None, device: Optional[Union[BDeviceType, Any]] = None) -> Union[BArrayType, Any]
Convert data to another backend.
DictSpace(backend: ComputeBackend[Any, BDeviceType, BDtypeType, BRNGType], spaces: Optional[Union[Dict[str, Space[Any, BDeviceType, BDtypeType, BRNGType]], Sequence[Tuple[str, Space[Any, BDeviceType, BDtypeType, BRNGType]]]]] = None, device: Optional[BDeviceType] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[Dict[str, Any], BDeviceType, BDtypeType, BRNGType]
Cartesian product of named subspaces represented as a mapping.
Create a dictionary-valued space from named child spaces.
metadata describes the dict as a whole; every child keeps its own.
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
to(backend: Optional[ComputeBackend] = None, device: Optional[Union[BDeviceType, Any]] = None) -> Union[DictSpace[BDeviceType, BDtypeType, BRNGType], DictSpace]
clip(x: Mapping[str, Any]) -> Dict[str, Any]
Project x onto this dict space (see Space.clip).
Keys of x that are not in this space are dropped and every value is
clipped recursively with its subspace; the result is a new dict
with exactly this space's keys. Raises ValueError if x is not a
mapping or lacks one of this space's keys (it cannot be invented).
is_subspaceeq(other: Any) -> bool
Return whether this dict space is a non-strict subspace of other (⊆).
True iff other is a DictSpace on the same backend, every key of
self is present in other (self.keys() ⊆ other.keys()), and
for every shared key self.spaces[k].is_subspaceeq(other.spaces[k])
holds recursively. Unlike contains, other is permitted to
expose EXTRA keys beyond those of self — this is the controller
required-observation-space use case where the environment may provide
additional observation entries. device is ignored.
get_repr(abbreviate: bool = False, include_backend: bool = True, include_device: bool = True, include_dtype: bool = True) -> str
GraphSpace(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], node_feature_space: Optional[BoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType]], edge_feature_space: Optional[BoxSpace[BArrayType, BDeviceType, BDtypeType, BRNGType]] = None, is_edge: bool = False, min_nodes: int = 1, max_nodes: Optional[int] = None, min_edges: int = 1, max_edges: Optional[int] = None, batch_shape: Sequence[int] = (), device: Optional[BDeviceType] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[GraphInstance[BArrayType], BDeviceType, BDtypeType, BRNGType], Generic[BArrayType, BDeviceType, BDtypeType, BRNGType]
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
instance-attribute
¶
node_feature_space = node_feature_space if (device is None or node_feature_space is None) else node_feature_space.to(device=device)
instance-attribute
¶
edge_feature_space = edge_feature_space if (device is None or edge_feature_space is None) else edge_feature_space.to(device=device)
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
to(backend: Optional[ComputeBackend] = None, device: Optional[Union[BDeviceType, Any]] = None) -> Union[GraphSpace[BArrayType, BDeviceType, BDtypeType, BRNGType], GraphSpace]
contains(x: GraphInstance[BArrayType]) -> bool
Return boolean specifying if x is a valid member of this space.
clip(x: GraphInstance[BArrayType]) -> GraphInstance[BArrayType]
Project the (possibly batched) graph x onto this graph space (see Space.clip).
- Graphs with more than
max_nodesnodes keep their firstmax_nodesnodes; edges touching a dropped node are dropped too. Kept nodes keep their index, so the remaining edges stay valid. - Graphs with more than
max_edges(remaining) edges keep their firstmax_edgesedges, in order. The edge arrays are compacted per graph (dropped edges move into the padding). - Node / edge features are clamped with the feature spaces (padding
included, as
containschecks it); edge features are dropped if this space has no edge feature space.
Every graph of a batch (batch_shape) is handled separately; the
padded node / edge axes shrink to the largest remaining count. Returns
a new GraphInstance; unchanged arrays are shared with x.
Raises ValueError if x is not a GraphInstance of this
batch shape, lacks node features or (for spaces with edges) edges, or
if some graph has fewer than min_nodes nodes or, after dropping,
fewer than min_edges edges (nodes and edges cannot be invented).
get_repr(abbreviate=False, include_backend=True, include_device=True, include_dtype=True)
is_subspaceeq(other: Any) -> bool
Return whether this graph space is a non-strict subspace of other (⊆).
True iff other is a GraphSpace on the same backend, the node
and edge feature spaces are subspaces recursively, and self's
node/edge count ranges are contained within other's:
other.min_nodes <= self.min_nodesandself.max_nodes <= other.max_nodes(treatingNoneas +∞);- likewise for edges, with the additional requirement that
self.is_edgeandother.is_edgeagree (an edge-less space cannot be a subspace of an edge-bearing one and vice versa); the edge counts are ignored for edge-less spaces.
device is ignored.
dataclass
¶
GraphInstance(n_nodes: BArrayType, n_edges: Optional[BArrayType] = None, nodes_features: Optional[BArrayType] = None, edges_features: Optional[BArrayType] = None, edges: Optional[BArrayType] = None)
Bases: Generic[BArrayType]
class-attribute
instance-attribute
¶
n_edges: Optional[BArrayType] = None
Number of edges in the graph, shape (*batch_shape) or None if no edges are present.
class-attribute
instance-attribute
¶
nodes_features: Optional[BArrayType] = None
Node features, shape (*batch_shape, max(n_nodes), *node_feature_space.shape) if node_feature_space is not None, otherwise None.
class-attribute
instance-attribute
¶
edges_features: Optional[BArrayType] = None
Edge features, shape (*batch_shape, max(n_edges), *edge_feature_space.shape) if edge_feature_space is not None, otherwise None.
class-attribute
instance-attribute
¶
edges: Optional[BArrayType] = None
Edges in the graph, shape (*batch_shape, max(n_edges), 2) where each edge is represented by a pair of node indices, or None if no edges are present.
TextSpace(backend: ComputeBackend[Any, BDeviceType, BDtypeType, BRNGType], max_length: int, *, min_length: int = 0, charset: Optional[FrozenSet[str] | str] = None, device: Optional[BDeviceType] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[str, BDeviceType, BDtypeType, BRNGType]
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
clip(x: str) -> str
Project the string x onto this text space (see Space.clip).
Characters outside this space's charset (if it has one) are removed and
the result is truncated to max_length. Raises ValueError if
x is not a string or the result is shorter than min_length
(characters cannot be invented).
get_repr(abbreviate=False, include_backend=True, include_device=True, include_dtype=True)
is_subspaceeq(other: Any) -> bool
Return whether this text space is a non-strict subspace of other (⊆).
True iff other is a TextSpace on the same backend, self's
charset is a subset of other's charset (a None charset on
self is a subset of any charset; a None charset on other
only contains a None charset on self), self.min_length >=
other.min_length and self.max_length <= other.max_length.
device is ignored.
TupleSpace(backend: ComputeBackend[Any, BDeviceType, BDtypeType, BRNGType], spaces: Iterable[Space[Any, BDeviceType, BDtypeType, BRNGType]], device: Optional[BDeviceType] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[Tuple[Any, ...], BDeviceType, BDtypeType, BRNGType]
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
instance-attribute
¶
spaces: Tuple[Space[Any, BDeviceType, BDtypeType, BRNGType], ...] = tuple(new_spaces)
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
clip(x: Tuple[Any, ...]) -> Tuple[Any, ...]
Clip every element of x with the subspace at the same index (see Space.clip).
Raises ValueError if x is not a tuple with one element per subspace.
get_repr(abbreviate=False, include_backend=True, include_device=True, include_dtype=True)
is_subspaceeq(other: Any) -> bool
Return whether this tuple space is a non-strict subspace of other (⊆).
True iff other is a TupleSpace on the same backend, the two
have the same arity, and for every index i the child
self.spaces[i].is_subspaceeq(other.spaces[i]) holds recursively.
device is ignored.
UnionSpace(backend: ComputeBackend[Any, BDeviceType, BDtypeType, BRNGType], spaces: Iterable[Space[Any, BDeviceType, BDtypeType, BRNGType]], device: Optional[BDeviceType] = None, metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[Tuple[int, Any], BDeviceType, BDtypeType, BRNGType]
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
instance-attribute
¶
spaces = tuple(spaces if device is None else [space.to(device=device) for space in spaces])
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
contains(x: Tuple[int, Any]) -> bool
Return boolean specifying if x is a valid member of this space.
clip(x: Tuple[int, Any]) -> Tuple[int, Any]
Clip a union value (index, value) with the alternative at index (see Space.clip).
A superset union has the same number of alternatives, each a superset
of the alternative at the same index here, so the index is kept and
only the value is clipped. Raises ValueError if x is not an
(index, value) pair with a valid index.
get_repr(abbreviate: bool = False, include_backend=True, include_device=True, include_dtype=True)
is_subspaceeq(other: Any) -> bool
Return whether this union space is a non-strict subspace of other (⊆).
Kept simple: True iff other is a UnionSpace on the same backend
with the same number of alternatives and, for every index i,
self.spaces[i].is_subspaceeq(other.spaces[i]) holds (pairwise,
same-index recursion). This is a conservative structural check — it
does not attempt to reason about alternative reordering or set-union
semantics. device is ignored.
BatchedSpace(single_space: Space[SpaceDataT, BDeviceType, BDtypeType, BRNGType], batch_shape: Sequence[int], metadata: Optional[Mapping[str, Any]] = None)
Bases: Space[ndarray, BDeviceType, BDtypeType, BRNGType]
This space represents a batch of
property
¶
metadata: SpaceMetadata
Read-only description of the data (see "Metadata" in the class docstring); empty if none was given.
property
¶
shape: tuple[int, ...] | None
Return the shape of the space as an immutable property.
with_metadata(metadata: Optional[Mapping[str, Any]] = None, /, *, replace: bool = False, **updates: Any) -> Space[SpaceDataT, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]
Return a copy of this space with updated metadata (this space is left unchanged).
metadata and updates are merged into the current metadata, or
replace it entirely with replace=True (space.with_metadata(replace=True)
clears it). Nested spaces are shared with this space, not copied.
is_subspace(other: Space) -> bool
Return whether this space is a STRICT subspace of other (self ⊂ other).
Defined uniformly for all spaces as::
self.is_subspace(other) ⟺ self.is_subspaceeq(other) and not other.is_subspaceeq(self)
I.e. self ⊆ other holds but other ⊆ self does not, so self
is a PROPER (strict) subspace of other. This is the ⊂ relation
versus the non-strict ⊆ provided by :meth:is_subspaceeq.
This definition is used instead of relying on __eq__ because some
space classes only have identity __eq__; defining strict
containment via the symmetric non-strict check works uniformly for all
classes regardless of their __eq__ implementation.
For structurally-distinct-but-mutually-containing spaces (which should
not occur under the strict dtype/shape policies enforced by the
per-class is_subspaceeq implementations) this degrades gracefully
to False: if both self.is_subspaceeq(other) and
other.is_subspaceeq(self) hold, the two spaces are considered
equivalent and neither is a STRICT subspace of the other.
If either side's is_subspaceeq is not implemented (the base
:meth:is_subspaceeq raises NotImplementedError), the exception
propagates to the caller — it is NOT swallowed into False so that
callers can tell that the comparison is unsupported.
Note: controller-required-space checks should typically use
:meth:is_subspaceeq (a controller's required space may exactly equal
the env space, in which case the strict is_subspace would return
False).
staticmethod
¶
abbr_device(spaces: Iterable[Space[Any, _SpaceBDeviceT, _SpaceBDTypeT, _SpaceBDRNGT]]) -> Optional[_SpaceBDeviceT]
Return the shared device across spaces, or None if mixed/empty.
clip(x: ndarray) -> np.ndarray
Clip every element of the object array x with single_space (see Space.clip).
Returns a new object array of batch_shape. Raises ValueError if
x is not an object array of batch_shape, or if any element
cannot be clipped.
get_repr(abbreviate=False, include_backend=True, include_device=True, include_dtype=True)
is_subspaceeq(other: Any) -> bool
Return whether this batched space is a subspace of other.
True iff other is a BatchedSpace on the same backend with equal
batch_shape and self.single_space.is_subspaceeq(other.single_space)
holds recursively (non-strict ⊆). device is ignored.
reshape_batch_size(space: Space, old_batch_shape: Tuple[int], new_batch_shape: Tuple[int]) -> Space
reshape_batch_size_in_data(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], data: Any, old_batch_shape: Tuple[int], new_batch_shape: Tuple[int]) -> Any
swap_batch_dims_in_data(backend: ComputeBackend[BArrayType, BDeviceType, BDtypeType, BRNGType], data: Any, dim1: int, dim2: int) -> Any
batch_differing_spaces(spaces: Sequence[Space], device: Optional[Any] = None) -> Space