Implementing the service

Now that we have everything in place, let’s start implementing the service.

Here, we will start small, with a working end-to-end service first that we will progressively enhance as we go. Of course, you are free to take the approach you feel most comfortable in, and jump directly to the end of this series to the completed project. In this tutorial series, we’ll take the approach we generally tend to recommend when designing services, where we progressively refine the specifications over multiple iterations. When possible to use this method, this approach generally tends to allow for greater success ratio, as it makes it possible to adapt to changing requirements over the lifespan of a project, and allows for greater flexibility to react and adapt to new information during the project. Greater transparency in communicating the status and progress of the project makes it easier for stakeholders to make informed decisions, which tend to promote efficiency and reduce waste in the project.

To illustrate humble beginnings, let’s start with a minimalistic design and implementation of our service so we can get a working end-to-end solution deployed and tested early.

Note

As we are walking through the process of defining and implementing a Service with you, we have annotated the code samples you will find in this guide with additional comments and insights about the design choices we made.

Please note these comments have been added for the benefit of providing context to our samples, as we will be iterating over the design and providing this information about where we are heading should help provide a sense of direction.

Defining the API

In order to iterate over our service, and have a functional prototype operating early, let’s define a minimal set of requirements that we would like to have in order to demonstrate the capabilities of our service.

We’ll start with a minimal API that will allow Users to capture images from the viewport of our Omniverse application. For this, we will define a REST API that will allow reception of payloads with the following JSON information:

As input:

usd_stage_path: The full path of the USD stage that we’ll want to capture.

As output:

success: A flag indicating whether the operation completed successfully.

captured_image_path: A string informing clients where captured images can be retrieved.

error_message: A string informing clients of any potential error details in case the operation did not complete successfully.

In subsequent steps, after completing an initial proof of concept demonstrating our ability to capture an image from a given USD stage, we could consider adding support for additional options such as image resolution, timestamp, camera, viewport overlay information and other features for which we may also wish to provide controls to Users.

Initial implementation

Let’s define the extension’s main file, its TOML definition. If you already created Omniverse extensions before, there should be nothing surprising in this extension’s configuration file:

exts/omni.services.example.viewport_capture.core/config/extension.toml

[package]
# Semantic versioning is used: https://semver.org
version = "1.0.0"

# The title and description fields are primarily for displaying extension info in UI:
title = "Viewport capture service"
description = "Sample service example demonstrating the creation of microservices using Omniverse."

# Path (relative to the root), or content of the extension's "readme" file, in Markdown format:
readme = "docs/README.md"

# Path (relative to the root) of the extension's changelog file, in Markdown format:
changelog = "docs/CHANGELOG.md"

# URL of the extension source repository:
repository = "https://github.com/NVIDIA-Omniverse/kit-extension-template"

# Category of the extension:
category = "services"

# Keywords of the extension:
keywords = ["kit", "service", "example"]

# Icon to show in the extension manager:
icon = "data/icon.png"

# Preview to show in the extension manager:
preview_image = "data/preview.png"

# Use omni.ui to build simple UI
[dependencies]
"omni.kit.menu.edit" = {}
"omni.kit.actions.core" = {}
"omni.services.core" = {}
"omni.services.transport.server.http" = {}
"omni.usd" = {}

# Main Python module this extension provides, it will be publicly available as
# "import omni.services.example.viewport_capture.core":
[[python.module]]
name = "omni.services.example.viewport_capture.core"

# Settings of our extension:
[settings.exts."omni.services.example.viewport_capture.core"]
# URL prefix where the service will be mounted, where our API will be available to handle incoming requests.
#
# Defining this as a setting makes it easy to change or rebrand the endpoint using only command-line or KIT-file
# configuration instructions, should extensions ever feature conflicting endpoint naming conventions.
url_prefix = "/viewport-capture"

# Path from where the captured images will be served from, when exposed to clients.
#
# This path will be mounted as a child of the `url_prefix` setting, and expressed as a formatted join of the
# `{url_prefix}{capture_path}` settings.
capture_path = "/static"

# Name of the directory on the server where captured images will be stored:
capture_directory = "captured_stage_images"

Just as we did in the step before when confirming the extension was properly registered with our Omniverse host application, we will also confirm adequate end-to-end functionality of this development iteration by displaying debugging information to the console when receiving a request from the network. This will allow us to confirm that we properly received incoming requests, and that all components of our extension are correctly connected and cooperating together.

So let’s now define our extension’s API wrapper, which we’ll later register in our extension entrypoint:

exts/omni.services.example.viewport_capture.core/omni/services/example/viewport_capture/core/services/capture.py

from typing import Optional

from pydantic import BaseModel, Field

import carb

from omni.services.core import routers


