Using the Public Streaming API

The following guide contains information pertaining to publishing custom Omniverse applications to the public streaming API in order to support hosting custom-built applications for use by Users subscribed to the Omniverse public streaming API. This guide assumes your organization is already registered to the Omniverse public streaming API, allowing your Users to access collections of Omniverse workflows from their web browsers, streaming directly their Nucleus scenes from their storage locations.

Setting up a Sample Extension

To get started with the code samples used for illustrative purposes in this guide, you may wish to create a custom Omniverse extension which will be used for forwarding communications between your local installation and the remote server hosting the streaming sessions.

Should you require additional information regarding building custom Omniverse extensions, consider referencing the development guide describing getting started with Omniverse extensions.

exts/omni.services.streaming.application.example/config/extension.toml

[package]
authors = ["Name <name@email.com>"]
category = "Services"
changelog = "docs/CHANGELOG.md"
description = "Application streaming integration sample"
icon = "data/icon.png"
keywords = ["kit", "services", "application", "streaming", "example"]
readme  = "docs/README.md"
repository = "https://example.com/kit-extensions/kit-streaming-service-sample"
title = "Application streaming integration sample"
version = "1.0.0"

[dependencies]
"omni.services.core" = {}
"omni.services.client" = {}
"omni.services.transport.server.http" = {}
"omni.services.transport.client.http_async" = {}

[[python.module]]
name = "omni.services.streaming.application.example"

[settings]
exts."omni.services.transport.server.http".http.enabled = true
exts."omni.services.transport.server.http".host = "0.0.0.0"
exts."omni.services.transport.server.http".port = 8011

exts/omni.services.streaming.application.example/models/streaming.py

# Copyright (c) 2023, NVIDIA CORPORATION.  All rights reserved.
#
# NVIDIA CORPORATION and its licensors retain all intellectual property
# and proprietary rights in and to this software, related documentation
# and any modifications thereto.  Any use, reproduction, disclosure or
# distribution of this software and related documentation without an express
# license agreement from NVIDIA CORPORATION is strictly prohibited.

from enum import Enum

from fastapi import BaseModel, Field


class SortBy(str, Enum):
   """Fields by which to sort results by."""
   created_time = "created_at"
   username = "username"

class OrderBy(str, Enum):
   """Sorting direction for the field by which to order results by."""
   asc = "ASC"
   desc = "DESC"

class PortType(str, Enum):
   """Supported Omniverse application port types."""
   tcp = "TCP"
   udp = "UDP"

class Port(BaseModel):
   """Omniverse application port specifications."""
   name: str = Field(..., description="Application port name.")
   number: int = Field(..., ge=0, le=65535, description="Application port number.")
   type: PortType = Field(..., description="Application port type.")

class StreamingMode(str, Enum):
   """Supported Omniverse application streaming mode."""
   secure = "secure"
   insecure = "insecure"

class StreamRequestSpec(BaseModel):
   """Extensible specification that is passed to the Streaming Controller endpoint."""
   kit_args: List[str] = Field(
      default=[],
      title="Omniverse application arguments.",
      description="Array of strings to be appended to the Omniverse application argument inputs.",
   )
   kit_env: Dict[str, str] = Field(
      default={},
      title="Omniverse application environment variables.",
      description="Dictionary containing environment variables for the Omniverse application container.",
   )
   streaming_mode: StreamingMode = Field(
      default=StreamingMode.secure,
      title="Omniverse application streaming mode.",
      description="Streaming mode for the Omniverse application.",
   )
   image: str = Field(
      ...,
      title="Omniverse application Docker image hosted on NGC.",
      description="Full path to Omniverse application Docker image hosted on NGC, along with its tag.",
   )
   user_ports: List[Port] = Field(
      default=[],
      title="Omniverse application ports.",
      description="List of additional container ports to expose on the streaming application container.",
   )
   user_labels: Dict[str, str] = Field(
      default={},
      title="Omniverse application pod labels.",
      description="Optional list of labels to assign to the Omniverse application pod.",
   )
   node_labels: Dict[str, str] = Field(
      default={},
      title="Omniverse application node labels.",
      description="Optional labels to set on the node for for scheduling purposes.",
   )

