Skip to content
R2Kernel Runtime Kernel for Robotics v1.0.0rc1
R2Kernel Runtime Kernel for Robotics v1.0.0rc1

    Subsystem SDK API Guide

    The subsystem SDK API is the subsystem-author API for Python and C++.

    Use this page when writing code that subclasses or wraps a R2Kernel subsystem. Do not use this API from browser clients or outside-runtime tools; those clients should use the Host Control REST API or Host Control WebSocket API.

    This page explains the subsystem-author flow. For the table-oriented public class and method catalog, see Subsystem SDK Reference.

    Scope

    The SDK API owns the following.

    • subsystem construction and lifecycle hooks
    • Event publish/subscribe registration and publishing
    • Control endpoint registration and timed exchange calls
    • Procedure registration/calls
    • Operation registration/start/cancel/query
    • typed payload refs and endpoint timing policy

    API Contract, component, and message definitions live under Spec. This page shows how subsystem code uses those contracts.

    Construction

    Use the role-specific subclass that matches subsystem.role in subsystem.yaml.

    Role Python/C++ subclass
    gateway GatewaySubsystem
    component ComponentSubsystem
    compound CompoundSubsystem
    app AppSubsystem

    Subsystem remains the common base class. In Python, the role-specific classes check the loaded manifest role during construction.

    Python example.

    ComponentSubsystem(
    name="controller",
    log_level="INFO",
    user_data_path="./userdata",
    manifest_path="./subsystem.yaml",
    heartbeat_port=40000,
)

    C++ example.

    r2kernel::wrapper::SubsystemOptions options;
options.name = "controller";
options.log_level = "INFO";
options.user_data_path = "./userdata";
options.manifest_path = "./subsystem.yaml";
options.heartbeat_port = 40000;

MyComponentSubsystem subsystem(std::move(options));

    When a subsystem is started with r2kernel launch run or r2kernel deploy up, the package launcher also sets R2KERNEL_SUBSYSTEM_MANIFEST and R2KERNEL_BINDINGS. That lets C++ subsystems report the same manifest, interfaces, and bindings status shape as Python subsystems.

    Lifecycle

    A subsystem implements required startup hooks and may implement runtime lifecycle hooks. Runtime lifecycle commands are requests; user code does not set the subsystem state directly.

    Python example.

    from r2kernel.wrapper import ComponentSubsystem


class MySubsystem(ComponentSubsystem):
    def on_initialize(self) -> bool:
        return True

    def on_start(self) -> bool:
        return True

    def on_terminate(self) -> bool:
        return True

    def on_reset(self) -> bool:
        return True

    def on_pause(self) -> bool:
        return True

    def on_resume(self) -> bool:
        return True

    def on_fault(self, reason: str | None = None) -> bool:
        return True

    C++ example.

    class MySubsystem final : public r2kernel::wrapper::ComponentSubsystem {
protected:
    bool on_initialize() override { return true; }
    bool on_start() override { return true; }
    bool on_terminate() override { return true; }
    bool on_reset() override { return true; }
    bool on_pause() override { return true; }
    bool on_resume() override { return true; }
    bool on_fault(const std::string& reason) override {
        (void)reason;
        return true;
    }
};

    Canonical subsystem lifecycle states are documented in Subsystem Lifecycle. The common operator commands are these.

    r2kernel subsystem pause <subsystem>
r2kernel subsystem resume <subsystem>
r2kernel subsystem reset <subsystem>
r2kernel subsystem fault <subsystem> --reason "fault text"

    Event API

    For Event methods, topic means the Event channel key.

    Python methods.

    • register_event_publish(...)
    • register_event_subscribe(...)
    • publish_event(...)
    • start_event_transport()
    • stop_event_transport()

    C++ methods use the same conceptual operations with EventPublishSpec, EventSubscribeSpec, and EventPayload.

    register_event_publish accepts an optional Event queue_depth in Python, and EventPublishSpec::queue_depth in C++. A value of 0 or omission selects the release default. Set it explicitly for reproducible Event streaming or telemetry workloads where short bursts must be absorbed before samples are dropped.

    For the method catalog, see Subsystem SDK Reference.

    Control API

    Control can be raw, typed, or typed/timed. Use the control method names in subsystem code. Low-level wire protocol details are reserved for internal protocol documentation.

    The public SDK argument is named topic. For Control calls, treat this value as the Control endpoint key, not as an Event channel.

    Python example.

    target = self.binding_target("gateway")
