Python Usage#
Setup and API choice#
Read the developer guide for profiles, coordinates, input devices, and components. Enable XR Core and the runtime/profile appropriate to your headset before using tracked devices.
These examples use XR Core’s raycast interface for XR interactions. For non-XR queries, sequence polling, units, and filtering, use the Raycast Query Python guide.
In Kit’s Script Editor:
import asyncio
import runpy
from pathlib import Path
import omni.kit.app
from omni.kit.xr.core import XRCore
manager = omni.kit.app.get_app().get_extension_manager()
extension = manager.get_extension_path_by_module("omni.kit.xr.core")
folder = Path(extension) / "example"
ex = runpy.run_path(str(folder / "controller_rays.py"))
core = XRCore.get_singleton()
One controller aim ray#
See input devices and coordinate spaces. The virtual-world aim pose already accounts for XR navigation into stage space; make sure you don’t apply the physical-to-world transform again. Local negative Z is the pointing axis. Missing or untracked poses produce no query in this code.
def aim_ray(device):
"""Return a normalized stage-space ray, or None without a tracked aim pose."""
if device is None or not device.has_pose("aim"):
return None
pose = device.get_virtual_world_pose_desc("aim")
required = (
XRPoseValidityFlags.POSITION_VALID
| XRPoseValidityFlags.ORIENTATION_VALID
| XRPoseValidityFlags.POSITION_TRACKED
| XRPoseValidityFlags.ORIENTATION_TRACKED
)
if pose.validity_flags & required != required:
return None
matrix = pose.pose_matrix
origin = matrix.ExtractTranslation()
direction = matrix.TransformDir(Gf.Vec3d(0, 0, -1))
if direction.GetLength() == 0:
return None
return XRRay(origin, direction.GetNormalized(), 0.0, float("inf"))
Run task = asyncio.ensure_future(ex["query_controller_once"]()).
This prints one left-hand hit or miss, without creating a tool or binding input.
Pass "/user/hand/right" to query the other hand. A missing tracked pose is
reported without submitting. Call task.cancel() to stop waiting and ignore
late delivery; there are no subscriptions to release.
async def query_controller_once(hand="/user/hand/left", core=None,
report=report_result, frames=120):
"""One aim ray: no input bindings, update subscriptions, or tool instance."""
core = core or XRCore.get_singleton()
stage = omni.usd.get_context().get_stage()
if stage is None:
raise RuntimeError("Open a stage first")
ray = aim_ray(core.get_input_device(hand))
if ray is None:
print(hand, "tracked aim pose unavailable")
return None
future = asyncio.get_running_loop().create_future()
def completed(completed_ray, result):
if not future.done():
future.set_result((completed_ray, result))
try:
core.submit_raycast_query(ray, completed)
for _ in range(frames):
if omni.usd.get_context().get_stage() != stage:
raise RuntimeError("Stage changed during the query")
if future.done():
completed_ray, result = future.result()
report("single", hand, completed_ray, result)
return completed_ray, result
await omni.kit.app.get_app().next_update_async()
raise TimeoutError("No raycast result; ensure RTX is rendering")
finally:
future.cancel() # Ignore late delivery; this does not cancel GPU work.
Independent controllers#
Start demo = ex["ControllerRays"]().start() for continuous two-hand preview.
demo.preview stores each hand’s most recent valid hit (or None on a miss).
Call demo.stop() to finish. No callback-order assumption is made; source
identity includes both the hand and the request purpose. Each source allows at
most one pending request, skipping pose updates until completion.
def query_once(self, hand):
"""Submit one ray for a selected hand after start(preview=False)."""
return self._submit("single", hand)
def _submit(self, kind, hand):
if not self._active or hand not in self.hands:
return False
if omni.usd.get_context().get_stage() != self._stage:
self.stop()
return False
ray = aim_ray(self.core.get_input_device(hand))
if ray is None:
self._forget_hand(hand)
return False
key = (kind, hand)
if key in self.pending:
return False # Example backpressure policy, not an XR API rule.
token = object()
self.pending[key] = token
stage = self._stage
def completed(completed_ray, result):
if self.pending.get(key) is not token:
return
del self.pending[key]
if (not self._active or omni.usd.get_context().get_stage() != stage
or aim_ray(self.core.get_input_device(hand)) is None):
return
if kind == "preview":
# A miss clears the preview; it never fabricates a point.
self.preview[hand] = (completed_ray, result) if result.valid else None
self.report(kind, hand, completed_ray, result)
try:
self.core.submit_raycast_query(ray, completed)
except Exception:
self.pending.pop(key, None)
raise
return True
Input-driven interaction#
See native input events and
components. Inspect available names by using
core.get_input_device("/user/hand/left").get_input_names() (repeat for right).
Create an inputs dictionary mapping each hand path to an actual input name
reported by that device, then start
demo = ex["ControllerRays"]().start(inputs=inputs).
This example explicitly binds physical inputs using the existing device mechanism.
Bindings can temporarily suspend another tool’s binding on the same input;
so you should use an isolated test profile and stop the demo before resuming other
tools. It does not intercept xr_selection_default.release.
The native payload’s input_device identifies its hand. Releasing the input
submits one more fresh query from that hand’s current pose, independently of
preview geometry; it doesn’t commit cached preview data. The example reports a
commit hit or miss without creating geometry, but you could use that to do
whatever work you need.
def _bind_input(self, hand, device, input_name):
if input_name not in [str(name) for name in device.get_input_names()]:
raise ValueError(
f"{hand} has no input {input_name!r}; inspect device.get_input_names()"
)
event_name = f"raycast_example_{id(self)}_{hand.rsplit('/', 1)[-1]}"
bus = self.core.get_message_bus()
self._subscriptions.append(bus.create_subscription_to_pop_by_type(
carb.events.type_from_string(event_name + ".release"),
lambda event, hand=hand: self.on_release(hand, event), name=event_name))
generator = device.bind_event_generator(input_name, event_name, ("release",))
if generator is None:
raise RuntimeError(f"Could not bind {input_name!r} on {hand}")
# Keep it alive; explicitly release it on stop or device replacement.
self._generators[hand] = generator
self._bound_devices[hand] = device
def on_release(self, hand, event):
if not self._active or hand not in self._inputs:
return False
if str(event.payload.get("input_device", "")) != hand:
return False
# Do not commit the preview cache. Sample this hand's pose for a new
# query, with separate in-flight state from its continuous preview.
return self._submit("commit", hand)
Lifecycle cleanup#
This example shows some best practices when cleaning up after your tools.
demo.stop() releases update/stage/event subscriptions, input generators,
previews, and pending delivery tokens. Stage replacement or XR disable also
stops it. Pose loss clears that hand’s pending deliveries and preview;
replacement of an input-bound device stops the tool so it can explicitly rebind
after restart.
Late callbacks are ignored, but may still be executed on the GPU. We keep the
demo object alive until stopping it, since production components should own it
in their lifecycle.
def start(self, preview=True, inputs=None):
"""inputs maps hand paths to device input names; None is preview-only.
Binding a physical input can suspend another tool's generator. Use
an isolated test app/profile; stop() releases the bindings.
"""
self.stop()
self.core = self.core or XRCore.get_singleton()
self._stage = omni.usd.get_context().get_stage()
if self.core is None or self._stage is None:
raise RuntimeError("Open a stage and enable XR Core first")
self._active = True
self._preview_enabled = preview
self._inputs = dict(inputs or {})
try:
for hand, input_name in self._inputs.items():
device = self.core.get_input_device(hand)
if hand not in self.hands or device is None:
raise ValueError(f"Controller unavailable: {hand}; enter XR first")
self._bind_input(hand, device, input_name)
stage_events = omni.usd.get_context().get_stage_event_stream()
self._subscriptions.append(stage_events.create_subscription_to_pop(
self._on_stage, name="xr.example.stage"))
bus = self.core.get_message_bus()
self._subscriptions.append(bus.create_subscription_to_pop_by_type(
carb.events.type_from_string("xr.disable"), lambda event: self.stop(),
name="xr.example.disable"))
updates = omni.kit.app.get_app().get_update_event_stream()
self._subscriptions.append(updates.create_subscription_to_pop(
self.update, name="xr.example.update"))
except Exception:
self.stop()
raise
return self
def update(self, _event=None):
if not self._active:
return
if omni.usd.get_context().get_stage() != self._stage:
self.stop()
return
for hand in self.hands:
device = self.core.get_input_device(hand)
if hand in self._bound_devices and device != self._bound_devices[hand]:
# Device handles/bindings must not survive disconnect/reconnect.
self.stop() # Explicit restart rebinds the new device.
return
if aim_ray(device) is None:
self._forget_hand(hand)
elif self._preview_enabled:
self._submit("preview", hand)
def _forget_hand(self, hand):
self.preview.pop(hand, None)
for key in list(self.pending):
if key[1] == hand:
del self.pending[key]
def _on_stage(self, event):
closing, opened = omni.usd.StageEventType.CLOSING, omni.usd.StageEventType.OPENED
if event.type in (int(closing), int(opened)):
self.stop()
def stop(self):
self._active = False
self.pending.clear() # Ignore late callbacks; GPU work may still finish.
self.preview.clear()
self._subscriptions.clear()
self._generators.clear()
self._bound_devices.clear()
self._inputs.clear()