class CreateStreamRequest(BaseModel):
   """Model of a request to create a new streaming instance."""
   url: str = Field(
      ...,
      title="USD Stage URL.",
      description="URL of the Nucleus USD Stage to open (e.g: `omniverse://example.com/path/to/stage.usd`)",
   )
   access_token: str = Field(
      ...,
      title="Nucleus access token.",
      description="Access token that the Omniverse application will use to connect to Nucleus.",
   )
   username: str = Field(
      ...,
      title="Username of the authenticated User for the session.",
      description="Authenticated username to use for the session.",
   )
   spec: StreamRequestSpec = Field(
      default=StreamRequestSpec(),
      title="Application stream specification.",
      description="Extensible Omniverse application stream specification.",
   )

class ErrorMessage(BaseModel):
   """Base application error message."""
   message: str = Field(
      ...,
      title="Error message.",
      description="Detail about the error that occurred.",
   )

exts/omni.services.streaming.application.example/python_ext.py

# Copyright (c) 2023, NVIDIA CORPORATION.  All rights reserved.
#
# NVIDIA CORPORATION and its licensors retain all intellectual property
# and proprietary rights in and to this software, related documentation
# and any modifications thereto.  Any use, reproduction, disclosure or
# distribution of this software and related documentation without an express
# license agreement from NVIDIA CORPORATION is strictly prohibited.

import omni.ext

from omni.services.core.main import deregister_router, register_router

from .services.streaming import router


class StreamingApplicationExampleExtension(omni.ext.IExt):
   """Streaming application example extension."""

   def on_startup(self, ext_id: str) -> None:
      register_router(router=router, tags=["examples"])

   def on_shutdown(self) -> None:
      deregister_router(router=router)

API Code Samples

Healthcheck

Before attempting to communicate with remote endpoints in order to schedule Omniverse application streams, you may wish to first reach the healthcheck API available on the service in order to ensure resources are in a state allowing them to serve incoming requests. Proceeding by first establishing that components are exhibiting standard operational behavior can greatly help reduce potential investigation times in cases where resources may be limited on remote clusters, or if network connectivity issues may be temporarily affecting provisioning due to maintenance windows.

