socket_transport

socket_transport

Concrete transport implementations for inference client communication.

SocketObservationTransport

SocketObservationTransport(server_address='127.0.0.1', server_port=5555, request_timeout_seconds=None)

Observation transport using tso_robotics_sockets.SocketClient.

Handles registration, observation requests, and image decompression over a ZMQ socket connection.

Initialize socket observation transport.

Parameters:

Name	Type	Description	Default
server_address	str	Address of the environment server.	'127.0.0.1'
server_port	int	Port of the environment server.	5555
request_timeout_seconds	float | None	Per-request timeout after which receive raises TimeoutError instead of blocking forever on a dead server. None keeps the fully blocking behavior.	None

Source code in src/versatil/inference/socket_transport.py

def __init__(
    self,
    server_address: str = "127.0.0.1",
    server_port: int = 5555,
    request_timeout_seconds: float | None = None,
):
    """Initialize socket observation transport.

    Args:
        server_address: Address of the environment server.
        server_port: Port of the environment server.
        request_timeout_seconds: Per-request timeout after which receive
            raises ``TimeoutError`` instead of blocking forever on a dead
            server. ``None`` keeps the fully blocking behavior.
    """
    self.socket = SocketClient(
        server_address=server_address,
        server_port=server_port,
        request_timeout_seconds=request_timeout_seconds,
    )

receive

receive(requested_keys, compression_type)

Request observations from the server.

Parameters:

Name	Type	Description	Default
requested_keys	list[str]	Observation keys to request.	required
compression_type	str	Compression format for image data.	required

Returns:

Type	Description
dict	Server response dict with observation data and status.

Source code in src/versatil/inference/socket_transport.py

def receive(self, requested_keys: list[str], compression_type: str) -> dict:
    """Request observations from the server.

    Args:
        requested_keys: Observation keys to request.
        compression_type: Compression format for image data.

    Returns:
        Server response dict with observation data and status.
    """
    return self.socket.send_request(
        route_name=ServerRoute.GET_OBSERVATION.value,
        dict_data={
            InferenceRequestKey.REQUESTED_KEYS.value: requested_keys,
            InferenceRequestKey.COMPRESSION_TYPE.value: compression_type,
        },
    )

register

register(client_name)

Register the client with the server.

Parameters:

Name	Type	Description	Default
client_name	str	Identifier for this client.	required

Returns:

Type	Description
dict	Server acknowledgement response.

Source code in src/versatil/inference/socket_transport.py

def register(self, client_name: str) -> dict:
    """Register the client with the server.

    Args:
        client_name: Identifier for this client.

    Returns:
        Server acknowledgement response.
    """
    return self.socket.send_request(
        route_name=ServerRoute.REGISTER_CLIENT.value,
        dict_data={
            InferenceRequestKey.CLIENT_NAME.value: client_name,
        },
    )

close

close()

Close the socket connection without blocking.

Source code in src/versatil/inference/socket_transport.py

def close(self) -> None:
    """Close the socket connection without blocking."""
    self.socket.request_socket.setsockopt(zmq.LINGER, 0)
    self.socket.close()

SocketActionTransport

SocketActionTransport(server_address='127.0.0.1', server_port=5555, request_timeout_seconds=None)

Action transport using tso_robotics_sockets.SocketClient.

Sends raw action predictions plus metadata to the server. The server handles any coordinate conversion (e.g. delta computation).

Initialize socket action transport.

Parameters:

Name	Type	Description	Default
server_address	str	Address of the environment server.	'127.0.0.1'
server_port	int	Port of the environment server.	5555
request_timeout_seconds	float | None	Per-request timeout after which send raises TimeoutError instead of blocking forever on a dead server. None keeps the fully blocking behavior.	None

Source code in src/versatil/inference/socket_transport.py

def __init__(
    self,
    server_address: str = "127.0.0.1",
    server_port: int = 5555,
    request_timeout_seconds: float | None = None,
):
    """Initialize socket action transport.

    Args:
        server_address: Address of the environment server.
        server_port: Port of the environment server.
        request_timeout_seconds: Per-request timeout after which send
            raises ``TimeoutError`` instead of blocking forever on a dead
            server. ``None`` keeps the fully blocking behavior.
    """
    self.socket = SocketClient(
        server_address=server_address,
        server_port=server_port,
        request_timeout_seconds=request_timeout_seconds,
    )

send

send(actions, action_metadata)

Send actions and metadata to the server.

Parameters:

Name	Type	Description	Default
actions	dict	Dict mapping environment index to a structured action dict keyed by ActionComponent value, as produced by ActionPostprocessor.format_action.	required
action_metadata	dict	Dict describing the action space, keyed by action component with ActionMetadataField entries.	required

Returns:

Type	Description
dict	Server acknowledgement response.

Source code in src/versatil/inference/socket_transport.py

def send(self, actions: dict, action_metadata: dict) -> dict:
    """Send actions and metadata to the server.

    Args:
        actions: Dict mapping environment index to a structured action
            dict keyed by ``ActionComponent`` value, as produced by
            ``ActionPostprocessor.format_action``.
        action_metadata: Dict describing the action space, keyed by action
            component with ``ActionMetadataField`` entries.

    Returns:
        Server acknowledgement response.
    """
    return self.socket.send_request(
        route_name=ServerRoute.SEND_ACTION.value,
        dict_data={
            InferenceRequestKey.ACTIONS.value: actions,
            InferenceRequestKey.ACTION_METADATA.value: action_metadata,
        },
    )

close

close()

Close the socket connection without blocking.

Source code in src/versatil/inference/socket_transport.py

def close(self) -> None:
    """Close the socket connection without blocking."""
    self.socket.request_socket.setsockopt(zmq.LINGER, 0)
    self.socket.close()