Robot

This module contains the Robot class which integrates all features and handles operations.

class pymirokai.robot.BaseRobot(retry_interval=5, api_version='v1', loglevel='INFO')

Bases: object

Base class for the robot, handling the integration of various features and API interactions.

This class manages the connection to both REST and WebSocket APIs, allowing for the execution of skills and management of robot features.

Parameters:

  • retry_interval (int)

  • api_version (str)

  • loglevel (str)

retry_interval

Interval between connection retry attempts.

Type:

int

rest_api

Instance for REST API communication.

Type:

RestAPI

websocket_api

Instance for WebSocket API communication.

Type:

WebSocketAPI

running

Event to control the running state of the robot.

Type:

asyncio.Event

loop

Event loop for asynchronous operations.

Type:

Optional[asyncio.AbstractEventLoop]

thread

Thread for running asynchronous tasks.

Type:

Optional[threading.Thread]

connection

Manages the connection state and operations.

Type:

Optional[BaseRobot.Connection]

class Connection(robot, rest_api_url, websocket_api_url, video_server_url, api_key)

Bases: object

Handles the robot’s connection to the REST and WebSocket APIs.

Manages the start, stop, and monitoring of connections to the robot’s APIs.

Parameters:

  • robot (BaseRobot)

  • rest_api_url (str)

  • websocket_api_url (str)

  • video_server_url (str)

  • api_key (str)

robot

The parent robot instance.

Type:

BaseRobot

rest_api_url

URL for the REST API.

Type:

str

websocket_api_url

URL for the WebSocket API.

Type:

str

rest_api

Task for managing REST API connection.

Type:

Optional[asyncio.Task]

websocket_api

Task for managing WebSocket API connection.

Type:

Optional[asyncio.Task]

finished

Future indicating the completion of connection tasks.

Type:

asyncio.Future

cancel()

Cancel the connection tasks and stop monitoring APIs.

Return type:

None

async cancel_and_complete()

Cancel the connection tasks and ensure completion.

Return type:

None

async completed()

Wait for the completion of all connection tasks.

Return type:

None

async connected()

Wait for the robot to connect to both REST and WebSocket APIs.

Return type:

Connection

Returns:

BaseRobot.Connection – The connection instance, indicating successful connection.

get_jwt_token()

Retrieve the latest JWT token.

Return type:

str

async run()

Start monitoring and managing data fetching from the APIs.

Return type:

None

async run_without_cancelled_error()

Run the connection tasks, handling cancellation errors gracefully.

Return type:

None

set_jwt_token(token)

Update the shared JWT token.

Parameters:

token (str)

cancel_and_wait_for_mission(id_mission)

Cancel a mission and wait for its completion.

Parameters:

id_mission (str) – The mission ID to cancel.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API indicating whether the mission was cancelled.

cancel_client_missions()

Cancel a mission and wait for its completion. :rtype: Dict[str, Any] :returns: Dict[str, Any] – The response from the REST API indicating whether the mission was cancelled.

Return type:

Dict[str, Any]

cancel_skill(id_mission)

Cancel a specific skill.

Parameters:

id_mission (str) – The mission ID to cancel.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API indicating whether the mission was cancelled.

connect(ip, api_key)

Initiate the connection to the robot’s APIs.

Starts monitoring and fetching data from the REST and WebSocket APIs.

Parameters:

  • ip (str) – Ip of the robot.

  • api_key (str) – API key of the robot

Return type:

Connection

Returns:

BaseRobot.Connection – The connection instance managing the APIs.

get_data()

Get the latest data from the WebSocket.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The latest data received from the WebSocket.

get_service_log(system, service)

Get logs for a specific service.

Parameters:

  • system (str) – The system to get the service logs from (e.g., ‘head’, ‘base’).

  • service (str) – The service name.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The logs from the specified service.

get_services_list()

Get the list of available services for logs.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The list of available services.

get_skills()

Get the list of available skills.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

publish(topic, data)

