Core

Here you can find a description of all the scripts that compose the core, their specific services and topics, and the documentation of their methods.

Commander node

Python script to implement a commander node, which is the main entry point for the system. It is responsible for managing execution nodes and cognitive nodes, as well as loading and saving configurations.

Specific services

/commander/add_executor => Create a new execution node.

/commander/delete_executor => Delete an execution node by its identifier.

/commander/load_config => Load execution node configuration from a yaml file.

/commander/kill => Kill the commander node to shutdown the cognitive architecture.

/commander/move_cognitive_node_to => Move a cognitive node from one execution node to another

/commander/create => Create a new cognitive node.

/commander/delete => Delete an existing cognitive node by its name.

/commander/read => Retrieve the data of a cognitive node.

/commander/save => Save the data of a cognitive node to a file.

/commander/load => Load a cognitive node from a yaml file.

/commander/save_config => Save the current experiment state to a file.

/commander/stop_execution => Stop the execution of the whole system.

/commander/load_experiment => Load an experiment configuration from a YAML file.

class core.commander_node.CommanderNode(*args: Any, **kwargs: Any)

Bases: Node

This class is responsible for handling the requests from the user and sending them to the execution nodes.

Attributes:

cli (rclpy.client.Client): A client for the ‘SendCommand’ service.

add_execution_node(threads=1)

Adds a new execution node to the system.

This method increments the last used ID, creates a new execution node with the updated ID, and returns the assigned ID in the response. The execution nodes are created using the multiprocessing library. The process object is stored in the self.executors dictionary.

Parameters:

threads (int) – The number of threads for the new execution node.

Returns:

The assigned ID for the new execution node.

Return type:

str

add_execution_node_callback(request, response)

Callback for adding a new node to the system with a service call.

Parameters:

• request (core_interfaces.srv.AddExecutionNode.Request) – The request to add a new execution node.

• response (core_interfaces.srv.AddExecutionNode.Response) – The response with the assigned ID for the new execution node.

Returns:

The response with the assigned ID for the new execution node.

Return type:

core_interfaces.srv.AddExecutionNode.Response

configure_services()

Generates additional services related to execution and cognitive nodes.

create_node(request, response)

Handle the creation of a cognitive node.

This method performs the following steps: 1. Extracts information from the creation request, including node name, class name, and parameters. 2. Checks if the specified node already exists. 3. Determines the executor with the lowest load to distribute the creation request. 4. Sends a create request to the selected executor to create the cognitive node. 5. Registers the created node in the system.

Parameters:

• request (core_interfaces.srv.CreateNode.Request) – The creation request.

• response (core_interfaces.srv.CreateNode.Response) – The response to the creation request.

Returns:

The response to the creation request.

Return type:

core_interfaces.srv.CreateNode.Response

delete_execution_node(request, response)

Deletes an execution node and redistributes its cognitive nodes to other executors.

This method performs the following steps: 1. Saves all cognitive nodes of the specified execution node. 2. Deletes the execution node and stops its execution. 3. Loads the saved cognitive nodes into other executors.

Parameters:

• request (core_interfaces.srv.DeleteExecutionNode.Request) – The request to delete an execution node.

• response (core_interfaces.srv.DeleteExecutionNode.Response) – The response indicating whether the deletion was successful.

Returns:

The response indicating whether the deletion was successful.

Return type:

core_interfaces.srv.DeleteExecutionNode.Response

delete_node(request, response)

Handle the deletion of a cognitive node.

This method performs the following steps: 1. Extracts the node name from the delete request. 2. Checks if the specified node exists. 3. Retrieves the executor associated with the node. 4. Sends a delete request to the executor to remove the cognitive node. 5. Removes the node from the local data structures.

Parameters:

• request (core_interfaces.srv.DeleteNode.Request) – The delete request.

• response (core_interfaces.srv.DeleteNode.Response) – The response to the delete request.

Returns:

The response to the delete request.

Return type:

core_interfaces.srv.DeleteNode.Response

get_executor_for_node(name)

Get the executor that manages a specific node.

Parameters:

name (str) – The name of the node.

Returns:

The executor ID or None if the node is not found.

Return type:

int or None

get_lowest_load_executor()

