Metrics

class pyniryo_nate.components.Metrics(http_client: HttpClient, mqtt_client: MqttClient, correlation_id: str)

Bases: BaseAPIComponent

Metrics component for declaring and publishing custom metrics.

This component allows you to define custom metrics that can be monitored and tracked during robot operations. Metrics are automatically published when their values change.

Methods

close	Close the component and release any resources if needed.
get_metrics	Retrieve custom metrics by their metrics ID.
on_api_metrics	Subscribe to the metrics the robot publishes about the requests its API serves, one message per request.
on_system_metrics	Subscribe to the device's system metrics (host resources and Docker container/volume metrics).
register_metrics	Register custom metrics from an object instance.

Parameters:

• http_client (HttpClient)

• mqtt_client (MqttClient)

• correlation_id (str)

register_metrics(inst: object) → None

Register custom metrics from an object instance.

This method inspects the provided object for Metric descriptors and registers them with the system. After registration, any changes to metric values will be automatically published.

Parameters:

inst (object) – An object instance containing Metric descriptors.

Raises:

DataValidationError – If the correlation id is not one the robot takes as a metrics id.

Return type:

None

on_system_metrics(callback: Callable[[SystemMetrics], None]) → Callable[[], None]

Subscribe to the device’s system metrics (host resources and Docker container/volume metrics).

Parameters:

callback (Callable[[SystemMetrics], None]) – The callback function that takes a SystemMetrics parameter.

Return type:

Callable[[], None]

on_api_metrics(callback: Callable[[dict[str, Any]], None]) → Callable[[], None]

Subscribe to the metrics the robot publishes about the requests its API serves, one message per request. The specification leaves the payload as a bare object, so it is passed as a dict.

Parameters:

callback (Callable[[dict[str, Any]], None]) – The callback function that takes the payload as a dict.

Return type:

Callable[[], None]

get_metrics(metrics_id: str) → list[Metric]

Retrieve custom metrics by their metrics ID.

Parameters:

metrics_id (str) – The ID of the metrics to retrieve.

Returns:

A list of Metric objects.

Return type:

list[Metric]