Champion IO

Prompt to Physics: Simulating a Queue System in Isaac with USD and n8n

May 26, 2025

🤖 To simulate intelligent behaviors in robotics or digital twins, it helps to first understand the structure of a queueing system.

Most systems can be broken down into three core components:

Arrival Process
This describes how units (like drones, robots, or jobs) enter the system. A common assumption is exponential interarrival times, but other distributions can be used.
Service Mechanism
This includes the number of servers and how long it takes each to serve a unit. In physical AI systems, this could mean a robotic arm completing a task or a gate processing vehicles.
Queue Discipline
This determines the order of service.
- FIFO (First-In, First-Out) is most common
- LIFO (Last-In, First-Out) is less typical but used in stacks
- Priority-based systems serve units according to importance or urgency

Now imagine this:
What if your AI copilot could understand a prompt like:

"Place 5 drones in a queue, serve 1 every 10 seconds.

To make this possible, you need a setup where:

n8n captures the user's natural language prompt
Scene metadata (like prim paths, types, and schemas) is sent to the AI
USD is used to define and structure the 3D scene
Isaac Sim runs the physics-based simulation

With this configuration, an AI copilot can take a plain-language instruction, understand what objects it needs to act on, and generate the appropriate Python code to build and run the simulation—no manual scripting required.

Let’s apply these concepts using simulation. Suppose you want to model a GI/G/1 queue—a single-server system with general (non-exponential) arrival and service distributions.

n8n receives the prompt
USD structures the scene
Isaac Sim executes the physics in real-time

Together, they simulate a GI/G/1 queue—a single-server system with flexible interarrival and service time distributions—entirely agent-driven.

Simulating a Queue System in Isaac with USD

Step 1: Prompt Input from n8n

n8n Node Prompt:

{
  "prompt": "Simulate a queue of 5 drones arriving randomly, served one by one"
}

This triggers a webhook that converts the prompt to parameters:

n_customers = 5
interarrival_dist = exponential
service_time = 10 sec

Step 2: Generate USD Scene via Python

In this step, we’re using Python and the USD API to programmatically define a scene inside Isaac Sim. The arrival times are modeled using an exponential distribution to simulate realistic, stochastic input. This script creates the necessary structure in the USD stage so that Isaac Sim can later apply physics and visualize the drone behavior.

from pxr import Usd, UsdGeom, Sdf
import random

# Create a new USD stage
stage = Usd.Stage.CreateNew("queue_sim.usda")

# Root
UsdGeom.Xform.Define(stage, Sdf.Path("/World"))

# Spawn 5 drones at random arrival times
for i in range(5):
    drone = UsdGeom.Xform.Define(stage, Sdf.Path(f"/World/Drone_{i}"))
    arrival_time = random.expovariate(1/5)  # λ=1/5
    x = 0
    y = i * -2  # queue position
    z = 0
    drone.AddTranslateOp().Set(value=(x, y, z))

stage.GetRootLayer().Save()

Step 3: Service Logic with Drive API in Isaac

In this step, we define how each drone is "served" using Isaac Sim’s Drive API. The Drive API allows us to apply motion or force to specific parts of a USD prim—such as rotating a joint or moving a rigid body. We loop through each drone in the queue and apply a targetPosition one at a time, introducing a delay (e.g., 10 seconds) between each to simulate sequential service. This mimics a real queue where only one unit is processed at a time by a single server.

from pxr import UsdPhysics
import omni
import asyncio

stage = omni.usd.get_context().get_stage()

async def serve_queue():
    for i in range(5):
        path = f"/World/Drone_{i}"
        prim = stage.GetPrimAtPath(path)
        if prim.IsValid():
            print(f"Serving {path}")
            drive = UsdPhysics.DriveAPI.Apply(prim, "linear")
            drive.CreateTargetPositionAttr().Set((10, 0, 0))  # move forward
            await asyncio.sleep(10)  # simulate service time

asyncio.ensure_future(serve_queue())

This coroutine simulates drone service, one every 10 seconds, in FIFO order.

Step 4: Visual Loop in n8n

Your n8n canvas might look like:

[Prompt Node]
     |
     v
[Webhook] --> [Python Node: USD Gen] --> [Run Isaac Sim Container]
                                     |
                               [Trigger serve_queue()]

Tools Become Agents

This is what we call Agentic USD—USD prims and physics objects aren’t passive files anymore. They respond, simulate, and act—just like an agent.

Available Now

Book a demo and get early access. Free trial!

Thank you! Your submission has been received!

Oops! Something went wrong while submitting the form.

Thank you! Your submission has been received!

Oops! Something went wrong while submitting the form.