Usage Examples

Get Adapter Registry and Register Stage Adapter

from omni.kit.property.adapter.core import get_adapter_registry, StageAdapter, PropertyType
from pxr import Usd

# Get the global adapter registry instance
registry = get_adapter_registry()

# Create a custom stage adapter by inheriting from StageAdapter
class MyCustomStageAdapter(StageAdapter):
    def __init__(self, stage):
        super().__init__(stage)
        self._name = "my_custom_adapter"
        self._priority_read = 100
        self._priority_write = 100

    @property
    def name(self):
        return self._name

    @property
    def priority_read(self):
        return self._priority_read

    @property
    def priority_write(self):
        return self._priority_write

    def GetPrimAtPath(self, path):
        return self.stage.GetPrimAtPath(path)

    def GetAttributeAtPath(self, path):
        return self.stage.GetAttributeAtPath(path)

    def CreateChangeTracker(self, attr_names, prim_paths, callback):
        # Implementation would create a change tracker
        pass

    def convert_data(self, data, dst_adapter_name):
        # Implementation would convert data between adapter formats
        return data

    def resolve_path_array(self, path, resolve_path, path_list, index):
        # Implementation would resolve path arrays
        pass

    def get_notice_paths(self, stage, notice):
        # Implementation would extract notice paths
        return []

# Register the custom stage adapter
registry.register_stage_adapter("my_custom_adapter", MyCustomStageAdapter)

# Get all registered adapters
adapters_dict = registry.registered_adapters
print(f"Registered adapters: {list(adapters_dict.keys())}")

---

Create Property Adapters

from omni.kit.property.adapter.core import AttributeAdapter, PrimAdapter, PropertyType
from pxr import Usd, Sdf

# Create a custom attribute adapter
class MyAttributeAdapter(AttributeAdapter):
    def __init__(self, attribute):
        super().__init__(attribute)

    def GetPrim(self):
        # Return the prim that owns this attribute
        return self.attribute.GetPrim()

    def GetPropertyType(self):
        # Return the property type for this attribute
        return PropertyType.ATTRIBUTE

# Create a custom prim adapter
class MyPrimAdapter(PrimAdapter):
    def __init__(self, prim):
        super().__init__(prim)

    def custom_method(self):
        return f"Custom method for prim: {self.prim.GetPath()}"

# Use PropertyType enum
print(f"Attribute property type: {PropertyType.ATTRIBUTE}")
print(f"Relationship property type: {PropertyType.RELATIONSHIP}")

# Example usage with USD objects (assuming you have a stage and prim)
stage = Usd.Stage.CreateInMemory()
prim = stage.DefinePrim("/MyPrim", "Cube")
attribute = prim.CreateAttribute("myAttr", Sdf.ValueTypeNames.Float)

# Create adapter instances
attr_adapter = MyAttributeAdapter(attribute)
prim_adapter = MyPrimAdapter(prim)

---

Monitor Registry Events

from omni.kit.property.adapter.core import get_adapter_registry, RegistryEventType, StageAdapter, RegistryGlobalEvent
from carb.eventdispatcher import get_eventdispatcher
from pxr import Usd

# Get the adapter registry
registry = get_adapter_registry()

# Define callback for registry events
def on_registry_event(event):
    print(f"Registry event received: {event.event_name}")
    if hasattr(event, 'payload') and event.payload:
        print(f"Event payload: {event.payload}")

# Subscribe to registry events using event dispatcher
event_dispatcher = get_eventdispatcher()

# Subscribe to adapter added events
adapter_added_sub = event_dispatcher.observe_event(
    observer_name="my_registry_observer",
    event_name=RegistryGlobalEvent.ADAPTER_ADDED,
    on_event=on_registry_event
)

# Subscribe to adapter removed events
adapter_removed_sub = event_dispatcher.observe_event(
    observer_name="my_registry_observer",
    event_name=RegistryGlobalEvent.ADAPTER_REMOVED,
    on_event=on_registry_event
)

# Create a simple test adapter
class TestAdapter(StageAdapter):
    def __init__(self, stage):
        super().__init__(stage)

    @property
    def name(self):
        return "test_adapter"

    @property
    def priority_read(self):
        return 50

    @property
    def priority_write(self):
        return 50

    def GetPrimAtPath(self, path):
        return self.stage.GetPrimAtPath(path)

    def GetAttributeAtPath(self, path):
        return self.stage.GetAttributeAtPath(path)

    def CreateChangeTracker(self, attr_names, prim_paths, callback):
        pass

    def convert_data(self, data, dst_adapter_name):
        return data

    def resolve_path_array(self, path, resolve_path, path_list, index):
        pass

    def get_notice_paths(self, stage, notice):
        return []

# Register adapter (this will trigger ADAPTER_ADDED event)
registry.register_stage_adapter("test_adapter", TestAdapter)

# Unregister adapter (this will trigger ADAPTER_REMOVED event)
registry.unregister_stage_adapter("test_adapter")

---

Instantiate Stage Adapters

from omni.kit.property.adapter.core import get_adapter_registry, StageAdapter
from pxr import Usd, Sdf

# Get the adapter registry
registry = get_adapter_registry()

# Create an in-memory USD stage
stage = Usd.Stage.CreateInMemory()

# Create some basic prims and attributes
prim = stage.DefinePrim("/World", "Xform")
cube_prim = stage.DefinePrim("/World/Cube", "Cube")
cube_prim.CreateAttribute("size", Sdf.ValueTypeNames.Float).Set(2.0)

# Instantiate all registered stage adapters for this stage
adapter_instances = registry.instantiate_all_stage_adapters(stage)

print(f"Created {len(adapter_instances)} adapter instances")
for adapter_name, adapter_instance in adapter_instances.items():
    print(f"Adapter: {adapter_name}, Type: {type(adapter_instance)}")
    print(f"  Name: {adapter_instance.name}")
    print(f"  Read Priority: {adapter_instance.priority_read}")
    print(f"  Write Priority: {adapter_instance.priority_write}")

    # Test basic adapter functionality
    test_prim = adapter_instance.GetPrimAtPath(Sdf.Path("/World/Cube"))
    if test_prim:
        print(f"  Successfully retrieved prim: {test_prim.GetPath()}")