The InOrbit Edge SDK allows Python programs to communicate with InOrbit
platform on behalf of robots - providing robot data and handling robot actions.
Its goal is to ease the integration between InOrbit and any other software
that handles robot data.
- Robot session handling through a
RobotSessionPool. - Publish key-values.
- Publish robot poses.
- Publish robot odometry.
- Publish robot path.
- Publish robot laser.
- Execute callbacks on Custom Action execution.
- Execute scripts (or any program) in response to Custom Action execution.
- Stream camera frames from RTSP (or anything OpenCV opens).
from inorbit_edge.robot import RobotSessionFactory, RobotSessionPool
def my_command_handler(robot_id, command_name, args, options):
"""Callback for processing custom command calls.
Args:
robot_id (str): InOrbit robot ID
command_name (str): InOrbit command e.g. 'customCommand'
args (list): Command arguments
options (dict): object that includes
- `result_function` can be called to report command execution
result with the following signature: `result_function(return_code)`
- `progress_function` can be used to report command output with
the following signature: `progress_function(output, error)`
- `metadata` is reserved for the future and will contain additional
information about the received command request.
"""
if command_name == "customCommand":
print(f"Received '{command_name}' for robot '{robot_id}'!. {args}")
# Return '0' for success
options["result_function"]("0")
robot_session_factory = RobotSessionFactory(
api_key="<YOUR_API_KEY>"
)
# Register commands handlers. Note that all handlers are invoked.
robot_session_factory.register_command_callback(my_command_handler)
robot_session_factory.register_commands_path("./user_scripts", r".*\.sh")
robot_session_pool = RobotSessionPool(robot_session_factory)
robot_session = robot_session_pool.get_session(
robot_id="my_robot_id_123", robot_name="Python SDK Quick Start Robot"
)
robot_session.publish_pose(x=0.0, y=0.0, yaw=0.0)Stable Release: pip install inorbit-edge
Development Head: pip install git+https://github.com/inorbit-ai/edge-sdk-python.git
For full package documentation please visit InOrbit Developer Portal.
See CONTRIBUTING.md for information related to developing the code.
-
pip install -e .[dev]This will install your package in editable mode with all the required development dependencies (i.e.
tox). -
make buildThis will run
toxwhich will run all your tests in Python 3.10 - 3.13 as well as linting your code. -
make cleanThis will clean up various Python and build generated files so that you can ensure that you are working in a clean environment.
Install the optional video extra (see requirements-video.txt), which pulls
in OpenCV:
pip install inorbit-edge[video]
Register a camera on a session. Frames are streamed only while the platform asks
for video -- for example when a user opens a camera view -- and the camera id is
the topic id the InOrbit camera must be configured with ("0" for the first
one):
from inorbit_edge.video import OpenCVCamera
session.register_camera(
"0", OpenCVCamera("rtsp://user:pass@192.0.2.10:554/stream1", rate=5)
)For RTSP, set OpenCV's FFmpeg options before the first capture is opened
(they are read by OpenCV when it opens the stream, so export them or set them in
os.environ at import time):
export OPENCV_FFMPEG_CAPTURE_OPTIONS="rtsp_transport;tcp|timeout;3000000"rtsp_transport;tcp because some cameras reject UDP, and timeout
(microseconds) bounds socket reads: without it, a camera that stops answering
mid-stream is only noticed after OpenCV's 30s watchdog, which delays both the
reopen and shutdown.
OpenCVCamera settings, all optional:
| Setting | Default | Meaning |
|---|---|---|
rate |
10 |
Frames per second published |
scaling |
0.3 |
Downscale factor applied before JPEG encoding |
quality |
35 |
JPEG quality, 1-100 |
stale_frame_seconds |
3.0 |
Stop serving the buffered frame once it is older than this, so a stream that died shows no video instead of a frozen picture. None keeps serving the last frame |
api_preference |
auto | OpenCV backend to open with; URL sources default to cv2.CAP_FFMPEG |
REOPEN_BACKOFF_SECONDS |
0.5s to 10s |
Class attribute: delay before each attempt to rebuild a capture whose grabs are failing |
HEALTH_LOG_SECONDS |
60.0 |
Class attribute: how often the capture health line below is logged |
Each camera logs one health line per window, which is usually enough to tell where video stopped:
Capture health: grabbed=1800 served=60 stale=0 reopens=0 in the last 60s
No line at all means the platform never requested video; grabbed=0 means the
stream is unreachable; frames grabbed but not served means nothing is consuming
them; frames served with nothing visible in the platform points at the MQTT
side. The same signals are exported as the video_frames_grabbed,
video_frames_stale and video_capture_reopens counters (see Metrics below).
The SDK is capable of collecting internal metrics such as number of calls to publishing functions. It uses OpenTelemetry, which supports various exporting mechanisms. Connectors are responsible for configuring the exporter of their choice; as well as adding more metrics if they chose to do so.
Install the optional telemetry extra (see requirements-telemetry.txt) so
the SDK records real OpenTelemetry metrics. Without it, built-in metrics are
no-ops and the base package has no OpenTelemetry dependency:
pip install inorbit-edge[telemetry]
To export to Prometheus, the extra above includes opentelemetry-exporter-prometheus
and prometheus-client. The following is an example initialization code that enables a
Prometheus HTTP endpoint, where all SDK metrics
(including system metrics such as CPU usage) and any metric added by the
connector can be scraped and exported to any external system (Grafana,
StackDriver, etc.)
from inorbit_edge.metrics import setup_prometheus_meter_provider
from prometheus_client import start_http_server
# ...
if setup_prometheus_meter_provider(
service_name="my-connector",
service_instance_id="robot-123",
service_version="1.2.3",
):
start_http_server(port=9464, addr="0.0.0.0")Custom metrics can use the same meter provider. Define instruments once during module initialization, then record values where the connector does the work:
from inorbit_edge.metrics import get_meter
meter = get_meter("my_connector")
messages_processed_counter = meter.create_counter(
"messages_processed",
unit="1",
description="Number of input messages processed by the connector",
)
def process_message(robot_id, message):
# ... connector-specific processing ...
messages_processed_counter.add(1, {"robot_id": robot_id})When exported to Prometheus with service_name="my-connector", this appears as
my_connector_messages_processed_total with a robot_id label. Without the
telemetry extra installed, the same code is safe to run but records no data.
For call-count metrics, the SDK also provides a decorator. This keeps the increment close to the function being counted:
from inorbit_edge.metrics import get_meter, with_counter_metric
meter = get_meter("my_connector")
command_handler_counter = meter.create_counter(
"command_handler_calls",
unit="1",
description="Number of command handler invocations",
)
@with_counter_metric(command_handler_counter, attributes={"command": "dock"})
def handle_dock_command(command_payload):
# ... handle the command ...
return "accepted"If attributes depend on the function arguments, pass a callable instead of a static dictionary:
@with_counter_metric(
command_handler_counter,
attributes=lambda robot_id, command_payload: {"robot_id": robot_id},
)
def handle_command(robot_id, command_payload):
# ... handle the command ...
return "accepted"