roboto.domain.metrics.metric#

Module Contents#

class roboto.domain.metrics.metric.Metric(record, roboto_client)#

A summary value recorded for one session under a metric definition.

Each Metric stores exactly one value per (metric, session) pair. Calling publish() a second time for the same metric name and session_id replaces the previous value (upsert semantics). This makes metrics suitable for recording per-session summary statistics that are computed once (or updated as reprocessing happens), not for streaming time-series data.

Recording a metric (publish()) stores the value under the MetricDefinition with the given name.

Querying metrics (query()) returns the data points with a session timestamp in the given range. Aggregating metrics (aggregate()) groups sessions by the calendar period their stored timestamp falls into and applies a summary function (sum, mean, max, min, or count) across the values in each period.

Note

Metric instances should not be constructed directly. Obtain them via publish() or query().

Parameters:

record (roboto.domain.metrics.record.MetricRecord)
roboto_client (Optional[roboto.http.RobotoClient])

classmethod aggregate(name, period, aggregation, start_time, end_time, time_filter=MetricTimeFilter.EndTime, include_device_ids=NotSet, include_session_ids=NotSet, include_invocation_ids=NotSet, condition=None, group_by=None, owner_org_id=None, roboto_client=None)#

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.

Parameters:

name (str) – Name of the metric definition to aggregate.
period (roboto.domain.metrics.record.AggregationPeriod) – Calendar bucket size to group observations by.
aggregation (roboto.domain.metrics.record.NumericAggregation) – Function to apply to values in each bucket.
start_time (roboto.time.Time) – Inclusive start of the aggregation window. Accepts any Time value.
end_time (roboto.time.Time) – Exclusive end of the aggregation window. Same input shape as start_time.
time_filter (roboto.domain.metrics.record.MetricTimeFilter) – Whether to match the window against each session’s start time or end time. Defaults to end time.
include_device_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific device IDs, or None to match only rows with no device_id.
include_session_ids (Union[list[str], roboto.sentinels.NotSetType]) – Restrict to specific session IDs.
include_invocation_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific invocation IDs, or None to match only rows with no invocation_id.
condition (Optional[roboto.query.ConditionType]) – Restrict the aggregated data points to those whose session, producing device, or session’s collections match this Condition or ConditionGroup. It narrows what each bucket aggregates without moving the window’s period boundaries; a bucket left with no matching data points is omitted from the result. See condition for the accepted fields, and for how collection conditions evaluate when a session belongs to several collections or to none.
group_by (Optional[str]) – Split each period bucket by the distinct values of this field, one record per (period, value) pair. Accepts device.device_id and String, Enum, or Boolean custom fields on sessions and devices (session.custom.<name>, device.custom.<name>); any other field is rejected. Data points carrying no value for the field come back under a null group_key rather than being dropped. Defaults to no split.
owner_org_id (Optional[str]) – Organization that owns the metric data. Defaults to the authenticated caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

One NumericAggregateMetricRecord per period bucket that contains at least one observation, sorted by start_time ascending. Under group_by, one per (bucket, distinct value) pair instead, each naming its value in group_key.

Raises:

RobotoNotFoundException – No metric with this name exists in the organization.
RobotoIllegalArgumentException – condition references a field or comparator the request does not accept; condition or group_by references a custom field that is either undefined in your organization or defined but not in the Ready state; or group_by names a field that cannot be a series.

Return type:

list[roboto.domain.metrics.record.NumericAggregateMetricRecord]

Examples

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)

property device_id: str | None#

Return type:: Optional[str]

classmethod get_by_session(session_id, roboto_client=None)#

Return every metric published to session_id.

Parameters:

session_id (str) – Session whose metrics to fetch.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

One Metric per matching (metric_definition, session) pair. May be empty. Order is unspecified.

Return type:

list[Metric]

Examples

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

property group_key: str | None#

Return type:: Optional[str]

property invocation_id: str | None#

Return type:: Optional[str]

property max_timestamp_ns: int | None#

Return type:: Optional[int]

property metric_id: str#

Return type:: str

property min_timestamp_ns: int | None#

Return type:: Optional[int]

property name: str#

Return type:: str

property org_id: str#

Return type:: str

classmethod publish(session_id, metrics, device_id=NotSet, caller_org_id=None, roboto_client=None)#

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.

Parameters:

session_id (str) – Session to attach every published value to.
metrics (list[roboto.domain.metrics.record.MetricEntry]) – Metric names and values to record. An empty list returns an empty response without contacting the platform.
device_id (Union[roboto.sentinels.NotSetType, Optional[str]]) – Device to associate with every published value, or None to associate none. When omitted, Roboto infers the device from the session’s attached devices, which succeeds only when exactly one device is attached.
caller_org_id (Optional[str]) – Organization context for the request. Defaults to the authenticated caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

A BatchResponse holding one element per entry, in request order, carrying either the recorded Metric or the exception the platform refused that entry with. An entry whose value is not finite, or whose name uses a character other than an ASCII letter, a digit, -, ., _, or ~, is refused on its own; the remaining entries are recorded. A database error while recording stores none of the entries and is reported on every entry not already refused.

Raises:

RobotoNotFoundException – session_id does not exist in the caller’s organization.
RobotoInvalidRequestException – device_id was omitted and the session has no attached device or more than one.

Return type:

roboto.http.BatchResponse[Metric]

Examples

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,
... )

property published: datetime.datetime#

Return type:: datetime.datetime

property published_by: str#

Return type:: str

