grutopia.core.datahub¶

datahub¶

class grutopia.core.datahub.IsaacData[source]¶

isaac status in grutopia

There are two types of isaac status:

Action
Observation

structure of isaac status like this:

{
    actions: {
        [
            {
                robot_1: {
                    cap: param,
                }
            }
        ]
    },
    observations: {
        [
            {
                robot_1: {
                    obs_1: data,
                    obs_2: data
                }
            }
        ]
    }
}

classmethod add_actions(actions: List[ActionData])[source]¶

Add actions

Parameters:: actions – action list

Returns:

classmethod get_action_by_id(task_id: int) → None | Dict[Any, Any][source]¶

Get action by id

Returns:: {controller_name: param}})
Return type:: action(dict like {robot_name

classmethod get_actions() → None | List[Dict[Any, Any]][source]¶

Get actions

Returns:: {controller_name: param}}) list
Return type:: action(dict like {robot_name

classmethod get_obs() → List[Dict[str, Any]][source]¶

Get isaac observation data

Returns:: isaac observation data list

classmethod get_obs_by_id(task_id: int) → Dict[str, Any][source]¶

Get isaac observation by id

Parameters:: task_id – isaac task id
Returns:: isaac observation data

local api¶

grutopia.core.datahub.api.get_actions() → None | Dict[Any, Any][source]¶

Get all actions

Returns:: action data dict
Return type:: Dict[str, Any]

grutopia.core.datahub.api.get_all_obs() → List[Dict[str, Any]][source]¶

Get all observation data.

Returns:: sensor data dict
Return type:: List[Dict[str, Any]]

```

grutopia.core.datahub.api.get_obs_by_id(task_id: int) → Dict[str, Any][source]¶

Get observation by task_id

Returns:: obs data dict
Return type:: Dict[str, Any]

grutopia.core.datahub.api.send_actions(actions: List[ActionData])[source]¶: send actions to datahub :param actions: list of [dict of {robot_id: ActionData}] :type actions: List[ActionData]

grutopia.core.datahub.api.set_obs_data(obs: List[Dict[str, Any]]) → None[source]¶

Flush observation data.

Parameters:: obs (List[Dict[str, Any]]) – observation data

web api¶

Includes web ui interactive

grutopia.core.datahub.web_ui_api.clear(uuid: str = 'none')[source]¶: Clear all data in webui.

grutopia.core.datahub.web_ui_api.get_chat_control(uuid: str = 'none') → Dict[str, Any] | None[source]¶

Get chat control data.

Parameters:: uuid (str) – chat control uuid. default: none.
Returns:: chat control data.
Return type:: chat_control (List[Dict[str, Any]])

grutopia.core.datahub.web_ui_api.get_log_data(uuid: str = 'none') → Dict[str, Any] | None[source]¶

Get log data.

Parameters:: uuid (str) – log data uuid. default: none.
Returns:: log data.
Return type:: log_data (list[dict])

grutopia.core.datahub.web_ui_api.send_chain_of_thought(cot: str, uuid: str = 'none') → None[source]¶

chain of thought data

Parameters:

uuid (str) – uuid of chain of thought data, defaults to “none”.
cot (str) – chain of thought data.

grutopia.core.datahub.web_ui_api.send_chat_control(nickname: str, text: str, img: str | None = None, role: str = 'user', uuid: str = 'none') → None[source]¶

Send a new message to the chatbox.

Parameters:

nickname (str) – nickname displayed in the chatbox.
text (str) – text to send to the chatbox.
img (str, optional) – image to send to the chatbox. Defaults to None.
role (str, optional) – role name, user or agent. Defaults to “user”.
uuid (str, optional) – uuid of the message. Defaults to ‘none’.

grutopia.core.datahub.web_ui_api.send_log_data(log_data: str, log_type: str = 'message', user: str = 'Bob', photo_url: str = 'http://127.0.0.1:8080/static/avatar_default.jpg', uuid: str = 'none') → None[source]¶

Send log data.

Parameters:

uuid (str) – uuid of log, default is none.
log_data (str) – log data.
log_type (str) – type of log. ‘message’ or ‘user’.
user (str) – logger name. default: Bob.
photo_url (str) – log photo url path.

Includes web api endpoints

grutopia.core.datahub.web_api.get_actions_by_id(task_id: int)[source]¶

Get actions by task id(int)

Parameters:: task_id (int) – id of task
Returns:: msg str(or None) data: data
Return type:: msg

grutopia.core.datahub.web_api.get_all_obs() → List[Dict[str, Any]] | None[source]¶: Get all observation data :returns: List of all observation data :rtype: obs (List[Dict[str, Any]])

grutopia.core.datahub.web_api.get_obs_by_id(task_id: int) → Any | None[source]¶

Get observation by id :param task_id: id of observation data :type task_id: int

Returns:: Observation data
Return type:: obs (Any)

grutopia.core.datahub.web_api.send_actions(actions: List[ActionData]) → bool[source]¶

send actions :param actions: action data list :type actions: List[ActionData]

Returns:: Send message successfully or not

grutopia.core.datahub.web_api.set_obs_data(obs: List[Dict[str, Any]]) → bool[source]¶

Set observation data web API :param obs: isaac observation data :type obs: List[Dict[str, Any]]

Returns:: OK if set successfully