Publish data to a WebSocket topic.

Parameters:

  • topic (str) – The topic to publish to.

  • data (str) – The data to publish (JSON format).

Return type:

None

register_callback(topic, callback)

Register a callback function for a WebSocket topic.

Parameters:

  • topic (str) – The topic to register the callback for.

  • callback (Callable[[Dict[str, Any]], None]) – The callback function to call when a

  • topic. (message is received for the)

Return type:

None

send_boolean(command, value)

Send a boolean command to the robot.

Parameters:

  • command (str) – The command to send.

  • value (bool) – The boolean data.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

send_command(command)

Send a simple command with no arguments to the robot.

Parameters:

command (str) – The command to send.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

send_dict(command, value)

Send a dictionary command to the robot.

Parameters:

  • command (str) – The command to send.

  • value (Dict[str, Any]) – The dictionary data.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

send_float(command, value)

Send a float command to the robot.

Parameters:

  • command (str) – The command to send.

  • value (float) – The float data.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

send_int(command, value)

Send an integer command to the robot.

Parameters:

  • command (str) – The command to send.

  • value (int) – The integer data to send.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

async send_int_async(command, value)

Send an integer command to the robot asynchronously.

Parameters:

  • command (str) – The command to send.

  • value (int) – The integer data to send.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

send_list(command, value)

Send a list command to the robot.

Parameters:

  • command (str) – The command to send.

  • value (list[Any]) – The list data.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

send_string(command, value)

Send a string command to the robot.

Parameters:

  • command (str) – The command to send.

  • value (str) – The string data.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

skill(skill, **kwargs)

Trigger a skill on the robot.

Parameters:

  • skill (str) – The name of the skill to trigger.

  • **kwargs (Any) – Additional arguments for the skill.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API containing the command and mission ID.

skills_activation(activate)

Activate or deactivate skills.

Parameters:

activate (bool) – Whether to activate (True) or deactivate (False) skills.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API.

subscribe(topic)

Subscribe to a WebSocket topic.

Parameters:

topic (str) – The topic to subscribe to.

Return type:

None

unsubscribe(topic)

Unsubscribe from a WebSocket topic.

Parameters:

topic (str) – The topic to unsubscribe from.

Return type:

None

wait_for_skill(id_mission)

Wait for a specific skill to complete.

Parameters:

id_mission (str) – The mission ID to wait for.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API indicating whether the mission was awaited.

wait_for_skill_started(id_mission)

Wait for a skill to start.

Parameters:

id_mission (str) – The mission ID to wait for.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The response from the REST API indicating whether the mission was started.

class pymirokai.robot.Robot(retry_interval=5, api_version='v1', loglevel='INFO')

Bases: BaseRobot, Animate, Behavior, Freeze, Generic, Interact, Interaction, Manipulate, Navigate, Navigation, Runes, ShowOff, System, Transport

Robot class that integrates all features and skills.

This class extends BaseRobot and includes abilities and skills such as animation, behavior, interaction, manipulation, navigation, runes, and more, providing a comprehensive interface for interacting with the robot.

Parameters:

  • retry_interval (int) – Time interval between retry attempts for API connections.

  • api_version (str) – API version to use for communication.

  • loglevel (str) – The level of logging detail.

pymirokai.robot.connect(ip, api_key)

Initiate the connection to the robot’s APIs.

Starts monitoring and fetching data from the REST and WebSocket APIs.

Parameters:

  • ip (str) – Ip of the robot.

  • api_key (str) – API key of the robot

Return type:

Connection

Returns:

BaseRobot.Connection – The connection instance managing the APIs. Do a async with of that object to have the connected robot object.

pymirokai.robot.initialize_coloredlogs(loglevel='DEBUG')

Initialize logging with colored logs.

This function sets up the logging configuration using the coloredlogs library, which provides colored output for log messages.

Parameters:

loglevel (str) – The logging level to use (e.g., ‘DEBUG’, ‘INFO’, ‘WARNING’).

Return type:

None