router = routers.ServiceAPIRouter()


# Let's define a model to handle the parsing of incoming requests.
#
# Using `pydantic` to handle data-parsing duties makes it less cumbersome for us to do express types, default values,
# minimum/maximum values, etc. while also taking care of documenting input and output properties of our service using
# the OpenAPI specification format.
class ViewportCaptureRequestModel(BaseModel):
    """Model describing the request to capture a viewport as an image."""

    usd_stage_path: str = Field(
        ...,
        title="Path of the USD stage for which to capture an image",
        description="Location where the USD stage to capture can be found.",
    )
    # If required, add additional capture response options in subsequent iterations.
    # [...]

# We will also define a model to handle the delivery of responses back to clients.
#
# Just like the model used to handle incoming requests, the model to deliver responses will not only help define
# default values of response parameters, but also in documenting the values clients can expect using the OpenAPI
# specification format.
class ViewportCaptureResponseModel(BaseModel):
    """Model describing the response to the request to capture a viewport as an image."""

    success: bool = Field(
        default=False,
        title="Capture status",
        description="Status of the capture of the given USD stage.",
    )
    captured_image_path: Optional[str] = Field(
        default=None,
        title="Captured image path",
        description="Path of the captured image, hosted on the current server.",
    )
    error_message: Optional[str] = Field(
        default=None,
        title="Error message",
        description="Optional error message in case the operation was not successful.",
    )
    # If required, add additional capture response options in subsequent iterations.
    # [...]


# Using the `@router` annotation, we'll tag our `capture` function handler to document the responses and path of the
# API, once again using the OpenAPI specification format.
@router.post(
    path="/capture",
    summary="Capture a given USD stage",
    description="Capture a given USD stage as an image.",
    response_model=ViewportCaptureResponseModel,
)
async def capture(request: ViewportCaptureRequestModel,) -> ViewportCaptureResponseModel:
    # For now, let's just print incoming request to the log to confirm all components of our extension are properly
    # wired together:
    carb.log_warn(f"Received a request to capture an image of \"{request.usd_stage_path}\".")

    # Let's return a JSON response, indicating that the viewport capture operation failed to avoid misinterpreting the
    # current lack of image output as a failure:
    return ViewportCaptureResponseModel(
        success=False,
        captured_image_path=None,
        error_message="Image not yet captured.",
    )

For the final step of this iteration, let’s register our API in our extension entrypoint, whose principal responsibility will be to register the service that will receive incoming requests to capture images from USD stages from the network:

exts/omni.services.example.viewport_capture.core/omni/services/example/viewport_capture/core/extension.py

import carb
import omni.ext

from omni.services.core import main

# As most of the features of our API are implemented by the means of function handlers in the `/services` sub-folder,
# the main duty of our extension entrypoint is to register our Service's `@router` and dictate when its capability
# should be enabled or disabled under the guidance of the User or the dependency system.
from .services.capture import router


# Any class derived from `omni.ext.IExt` in the top level module (defined in the `python.module` section of the
# `extension.toml` file) will be instantiated when the extension is enabled, and its `on_startup(ext_id)` method
# will be called. When disabled or when the application is shut down, its `on_shutdown()` will be called.
class ViewportCaptureCoreExtension(omni.ext.IExt):
    """Sample extension illustrating registration of a service."""

    # `ext_id` is the unique identifier of the extension, containing its name and semantic version number. This
    # identifier can be used in conjunction with the Extension Manager to query for additional information, such
    # as the extension's location on the filesystem.
    def on_startup(self, ext_id: str) -> None:
        ext_name = ext_id.split("-")[0]
        carb.log_info("ViewportCaptureCoreExtension startup")

        # At this point, we register our Service's `router` under the prefix we gave our API using the settings system,
        # to facilitate its configuration and to ensure it is unique from all other extensions we may have enabled:
        url_prefix = carb.settings.get_settings().get_as_string(f"exts/{ext_name}/url_prefix")
        main.register_router(router=router, prefix=url_prefix, tags=["Viewport capture"],)

    def on_shutdown(self) -> None:
        carb.log_info("ViewportCaptureCoreExtension shutdown")

        # When disabling the extension or shutting down the instance of the Omniverse application, let's make sure we
        # also deregister our Service's `router` in order to avoid our API being erroneously advertised as present as
        # part of the OpenAPI specification despite our handler function no longer being available:
        main.deregister_router(router=router)

Next, we’ll confirm all is set and ready for us to iterate on the core feature by submitting a request to our service in its current early stage.

If you remember from the function handler we just defined in our Service layer, we should expect to see the path of the USD stage we expect to capture, along with a flag indicating whether the operation was successful.


« Previous section: Getting started	Next section: Exploring the API »