Core

Core is the ROS-facing API surface: namespace discovery, topic inspection, publish/subscribe, image helpers, parameter handling, and service calls.

Image, CompressedImage, and PointCloud2 topics carry their data as native protobuf bytes in both directions. Other ROS messages use the generic structured representation.

Python

Core topic/param wrapper over olo.core.v1.Core.

Client interfaces

class olo.core.Core

Bases: object

__init__(session)

Bind to a channel session.

call_service(service, request=None, *, timeout=10.0)

Call a ROS service and return the response as a plain dict.

Return type:

Dict[str, Any]

Parameters:

• service (ServiceInfo)

• request (Dict[str, Any] | None)

• timeout (float)

get_image(topic, *, timeout=5.0)

Wait for the next image on a topic and decode it into an Image.

Handles both sensor_msgs/msg/Image and sensor_msgs/msg/CompressedImage transparently, driven by topic.msg_type.

Return type:

Image

Parameters:

• topic (TopicInfo)

• timeout (float)

get_latest(topic, *, timeout=5.0)

Wait for the next message on a topic and return it as a raw dict. It resolves once a message is published (or the timeout elapses). Use getImage to decode image topics into an Image.

Return type:

Dict[str, Any]

Parameters:

• topic (TopicInfo)

• timeout (float)

get_params(node, names, *, timeout=10.0)

Read ROS parameters from a node.

Values come back as natives matching the ROS parameter type (bool, int, float, str, bytes, or a homogeneous list). Unset parameters are returned as None.

Return type:

Dict[str, Any]

Parameters:

• node (str)

• names (List[str])

• timeout (float)

handle(namespace=None)

Return a namespace-scoped core handle.

Return type:

CoreHandle

Parameters:

namespace (str | None)

list_robot_namespaces(*, timeout=10.0)

List available robot namespaces.

Return type:

RobotNamespaces

Parameters:

timeout (float)

list_services(*, timeout=10.0)

List available ROS services.

Return type:

List[ServiceInfo]

Parameters:

timeout (float)

list_topics(*, timeout=10.0)

List available ROS topics.

Return type:

List[TopicInfo]

Parameters:

timeout (float)

publish(topic, message, *, timeout=10.0)

Overloads:

• self, topic (TopicInfo), message (Dict[str, Any]), timeout (Optional[float]) → None

• self, topic (TopicInfo), message (Image), timeout (Optional[float]) → None

Parameters:

• topic (TopicInfo)

• message (Dict[str, Any] | Image)

• timeout (float | None)

Return type:

None

Publish a ROS message on a topic.

message is either a plain ROS message dict or an Image. When an Image is given, it is encoded according to topic.msg_type (sensor_msgs/msg/Image -> raw, sensor_msgs/msg/CompressedImage -> compressed), mirroring how get_latest() decodes those types into an Image.

publisher(topic)

Create a reusable publisher handle.

Return type:

Publisher

Parameters:

topic (TopicInfo)

set_params(node, params, *, timeout=10.0)

Write ROS parameters on a node.

The Python type of each value determines the ROS parameter type (bool, int, float, str, bytes, or a homogeneous list thereof).

Return type:

None

Parameters:

• node (str)

• params (Dict[str, Any])

• timeout (float)

subscribe(topic, *, timeout=None)

Subscribe to a topic stream.

Return type:

Subscription

Parameters:

• topic (TopicInfo)

• timeout (float | None)

class olo.core.CoreHandle

Bases: object

Namespace-scoped handle for common core operations.

__init__(core, namespace)

Bind to a core client and resolved robot namespace.

Parameters:

• core (Core)

• namespace (str)

call_service(service, request=None, *, timeout=10.0)

Call a service resolved against this namespace.

Return type:

Dict[str, Any]

Parameters:

• service (ServiceInfo)

• request (Dict[str, Any] | None)

• timeout (float)

get_image(topic, *, timeout=5.0)

Wait for the next image on a topic resolved against this namespace.

Return type:

Image

Parameters:

• topic (TopicInfo)

• timeout (float)

get_latest(topic, *, timeout=5.0)

Wait for the next message on a topic resolved against this namespace.

Return type:

Dict[str, Any]

Parameters:

• topic (TopicInfo)

• timeout (float)

get_params(node, names, *, timeout=10.0)

Read ROS parameters from a namespace-relative node.

Return type:

Dict[str, Any]

