Cognitive Processes API Documentation

Here you can find a description of the Main Loop script, its specific topics, and the documentation of its methods.

Cognitive Processes

Base class for all cognitive processes in the architecture. It provides common functionalities and interfaces for different cognitive processes.

class cognitive_processes.cognitive_process.CognitiveProcess(*args: Any, **kwargs: Any)

Bases: Node

CognitiveProcess class, base class for cognitive processes in the architecture.

add_antipoint(name, sensing, node_type='pnode')

Sends the request to add an antipoint to a P-Node.

Parameters:

• name (str) – Name of the P-Node.

• sensing (dict) – Sensorial data to be added as a antipoint.

Returns:

Success status received from the P-Node.

Return type:

bool

add_neighbor(node_name, neighbor_name)

This method adds a neighbor to a node in the LTM.

Parameters:

• node_name (str) – Name of the node to which the neighbor will be added.

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

Returns:

True if the neighbor was added successfully, False otherwise.

Return type:

bool

add_point(name, sensing, node_type='pnode')

Sends the request to add a point to a P-Node.

Parameters:

• name (str) – Name of the P-Node.

• sensing (dict) – Sensorial data to be added as a point.

Returns:

Success status received from the P-Node.

Return type:

bool

configure_perceptions()

Reads the LTM cache and populates the perception subscribers and perception cache dictionaries.

create_activation_input(name, node_type)

This method creates a new activation input for a node.

Parameters:

• name (str) – Name of the node to be added as an activation input.

• node_type (str) – Type of the node to be added as an activation input.

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(name)

This method deletes an activation input for a node.

Parameters:

name (str) – Name of the node to be deleted as an activation input.

execute_actuation(actuation: dict)

Execute an action by sending it to the action service.

Parameters:

actuation (dict) – The actuation dictionary containing the action details.

Returns:

The response from the action service.

Return type:

cognitive_node_interfaces.srv.Execute.Response

execute_policy(perception, policy)

Execute a policy. This method sends a request to the policy to be executed.

Parameters:

• perception (dict) – The perception to be used in the policy execution.

• policy (str) – The policy to execute.

Returns:

The response from executing the policy.

Return type:

The executed policy.

get_all_active_nodes(node_type, ltm_cache)

This method retrieves all active nodes of a given type from the LTM cache.

Parameters:

• node_type (str) – Type of the nodes to be selected (e.g., “WorldModel”, “Goal”, etc.).

• ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

List of active nodes of the specified type.

Return type:

list

get_current_world_model()

This method selects the world model with the highest activation in the LTM cache.

Returns:

World model with highest activation.

Return type:

str

get_goals(ltm_cache)

This method retrieves all active goals from the LTM cache.

Parameters:

ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

List of active goals.

Return type:

list

get_goals_reward(old_sensing, sensing, ltm_cache)

This method retrieves the rewards for each active goal based on the old and current sensing.

Parameters:

• old_sensing (dict) – Old sensing data.

• sensing (dict) – Current sensing data.

• ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

Dictionary with goal names as keys and their corresponding rewards as values.

Return type:

dict

get_max_activation_node(node_type)

This method retrieves the node with the maximum activation of a given type from the LTM cache.

Parameters:

node_type (str) – Type of the node to be selected (e.g., “WorldModel”, “Goal”, etc.).

Returns:

Tuple containing the name of the node with the maximum activation and a dictionary with all activations of that type.

Return type:

tuple

get_node_activations_by_list(node_list, ltm_cache)

This method retrieves the activations of a list of nodes from the LTM cache.

Parameters:

• node_list (list) – List of node names to retrieve activations for.

• ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

Dictionary with node names as keys and their activations as values, sorted by activation.

Return type:

dicts

get_node_activations_by_type(node_type, ltm_cache)

This method retrieves the activations of all nodes of a given type from the LTM cache.

Parameters:

• node_type (str) – Type of the nodes to be selected (e.g., “WorldModel”, “Goal”, etc.).

• ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

Dictionary with node names as keys and their activations as values, sorted by activation.

Return type:

dict

get_node_data(node_name, ltm_cache)

This method retrieves the data of a node from the LTM cache.

Parameters:

• node_name (str) – Name of the node to retrieve data for.

• ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

Data of the node as a dictionary.

Return type:

dict

get_node_type(node_name, ltm_cache)

This method retrieves the type of a node from the LTM cache.

Parameters:

• node_name (str) – Name of the node to retrieve the type for.

• ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

Data of the node as a dictionary.

Return type:

dict

get_purpose_satisfaction(purpose_list, timestamp)

This method retrieves the satisfaction of each purpose in the purpose_list.

Parameters:

• purpose_list (list) – List of purposes.

• timestamp (rclpy.time.Time) – Timestamp to be used for the request.

Returns:

Dictionary with purpose names as keys and their satisfaction status as values.

Return type:

dict

get_purposes(ltm_cache)

This method retrieves all active purposes from the LTM cache.

Parameters:

ltm_cache (dict) – LTM cache containing the nodes and their data.

Returns:

List of active purposes.

Return type:

list

get_unlinked_drives()

This method retrieves the drives that are not linked to any goal in the LTM cache.

Returns:

List of unlinked drives. If there are no unlinked drives, it returns an empty list.

Return type:

list

ltm_change_callback(msg)

PENDING METHOD: This method will read changes made in the LTM external to the cognitive process and update the LTM and perception caches accordingly.

Parameters:

msg (std_msgs.msg.String) – Message containing the LTM dump.

new_cnode(perception, goal, policy)

This method creates a new C-Node/P-Node pair.

Parameters:

• perception (dict) – Perception to be added as the first point in the P-Node.

• goal (str) – Goal that will be linked to the C-Node.

• policy (str) – Policy that will be linked to the C-Node.

new_goal(perception, drive)

This method creates a new Goal node linked to a Drive.

Parameters:

• perception (dict) – Perception to be used in the Goal creation.

• drive (str) – Drive to which the Goal will be linked.

Returns:

Name of the created Goal node.

Return type:

str

publish_episode()

Publish the current episode data to the episode topic.

read_activation_callback(msg: cognitive_node_interfaces.msg.Activation)

This method receives a message from an activation topic, processes the message and updates the activation in the LTM cache.

Parameters:

msg (cognitive_node_interfaces.msg.Activation) – Message that contains the activation information.

read_ltm(ltm_dump=None)

Makes an empty call for the LTM get_node service which returns all the nodes stored in the LTM. Then, a LTM cache dictionary is populated with the data.

Parameters:

ltm_dump (dict) – Optional LTM dump to be used instead of requesting it from the current one.

read_perceptions()

Reads the perception cache and returns the latest perception.

This method iterates the perception cache dictionary. For each perception waits for the flag that signals that the value has been updated. Then, the value is copied in the sensing dictionary.

When the whole cache is processed, the sensing dictionary is returned.

Returns:

Latest sensing.

Return type:

dict

receive_perception_callback(msg)

Receives a message from a perception value topic, processes the message and copies it to the perception cache. Finally sets the flag to signal that the value has been updated.

Parameters:

msg (cognitive_node_interfaces.msg.Perception) – Message that contains the perception.

request_activation(name, sensing)

This method calls the service to get the activation of a node.

Parameters:

• name (str) – Name of the node.

• sensing (dict) – Sensing used to calculate the activation.

Returns:

Activation value.

Return type:

float

request_ltm()

Requests the LTM dump from its service.

Returns:

LTM dump as a dictionary.

Return type:

dict

request_neighbors(name)

This method calls the service to get the neighbors of a node.

Parameters:

name (str) – Node name.

Returns:

List of dictionaries with the information of each neighbor of the node. [{‘name’: <Neighbor Name>, ‘node_type’: <Neighbor type>},…].

Return type:

list

run()

Main loop of the cognitive process. This method is executed in a separate thread. It runs the cognitive process, reading perceptions, updating activations, and processing episodes.

sensorial_changes(sensing, old_sensing, threshold=0.01)

Return false if all perceptions have the same value as the previous step. True otherwise.

Parameters:

• sensing (dict) – Sensing in the current iteration.

• old_sensing (dict) – Sensing in the last iteration.

Returns:

Boolean that indicates if there is a sensorial change.

Return type:

bool

setup()

Initial configuration of the MainLoop node. This method sets up the LTM, perceptions, files, connectors, control channel, etc.

setup_connectors()

Configures the default classes for the cognitive nodes.

setup_control_channel()

Configures the control channel.

setup_ltm_suscription()

Sets up a subscription to the LTM state topic.

update_activations()

This method updates the activations of the nodes in the LTM cache.

update_ltm(stm: cognitive_nodes.episode.Episode)

Placeholder method to update the LTM with the current STM data.

Main loop

