Publish Position of object in the Sim Externally by Socket

Omniverse Isaac Sim
1

I have created a vehicle that updates its position in the Isaac Sim environment based on position data published externally via a socket. What is the simplest method to do it?

3

Hi @lnxguyz ,

Welcome to the forums!

Simplest approach: a non-blocking UDP socket drained once per physics step, applying only the newest packet. No threads, no asyncio. Paste into Window → Script Editor and press Play. (Tested on Isaac Sim 6.0.1.)

First, set physics:kinematicEnabled = true on the vehicle’s rigid body root — otherwise PhysX fights your writes and you get jitter. If it’s a plain Xform, use XformPrim instead of RigidPrim below.

import json
import socket
import numpy as np

from isaacsim.core.experimental.prims import RigidPrim
from isaacsim.core.simulation_manager import SimulationEvent, SimulationManager


class UdpPoseReceiver:
    def __init__(self, prim_path="/World/Vehicle", port=5005):
        self.vehicle = RigidPrim(prim_path)

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("0.0.0.0", port))
        self.sock.setblocking(False)

        self.callback_id = SimulationManager.register_callback(
            self._on_physics_step, event=SimulationEvent.PHYSICS_PRE_STEP
        )

    def _on_physics_step(self, dt, context):
        # Drain the socket, keeping only the most recent packet.
        newest = None
        while True:
            try:
                newest, _ = self.sock.recvfrom(2048)
            except BlockingIOError:
                break
        if newest is None:
            return

        try:
            msg = json.loads(newest.decode("utf-8"))
            position = np.array([[msg["x"], msg["y"], msg["z"]]], dtype=np.float32)
            # Isaac Sim quaternion order is w, x, y, z
            orientation = np.array(
                [[msg["qw"], msg["qx"], msg["qy"], msg["qz"]]], dtype=np.float32
            )
        except (ValueError, KeyError, TypeError):
            print("Invalid pose packet:", newest)
            return

        self.vehicle.set_world_poses(positions=position, orientations=orientation)

    def close(self):
        SimulationManager.deregister_callback(self.callback_id)
        self.sock.close()


try:            # safe to re-run
    receiver.close()
except NameError:
    pass

receiver = UdpPoseReceiver(prim_path="/World/Vehicle", port=5005)

Sender side:

import json, socket

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
pose = {"x": 1.0, "y": 2.0, "z": 0.25, "qx": 0.0, "qy": 0.0, "qz": 0.0, "qw": 1.0}
sock.sendto(json.dumps(pose).encode("utf-8"), ("127.0.0.1", 5005))

Three things that bite people:

  • Never block on the sim thread. The setblocking(False) + drain loop is what keeps Kit responsive. Draining also means you get the newest pose rather than working through a backlog.
  • Quaternion order. USD is (w, x, y, z), most other libraries give you (x, y, z, w). The mismatch produces a rotation that looks almost right.
  • set_world_poses() teleports, discarding suspension, friction, and collision response. If you want the vehicle to behave physically, send a target velocity into the wheel/articulation controller instead.