Skip to main content

Samples

Local only

This feature is not available in the Cloud API.

The Nimbus agent repository includes sample clients for Local API Gateway REST message uploads, live streams, and fleet Custom KPI WebSocket flows.

Download the ready-to-run Python source bundle:

Python: Send One Message Through The Fleet Manager​

Download the standalone Python REST sample, also included in the Python sample ZIP. It runs with Python 3.10 or later and needs no third-party packages or installation.

Enable Local API on the Fleet Manager and target edge Agent. Register the edge robot in the Fleet Manager with its URL/password:

nimbus fleet add-robot <edge-agent-url> <edge-agent-password>

Use the Fleet Manager's password and URL below. The manager uses the registered edge password to forward REST requests; the client does not need to log in to the edge Agent separately.

1. List Robots​

python3 nimbus_apigateway_upload.py '<fleet-manager-password>' http://<fleet-manager-host>:19992

Copy the target edge robot's basicData.id. The list also includes the Fleet Manager itself.

2. List The Target Robot's Input Streams​

python3 nimbus_apigateway_upload.py \
  '<fleet-manager-password>' http://<fleet-manager-host>:19992 '<edge-robot-id>'

The sample lists input streams with source 1 (component), 2 (ROS1), or 3 (ROS2).

3. Send One JSON Message​

For a listed component input with componentName: data-publisher, streamName: gps, and streamType: Nimbus.Messages.geometry_msgs.Point:

python3 nimbus_apigateway_upload.py \
  '<fleet-manager-password>' http://<fleet-manager-host>:19992 '<edge-robot-id>' gps \
  --component-name data-publisher --source 1 --message '{"x":1,"y":2,"z":3}'

Replace the component, stream, and JSON payload with values for your discovered input. For a ROS2 Point input topic named /example_point, omit the component name:

python3 nimbus_apigateway_upload.py \
  '<fleet-manager-password>' http://<fleet-manager-host>:19992 '<edge-robot-id>' /example_point \
  --source 3 --message '{"x":1,"y":2,"z":3}'

For ROS1, use --source 2. These examples require the corresponding Point input stream; adapt the payload to the actual message type. The sample copies streamType into dataType and serializes the object into the required streamJson string before calling POST /ApiGateway/V2/robots/{robotId}/upload-stream.

Each run sends one message to one robot without retrying the upload. A successful response means the target Agent's upload endpoint returned successfully; it does not confirm completion of a robot action. Errors include the HTTP status and server response; an unreachable fleet member normally returns 503. There is no offline message queue in this path.

WebSocket forwarding to local-fleet members is unsupported. For continuous teleoperation, connect the WebSocket samples directly to the target Agent. See Upload Stream and Local-Fleet Behavior for the REST contract and forwarding details.

When running from the repository or extracted ZIP root, prefix the script path with Nimbus.ApiGateway.PythonSampleClient/. After installing the sample package, you can also run python3 -m nimbus_apigateway_upload or nimbus-apigateway-upload with the same arguments.

C# Desktop Sample​

Download the ready-to-run C# source bundle:

The sample targets .NET 10 and requires the .NET 10 SDK.

The C# sample uses:

  • Cogniteam.ApiGateway.Core DTOs and a sample-owned ReFit route interface for login, robot listing, and stream listing.
  • Cogniteam.ApiGateway.Client for normal robot streams through INimbusStreamWebSocketConnection, the Fleet Custom KPI WebSocket, and JSON deserialization.
  • ClientWebSocket only for the explicit self-signed HTTPS fallback.
  • a sample-owned protobuf for binary output MessageStream envelopes.
  • shared API Gateway DTOs for Fleet Custom KPI JSON frames.
  • Cogniteam.Nimbus.Messages for selected output parsers and teleoperation commands.

The explicit self-signed HTTPS test option keeps small sample-owned socket paths for robot streams and Fleet Custom KPIs. Normal trusted-certificate connections use Cogniteam.ApiGateway.Client.

Run it from the Nimbus Agent repository root:

dotnet run --project Nimbus.ApiGateway.SampleClient

The default base URL is:

http://<agent-host>:19992

Choose output to subscribe to binary MessageStream envelopes or direct typed JSON. Choose input to send a finite Twist or Ackermann command as JSON or raw protobuf. Motion defaults to zero, the command rate is limited to 20 Hz, and the sample sends zero before and after motion. Use Fleet Custom KPIs for the separate local-fleet JSON feed.

Python Sample​

The Python package contains separate commands for REST uploads, robot output, teleoperation input, and Fleet Custom KPIs. Robot output supports binary MessageStream envelopes or direct typed JSON. The teleoperation command selects a discovered Twist or Ackermann input and sends JSON or raw typed protobuf for a finite duration. Fleet Custom KPI frames remain direct JSON.

Install and run it from the Nimbus agent repository root, or from the extracted Python sample bundle root:

python3 -m pip install --target /tmp/nimbus-apigateway-python ./Nimbus.ApiGateway.PythonSampleClient

PYTHONPATH=/tmp/nimbus-apigateway-python \
python3 -m nimbus_apigateway_streams <local-user-token> <base-url> [filter]

Example:

PYTHONPATH=/tmp/nimbus-apigateway-python \
python3 -m nimbus_apigateway_streams 588e0a9f-8d2 http://<agent-host>:19992 pose

Request direct JSON output:

PYTHONPATH=/tmp/nimbus-apigateway-python \
python3 -m nimbus_apigateway_streams <local-user-token> <base-url> pose --format json

Send a finite command after selecting a discovered input descriptor:

PYTHONPATH=/tmp/nimbus-apigateway-python \
python3 -m nimbus_apigateway_teleoperation <local-user-token> <base-url> cmd_vel \
  --format json --speed 0.2 --turn 0.1 --rate 10 --duration 1

The command's speed and turn default to zero. Its maximum rate is 20 Hz and its maximum duration is five minutes. It sends zero before and after the command interval; the Agent also applies a 500 ms deadman.

The optional filter searches descriptor display name, component name, stream name, stream type, source type, URL, and endpoint.

To subscribe to fleet Custom KPI updates from the local fleet manager:

PYTHONPATH=/tmp/nimbus-apigateway-python \
python3 -m nimbus_apigateway_fleet_custom_kpis <local-user-token> <base-url> [fleet-id]

fleet-id defaults to local-fleet.