Get the executor with the lowest load.

Returns:

The executor with the lowest load.

Return type:

int

kill_commander(request, response)

Handle the termination of the commander node, shutting down the entire cognitive architecture.

Parameters:

• request (core_interfaces.srv.StopExecution.Request) – The kill request.

• response (core_interfaces.srv.StopExecution.Response) – The response to the kill request.

Returns:

The response indicating the success of the termination.

Return type:

core_interfaces.srv.StopExecution.Response

load_config(request, response)

Loads a yaml file with the configuration of the execution nodes.

Parameters:

• request (core_interfaces.srv.LoadConfig.Request) – Request for loading configuration file.

• response (core_interfaces.srv.LoadConfig.Response) – Response with success result of the configuration.

Returns:

Response with success result of the configuration.

Return type:

core_interfaces.srv.LoadConfig.Response

load_experiment(request, response)

Load an experiment from a file.

This method performs the following steps: 1. Extracts the experiment file path from the load request. 2. Checks if the specified experiment file exists. 3. Reads the YAML data from the experiment file. 4. Retrieves information about cognitive nodes from the loaded data. 5. Iterates over the cognitive nodes, creating and registering them in the system.

Parameters:

• request (core_interfaces.srv.LoadConfig.Request) – The load experiment request.

• response (core_interfaces.srv.LoadConfig.Response) – The response to the load experiment request.

Returns:

The response to the load experiment request.

Return type:

core_interfaces.srv.LoadConfig.Response

load_node(request, response)

Handle loading a cognitive node from a configuration.

This method performs the following steps: 1. Extracts information from the load request, including the node name and file path. 2. Checks if the specified node already exists. 3. Retrieves the executor with the lowest load. 4. Sends a load request to the executor to load the cognitive node from the specified file. 5. Registers the loaded node in the local data structures.

Parameters:

• request (core_interfaces.srv.LoadNode.Request) – The load request.

• response (core_interfaces.srv.LoadNode.Response) – The response to the load request.

Returns:

The response to the load request.

Return type:

core_interfaces.srv.LoadNode.Response

move_cognitive_node_to(request, response)

Moves a cognitive node from its current executor to a specified executor.

This method performs the following steps: 1. Checks if the specified node exists. 2. Saves the data of the node. 3. Removes the node from its current executor. 4. Loads the node into the specified executor.

Parameters:

• request (core_interfaces.srv.MoveCognitiveNodeToExecutionNode.Request) – The request to move a cognitive node.

• response (core_interfaces.srv.MoveCognitiveNodeToExecutionNode.Response) – The response indicating whether the movement was successful.

Returns:

The response indicating whether the movement was successful.

Return type:

core_interfaces.srv.MoveCognitiveNodeToExecutionNode.Response

node_exists(name)

Check if a cognitive node with the given name exists.

Parameters:

name (str) – The name of the node.

Returns:

True if the node exists, False otherwise.

Return type:

bool

process_shutdown()

Shutdown all execution nodes and terminate the commander node.

read_node(request, response)

Handle reading data from a cognitive node.

This method performs the following steps: 1. Extracts information the node name from the read request. 2. Checks if the specified node exists. 3. Retrieves the executor associated with the node. 4. Sends a read request to the executor to obtain data from the cognitive node. 5. Parses the received data and logs information. 6. Sets the response data with the information obtained from the cognitive node.

Parameters:

• request (core_interfaces.srv.ReadNode.Request) – The read request.

• response (core_interfaces.srv.ReadNode.Response) – The response to the read request.

Returns:

The response to the read request.

Return type:

core_interfaces.srv.ReadNode.Response

register_node(executor_id, node_name)

Register a node in the list of managed nodes for an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• node_name (str) – The name of the node.

remove_node_from_executor(executor_id, node_name)

Remove a node from the list of managed nodes for an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• node_name (str) – The name of the node to remove.

save_config(request, response)

Saves the state of the system to a YAML file.

This method performs the following steps: 1. Constructs the file path using the saved data directory and the specified file name. 2. Prepares the data structure to be saved, including the LTM nodes organized by type. 3. Iterates over each execution node to retrieve the data of all cognitive nodes. 4. Appends the retrieved node data to the corresponding type in the prepared data structure. 5. Writes the final data structure to a YAML file.