To query the /healthcheck API for the status of the streaming service APIs, navigate to the OpenAPI specification page of a production cluster (e.g. https://store-portal.ovx-prd31-sjc11.proxy.ace.ngc.nvidia.com/backend/api-docs), expanding the /healthcheck endpoint of the Swagger documentation page before clicking the Execute button and confirming that the endpoint responds with the successful application/json Content-Type of:

{
   "ok": true
}

Listing Streaming Sessions

In order to list existing streaming sessions started by Users subscribed to the service, an HTTP GET request can be issued against a service endpoint responsible for returning the list of existing streaming sessions. The most straightforward example illustrating the results of this operation can be implemented using the following code sample:

exts/omni.services.streaming.application.example/services/streaming.py

# Copyright (c) 2023, NVIDIA CORPORATION.  All rights reserved.
#
# NVIDIA CORPORATION and its licensors retain all intellectual property
# and proprietary rights in and to this software, related documentation
# and any modifications thereto.  Any use, reproduction, disclosure or
# distribution of this software and related documentation without an express
# license agreement from NVIDIA CORPORATION is strictly prohibited.

from typing import Any, Dict

from omni.services.client import AsyncClient
from omni.services.core import routers

router = routers.ServiceAPIRouter()


@router.get(
   path="/list-streams",
   summary="Return the list of streaming sessions.",
   description="Returns the list of streaming sessions currently active.",
)
async def list_streams() -> Dict[str, Any]:
   """
   Return metadata information about application streaming sessions.

   Args:
      None

   Returns:
      Dict[str, Any]: Metadata information about application streaming sessions.

   """
   client = AsyncClient(uri="https://cockpit.devtest.az.cloud.omniverse.nvidia.com")
   response = await client.session_service.session.get()
   return response

In the case of a successful response, the remote endpoint will respond with an HTTP 200 and an application/json Content-Type containing details about the streaming session that was created on behalf of the User:

{
   "results": [
      {
         "session_record": {
            "session_id": "99a13993-77d4-402b-81a8-a563858908d9",
            "username": "user@email.com",
            "source_address": "0.0.0.0",
            "destination_address": "12.34.56.78",
            "session_routes": "16249:TCP:NPORT:30136,15426:UDP:MEDIA:31306",
            "nucleus": "external.devtest.az.cloud.omniverse.nvidia.com",
            "stream_type": "app",
            "pod_name": "ovc-v1-kit-99a13993-77d4-402b-81a8-a563858908d9",
            "service_name": "ovc-v1-kit-99a13993-77d4-402b-81a8-a563858908d9",
            "configmap_name": "ovc-v1-kit-99a13993-77d4-402b-81a8-a563858908d9",
            "session_link": "https://devtest.cloud.omniverse.nvidia.com/webclient/index.html?signalingserver=customer.devtest.az.cloud.omniverse.nvidia.com&signalingport=48322&mediaserver=customer.devtest.az.cloud.omniverse.nvidia.com&mediaport=15426&sessionid=99a13993-77d4-402b-81a8-a563858908d9&mic=0&cursor=free&server=&nucleus=external.devtest.az.cloud.omniverse.nvidia.com&resolution=1920x1080&fps=60&autolaunch=true&backendurl=public-api.devtest.az.cloud.omniverse.nvidia.com&accessToken=<Nucleus Access Token>&terminateVerb=DELETE",
            "created_at": "2023-08-01T12:00:00.000000+00:00",
            "updated_at": "2023-08-01T12:00:00.000000+00:00"
         },
         "status": {
            "phase": "Running",
            "node_name": "cbg1-f06-ovx-05",
            "host_ip": "12.34.56.78",
            "pod_ip": "100.96.18.25",
            "start_time": "2023-08-01T12:00:00+00:00"
         }
      },
      // Include additional session metadata, for any available on the service.
      // [...]
   ]
}

For an example using additional filtering and sorting options when issuing queries against the remote service’s session API, the following code sample illustrates how one could formulate a query to list the 10 most recent sessions of a particular User:

from .models.streaming import OrderBy, SortBy

@router.get(
   path="/list-streams",
   summary="Return the list of streaming sessions.",
   description="Returns the list of streaming sessions currently active.",
)
async def list_streams() -> Dict[str, Any]:
   """
   Return metadata information about application streaming sessions.

   Args:
      None

   Returns:
      Dict[str, Any]: Metadata information about application streaming sessions.

   """
   client = AsyncClient(uri="https://cockpit.devtest.az.cloud.omniverse.nvidia.com")
   response = await client.session_service.session.get(
      # Maximum number of results to display:
      limit=10,
      # Offset of the first result to begin displaying:
      offset=0,
      # Username of the User whose sessions should be displayed:
      username="username@example.com",
      # Field by which to sort results by:
      sortBy=SortBy.created_time.value,
      # Sorting direction of the results:
      orderBy=OrderBy.desc.value,
   )
   return response

Creating Streaming Sessions

In order to create new streaming sessions, queries can be issued against the create-stream API, which accepts POST queries containing information such as the access token to use to authorize requests for the given username, along with data such as the URL of the NGC Docker container hosting the Omniverse application to stream:

exts/omni.services.streaming.application.example/services/streaming.py

# Copyright (c) 2023, NVIDIA CORPORATION.  All rights reserved.
#
# NVIDIA CORPORATION and its licensors retain all intellectual property
# and proprietary rights in and to this software, related documentation
# and any modifications thereto.  Any use, reproduction, disclosure or
# distribution of this software and related documentation without an express
# license agreement from NVIDIA CORPORATION is strictly prohibited.

from typing import Any, Dict

from omni.services.client import AsyncClient
from omni.services.core import routers

from .models.streaming import CreateStreamRequest, StreamingMode, StreamRequestSpec

router = routers.ServiceAPIRouter()


@router.post(
   path="/create-stream",
   summary="Create a new streaming session.",
   description="Returns metadata information about a newly-created streaming session.",
)
async def create_stream() -> Dict[str, Any]:
   """
   Send a request to create a streaming session using the given NGC Docker container on behalf of the User.

   Args:
      None

   Returns:
      Dict[str, Any]: Metadata information about the streaming session that was just created.

   """
   client = AsyncClient(uri="https://cockpit.devtest.az.cloud.omniverse.nvidia.com")
   create_stream_options = CreateStreamRequest(
      url="omniverse://store.devtest.az.cloud.omniverse.nvidia.com",
      # Replace with the Nucleus access token for the given User:
      access_token="<Nucleus Access Token>",
      # Replace with the email address identifying the User on whose behalf the stream is initiated:
      username="user@email.com",
      # Configure any required information about the streaming session, including Docker container image settings
      # from NGC, or environment variables to provide for the streaming session:
      spec=StreamRequestSpec(
         # Array of strings to be appended to the Omniverse application argument inputs:
         kit_args=[],
         # Omniverse application environment variables:
         kit_env={},
         # Omniverse application streaming mode:
         streaming_mode=StreamingMode.secure,
         # Omniverse application Docker image hosted on NGC:
         image="nvcr.io/nvidian/omniverse/ov-composer-streaming-webrtc-ovc:2023.1.0-beta.28",
         # Omniverse application ports:
         user_ports=[],
         # Omniverse application pod labels:
         user_labels={},
         # Omniverse application node labels:
         node_labels={},
      ),
   )
   response = await client.session_service.session.post(**create_stream_options.dict())
   return response

Note

The code sample above provides a link to an Omniverse USD Composer application instance featuring WebRTC streaming capabilities in order to enable remote development workflows. To publish a custom version of an application’s Docker container, host the desired features to your organization’s private registry on NGC.

For additional information about using the NGC private registry, consult the User Guide available online.

In the case of a successful stream creation, the remote endpoint will respond with an HTTP 200 and an application/json Content-Type containing the URL of the stream that was created on behalf of the User:

{
   "redirect": "https://devtest.cloud.omniverse.nvidia.com/webclient/index.html?signalingserver=customer.devtest.az.cloud.omniverse.nvidia.com&signalingport=48322&mediaserver=customer.devtest.az.cloud.omniverse.nvidia.com&mediaport=15426&sessionid=99a13993-77d4-402b-81a8-a563858908d9&mic=0&cursor=free&server=&nucleus=external.devtest.az.cloud.omniverse.nvidia.com&resolution=1920x1080&fps=60&autolaunch=true&backendurl=public-api.devtest.az.cloud.omniverse.nvidia.com&accessToken=<Nucleus Access Token>&terminateVerb=DELETE",
   "session_id": "99a13993-77d4-402b-81a8-a563858908d9",
   "session_routes": "16249:TCP:NPORT:30136,15426:UDP:MEDIA:31306"
}

Deleting Streaming Sessions

Once interaction with a streaming session has been completed, Users can proceed to terminate the application instance they have initiated in order to free resources on remote host and increase hardware availability for Users wishing to make use of greater node capacity. To terminate sessions, Users can emit DELETE requests against the streaming API and supplying the unique identifier of the session to terminate:

exts/omni.services.streaming.application.example/services/streaming.py

# Copyright (c) 2023, NVIDIA CORPORATION.  All rights reserved.
#
# NVIDIA CORPORATION and its licensors retain all intellectual property
# and proprietary rights in and to this software, related documentation
# and any modifications thereto.  Any use, reproduction, disclosure or
# distribution of this software and related documentation without an express
# license agreement from NVIDIA CORPORATION is strictly prohibited.

from typing import Any, Dict

from omni.services.client import AsyncClient
from omni.services.core import routers

from .models.streaming import ErrorMessage

router = routers.ServiceAPIRouter()


@router.delete(
   path="/{session_id}",
   summary="Terminate a session.",
   description="Terminate the session with the given unique identifier.",
   responses={
      404: {"model": ErrorMessage},
   },
)
async def delete_stream(session_id: str,) -> Dict[str, Any]:
   """
   Send a request to terminate the session with the given unique identifier.

   Args:
      session_id (str): Unique identifier of the session to terminate.

   Returns:
      Dict[str, Any]: Metadata information about the session that was just terminated.

   """
   client = AsyncClient(uri="https://cockpit.devtest.az.cloud.omniverse.nvidia.com")
   response = await getattr(client.session_service.session, session_id).delete()
   return response

In the case of a successful request, the remote endpoint will respond with an HTTP 200 and the content "OK".