Device

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

Bases: BaseAPIComponent

Device component for accessing robot device information and control.

This component provides methods for: - Getting the device identity, version and aggregated status - Checking device health and readiness - Setting the clock and the NTP configuration - Following and starting updates of the embedded software - Rebooting or shutting down the device - Reading the API specifications the robot serves

Methods

check_update	Ask the fleet now for a newer release, instead of waiting for the robot's next question.
clear_system	Free disk space by pruning unused Docker resources.
close	Close the component and release any resources if needed.
get_asyncapi_spec	Get the AsyncAPI specification of the MQTT API the robot serves, in YAML.
get_asyncapi_ui	Get the interactive documentation page of the MQTT API, as HTML.
get_datetime	Get the current date and time of the device.
get_identity	Get the identity of the robot: its id, name, model, arm serial number and the software release it runs.
get_ntp_config	Get the device's NTP configuration.
get_openapi_spec	Get the OpenAPI specification of the REST API the robot serves, in YAML.
get_openapi_ui	Get the interactive documentation page of the REST API, as HTML.
get_status	Get the aggregated status of the robot: whether it can be used right now, who holds it, and which robot it is.
get_update	Get where the robot stands on updating its embedded software.
get_version	Get the version and commit of the API binary.
id	Get the device ID of the robot.
is_healthy	Check if the robot device is healthy.
is_ready	Check if the API service is ready.
on_update_state	Set the callback to call when the update state changes.
reboot	Reboot the robot device.
set_datetime	Set the device's date and time.
set_ntp_config	Update the device's NTP configuration.
shutdown	Shutdown the robot device.
start_update	Install a release of the embedded software.
sync_clock	Synchronize the device's clock with the configured NTP server(s).

Parameters:

• http_client (HttpClient)

• mqtt_client (MqttClient)

• correlation_id (str)

id() → str

Get the device ID of the robot. :return: The device ID of the robot.

Return type:

str

get_identity() → DeviceIdentity

Get the identity of the robot: its id, name, model, arm serial number and the software release it runs.

Returns:

The identity of the robot.

Return type:

DeviceIdentity

is_healthy() → bool

Check if the robot device is healthy.

Returns:

True if the robot is healthy, False otherwise.

Return type:

bool

is_ready() → bool

Check if the API service is ready. This says nothing about the arm: see get_status() for whether the robot can be used right now.

Returns:

True if the service is ready, False otherwise.

Return type:

bool

get_version() → DeviceVersion

Get the version and commit of the API binary. Answers without a token.

Returns:

The version of the API binary.

Return type:

DeviceVersion

get_status() → DeviceStatus

Get the aggregated status of the robot: whether it can be used right now, who holds it, and which robot it is. Answers without a token.

Returns:

The aggregated status of the robot.

Return type:

DeviceStatus

get_update() → DeviceUpdate

Get where the robot stands on updating its embedded software. The robot answers from what it last knew and never waits on the network: a robot cut off from its fleet answers with stale set. The same state goes out on on_update_state().

Returns:

The update state.

Raises:

ServerError – If this robot does not manage its own software (code == 'updates_unavailable'). A client meeting this stops offering updates rather than retrying.

Return type:

DeviceUpdate

start_update(release_id: str) → None

Install a release of the embedded software. The call returns as soon as the robot has accepted the order. The state then reads DOWNLOADING, with a progress, then APPLYING, during which the robot stops answering for minutes. Conclude on reconnection, where the state reads SUCCESS or ERROR.

Nothing queues: a refused call has to be made again.

Parameters:

release_id (str) – The release to install, the target.release_id of get_update(). Asking for a release the fleet no longer offers is refused rather than served a different one.

Raises:

• ClientError – If a program runs or the arm moves (code == 'execution_running'), an update is already under way (code == 'update_running'), or the release is not the one the fleet offers (code == 'release_not_available').

• ServerError – If the robot cannot go through its fleet (code is 'updates_unavailable' or 'fleet_unreachable').

Return type:

None

check_update() → None

Ask the fleet now for a newer release, instead of waiting for the robot’s next question. The call returns at once and the answer lands on get_update(): the state reads CHECKING until the robot knows. A call made while the robot is already asking asks nothing again.

Raises:

ServerError – If the robot cannot go through its fleet (code is 'updates_unavailable' or 'fleet_unreachable').

Return type:

None

on_update_state(callback: Callable[[DeviceUpdate], None]) → Callable[[], None]

Set the callback to call when the update state changes. The state is retained, so the callback receives the current one as soon as the subscription is in place. The broker goes silent while a release is applied: reconnect and read the verdict then.

Parameters:

callback (Callable[[DeviceUpdate], None]) – The callback function that takes the new DeviceUpdate.

Return type:

Callable[[], None]

get_openapi_spec() → str

Get the OpenAPI specification of the REST API the robot serves, in YAML.

Return type:

str

get_asyncapi_spec() → str

Get the AsyncAPI specification of the MQTT API the robot serves, in YAML.

Return type:

str

get_openapi_ui() → str

Get the interactive documentation page of the REST API, as HTML.

Return type:

str

get_asyncapi_ui() → str

Get the interactive documentation page of the MQTT API, as HTML.

Return type:

str

get_datetime() → datetime

Get the current date and time of the device.

Returns:

The device’s current datetime.

Return type:

datetime

set_datetime(dt: datetime) → None

Set the device’s date and time.

Note: This will fail if NTP synchronization is enabled.

Parameters:

dt (datetime) – The datetime to set on the device, timezone-aware.

Raises:

DataValidationError – If dt carries no timezone.

Return type:

None

get_ntp_config() → NtpConfig

Get the device’s NTP configuration.

Returns:

The current NTP configuration.

Return type:

NtpConfig

set_ntp_config(enabled: bool | None = None, servers: list[str] | None = None, pools: list[str] | None = None) → NtpConfig

Update the device’s NTP configuration. The robot takes the whole configuration at once, so a parameter left to None is read from the current configuration first and sent back unchanged.

Parameters:

• enabled (bool | None) – Whether the device synchronizes its clock with NTP.

• servers (list[str] | None) – The NTP servers to synchronize with.

• pools (list[str] | None) – The NTP pools to get servers from.

Returns:

The updated NTP configuration.

Return type:

NtpConfig

sync_clock() → None

Synchronize the device’s clock with the configured NTP server(s).

Return type:

None

clear_system() → None

Free disk space by pruning unused Docker resources. The call returns once the prune has started.

Return type:

None

reboot() → None

Reboot the robot device. The call returns once the robot has accepted the order.

Note: The connection will be lost during the reboot process.

Return type:

None

shutdown() → None

Shutdown the robot device.

Note: The device will need to be manually powered on after shutdown.

Return type:

None