Parameters:

• request (core_interfaces.srv.SaveConfig.Request) – The save config request.

• response (core_interfaces.srv.SaveConfig.Response) – The response to the save config request.

Returns:

The response to the save config request.

Return type:

core_interfaces.srv.SaveConfig.Response

save_node(request, response)

Handle saving the state of a cognitive node.

This method performs the following steps: 1. Checks if the specified node exists. 2. Determines the current executor of the node. 3. Sends a save request to the executor to save the state of the node.

Parameters:

• request (core_interfaces.srv.SaveNode.Request) – The save request.

• response (core_interfaces.srv.SaveNode.Response) – The response to the save request.

Returns:

The response to the save request.

Return type:

core_interfaces.srv.SaveNode.Response

send_create_request_to_executor(executor_id, name, class_name, parameters)

Send a ‘create’ request to an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• name (str) – The name of the node to create.

• class_name (str) – The type of the node to create.

• parameters (str) – The parameters for the node creation.

Returns:

The response from the executor.

Return type:

core_interfaces.srv.CreateNode.Response

send_delete_request_to_executor(executor_id, name)

Send a ‘delete’ request to an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• name (str) – The name of the node to delete.

• class_name (str) – The type of the node to delete.

Returns:

The response from the executor.

Return type:

core_interfaces.srv.DeleteNode.Response

send_load_request_to_executor(executor_id, name, file_path)

Send a ‘load’ request to an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• name (str) – The name of the node to load.

• file_path (str) – The file path of the configuration to load.

Returns:

The response from the executor.

Return type:

core_interfaces.srv.LoadNode.Response

send_read_all_nodes_request_to_executor(executor_id)

Send a request to read data from all cognitive nodes in a specific execution node. :param executor_id: The ID of the executor. :type executor_id: int :return: The response containing data from all cognitive nodes. :rtype: core_interfaces.srv.ReadNode.Response

send_read_request_to_executor(executor_id, name)

Send a ‘read’ request to an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• name (str) – The name of the node to read.

Returns:

The response from the executor.

Return type:

core_interfaces.srv.ReadNode.Response

send_save_all_nodes_request_to_executor(executor_id)

Send a request to save the state of all cognitive nodes in a specific execution node.

Parameters:

executor_id (int) – The ID of the executor.

Returns:

The response indicating the success of saving all cognitive nodes.

Return type:

core_interfaces.srv.SaveNode.Response

send_save_request_to_executor(executor_id, name)

Send a ‘save’ request to an executor.

Parameters:

• executor_id (int) – The ID of the executor.

• name (str) – The name of the node to save.

Returns:

The response from the executor.

Return type:

core_interfaces.srv.SaveNode.Response

send_stop_request_to_executor(executor_id)

Send a request to stop the execution of all cognitive nodes in a specific execution node.

Parameters:

executor_id (int) – The ID of the executor.

Returns:

The response indicating the success of stopping the execution.

Return type:

core_interfaces.srv.StopExecution.Response

stop_execution(request, response)

Stop the execution of all execution nodes in the system.

This method performs the following steps: 1. Iterates over each execution node to send a stop request and stop its execution. 2. Clears the list of nodes associated with each execution node. 3. Publishes a stop message for each execution node.

Parameters:

• request (core_interfaces.srv.StopExecution.Request) – The stop execution request.

• response (core_interfaces.srv.StopExecution.Response) – The response to the stop execution request.

Returns:

The response to the stop execution request.

Return type:

core_interfaces.srv.StopExecution.Response

Execution node

Python script to implement an execution node is a container for cognitive nodes. It is responsible for managing the execution of cognitive nodes and their interactions.

ID is the identifier of the node, manually or automatically assigned. It is an integer.

Specific topics

/execution_node_<ID>/stop_execution_node => Stop the execution node.

Specific services

/execution_node_<ID>/create => Create a new cognitive node.

/execution_node_<ID>/read => Retrieve information about a node by its name.

/execution_node_<ID>/delete => Delete a cognitive node by its name.

/execution_node_<ID>/save => Save the state of a cognitive node.

/execution_node_<ID>/load => Load a cognitive node from a file.