classmethod query(name, start_time=None, end_time=None, time_filter=MetricTimeFilter.EndTime, max_results=MAX_METRIC_LIST_RESULTS, descending=False, include_device_ids=NotSet, include_session_ids=NotSet, include_invocation_ids=NotSet, condition=None, group_by=None, owner_org_id=None, roboto_client=None, sort_by=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.

Parameters:

name (str) – Name of the metric definition to query.
start_time (Optional[roboto.time.Time]) – Inclusive start of the query window. Accepts any Time value (int Unix-epoch nanoseconds, datetime, ISO 8601 string, decimal seconds, etc.). Defaults to None (the Unix epoch).
end_time (Optional[roboto.time.Time]) – Exclusive end of the query window. Same input shape as start_time. Defaults to None (now).
time_filter (roboto.domain.metrics.record.MetricTimeFilter) – Whether to match the window against the session’s start time or end time. Defaults to end time.
max_results (int) – Page size — number of data points per HTTP request. Total results are unbounded; pagination is automatic.
descending (bool) – Yield the largest sort_by values first instead of the smallest, so with the default sort_by the most recent sessions come first. Applies across the whole result set, not just within a page. Data points with no device_id or invocation_id then sort before every other value.
include_device_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific device IDs, or None to match only rows with no device_id.
include_session_ids (Union[list[str], roboto.sentinels.NotSetType]) – Restrict to specific session IDs.
include_invocation_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific invocation IDs, or None to match only rows with no invocation_id.
condition (Optional[roboto.query.ConditionType]) – Restrict to data points whose session, producing device, or session’s collections match this Condition or ConditionGroup. See condition for the accepted fields, and for how collection conditions evaluate when a session belongs to several collections or to none.
group_by (Optional[str]) – Field whose value each yielded data point carries under group_key, for separating the points into a series per distinct value. Does not change which data points are returned; see group_by for the accepted fields.
owner_org_id (Optional[str]) – Organization that owns the metric data. Defaults to the authenticated caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.
sort_by (Optional[str]) – Field to order the data points by. See sort_by for the accepted fields. Defaults to the session time selected by time_filter.

Yields:

One Metric per matching session, sorted by sort_by — ascending by default, descending when descending is set — with session_id as a deterministic tiebreaker.

Raises:

RobotoNotFoundException – No metric with this name exists in the organization.
RobotoIllegalArgumentException – condition or group_by references a field or comparator the request does not accept, or a custom field that is either undefined in your organization or defined but not in the Ready state.

Return type:

collections.abc.Generator[Metric, None, None]

Examples

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)

property record: roboto.domain.metrics.record.MetricRecord#

Return type:: roboto.domain.metrics.record.MetricRecord

property session_id: str#

Return type:: str

property unit: str | None#

Return type:: Optional[str]

property value: float#

Return type:: float

class roboto.domain.metrics.metric.MetricDefinition(record, roboto_client)#

A named schema for a metric tracked across sessions and devices.

Metric definitions are org-scoped schemas that describe a single measurable quantity. They act as the registry entry that all Metric data points reference. Every metric definition has a unique name within an organization, and an optional human-readable description.

Metric definitions are created once per org and reused across many sessions. Use create() to register a definition the first time, and update() to change its description later. for_org() lists all definitions that belong to an organization.

Note

MetricDefinition instances should not be constructed directly. Always obtain them via create(), get(), or for_org().

Parameters:

record (roboto.domain.metrics.record.MetricDefinitionRecord)
roboto_client (Optional[roboto.http.RobotoClient])

classmethod create(name, description=None, unit=None, caller_org_id=None, roboto_client=None)#

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

Parameters:

name (str) – Unique metric name. Must contain only URL-safe characters (A–Z, a–z, 0–9, -, ., _, ~). Dots are conventional namespace separators, e.g. cpu.usage_pct.
description (Optional[str]) – Optional human-readable description of what the metric measures.
unit (Optional[str]) – Optional unit of measure for values recorded under this metric, e.g. "%", "ms", "m/s". Free-form and unvalidated. Omit for a unitless metric.
caller_org_id (Optional[str]) – Organization to create the definition in. Defaults to the authenticated caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

The newly created MetricDefinition.

Raises:

RobotoConflictException – A definition with this name already exists in the organization.

Return type:

MetricDefinition

Examples

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

delete()#

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

Warning

This operation is irreversible. All Metric data points recorded under this name will be permanently removed.

Examples

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

Return type:: None

property description: str | None#

Return type:: Optional[str]

classmethod for_org(owner_org_id, roboto_client=None)#

Yield all metric definitions belonging to an organization.

Parameters:

owner_org_id (str) – Organization that owns the metric definitions to enumerate.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Yields:

Each MetricDefinition belonging to owner_org_id.

Return type:

collections.abc.Generator[MetricDefinition, None, None]

Examples

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

classmethod get(name, owner_org_id=None, roboto_client=None)#

Retrieve an existing metric definition by name.

Parameters:

name (str) – Name of the metric definition to retrieve. Must match exactly (case-sensitive) the name used when the definition was created.
owner_org_id (Optional[str]) – Organization that owns the definition. Defaults to the authenticated caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

The MetricDefinition with the given name.

Raises:

RobotoNotFoundException – No definition with this name exists in the organization.

Return type:

MetricDefinition

Examples

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

property metric_id: str#

Return type:: str

property name: str#

Return type:: str

property org_id: str#

Return type:: str

property unit: str | None#

Return type:: Optional[str]

update(description=NotSet, unit=NotSet)#

Update the mutable attributes of this definition.

Parameters:

description (Optional[Union[roboto.sentinels.NotSetType, str]]) – New human-readable description, None to clear, or NotSet to leave unchanged.
unit (Optional[Union[roboto.sentinels.NotSetType, str]]) – New unit of measure, None to clear, or NotSet to leave unchanged.

Return type:

None

Examples

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