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)

Bases: object

Class to handle REST API communication.

Parameters:

  • api_version (str)

  • retry_interval (int)

async authentify()

Retrieve the mission ID and JWT token using the API key.

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 get_data_and_update_token_and_client_id(response)

From a response from the API, recursively pare 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 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.

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

Start the REST API session and connection check.

Parameters:

  • api_url (str) – The API URL on which to connect

  • api_key (str) – The API key to use.

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

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

Return type:

None

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