/execution_node_<ID>/read_all_nodes => Read the data from all cognitive nodes in this execution node.

/execution_node_<ID>/save_all_nodes => Save data from all cognitive nodes in this execution node.

/execution_node_<ID>/stop_execution => Stop the execution of all cognitive nodes in this execution node.

class core.execution_node.ExecutionNode(*args: Any, **kwargs: Any)

Bases: Node

This class represents an execution node, which can execute several cognitive nodes. It is subscribed to the topic where the commander node sends commands.

async create_node(request, response)

Create a new cognitive node. If the node doesn’t have previous data, it is created with the default values. In other case, the existent data is loaded.

Parameters:

• request (core_interfaces.srv.CreateNode.Request) – The request to create a new node.

• response (core_interfaces.srv.CreateNode.Response) – The response indicating the success of the creation.

Returns:

The response indicating the success of the creation.

Return type:

core_interfaces.srv.CreateNode.Response

delete_node(request, response)

Delete a cognitive node by its name.

Parameters:

• request (core_interfaces.srv.DeleteNode.Request) – The request containing the name of the node to delete.

• response (core_interfaces.srv.DeleteNode.Response) – The response indicating the success of the deletion.

Returns:

The response indicating the success of the deletion.

Return type:

core_interfaces.srv.DeleteNode.Response

load_node(request, response)

Load a cognitive node from a file.

Parameters:

• request (core_interfaces.srv.LoadNode.Request) – The request containing the name and file path of the node to load.

• response (core_interfaces.srv.LoadNode.Request) – The request containing the name and file path of the node to load.

Returns:

The response indicating the success of the loading.

Return type:

core_interfaces.srv.LoadNode.Response

read_all_nodes(_, response)

Read the data from all cognitive nodes in this execution node.

Parameters:

response (core_interfaces.srv.ReadNode.Response) – The response containing data from all nodes.

Returns:

The response containing data from all nodes.

Return type:

core_interfaces.srv.ReadNode.Response

read_node(request, response)

Retrieve information about a node by its name.

Parameters:

• request (core_interfaces.srv.ReadNode.Request) – The request containing the name of the node to read.

• response (core_interfaces.srv.ReadNode.Response) – The response with the requested node data.

Returns:

The response with the requested node data.

Return type:

core_interfaces.srv.ReadNode.Response

save_all_nodes(_, response)

Save data from all cognitive nodes in this execution node.

Parameters:

response (core_interfaces.srv.SaveNode.Response) – The response indicating the success of the operation.

Returns:

The response indicating the success of the operation.

Return type:

core_interfaces.srv.SaveNode.Response

save_node(request, response)

Save the state of a cognitive node.

Parameters:

• request (core_interfaces.srv.SaveNode.Request) – The request containing the name of the node to save.

• response (core_interfaces.srv.SaveNode.Response) – The response indicating the success of the saving.

Returns:

The response indicating the success of the saving.

Return type:

core_interfaces.srv.SaveNode.Response

stop_execution(request, response)

Stop the execution of all cognitive nodes in this execution node.

Parameters:

• request (core_interfaces.srv.StopExecution.Request) – The request to stop execution.

• response (core_interfaces.srv.StopExecution.Response) – The response indicating the success of stopping execution.

Returns:

The response indicating the success of stopping execution.

Return type:

core_interfaces.srv.StopExecution.Response

stop_execution_callback(msg)

Callback method to stop the execution of this execution node.

Parameters:

msg (std_msgs.msg.String) – The message containing the ID of the execution node to stop.

core.execution_node.create_execution_node(id: int, threads: int, args=None)

Create and run an execution node.

Parameters:

• id (int) – The identifier for the execution node.

• threads (int) – The number of threads for the executor.

• args (list) – Additional arguments for rclpy initialization, defaults to None.

Long-Term Memory (LTM)

Python script to implement the Long-Term Memory (LTM), which is a persistent storage for cognitive nodes and their relationships.

ID is the identifier of the node, manually or automatically assigned. It is an integer.

Specific topics

/ltm_<ID>/state => Modifications to the content of the LTM are published here for whoever needs them.

Specific services

/ltm_<ID>/add_node => Add a node to the LTM.

