roboto.domain.metrics.metric

Aggregate a metric across sessions, grouped by calendar period.

Sessions whose session_min_timestamp_ns or session_max_timestamp_ns (selected via time_filter) falls inside the [start_time, end_time) window are grouped into UTC calendar buckets sized by period, and the chosen NumericAggregation is applied to the values in each bucket.

The server snaps the requested window outward to whole-period boundaries to guarantee apples-to-apples comparisons. All time period buckets always cover their complete calendar period. For example, * a monthly aggregation requested between Jan 15 – Mar 15 will return aggregated data for all of January, February, and March. * a quarterly aggregation from Apr 27 - Dec 28 will return aggregated data for all of Q2, Q3, and Q4.

Usage

Daily max CPU usage over a month, passing datetime directly:

import datetime
from roboto.domain.metrics import (
    AggregationPeriod,
    Metric,
    NumericAggregation,
)
for bucket in Metric.aggregate(
    name="cpu.usage_max",
    period=AggregationPeriod.Daily,
    aggregation=NumericAggregation.Max,
    start_time=datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc),
    end_time=datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc),
):
    print(bucket.start_time, bucket.value)

The same aggregation over data points published from a device in the delivery fleet. The Device custom field fleet must be defined by your organization and moved to Ready; the aggregation raises if it has not been. A data point published without a device carries no fleet, so Equals drops it, and NotEquals drops it too: only IsNull and NotExists match a data point with no device. See condition for the full field and comparator rules:

from roboto.query import Comparator, Condition
delivery_buckets = Metric.aggregate(
    name="cpu.usage_max",
    period=AggregationPeriod.Daily,
    aggregation=NumericAggregation.Max,
    start_time="2026-05-01T00:00:00Z",
    end_time="2026-06-01T00:00:00Z",
    condition=Condition(
        field="device.custom.fleet",
        comparator=Comparator.Equals,
        value="delivery",
    ),
)
for bucket in delivery_buckets:
    print(bucket.start_time, bucket.end_time, bucket.value, bucket.total)

One line per robot rather than one line for the fleet. Buckets aggregating data points published without a device come back with group_key set to None:

per_device = Metric.aggregate(
    name="cpu.usage_max",
    period=AggregationPeriod.Daily,
    aggregation=NumericAggregation.Max,
    start_time="2026-05-01T00:00:00Z",
    end_time="2026-06-01T00:00:00Z",
    group_by="device.device_id",
)
for bucket in per_device:
    print(bucket.group_key, bucket.start_time, bucket.value)

Return every metric published to session_id.

Usage

from roboto.domain.metrics import Metric
for m in Metric.get_by_session("ss_abc123"):
    print(m.metric_id, m.value)

Record metric values for a session in a single network call.

Each (metric, session) pair is upserted: republishing under the same name and session_id replaces the previous value. Repeating a metric name within one call stores the last value given for it that the platform accepted, and every accepted entry naming that metric reports the stored value.

A metric definition is created for any name the org does not already have one for. When called from within a Roboto action, every recorded value is linked to that action invocation.

Usage

Publish with an explicit device:

from roboto.domain.metrics import Metric, MetricEntry
published = Metric.publish(
    session_id="ss_abc123",
    metrics=[MetricEntry(name="cpu.usage_max", value=87.2)],
    device_id="robot01",
)
len(published.succeeded)
# 1

Let the server infer the device from the session’s single attached device:

Metric.publish(
    session_id="ss_abc123",
    metrics=[MetricEntry(name="memory.peak_mb", value=2048.0)],
)

Record values that are not tied to any device:

Metric.publish(
    session_id="ss_abc123",
    metrics=[MetricEntry(name="run.duration_s", value=42.0)],
    device_id=None,
)

Yield stored metric values whose session time falls in a range.

The time window is matched against either session_min_timestamp_ns or session_max_timestamp_ns on each metric row depending on time_filter.

This method auto-paginates: max_results is the page size (capped at MAX_METRIC_LIST_RESULTS), not a total result cap. The generator continues fetching pages until the server reports no more data.

Usage

Query a metric over a single day, passing datetime directly:

import datetime
from roboto.domain.metrics import Metric
for m in Metric.query(
    name="cpu.usage_max",
    start_time=datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc),
    end_time=datetime.datetime(2026, 5, 2, tzinfo=datetime.timezone.utc),
):
    print(m.session_id, m.value)

Or with an ISO 8601 string:

all_records = list(
    Metric.query(
        name="cpu.usage_max",
        start_time="2026-05-01T00:00:00Z",
        end_time="2026-05-02T00:00:00Z",
    )
)

Take just the 10 most recent sessions:

import itertools
recent = list(itertools.islice(Metric.query(name="cpu.usage_max", descending=True), 10))

Take the 10 sessions with the highest value:

highest = list(
    itertools.islice(
        Metric.query(name="cpu.usage_max", sort_by="value", descending=True),
        10,
    )
)

Restrict to data points from production-tagged sessions that either ran in the EMEA region or were produced by a device at the Berlin site. Both custom fields, region on Sessions and site on Devices, must be defined by your organization and moved to Ready; the query raises if either has not been:

from roboto.query import Comparator, Condition, ConditionGroup, ConditionOperator
for m in Metric.query(
    name="cpu.usage_max",
    condition=ConditionGroup(
        operator=ConditionOperator.And,
        conditions=[
            Condition(
                field="session.tags",
                comparator=Comparator.Contains,
                value="production",
            ),
            ConditionGroup(
                operator=ConditionOperator.Or,
                conditions=[
                    Condition(
                        field="session.custom.region",
                        comparator=Comparator.Equals,
                        value="emea",
                    ),
                    Condition(
                        field="device.custom.site",
                        comparator=Comparator.Equals,
                        value="berlin",
                    ),
                ],
            ),
        ],
    ),
):
    print(m.session_id, m.value)

Separate the data points by the device that published them:

from collections import defaultdict
by_device = defaultdict(list)
for m in Metric.query(name="cpu.usage_max", group_by="device.device_id"):
    by_device[m.group_key].append(m.value)

Create a new metric definition in the caller’s organization.

Usage

MetricDefinition.create(
    name="cpu.usage_max",
    description="Peak CPU usage recorded during the session.",
    unit="%",
)

Delete this metric definition and all of its associated data points.

Usage

definition = MetricDefinition.get("cpu.usage_max")
definition.delete()

Yield all metric definitions belonging to an organization.

Usage

for definition in MetricDefinition.for_org("og_myorg"):
    print(definition.name, "-", definition.description)

Retrieve an existing metric definition by name.

Usage

definition = MetricDefinition.get("cpu.usage_max")

Update the mutable attributes of this definition.

Usage

definition = MetricDefinition.get("cpu.usage_max")
definition.update(description="Peak CPU usage recorded during the session.", unit="%")