Python class that implements the Main Cognitive Loop of the system. The Main Loop is responsible for:

• Reading perceptions from the environment.

• Updating cognitive node activations.

• Selecting and executing policies.

• Evaluating rewards.

• Creating and managing cognitive nodes dynamically.

• Managing the world state and resetting it when necessary.

Specific topics

/{control_topic} => Publishes control messages with information about iterations, world state, and commands. The name of the topic is defined in the experiment configuration file (usually main_loop/control).

/{episodes_topic} => Publishes episode data including perceptions, policies, actuations, and rewards for each execution cycle. The name of the topic is defined in the experiment configuration file (usually main_loop/episodes).

class cognitive_processes.main_loop.MainLoop(*args: Any, **kwargs: Any)

Bases: CognitiveProcess

MainLoop class for managing the main loop of the system.

add_antipoint(name, sensing, node_type='pnode')

Adds an antipoint to the specified PNode.

Parameters:

• name (str) – Name of the PNode to which the antipoint is added.

• sensing (dict) – Sensing data to be used for the antipoint.

add_file(file_item)

Process a file entry (create the corresponding object) in the configuration.

Parameters:

file_item (dict) – Dictionary with the file information.

add_point(name, sensing, node_type='pnode')

Sends the request to add a point to a P-Node.

Parameters:

• name (str) – Name of the P-Node.

• sensing (dict) – Sensorial data to be added as a point.

Returns:

Success status received from the P-Node.

Return type:

bool

close_files()

Close all files when execution is finished.

publish_iteration()

Method for publishing execution data in the control topic in each iteration.

random_policy()

Selects a random policy.

Returns:

The selected policy.

Return type:

str

read_activation_callback(msg: cognitive_node_interfaces.msg.Activation)

This method receives a message from an activation topic, processes the message and updates the activation in the LTM cache.

Parameters:

msg (cognitive_node_interfaces.msg.Activation) – Message that contains the activation information.

reset_world(check_finish=True)

Reset the world if necessary, according to the experiment parameters.

Parameters:

check_finish (bool) – If True, checks if the world has finished before deciding to reset. If False, only the trial/iteration count is considered.

Returns:

True if the world was reset, False otherwise.

Return type:

bool

run()

Run the main loop of the system.

select_max_policy(policy_activations: dict)

Selects the policy with the maximum activation.

Parameters:

policy_activations (dict) – Dictionary with policy names as keys and their activations as values.

Returns:

The name of the policy with the maximum activation.

Return type:

str

select_policy(softmax=False)

Selects the policy with the higher activation. If softmax is True, it selects the policy using a softmax function. If no policy is selected, it selects a random policy.

Parameters:

softmax (bool) – If True, selects the policy using a softmax function, defaults to False.

Returns:

The selected policy.

Return type:

str

select_policy_softmax(policy_activations: dict, temperature=1)

Selects a policy using the softmax function.

Parameters:

• policy_activations (dict) – Dictionary with policy names as keys and their activations as values.

• temperature (int) – Temperature parameter for the softmax function, defaults to 1.

Returns:

The name of the selected policy.

Return type:

str

setup()

Initial configuration of the MainLoop node. This method sets up the LTM, perceptions, files, connectors, control channel, etc.

setup_control_channel()

Configures the control channel.

setup_files()

Configures the output files.

update_ltm(stm: cognitive_nodes.episode.Episode)

This method updates the LTM with the perception changes, policy executed and reward obtained.

Parameters:

stm (cognitive_processes.main_loop.Episode) – Episode object containing the information to update the LTM.

update_pnodes_reward_basis(old_perception, perception, policy, reward_list, ltm_cache)

This method creates or updates CNodes and PNodes according to the executed policy, current goal and reward obtained.

The method follows these steps: 1. Obtain the CNode(s) linked to the policy. -If there are CNodes linked to the policy, for each CNode: 2. Obtain WorldModel, Goal and PNode activation 3. Check if the WorldModel and Goal are active 4. If there is a reward an antipoint is added, if there is no reward and the PNode is active, an antipoint is added.

-If there are no CNodes connected to the policy a new CNode is created if there is reward.

Parameters:

• old_perception (dict) – Perception before the execution of the policy.

• perception (dict) – Perception after the execution of the policy.

• policy (str) – Policy executed.

• reward_list (dict) – Dictionary with the rewards obtained for each goal after the execution of the policy.

• ltm_cache (dict) – LTM cache containing the nodes and their data.