/ltm_<ID>/replace_node => Replace a node in the LTM by another one.

/ltm_<ID>/delete_node => Delete a node in the LTM.

/ltm_<ID>/get_node => Get node(s) information. This service can be used to get information about a specific node, all the nodes of a given type, or the whole LTM.

/ltm_<ID>/set_changes_topic => Activate / deactivate automatic publishing of changes in the LTM.

/ltm_<ID>/update_neighbor => Update the neighbor relatioship between two cognitive nodes in the LTM.

class core.ltm.LTM(*args: Any, **kwargs: Any)

Bases: Node

The Long-Term Memory (LTM) node in the cognitive architecture.

This node is responsible for storing and managing cognitive nodes of various types. It provides services for adding, replacing, deleting, and retrieving these nodes, as well as publishing changes.

Attributes:

id (str): An identifier for the LTM instance. changes_topic (bool): Flag to indicate if changes are being published. cognitive_nodes (dict): A dictionary to store cognitive nodes by type. state_publisher (Publisher): Publisher for the state of the LTM. add_node_service (Service): Service to add new cognitive nodes. replace_node_service (Service): Service to replace existing cognitive nodes. delete_node_service (Service): Service to delete cognitive nodes. update_neighbors_service (Service): Service to update the neighbors list of a node. get_node_service (Service): Service to retrieve data of cognitive nodes. set_changes_topic_service (Service): Service to set the changes topic.

async add_neighbor(neighbor_name, neighbor_type, service_node_name)

Add a neighbor to a cognitive node by calling its ‘add_neighbor’ service.

Parameters:

• neighbor_name (str) – The name of the neighbor to be added.

• neighbor_type (str) – The type of the neighbor to be added.

• service_node_name (str) – The name of the cognitive node whose ‘add_neighbor’ service will be called.

Returns:

The result of the service call.

Return type:

cognitive_node_interfaces.srv.AddNeighbor.Response

async add_node(node_type, node_name, node_data)

Add a cognitive node to the LTM.

Parameters:

• node_type (str) – The type of the cognitive node.

• node_name (str) – The name of the cognitive node.

• node_data (dict) – The dictionary containing the data of the cognitive node.

async add_node_callback(request, response)

Callback function for the ‘add_node’ service. Adds a cognitive node to the LTM.

This method checks if the node already exists in the LTM. If it does, it sets the response ‘added’ attribute to False. If the node does not exist, it adds the new node to the LTM and sets the ‘added’ attribute to True.

Parameters:

• request (core_interfaces.srv.AddNodeToLTM.Request) – The service request containing the node’s name, type, and data.

• response (core_interfaces.srv.AddNodeToLTM.Response) – The service response.

Returns:

The response indicating whether the node was added successfully.

Return type:

core_interfaces.srv.AddNodeToLTM.Response

async delete_neighbor(neighbor_name, neighbor_type, service_node_name)

Remove a neighbor from a cognitive node by calling its ‘delete_neighbor’ service.

Parameters:

• neighbor_name (str) – The name of the neighbor to be removed.

• neighbor_type (str) – The type of the neighbor to be removed.

• service_node_name (str) – The name of the cognitive node whose ‘delete_neighbor’ service will be called.

Returns:

The result of the service call.

Return type:

cognitive_node_interfaces.srv.DeleteNeighbor.Response

delete_node(node_type, node_name)

Delete a cognitive node from the LTM.

Parameters:

• node_type (str) – The type of the cognitive node.

• node_name (str) – The name of the cognitive node.

delete_node_callback(request, response)

Callback function for the ‘delete_node’ service. Deletes a cognitive node from the LTM.

This method iterates over all node types in the LTM to find the node with the given name. If the node is found, it is deleted, and the response ‘deleted’ attribute is set to True. If the node is not found, the ‘deleted’ attribute is set to False.

Parameters:

• request (core_interfaces.srv.DeleteNodeFromLTM.Request) – The service request containing the name of the node to be deleted.

• response (core_interfaces.srv.DeleteNodeFromLTM.Response) – The service response.

Returns:

The response indicating whether the node was deleted successfully.

Return type:

core_interfaces.srv.DeleteNodeFromLTM.Response

property drives

Get all cognitive nodes of type ‘Drive’ from the LTM.

