Usage Examples#
Initialize and Use TaggingClientContextAsync#
import asyncio
import omni.kit.tagging
from omni.kit.async_engine import run_coroutine
async def initialize_tagging_client():
# Create a tagging client context with a server host
client_context = omni.kit.tagging.TaggingClientContextAsync(ov_server="your_server_host")
# Initialize the client to connect to the tagging service
await client_context.init_client()
# Use the client for tagging operations...
print(f"Client initialized with ID: {client_context.client_id}")
# Close the client when done with a custom message
await client_context.close_client(client_msg="Client closed successfully")
# Run the async function using run_coroutine
run_coroutine(initialize_tagging_client())
Create and Set a Custom Tagging Delegate#
import omni.kit.tagging
# Create a custom delegate class to handle tagging events
class MyTaggingDelegate(omni.kit.tagging.OmniKitTaggingDelegate):
def updated_tags_for_url(self, url):
# Called when tags for a URL are updated
print(f"Tags updated for URL: {url}")
def finished_getting_tags(self, urls):
# Called when tags for URLs have been fetched
print(f"Finished getting tags for URLs: {urls}")
# Get the tagging instance
tagging_instance = omni.kit.tagging.get_tagging_instance()
# Set our custom delegate to handle tagging events
tagging_instance.set_delegate(MyTaggingDelegate())
Handle Transport Errors in Tagging Operations#
import omni.kit.tagging
from idl.connection.transport import TransportError
# Try to perform tagging operations that might fail
try:
# Get the tagging instance
tagging_instance = omni.kit.tagging.get_tagging_instance()
# Attempt an operation that might cause a transport error
url = "omniverse://non_existent_server/path"
tagging_instance.get_tags([url], lambda: print("Tags retrieved"))
except TransportError as e:
# Handle the transport error
print(f"Transport error occurred: {e.text}, code: {e.code}")