Parameters:

• node (str)

• names (List[str])

• timeout (float)

list_services(*, timeout=10.0)

List services visible through this handle using namespace-relative names.

Return type:

List[ServiceInfo]

Parameters:

timeout (float)

list_topics(*, timeout=10.0)

List topics visible through this handle using namespace-relative names.

Return type:

List[TopicInfo]

Parameters:

timeout (float)

node(name)

Build a namespace-prefixed node name.

Return type:

str

Parameters:

name (str)

publish(topic, message, *, timeout=10.0)

Overloads:

• self, topic (TopicInfo), message (Dict[str, Any]), timeout (Optional[float]) → None

• self, topic (TopicInfo), message (Image), timeout (Optional[float]) → None

Parameters:

• topic (TopicInfo)

• message (Dict[str, Any] | Image)

• timeout (float | None)

Return type:

None

Publish a ROS message on a topic resolved against this namespace.

publisher(topic)

Create a publisher for a topic resolved against this namespace.

Return type:

Publisher

Parameters:

topic (TopicInfo)

service(name)

Resolve a service name against the robot namespace (ROS-style).

Return type:

str

Parameters:

name (str)

set_params(node, params, *, timeout=10.0)

Write ROS parameters on a namespace-relative node.

Return type:

None

Parameters:

• node (str)

• params (Dict[str, Any])

• timeout (float)

subscribe(topic, *, timeout=None)

Subscribe to a topic resolved against this namespace.

Return type:

Subscription

Parameters:

• topic (TopicInfo)

• timeout (float | None)

topic(name, msg_type)

Resolve a topic name against the robot namespace (ROS-style).

Leading / means absolute: the name is used as-is. Otherwise the name is resolved relative to this handle’s namespace.

Return type:

TopicInfo

Parameters:

• name (str)

• msg_type (str)

class olo.core.Publisher

Bases: object

Lightweight handle for repeated Publish calls (stateless on the appliance).

__init__(core, topic)

Capture the publish target.

Parameters:

• core (Core)

• topic (TopicInfo)

publish(message, *, timeout=None)

Overloads:

• self, message (Dict[str, Any]), timeout (Optional[float]) → None

• self, message (Image), timeout (Optional[float]) → None

Parameters:

• message (Dict[str, Any] | Image)

• timeout (float | None)

Return type:

None

Publish a message on this topic.

class olo.core.Subscription

Bases: object

Context-managed sync iterator over a Core.Subscribe server stream.

__init__(stream)

Wrap a Subscribe server stream.

close()

Cancel the stream and mark it closed.

Return type:

None

Types

class olo.core.RobotNamespaces

Bases: object

Robot namespaces available through the appliance.

__init__(has_global=False, namespaces=())

Parameters:

• has_global (bool)

• namespaces (tuple[str, ...])

Return type:

None

resolve_default()

Return the unambiguous default robot namespace, or raise.

Return type:

str

class olo.core.TopicInfo

Bases: object

A ROS topic name and message type.

__init__(name, msg_type)

Parameters:

• name (str)

• msg_type (str)

Return type:

None

class olo.core.ServiceInfo

Bases: object

A ROS service name and service type.

__init__(name, service_type)

Parameters:

• name (str)

• service_type (str)

Return type:

None

class olo.core.Image

Bases: object

Decoded image data; color images use OpenCV BGR layout.

__init__(array, encoding='', format='', header=None, is_bigendian=False)

Parameters:

• array (ndarray)

• encoding (str)

• format (str)

• header (Dict[str, Any] | None)

• is_bigendian (bool)

Return type:

None

TypeScript

Import from olo/core:

Client interfaces

• Core — service wrapper for low-level Core RPCs.

• CoreHandle — namespace-scoped helper for common robot operations.

• Publisher — reusable topic publisher.

• Subscription — async iterable subscription stream.

Types

• TopicInfo — topic name and ROS message type.

• ServiceInfo — service name and ROS service type.

• RosMessage — ROS-shaped object; binary sensor fields are Uint8Array values.

• Image — dependency-free image payload.

• CallOptions — per-call timeout options.

• ParamNative — native value union for ROS parameters.

• RobotNamespaces — namespace discovery result and default-resolution helper.

Images and namespaces

• imageMessageToFrame

• compressedMessageToFrame

• imageToMessage

• compressedImageToMessage

• normalizeRobotNamespace