Returns:

A list of ‘Drive’ nodes.

Return type:

list

get_node_callback(request, response)

Callback function for the ‘get_node’ service. Retrieves data of a specific cognitive node from the LTM.

This method iterates over all node types in the LTM to find the node with the given name. If the node is found, its data is serialized into YAML format and returned in the response. If the node is not found, an empty string is returned.

Parameters:

• request (core_interfaces.srv.GetNodeFromLTM.Request) – The service request containing the name of the node to retrieve.

• response (core_interfaces.srv.GetNodeFromLTM.Response) – The service response containing the node data if found.

Returns:

The response with the node data in YAML format or an empty string.

Return type:

core_interfaces.srv.GetNodeFromLTM.Response

get_node_dict(name, default=None)

Retrieve the dictionary of a cognitive node by its name.

This method searches for a cognitive node with the given name across all node types in the LTM. If the node is found, its dictionary is returned. If the node is not found, the provided default value is returned.

Parameters:

• name (str) – The name of the cognitive node to retrieve.

• default (any) – The default value to return if the node is not found, defaults to None.

Returns:

The dictionary of the cognitive node if found, otherwise the default value.

Return type:

dict or any

get_node_type(name)

Retrieve the type of a cognitive node by its name.

This method searches for a cognitive node with the given name across all node types in the LTM. If the node is found, its type is returned. If the node is not found, None is returned.

Parameters:

name (str) – The name of the cognitive node to retrieve the type for.

Returns:

The type of the cognitive node if found, otherwise None.

Return type:

str or None

property goals

Get all cognitive nodes of type ‘Goal’ from the LTM.

Returns:

A list of ‘Goal’ nodes.

Return type:

list

node_exists(node_type, node_name)

Check if a cognitive node exists in the LTM.

Parameters:

• node_type (str) – The type of the cognitive node.

• node_name (str) – The name of the cognitive node.

Returns:

True if the node exists, False otherwise.

Return type:

bool

node_type_exists(node_type)

Check if cognitive nodes of a specific type exist in the LTM.

Parameters:

node_type (str) – The type of cognitive node to check.

Returns:

True if the specified type exist, False otherwise.

Return type:

bool

property pnodes

Get all cognitive nodes of type ‘PNode’ from the LTM.

Returns:

A list of ‘PNode’ nodes.

Return type:

list

property policies

Get all cognitive nodes of type ‘Policy’ from the LTM.

Returns:

A list of ‘Policy’ nodes.

Return type:

list

publish_state()

Publishes the LTM state in the state topic.

property purposes

Get all cognitive nodes of type ‘RobotPurpose’ from the LTM.

Returns:

A list of ‘RobotPurpose’ nodes.

Return type:

list

replace_node_callback(request, response)

Callback function for the ‘replace_node’ service. Replaces an existing cognitive node in the LTM.

This method first checks if the original node exists in the LTM. If it doesn’t, it sets the response ‘replaced’ attribute to False. If the original node exists, it then checks if the new name is already taken. If not, it replaces the node with the new data and sets the ‘replaced’ attribute to True.

Parameters:

• request (core_interfaces.srv.ReplaceNodeFromLTM.Request) – The service request containing the original and new name of the node, its type, and the new data for the node.

• response (core_interfaces.srv.ReplaceNodeFromLTM.Response) – The service response.

Returns:

The response indicating whether the node was replaced successfully.

Return type:

core_interfaces.srv.ReplaceNodeFromLTM.Response

set_changes_topic_callback(request, response)

Callback function for the ‘set_changes_topic’ service. Sets the topic for tracking changes in the LTM.

Parameters:

• request (core_interfaces.srv.SetChangesTopic.Request) – The service request containing the boolean value for the changes topic.

• response (core_interfaces.srv.SetChangesTopic.Response) – The service response confirming the updated changes topic.

Returns:

The response with the updated changes topic value.

Return type:

core_interfaces.srv.SetChangesTopic.Response

async update_neighbor_callback(request, response)

Callback function for the ‘update_neighbor’ service. Updates the neighbor relationship between two cognitive nodes.

Parameters:

• request (core_interfaces.srv.UpdateNeighbor.Request) – The service request containing the node name, neighbor name, and operation type.