response = self.transact_control(
    target=target,
    topic="timed.request",
    type="bytes",
    value={"value": 1},
    request_type="msg.core.common.primitive.v1/Int32Value",
    response_type="msg.core.common.primitive.v1/Int32Value",
    timing={"execute_after_ns": 500_000},
    period_ms=0.0,
    deadline_ms=1000.0,
)

    For the method catalog, see Subsystem SDK Reference.

    C++ example.

    r2kernel::wrapper::ControlTransactOptions options;
options.request_type = "msg.core.common.primitive.v1/Int32Value";
options.response_type = "msg.core.common.primitive.v1/Int32Value";
options.timing = r2kernel::wrapper::ControlTimingOptions{};
options.timing->execute_after_ns = 500'000;

const auto response = transact_control(
    "gateway",
    "timed.request",
    "{\"value\":1}",
    options
);

    Procedure API

    Basic procedure calls remain available. Typed/timed calls add request and response message refs.

    Python example.

    response = self.call_bound_procedure(
    binding="io_gateway",
    procedure="set_output",
    params={"index": 1, "value": True},
    request_type="msg.core.common.primitive.v1/IndexedBoolValue",
    response_type="msg.core.common.primitive.v1/BoolValue",
    timing={"execute_after_ns": 500_000},
)

    C++ example.

    r2kernel::wrapper::ProcedureCallOptions options;
options.request_type = "msg.core.common.primitive.v1/IndexedBoolValue";
options.response_type = "msg.core.common.primitive.v1/BoolValue";
options.timing = r2kernel::wrapper::ProcedureTimingOptions{};
options.timing->execute_after_ns = 500'000;

auto response = call_procedure(
    "io_gateway",
    "set_output",
    "{\"index\":1,\"value\":true}",
    options
);

    For the method catalog, see Subsystem SDK Reference.

    Operation API

    Operation uses the control exchange path for start/cancel/query and Event for state/feedback/result.

    Python example.

    operation = self.start_bound_operation(
    binding="arm",
    operation="move_j",
    params={"positions": [0.0, 0.1, 0.2]},
    request_type="msg.motion.articulated.v1/JointWayPoint",
    timing={"execute_after_ns": 1_000_000},
)

    C++ example.

    r2kernel::wrapper::OperationStartOptions options;
options.request_type = "msg.motion.articulated.v1/JointWayPoint";
options.timing = r2kernel::wrapper::OperationTimingOptions{};
options.timing->execute_after_ns = 1'000'000;

auto response = start_operation("arm", "move_j", "{...}", options);

    For the method catalog, see Subsystem SDK Reference.

    Endpoint Timing Policy

    Endpoint registration accepts the following.

    • timing_support = "none"
    • timing_support = "optional"
    • timing_support = "required"

    This applies to Control endpoints, Procedures, and Operations.

    For the option fields used by Python and C++, see Subsystem SDK Reference.

    Spatial Frame API

    Spatial ports are declared in subsystem.yaml. SDK methods report the dynamic relationships owned by the implementation; they do not create Event endpoints.

    Python.

    self.set_frame_transform(
    parent_frame="odom",
    child_frame="base_link",
    translation=(x, y, z),
    rotation_xyzw=(qx, qy, qz, qw),
    observed_at_sync_ns=self.now_ns(),
    stale_after_ms=500,
)

result = self.lookup_frame_transform(
    "component.mobile/odom",
    "component.arm/tool0",
)

    Use clear_frame_transform(child_frame) when this subsystem no longer owns a valid edge. frame_transforms() returns the current local snapshot.

    C++ uses FrameTransform, Subsystem::set_frame_transform(), Subsystem::clear_frame_transform(), and Subsystem::frame_transforms(). Host code can perform lookup through r2kernel frame lookup or Host Control REST.

    Dynamic updates use a lightweight control-backend command. Heartbeats carry the complete snapshot for recovery. stale_after_ms is required for dynamic edges; static edges use is_static=true and do not expire.

    See Spatial Model And Frame Graph.

    Managed Resource Handles

    Packages launched through R2Kernel receive R2KERNEL_RESOURCE_LEASE_ID and can ask the host resource service for approved handles. Hardware-facing code uses the public helpers such as open_udp_socket, open_tcp_socket, open_can_socket, and open_device.

    Managed package frontend backends use a narrower helper. If package.yaml declares frontend.context.api or frontend.context.ws, launch resolves that context into a loopback listener grant. The package process opens that listener through the broker.

    from r2kernel.core import ResourceBrokerClient

api_listener = ResourceBrokerClient().open_frontend_tcp_listener(kind="api").socket
ws_listener = ResourceBrokerClient().open_frontend_tcp_listener(kind="ws").socket

    The browser still connects to the control backend same-origin proxy. The package backend serves the accepted listener socket; it must not bind a separate frontend port directly under brokered network sandboxing.

    For resource helper methods, see Subsystem SDK Reference.