Omniverse Realtime Viewer

{"event_type": "<MessageType>", "payload": {}}

stream_server.on_connection = message_handler.on_connection
stream_server.on_message = message_handler.on_message
stream_server.on_input = message_handler.on_input
# Optional for composed text input:
# stream_server.on_unicode = message_handler.on_unicode

def on_input(self, event):
    import ovstream
    if event.type == ovstream.InputEventType.MOUSE:
        mouse = event.mouse
        if mouse.type == ovstream.MouseEventType.MOVE:
            server.camera.on_mouse_move(mouse.x, mouse.y)
        elif mouse.type == ovstream.MouseEventType.BUTTON:
            button = camera_button_from_ovstream(mouse.data, ovstream)
            if button is None:
                return
            is_down = mouse.button_state == ovstream.KeyState.DOWN
            if is_down:
                server.camera.on_mouse_button_down(mouse.x, mouse.y, button)
            else:
                was_click = server.camera.on_mouse_button_up(mouse.x, mouse.y, button)
                if button == 0 and was_click:
                    self._handle_click(mouse.x, mouse.y)
        elif mouse.type == ovstream.MouseEventType.WHEEL:
            server.camera.on_scroll(mouse.scroll_y or mouse.data)

{"event_type":"openStageResult","payload":{"url":"samples/samples_data/stage01.usd","result":"success","root_prim_path":"/World"}}

{"event_type":"getChildrenResult","payload":{"prim_path":"/World","children":[{"name":"Cube","path":"/World/Cube","type":"geom","children":true},{"name":"Light","path":"/World/Light","type":"light","children":false}]}}

{"event_type":"getPropertiesResponse","payload":{"prim_path":"/World/Cube","properties":{"typeName":"Mesh","visibility":"inherited","xformOp:translate":[0,1,0]},"truncated":false}}

{"event_type":"getVariantsResponse","payload":{"prim_path":"/World/Car","variants":{"color":{"options":["red","blue"],"selection":"red"}}}}

{"event_type":"changeAOVRequest","payload":{"aov":"NormalSD"}}

{"event_type":"activeAOVState","payload":{"active":"NormalSD","available":["LdrColor","HdrColor","NormalSD","InstanceSegmentationSD","SemanticSegmentationSD","DepthSD","DiffuseAlbedoSD"],"result":"success","previous":"LdrColor"}}

{"event_type":"availableAOVsResult","payload":{"aovs":["LdrColor","HdrColor","NormalSD","InstanceSegmentationSD","SemanticSegmentationSD","DepthSD","DiffuseAlbedoSD"],"available":["LdrColor","HdrColor","NormalSD","InstanceSegmentationSD","SemanticSegmentationSD","DepthSD","DiffuseAlbedoSD"]}}

{"event_type":"setViewportInputActive","payload":{"active":false}}

def _handle_select_prims(self, payload: Dict[str, Any]) -> None:
    paths = [p for p in payload.get("paths", []) if isinstance(p, str)]
    prev_selected = set(self.server.selected_prims)
    new_selected = set(paths)
    self.server.selected_prims = paths

    if self.server._highlight_mgr:
        self.server._highlight_mgr.update_selection(new_selected, prev_selected)
    if self.server._animator:
        for path in new_selected - prev_selected:
            self.server._animator.select(path)
        for path in prev_selected - new_selected:
            self.server._animator.deselect(path)

    self.send_message("stageSelectionChanged", {"prims": self.server.selected_prims})

sendMessage({ event_type: 'selectPrimsRequest', payload: { paths: selectedPaths } });

def cmd_get_root_prim_path() -> Dict[str, Any]:
    if not _stage:
        return {"ok": False, "error": "no stage loaded"}
    world = _stage.GetPrimAtPath("/World")
    if world.IsValid():
        return {"ok": True, "path": "/World"}

    default_prim = _stage.GetDefaultPrim()
    if default_prim and default_prim.IsValid():
        return {"ok": True, "path": str(default_prim.GetPath())}

    for child in _stage.GetPseudoRoot().GetChildren():
        return {"ok": True, "path": str(child.GetPath())}

    return {"ok": True, "path": "/"}

self.send_message("openStageResult", {
    "url": active_url,
    "result": "success",
    "root_prim_path": root_path,
})
children = server._pxr.get_children(root_path)
self.send_message("getChildrenResult", {"prim_path": root_path, "children": children})

if not force and self.current_stage_url:
    requested_key = os.path.normcase(os.path.abspath(url))
    current_key = os.path.normcase(os.path.abspath(self.current_stage_url))
    if requested_key == current_key:
        logger.info("Stage already loaded, skipping reload: %s", url)
        return True

def _handle_reset_stage(self, payload: Dict[str, Any]) -> None:
    if server.current_stage_url:
        server._load_stage(server.current_stage_url, force=True)

def _send_aov_state(self, extra: Optional[Dict[str, Any]] = None) -> None:
    available = server.get_available_aovs()
    active = getattr(server, "_active_aov", "LdrColor")
    payload = {"active": active, "available": available}
    if extra:
        payload.update(extra)
    self.send_message("activeAOVState", payload)
    self.send_message("availableAOVsResult", {"aovs": available, "available": available})

def _handle_change_aov(self, payload: Dict[str, Any]) -> None:
    requested = payload.get("aov") or payload.get("name")
    if not isinstance(requested, str) or not requested:
        self._send_aov_state({"result": "error", "reason": "Missing AOV name"})
        return

    previous = getattr(server, "_active_aov", "LdrColor")
    if server.set_active_aov(requested):
        self._send_aov_state({"result": "success", "previous": previous})
        return

    self._send_aov_state({
        "result": "error",
        "requested": requested,
        "reason": "AOV is not available for the current render product",
    })

case 'activeAOVState': {
  const payload = event.payload as ActiveAOVStatePayload;
  if (Array.isArray(payload.available) && payload.available.length > 0) {
    setAvailableAOVs(payload.available);
  }
  setActiveAOV(payload.active || 'LdrColor');
  break;
}
case 'availableAOVsResult': {
  const payload = event.payload as AvailableAOVsResultPayload;
  const names = payload.aovs || payload.available || [];
  if (names.length > 0) {
    setAvailableAOVs(names);
  }
  break;
}

MAX_MESSAGE_BYTES = 60000

def capped_event(event_type: str, payload: dict) -> tuple[dict, bool]:
    encoded = json.dumps({"event_type": event_type, "payload": payload}, default=str)
    if len(encoded.encode("utf-8")) <= MAX_MESSAGE_BYTES:
        return payload, False
    if "properties" in payload:
        trimmed = {}
        omitted = 0
        for key, value in payload["properties"].items():
            candidate = {**payload, "properties": {**trimmed, key: value}}
            size = len(json.dumps({"event_type": event_type, "payload": candidate}, default=str).encode("utf-8"))
            if size > MAX_MESSAGE_BYTES:
                omitted += 1
            else:
                trimmed[key] = value
        return {**payload, "properties": trimmed, "truncated": True, "omitted_count": omitted}, True
    return {**payload, "truncated": True, "error": "payload too large"}, True

def _handle_my_feature(self, payload):
    result = self._do_something(payload.get("some_param", ""))
    self._send_message("myFeatureResponse", {"result": result})

sendMessage({ event_type: 'myFeatureRequest', payload: { some_param: 'value' } });