from grid_sim_client.isaac import IsaacClient

Generic Zenoh pub/sub client for simulation scene interactions. Instead of dedicated methods like setObjectPose, this client exposes generic pub / sub / get operations over a predefined topic registry. Topics are looked up by key (e.g. "set_object_pose"), and the client handles CDR serialization using the topic’s declared ROS2 message type. Usage:: client = SimClient()

Publish a Pose to the “set_object_pose” topic

client.pub(“set_object_pose”, { “position”: {“x”: 1.0, “y”: 2.0, “z”: 0.5}, “orientation”: {“x”: 0.0, “y”: 0.0, “z”: 0.0, “w”: 1.0}, }) client.sub(“get_object_pose”) pose = client.get(“get_object_pose”)

Constructor

Signature

IsaacClient(
    topics: Optional[Dict[str, dict]] = None,
    *,
    connect: Optional[List[str]] = None,
    listen: Optional[List[str]] = None,
    mode: Optional[str] = None,
    config_file: Optional[str] = None,
    verbose: bool = False,
)

Methods

`add_topic()`

Signature

add_topic(key: str, zenoh_topic: str, msg_type: str, mode: str = 'both') -> None

required

Lookup key (e.g. "my_custom_topic").

required

Zenoh key expression (e.g. "/my/topic").

required

ROS2 message class name from ros2_types (e.g. "Pose").

default:"'both'"

"pub", "sub", or "both".

`close()`

Signature

close() -> None

Tear down all subscriptions and close the Zenoh session.

`get()`

Signature

get(topic_key: str, timeout: float = 1.0) -> Optional[Any]

Return the latest message on a subscribed topic. Call sub first to start receiving messages.

required

Registered topic key.

default:"1.0"

Seconds to wait for a message.

Returns: Deserialized message, or None if nothing received within timeout.

`pub()`

Signature

pub(topic_key: str, data: Any) -> None

Publish data to a predefined topic. data can be:

A dict matching the ROS2 message fields (auto-converted & CDR-serialized).
A primitive (bool, int, float) for single-field messages like Bool.
A pre-built pycdr2.IdlStruct instance (serialized directly).
A raw bytes object (published as-is).

required

Registered topic key.

required

Payload to publish.

`set_named_pose()`

Signature

set_named_pose(
    name: str,
    position: Union[Sequence[float], Tuple[float, float, float]],
    orientation: Union[Sequence[float], Tuple[float, float, float, float]] = (1, 0, 0, 0),
) -> None

Teleport a named rigid body to a given pose.

required

Scene object name (e.g. "box_pass_005").

required

Target (x, y, z) world position.

default:"(1, 0, 0, 0)"

Target (w, x, y, z) quaternion. Defaults to identity.

`sub()`

Signature

sub(topic_key: str, callback: Optional[Callable] = None) -> None

Subscribe to a predefined topic. If callback is provided it is invoked with the deserialized message each time a new sample arrives. Regardless of callbacks, the latest message is always available via get.

required

Registered topic key.

Optional fn(msg) invoked on each message.

​Publish a Pose to the “set_object_pose” topic

​Subscribe and poll

​Constructor

​Methods

​add_topic()

​close()

​get()

​pub()

​set_named_pose()

​sub()

Publish a Pose to the “set_object_pose” topic

Subscribe and poll

Constructor

Methods

`add_topic()`

`close()`

`get()`

`pub()`

`set_named_pose()`

`sub()`