Robot

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

class pymirokai.robot.BaseRobot(connection, jwt_token, retry_interval=5, api_version='v1', loglevel=None)

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:

• connection (Connection)

• jwt_token (str)

• retry_interval (int)

• api_version (str)

• loglevel (str)

retry_interval

Interval between connection retry attempts.

Type:

int

loop

Event loop for asynchronous operations.

Type:

asyncio.AbstractEventLoop | None

thread

Thread for running asynchronous tasks.

Type:

threading.Thread | None

connection

Manages the connection state and operations.

Type:

Connection | None

mission_id

The current mission_id attached to the robot instance.

Type:

str | None

cancel()

Cancel the connection tasks and stop monitoring APIs.

Return type:

None

async cancel_and_complete()

Cancel all active missions and shut down API connections.

Stops REST and WebSocket tasks and closes all active sessions gracefully.

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_mission(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.

cancel_missions_and_wait_for_cancelation()

Cancel a mission and wait for its completion.

Return type:

dict[str, Any]

Returns:

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

async completed()

Wait for the completion of all connection tasks.

Return type:

None

enable_skill_file(skill_name, enable)

Enable or disable a skill on the robot.

Parameters:

• skill_name (str) – The name of the skill to enable or disable.

• enable (bool) – Whether to enable (True) or disable (False) the skill.

Return type:

dict[str, Any]

Returns:

dict – The response from the REST API indicating the operation status.

enqueue_mission(skill, priority, restart_after_cancellation=None, log_level=None)

Queue a skill for execution with a specified priority.

If the skill’s priority allows it to run immediately, it will start. Otherwise, it will be queued according to the priority system.

Parameters:

• skill (str) – The skill object to enqueue. This should be serializable and understandable

• system. (by the mission)

• priority (str) – String representing the priority level under which the skill should be classified.

• restart_after_cancellation (bool, optional) – Whether the skill can restart if previously cancelled.

• True. (Defaults to)

• log_level (int, optional) – Logging level to associate with this skill. Must be one of the following logging constants: - logging.CRITICAL, logging.ERROR, logging.WARNING, - logging.INFO, logging.DEBUG, logging.NOTSET

Return type:

dict[str, Any]

Returns:

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

get_available_priorities()

Get the list of available priorities.

Return type:

dict[str, Any]

Returns:

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

get_missions()

Get the list of running skills.

Return type:

dict[str, Any]

Returns:

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

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.

list_skill_files()

Retrieve a list of skills available on the robot.

Return type:

dict[str, Any]

Returns:

dict – A list of skills from the REST API.

make_priority_level(priority_level_name, below, mission_conflict_strategy)

Send a ‘make_priority_level’ command to the robot to create a new priority level.

Parameters:

• priority_level_name (str) – The name of the new priority level.

• below (str) – The name of the existing priority level below which the new one will be inserted.

• mission_conflict_strategy (PreemptionStrategy) – Strategy to apply when two missions with the same

• conflict. (priority are in)

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

remove_skill_file(skill_name)

Remove a skill file from the robot.

Parameters:

skill_name (str) – The name of the skill to remove.

Return type:

dict[str, Any]

Returns:

dict – The response from the REST API indicating the removal status.

send_boolean(command, value)

Send a boolean command to the REST API.

Parameters:

• command (str) – The command to send.

• value (bool) – The boolean value to send.

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 REST API.

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 REST API.

Parameters:

• command (str) – The command to send.

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

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 REST API.

Parameters:

• command (str) – The command to send.

• value (float) – The float value to send.

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 REST API.

Parameters:

• command (str) – The command to send.

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

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 REST API.

Parameters:

• command (str) – The command to send.

• value (str) – The string value to send.

Return type:

dict[str, Any]

Returns:

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

skill(skill, **kwargs)

Trigger a robot skill via the REST API.

Parameters:

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

• **kwargs (Any) – Optional parameters to pass to the skill.

Return type:

dict[str, Any]

Returns:

A dictionary containing the API response and mission ID.

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

upload_skill_file(skill_path)

Upload a skill file to the robot.

Parameters:

skill_path (str) – The file path of the skill to upload.

Return type:

dict[str, Any]

Returns:

dict – The response from the REST API indicating the upload status.

wait_for_mission(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_mission_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(connection, jwt_token, retry_interval=5, api_version='v1', loglevel=None)

Bases: SkillWrapper, BaseRobot, Animate, Behavior, Freeze, Generic, Interact, Interaction, Manipulate, Navigate, Navigation, Planning, 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.

• connection (Connection)

• jwt_token (str)

pymirokai.robot.connect(api_key='', ip=None, timeout=None)

Create and connect a new Robot instance.

Establishes REST, WebSocket, and video connections either via local UDS or remote IP depending on the provided parameters.

Parameters:

• api_key (str) – API key for authentication.

• ip (Optional[str]) – Target robot IP address (if omitted, uses UDS).

• timeout (Optional[float]) – Optional connection timeout in seconds.

Return type:

Robot

Returns:

A connected Robot instance ready for async context management.

pymirokai.robot.initialize_coloredlogs(loglevel=None)

Configure colored log output.

Initializes colored log formatting with timestamps and levels.

Parameters:

loglevel (Optional[int]) – Numeric log level (e.g., logging.INFO). Defaults to env PYLOG_LEVEL or INFO.

Return type:

None