• response (core_interfaces.srv.UpdateNeighbor.Response) – The service response indicating the success of the operation.

Returns:

The response indicating whether the neighbor update was successful.

Return type:

core_interfaces.srv.UpdateNeighbor.Response

property utilitymodels

Get all cognitive nodes of type ‘UtilityModel’ from the LTM.

Returns:

A list of ‘UtilityModel’ nodes.

Return type:

list

property worldmodels

Get all cognitive nodes of type ‘WorldModel’ from the LTM.

Returns:

A list of ‘WorldModel’ nodes.

Return type:

list

Cognitive node

Python script to implement a cognitive node, which is a fundamental building block of the system. Each cognitive_node is one of the following list: Perception, PNode, CNode, Need, Drive, Goal, UtilityModel, WorldModel, Policy.

ID is the identifier of the node, manually or automatically assigned. It is usually a string.

Specific topics

/cognitive_node/<ID>/activation => If the node has been configured to do so, the current activation is published whenever there is a change or periodically using a timer.

Specific services

/cognitive_node/<ID>/get_activation => Get latest activation not newer than a given timestamp.

/cognitive_node/<ID>/get_information => Get information about the current state (useful for visualization).

/cognitive_node/<ID>/set_activation_topic => Activate / deactivate automatic publishing of activation.

/cognitive_node/<ID>/add_neighbor => Adds a node to the neighbor list of cognitive node.

/cognitive_node/<ID>/delete_neighbor => Deletes a node from the neighbor list of the cognitive_node.

class core.cognitive_node.CognitiveNode(*args: Any, **kwargs: Any)

Bases: Node

A base class for cognitive nodes in the system.

This class extends the rclpy.node.Node class and provides some common functionality for cognitive nodes.

add_neighbor_callback(request, response)

Add a neighbor to the nodes neighbors collection.

Parameters:

• request (cognitive_node_interfaces.srv.AddNeighbor.Request) – The request that contains the neighbor info.

• response (cognitive_node_interfaces.srv.AddNeighbor.Response) – The response that indicates if the neighbor was added.

Returns:

The response that indicates if the neighbor was added.

Return type:

cognitive_node_interfaces.srv.AddNeighbor.Response

add_neighbor_client(node_name, neighbor_name)

Client that adds a node as a neighbor of another node.

Parameters:

• node_name (str) – Node to which the neighbor will be added.

• neighbor_name (str) – Node that will be added as neighbor.

Returns:

Future with service’s response

Return type:

Future

calculate_activation(perception, activation_list)

Calculate the node’s activation for the given perception. :param perception: The perception for which the activation will be calculated. :type perception: dict :param activation_list: List of activations considered in the node :type activation_list: dict

calculate_activation_max(activation_list)

Calculates the activation of the node by multiplying the activation of the nodes in the activation_list. The timestamp of the resulting activation will be the oldest timestamp of the nodes in the list.

Parameters:

activation_list (dict) – Dictionary with the activation of multiple nodes.

calculate_activation_prod(activation_list)

Calculates the activation of the node by multiplying the activation of the nodes in the activation_list. The timestamp of the resulting activation will be the oldest timestamp of the nodes in the list.

Parameters:

activation_list (dict) – Dictionary with the activation of multiple nodes.

configure_activation_inputs(neighbor_list)

Populates the activation list from a list of neighbors.

Parameters:

neighbor_list (dict) – Dictionary with the information of the node [{‘name’: <name>, ‘node_type’: <node_type>}, …. ].

create_activation_input(node: dict)

Adds a node to the activation inputs list.

Parameters:

node (dict) – Dictionary with the information of the node {‘name’: <name>, ‘node_type’: <node_type>}.

create_node_client(name, class_name, parameters={})

This method calls the add node service of the commander.

Parameters:

• name (str) – Name of the node to be created.

• class_name (str) – Name of the class to be used for the creation of the node.

• parameters (dict) – Optional parameters that can be passed to the node, defaults to {}.

Returns:

Success status received from the commander.

Return type:

bool

delete_activation_input(node: dict)

Deletes a node from the activation inputs list.

Parameters:

node (dict) – Dictionary with the information of the node {‘name’: <name>, ‘node_type’: <node_type>}.

