`botpulse.init()`

Initialize and return a BotPulse client instance.

import botpulse

pulse = botpulse.init(
    service_name="navigation-node",
    robot_id="robot-01",
)

Parameters

Parameter	Type	Default	Description
`service_name`	`str`	required	Name of this service/node
`robot_id`	`str`	`None`	Robot identifier, added to all data
`endpoint`	`str`	`http://localhost:4317`	Collector endpoint
`batch_size`	`int`	`512`	Events buffered before flush
`flush_interval`	`float`	`5.0`	Seconds between flushes
`enabled`	`bool`	`True`	Toggle collection on/off

Returns

BotPulse — the main client instance.

Tracing

`pulse.trace(name, attributes=None)`

Start a new trace and return the root span. Use as a context manager.

with pulse.trace("navigate_to_waypoint") as span:
    span.set_attribute("waypoint.x", 12.5)

Parameter	Type	Description
`name`	`str`	Operation name
`attributes`	`dict`	Optional initial attributes

`pulse.child_span(name, attributes=None)`

Start a child span within the current trace. Must be called inside a trace() block.

with pulse.trace("navigate") as root:
    with pulse.child_span("plan_path") as span:
        span.set_attribute("algorithm", "a_star")

`span.set_attribute(key, value)`

Set a key-value attribute on a span.

Value types: str, int, float, bool

`span.add_event(name, attributes=None)`

Add a timestamped event within the span.

span.add_event("gripper_opened", {"force_n": 15.0})

`span.set_status(code, description=None)`

Set the span status. Use "error" to mark a span as failed.

span.set_status("error", "Motor timeout")

`span.record_exception(exception)`

Record an exception on the span, capturing the full traceback.

try:
    ...
except Exception as e:
    span.record_exception(e)

`span.duration_ms`

Read-only property. Returns the span duration in milliseconds (available after the span closes).

Logging

`pulse.log(message, level="info", fields=None)`

Emit a structured log entry. Automatically correlated with the active trace/span.

pulse.log("Waypoint reached", level="info", fields={"x": 12.5, "y": 3.2})

Parameter	Type	Default	Description
`message`	`str`	required	Log message
`level`	`str`	`"info"`	One of: `debug`, `info`, `warning`, `error`, `critical`
`fields`	`dict`	`None`	Additional structured data

`pulse.log_exception(exception, message=None)`

Log an exception with its full traceback.

except SomeError as e:
    pulse.log_exception(e, message="Operation failed")

Metrics

`pulse.metric(name, value, attributes=None)`

Emit a metric with auto-detected type.

pulse.metric("latency_ms", 142.3)

`pulse.counter(name, value=1, attributes=None)`

Increment a counter.

pulse.counter("objects_detected", value=3)

`pulse.gauge(name, value, attributes=None)`

Set a gauge value (can go up or down).

pulse.gauge("battery_soc_percent", value=82.5)

`pulse.histogram(name, value, attributes=None)`

Record a value in a histogram distribution.

pulse.histogram("planning_duration_ms", value=45.2)

Common metric parameters

Parameter	Type	Description
`name`	`str`	Metric name (use snake_case with units: `latency_ms`)
`value`	`float`	The measurement value
`attributes`	`dict`	Optional dimensions for filtering

Lifecycle

`pulse.flush()`

Force an immediate flush of all buffered events to the collector.

pulse.flush()

`pulse.shutdown()`

Flush remaining events and release resources. Call this before your process exits.

pulse.shutdown()

Parameters​

Returns​

Tracing

pulse.trace(name, attributes=None)​

pulse.child_span(name, attributes=None)​

span.set_attribute(key, value)​

span.add_event(name, attributes=None)​

span.set_status(code, description=None)​

span.record_exception(exception)​

span.duration_ms​