unienv_interface.remote¶
Lightweight RPC for UniEnv environments, worlds, nodes and data transformations.
Host objects with :class:RemoteServer and use them through
:class:RemoteClient proxies that implement the normal Env / World /
WorldNode / DataTransformation interfaces. Transports: TCP, Unix sockets, WebSocket (optional
websockets dependency) and in-memory. See docs/guides/remote-execution.md.
RemoteClient(transport: Transport, *, backend: ComputeBackend = NumpyComputeBackend, device: Optional[Any] = None, timeout: Optional[float] = None, compress: Optional[Dict[str, str]] = None)
A connection to a :class:RemoteServer. Thread-safe: calls may be concurrent.
Results are converted to backend arrays on device. A remote exception
is re-raised as :class:RemoteError; a lost connection raises
ConnectionError, in which case a call that was in flight may or may not
have run on the server. timeout (seconds, None = wait forever) bounds
each call; a timed-out call keeps running on the server.
instance-attribute
¶
resources: Dict[str, Dict[str, Any]] = hello['resources']
Descriptor of every hosted resource, by name.
classmethod
¶
connect(address: str, *, timeout: Optional[float] = None, connect_timeout: float = 10.0, backend: ComputeBackend = NumpyComputeBackend, device: Optional[Any] = None, compress: Optional[Dict[str, str]] = None, max_frame_size: int = wire.DEFAULT_MAX_FRAME_SIZE) -> RemoteClient
Connect to tcp://host:port, unix:///path or ws://host:port.
compress opts into lossy compression of server-to-client data, e.g.
{"image": "jpeg:90"} JPEG-encodes uint8 image leaves (spaces whose
metadata["kind"] == "image") at quality 90.
call(name: str, method: str, *args: Any, **kwargs: Any) -> Any
Call an allowed method of the hosted resource name.
resource(name: str) -> Union[RemoteEnv, RemoteWorld, RemoteWorldNode, RemoteDataTransformation]
The proxy of hosted resource name (one shared proxy per name).
world(name: str) -> RemoteWorld
Proxy of a hosted World (e.g. "sim/world" for a WorldEnv registered recursively as "sim").
node(name: str) -> RemoteWorldNode
Proxy of a hosted WorldNode (e.g. "sim/node", "sim/node/camera").
transformation(name: str) -> RemoteDataTransformation
Proxy of a hosted DataTransformation.
subscribe(name: str) -> Subscription
Follow the reset / reload / step results of hosted env name.
Results of calls by any client are delivered, latest wins: a slow subscriber skips results rather than slowing the server down.
Subscription(client: RemoteClient, name: str)
Iterator over the latest results of a hosted env (latest wins, never queued).
Use it as a context manager or call :meth:close; next(timeout=...)
raises TimeoutError if no new result arrives in time.
Bases: NamedTuple
The result of a reset / reload / step of a subscribed env.
RemoteError(type: str, message: str, remote_traceback: Optional[str] = None)
Bases: RuntimeError
An exception raised by the hosted object (or the server) while handling a call.
type is the remote exception's class name and remote_traceback its
formatted traceback; exception objects themselves never cross the wire.
RemoteDataTransformation(client: RemoteClient, name: str)
Bases: _Remote, DataTransformation
A DataTransformation hosted by a :class:RemoteServer.
transform and get_target_space_from_source run on the server: the
source space and the data are rebuilt there on the backend / device the
transformation was registered with, and the result comes back on the
client's. Use it to run a transformation whose dependencies live in another
environment (e.g. a grasp synthesizer pinned to an old CUDA toolchain).
The proxy has no inverse and cannot be serialized: it is bound to its connection (serialize the hosted transformation on the server instead).
direction_inverse(source_space: Optional[Space[SourceDataT, SourceBDeviceT, SourceBDTypeT, SourceBDRNGT]] = None) -> Optional[DataTransformation]
Return the inverse transformation when one exists.
classmethod
¶
deserialize_from(json_data: Dict[str, Any], source_space: Any = None) -> RemoteDataTransformation
RemoteEnv(client: RemoteClient, name: str)
Bases: _Remote, Env
An Env hosted by a :class:RemoteServer.
instance-attribute
¶
observation_space: Space[ObsType, BDeviceType, BDtypeType, BRNGType]
class-attribute
instance-attribute
¶
context_space: Optional[Space[ContextType, BDeviceType, BDtypeType, BRNGType]] = None
sample_context() -> Optional[ContextType]
Sample one context value if context_space is defined.
update_observation_post_reset(old_obs: ObsType, newobs_masked: ObsType, mask: BArrayType) -> ObsType
Merge masked reset observations back into a full batched observation.
update_context_post_reset(old_context: ContextType, new_context: ContextType, mask: BArrayType) -> ContextType
Merge masked reset contexts back into a full batched context.
has_wrapper_attr(name: str) -> bool
Checks if the attribute name exists in the environment.
set_wrapper_attr(name: str, value: Any)
Sets the attribute name on the environment with value.
get_node(nested_keys: Union[str, Sequence[str]]) -> Optional[RemoteWorldNode]
Node proxy by key path, for envs built on a WorldEnv.
RemoteWorld(client: RemoteClient, name: str)
Bases: _Remote, World
A World hosted by a :class:RemoteServer.
instance-attribute
¶
device: Optional[BDeviceType]
The world timestep in seconds, if None, the world is asynchronous (real-time)
class-attribute
instance-attribute
¶
world_subtimestep: Optional[float] = None
The number of parallel environments in this world
is_control_timestep_compatible(control_timestep: Optional[float]) -> bool
Check whether a node control period aligns with world_timestep.
RemoteWorldNode(client: RemoteClient, name: str)
Bases: _Remote, WorldNode
A WorldNode hosted by a :class:RemoteServer.
class-attribute
instance-attribute
¶
context_space: Optional[Space[ContextType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
observation_space: Optional[Space[ObsType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
action_space: Optional[Space[ActType, BDeviceType, BDtypeType, BRNGType]] = None
class-attribute
instance-attribute
¶
supported_render_modes: Sequence[str] = ()
class-attribute
instance-attribute
¶
after_reset_priorities: Set[int] = set()
class-attribute
instance-attribute
¶
after_reload_priorities: Set[int] = set()
class-attribute
instance-attribute
¶
pre_environment_step_priorities: Set[int] = set()
class-attribute
instance-attribute
¶
post_environment_step_priorities: Set[int] = set()
property
¶
effective_update_timestep: Optional[float]
Resolved update period, defaulting to control_timestep when unset.
get_nodes_by_type(node_type: Type[WorldNode]) -> list[WorldNode]
Return nodes in this subtree that are instances of node_type.
has_wrapper_attr(name: str) -> bool
Checks if the attribute name exists in the environment.
set_wrapper_attr(name: str, value: Any)
Sets the attribute name on the environment with value.
reset(*, priority: int = 0, seed: Optional[int] = None, mask: Any = None, **kwargs: Any) -> None
reload(*, priority: int = 0, seed: Optional[int] = None, mask: Any = None, **kwargs: Any) -> None
RemoteServer(*, max_frame_size: int = wire.DEFAULT_MAX_FRAME_SIZE)
Serve registered objects to :class:RemoteClient connections.
Example::
with RemoteServer() as server:
server.register("sim", env) # or register_recursive() to also host its world and nodes
server.serve("tcp://0.0.0.0:7000") # blocks; or listen() in the background
Each registered root runs on its own worker thread, shared by every hosted object
with the same world; calls on them are executed there one at a time, in arrival
order, whichever client sent them. Only the methods in protocol.METHODS can be
called.
register(name: str, obj: Hosted, *, close: bool = True, backend: Optional[ComputeBackend] = None, device: Optional[Any] = None) -> str
Host obj (an Env, World, WorldNode or DataTransformation) under name.
Only obj itself is exposed; use :meth:register_recursive to also expose
the world and nodes it is built from. Objects sharing a world (e.g. an env and
its world registered separately) run on the same worker. Registering an object
that is already hosted returns its existing name. With close=True the server
closes obj (on its worker) when it shuts down.
A DataTransformation has no backend of its own: backend / device
(default: numpy, no device) are where the source spaces and data of its
calls are rebuilt on the server. Other objects use their own.
register_recursive(name: str, obj: Hosted, *, close: bool = True) -> str
Like :meth:register, but also expose everything obj is built from.
An env built on a WorldEnv also exposes name/world and name/node
(the root node), and every node exposes its world (<node name>/world, unless
already hosted) and its children as <node name>/<child>, so clients can
compose them in components mode. Objects already hosted keep their first name.
listen(address: str = 'tcp://127.0.0.1:0') -> str
Accept connections on address in the background; return the bound address.
address is tcp://host:port, unix:///path or ws://host:port
(the latter requires websockets, from the network extra).
serve(address: str = 'tcp://127.0.0.1:0') -> None
Listen on address and block until :meth:close (or Ctrl-C).
connect(**client_options: Any) -> RemoteClient
An in-process client over a :class:MemoryTransport (mainly for tests).
close() -> None
Stop listening, drop every connection, close owned objects on their workers.
MemoryTransport(incoming: _Queue, outgoing: _Queue, max_frame_size: int = DEFAULT_MAX_FRAME_SIZE)
StreamTransport(sock: socket, max_frame_size: int = DEFAULT_MAX_FRAME_SIZE)
A reliable, ordered connection carrying whole frames.
send(parts: Sequence[Any], size: int) -> None
Send the frame parts (size bytes in total). Thread-safe.
WebSocketTransport(connection: Any, max_frame_size: int = DEFAULT_MAX_FRAME_SIZE)
Bases: ValueError
A frame is malformed or violates the wire format's limits.