delete_neighbor_callback(request, response)

Delete a neighbor to the nodes neighbors collection.

Parameters:

• request (cognitive_node_interfaces.srv.DeleteNeighbor.Request) – The request that contains the neighbor info.

• response (cognitive_node_interfaces.srv.DeleteNeighbor.Response) – The response that indicates if the neighbor was deleted.

Returns:

The response that indicates if the neighbor was deleted.

Return type:

cognitive_node_interfaces.srv.DeleteNeighbor.Response

delete_neighbor_client(node_name, neighbor_name)

Client that deletes a node as a neighbor of another node.

Parameters:

• node_name (str) – Node to which the neighbor will be deleted.

• neighbor_name (str) – Node that will be deleted as neighbor.

Returns:

Future with service’s response.

Return type:

Future

extract_oldest_timestamp(activation_list)

Helper method to obtain the oldest timestamp from all the activations that the node has recieved.

Parameters:

activation_list (dict) – Dictionary with the activation of multiple nodes.

Returns:

Timestamp message with the oldest activation and the node that produced it.

Return type:

Tuple (builtin_interfaces.msg.Time, str)

async get_activation_callback(request, response)

Callback method to calculate and return the node’s activations. This method calculates the activation of the node based on its perception.

Parameters:

• request (cognitive_node_interfaces.srv.GetActivation.Request) – The request containing the perception data.

• response (cognitive_node_interfaces.srv.GetActivation.Response) – The response that will contain the calculated activation.

Returns:

The response with the calculated activation.

Return type:

cognitive_node_interfaces.srv.GetActivation.Response

get_data()

Get the data associated with the node.

This method returns a dictionary containing the attributes of the node, excluding private attributes and the ‘subscription’ attribute found in the ANode class.

Returns:

A dictionary with node data.

Return type:

dict

get_information_callback(request, response)

Callback method to get information about the node.

This method retrieves information about the node, such as its current activation. The activation value is included in the response for external queries.

Parameters:

• request (cognitive_node_interfaces.srv.GetInformation.Request) – The request for node information.

• response (cognitive_node_interfaces.srv.GetInformation.Response) – The response that will contain the node’s information.

Returns:

The response with the node’s information.

Return type:

cognitive_node_interfaces.srv.GetInformation.Response

publish_activation(activation: cognitive_node_interfaces.msg.Activation)

Publish the activation of this node. :param activation: The activation to be published. :type activation: cognitive_node_interfaces.msg.Activation

async publish_activation_callback()

Timed publish of the activation value. This method will calculate the activation based on the neighbor’s activation, and then publish it in the corresponding topic.

read_activation_callback(msg: cognitive_node_interfaces.msg.Activation)

Callback to read the activation of a neighbor node.

Parameters:

msg (cognitive_node_interfaces.msg.Activation) – Activation message from a neighbor node.

async register_in_LTM(data_dic)

Requests registering the node in the LTM.

Parameters:

data_dic (dict) – A dictionary with the data to be saved.

Returns:

A future that will contain the response from the LTM service.

Return type:

rclpy.task.Future

remove_from_LTM()

Removes the node from the LTM. :return: True if the operation was succesful, False otherwise. :rtype: core_interfaces.srv.DeleteNodeFromLTM.Response

set_activation_topic_callback(request, response)

Callback method to control activation topic publishing for the node.

This method toggles the activation topic publishing for the node based on the provided request.

Parameters:

• request (cognitive_node_interfaces.srv.SetActivationTopic.Request) – True to publish the activation; False otherwise.

• response (cognitive_node_interfaces.srv.SetActivationTopic.Response) – True if the node will publish the activation; False otherwise.

Returns:

True if the node will publish the activation; False otherwise.

Return type:

cognitive_node_interfaces.srv.SetActivationTopic.Response

update_neighbor_client(node_name, neighbor_name, operation)

Calls the /update_neighbor service in the LTM.

Parameters:

• node_name (str) – Node whose neighbor list will be modified.

• neighbor_name (str) – Node that will be added/deleted as neighbor.

• operation (bool) – Selects between adding (True) or deleting (False) neighbor.

Returns:

Future with service’s response.

Return type:

Future