update_policies_to_test(policy=None)

Maintenance tasks on the pool of policies used to choose one randomly when needed.

When no policy is passed, the method will fill the policies to test list with all available policies in the LTM.

When a policy is passed, the policy will be removed from the policies to test list.

Parameters:

policy (str) – Policy to be removed, defaults to None.

update_status()

Method that writes the files with execution data.

world_finished()

Check if the world has finished.

Returns:

True if the world has finished, False otherwise.

Return type:

bool

class cognitive_processes.main_loop.MainLoopLight(*args: Any, **kwargs: Any)

Bases: MainLoop

MainLoopLight class for running the main loop with only action selection

execute_policy(perception, policy)

Execute a policy or utility model. This method sends a request to the policy to be executed.

Parameters:

• perception (dict) – The perception to be used in the policy execution.

• policy (str) – The policy to execute.

Returns:

The response from executing the policy.

Return type:

The executed policy.

run()

Run the main loop of the system.

select_policy(softmax=False)

Selects the policy with the higher activation. If softmax is True, it selects the policy using a softmax function. If no policy is selected, it selects a random policy.

Parameters:

softmax (bool) – If True, selects the policy using a softmax function, defaults to False.

Returns:

The selected policy.

Return type:

str

Deliberation

Python class that implements the Deliberation process of the cognitive architecture. The Deliberation process is responsible for selecting the most appropriate action or policy to execute based on the current state of the system, the predicted states and the current goal. It is instantiated in the Utility Models.

class cognitive_processes.deliberation.Deliberation(*args: Any, **kwargs: Any)

Bases: CognitiveProcess

Deliberation class: A cognitive process that allows the agent to deliberate on its actions and decisions.

generate_candidate_actions(old_perception=None, algorithm='latin')

Generates a list of candidate Episode objects based on the configured actuation dimensions. Each Episode will have its old_perception set to the argument and action.actuation set to the candidate action.

get_linked_goals()

Retrieves the goal linked to the parent node of the process

get_linked_world_model()

Retrieves the goal linked to the parent node of the process

predict_perceptions(world_model, input_episodes: list[cognitive_nodes.episode.Episode]) → list[cognitive_nodes.episode.Episode]

Predicts the expected utilities for the given input episodes using the Utility Model.

Parameters:

• world_model (dict) – The current world model.

• input_episodes (list[Episode]) – List of input episodes to predict utilities for.

Returns:

List of predicted utilities.

Return type:

list[float]

predict_utilities(input_episodes: list[cognitive_nodes.episode.Episode]) → list[float]

Predicts the expected utilities for the given input episodes using the Utility Model.

Parameters:

input_episodes (list[Episode]) – List of input episodes to predict utilities for.

Returns:

List of predicted utilities.

Return type:

list[float]

publish_episode()

Publish the current episode data to the episode topic.

run()

Main loop of the cognitive process. This method is executed in a separate thread. It runs the cognitive process, reading perceptions, updating activations, and processing episodes.

select_action(candidate_actions, predicted_episodes, expected_utilities)

Selects an action probabilistically using softmax over the expected utilities.

setup()

Initial configuration of the MainLoop node. This method sets up the LTM, perceptions, files, connectors, control channel, etc.

update_activations()

This method updates the activations of the nodes in the LTM cache.

update_ltm(stm: cognitive_nodes.episode.Episode)

This method updates the LTM with the perception changes, policy executed and reward obtained.

Parameters:

stm (cognitive_processes.main_loop.Episode) – Episode object containing the information to update the LTM.

update_pnodes_reward_basis(old_perception, perception, policy, reward_list, ltm_cache)

This method creates or updates CNodes and PNodes according to the executed policy, current goal and reward obtained.

The method follows these steps: 1. Obtain the CNode(s) linked to the policy. -If there are CNodes linked to the policy, for each CNode: 2. Obtain WorldModel, Goal and PNode activation 3. Check if the WorldModel and Goal are active 4. If there is a reward an antipoint is added, if there is no reward and the PNode is active, an antipoint is added.

-If there are no CNodes connected to the policy a new CNode is created if there is reward.

Parameters:

• old_perception (dict) – Perception before the execution of the policy.

• perception (dict) – Perception after the execution of the policy.

• policy (str) – Policy executed.

• reward_list (dict) – Dictionary with the rewards obtained for each goal after the execution of the policy.

• ltm_cache (dict) – LTM cache containing the nodes and their data.