Core Modules

This module contains the RestAPI class for communicating with the REST API.

class pymirokai.core.rest_api.RestAPI(api_version='v1', retry_interval=5, uds_path='/tmp/rest_api.sock', use_uds=False)

Bases: object

Class to handle REST API communication, supporting both Unix Domain Socket (UDS) and TCP/IP.

Parameters:

• api_version (str)

• retry_interval (int)

• uds_path (str)

• use_uds (bool)

async authentify()

Retrieve the JWT token for authentication. - If UDS, assume a default local authentication. - If TCP, authenticate normally.

Return type:

None

async close()

Close the REST API connection.

Return type:

None

async connect()

Check the connection status of the REST API and update the status.

Return type:

None

async enable_skill_file(skill_name, enable)

Enable or disable a skill on the robot.

Parameters:

• skill_name (str) – Name of the skill.

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

Return type:

dict

Returns:

dict – The response from the API.

async get_data_and_update_token_and_client_id(response)

From a response from the API, recursively parse JSON, update client_id and JWT token.

Return type:

dict

async get_services(system=None, service=None)

Either get the list of available services, or the service logs of a specific service.

Parameters:

• system (Optional[str]) – The system to get a specific service (head/base).

• service (Optional[str]) – The service name.

Return type:

Dict[str, Any]

Returns:

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

async list_skill_files()

Retrieve the list of uploaded skills.

Return type:

dict

Returns:

dict – The list of skills from the API.

async remove_skill_file(skill_name)

Remove a skill from the robot.

Parameters:

skill_name (str) – Name of the skill to remove.

Return type:

dict

Returns:

dict – The response from the API.

async send_command(endpoint, model)

Send a command to the specified endpoint.

Parameters:

• endpoint (str) – The endpoint to send the command to.

• model (BaseModel) – The command model.

Return type:

dict

Returns:

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

async setup_connection()

Get the JWT token and client_id, and retry until connected.

async start(api_url, api_key, set_jwt_token=None, get_jwt_token=None, use_uds=None)

Start the REST API session and connection check.

Parameters:

• api_url (str) – The API URL (ignored if UDS is used).

• api_key (str) – API key (for TCP only).

• set_jwt_token (callable) – Callback to set the JWT token.

• get_jwt_token (callable) – Callback to get the JWT token.

• use_uds (bool | None)

Return type:

None

async upload_skill_file(skill_path)

Upload a skill to the robot.

Parameters:

skill_path (str) – Path to the Python skill to upload.

Return type:

dict

Returns:

dict – The response from the API.

This module contains the WebSocketAPI class for communicating with the WebSocket.

class pymirokai.core.websocket_api.WebSocketAPI(retry_interval=5)

Bases: object

Class to handle WebSocket communication.

Parameters:

retry_interval (int)

async close()

Close the WebSocket connection.

Return type:

None

async connect()

Connect to the WebSocket and start the data retrieval.

Return type:

None

get_data()

Get the latest data received from the WebSocket.

Return type:

Dict[str, Any]

Returns:

Dict[str, Any] – The latest data.

async publish(topic, data)

Publish data to a 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 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

async start(ws_url, get_jwt_token)

Start the WebSocket connection.

Parameters:

• ws_url (str) – The url of the websocket

• get_jwt_token (callable) – Callback to get the JWT token

Return type:

None

async subscribe(topic)

Subscribe to a topic.

Parameters:

topic (str) – The topic to subscribe to.

Return type:

None

async unsubscribe(topic)

Unsubscribe from a topic.

Parameters:

topic (str) – The topic to unsubscribe from.

Return type:

None