Intrinsic Motivations

Here you can find a description of all the scripts that implement the different intrinsic motivation components, which drive autonomous exploration and learning in the cognitive architecture.

Novelty

Python script which implements components for exploration intrinsic motivation, based on novelty, which enables robots to explore their environment.

For the moment, only a random exploration is implemented.

class cognitive_nodes.novelty.DriveNovelty(*args: Any, **kwargs: Any)

Bases: Drive

DriveNovelty Class, represents a drive to explore the environment.

evaluate(perception=None)

Evaluation that always returns 1.0, as the drive is always .

Parameters:

perception (dict or Any.) – Unused perception.

Returns:

Evaluation of the Drive.

Return type:

cognitive_node_interfaces.msg.Evaluation

class cognitive_nodes.novelty.PolicyNovelty(*args: Any, **kwargs: Any)

Bases: Policy

PolicyNovelty Class, represents a policy that selects a random policy from the LTM and executes it.

configure_policies(ltm_cache)

Creates a list of eligible policies to be executed and shuffles it.

Parameters:

ltm_cache (dict) – LTM cache.

async execute_callback(request, response)

Callback that selects a policy and then executes it.

Parameters:

• request (cognitive_node_interfaces.srv.Execute.Request) – Execution request

• response (cognitive_node_interfaces.srv.Execute.Response) – Response of the execution. Includes the name of the selected policy.

Returns:

Response of the execution.

Return type:

cognitive_node_interfaces.srv.Execute.Response

ltm_change_callback(msg)

Callback that reads the LTM change and updates the policies accordingly.

Parameters:

msg (std_msgs.msg.String) – LTM change message.

request_ltm()

Requests data from the LTM.

Returns:

LTM dump.

Return type:

dict

select_policy()

Selects a policy from the queue. It begins by selecting the front policy and rotates the queue, once all the queue has been iterated, the policies are shuffled.

Returns:

Selected policy.

Return type:

str

setup()

Setup method that configures the PolicyNovelty node.

class cognitive_nodes.novelty.PolicyQueue

Bases: object

PolicyQueue Class, wrapper over the builtin deque object to create a policy queue.

dequeue()

Removes the last item from the queue.

Returns:

The last item in the queue.

Return type:

str

enqueue(item)

Adds an item to the front of the queue.

Parameters:

item (str) – Item to add to the queue.

Returns:

None

Return type:

NoneType

exists(item)

Checks if an item exists in the queue.

Parameters:

item (str) – Item to check for existence in the queue.

Returns:

True if the item exists in the queue, False otherwise.

Return type:

bool

find_differences(items)

Finds the differences between the current queue and the provided items.

Parameters:

items (list) – List of items to compare with the queue.

Returns:

Tuple containing the new items and the missing items.

Return type:

tuple (new_items, missing_items)

front()

Returns the first item in the queue.

Returns:

The first item in the queue.

Return type:

str

isEmpty()

Checks if the queue is empty.

Returns:

True if the queue is empty, False otherwise

Return type:

bool

merge(items)

Merges the current queue with the provided items. It adds new items and removes missing items.

Parameters:

items (list) – List of items to merge with the queue.

Returns:

True if there are changes, False otherwise.

Return type:

bool

rear()

Returns the last item in the queue.

Returns:

The last item in the queue.

Return type:

str

remove(item)

Removes an item from the queue.

Parameters:

item (str) – Item to remove from the queue.

Returns:

True if the item was removed, False otherwise.

Return type:

bool

select_policy()

Selects a policy from the queue. It begins by selecting the front policy and rotates the queue.

Returns:

Selected policy.

Return type:

str

shuffle(rng: numpy.random.Generator)

Shuffles the queue using the provided random number generator.

Parameters:

rng (np.random.Generator) – Random number generator to use for shuffling.

class cognitive_nodes.novelty.PolicyRandomAction(*args: Any, **kwargs: Any)

Bases: PolicyBlocking

WORK IN PROGRESS

PolicyRandomAction Class, represents a policy that executes a random low level action.

async execute_callback(request, response)

Makes a service call to the server that handles the execution of the policy.

Parameters:

• request (cognitive_node_interfaces.srv.Execute.Request) – The request to execute the policy.

• response (cognitive_node_interfaces.srv.Execute.Response) – The response indicating the executed policy.

Returns:

The response with the executed policy name.

Return type:

cognitive_node_interfaces.srv.Execute.Response

randomize_actuation()

Randomizes the actuation values.

Raises:

TypeError – Unknown type assigned to an actuator.

setup()

Setup method that configures the PolicyRandomAction node. :raises TypeError: Unknown type assigned to an actuator.

LLM Exploration

Python script which implements components for exploration intrinsic motivation, guided by a Large Language Model (LLM), which enables robots to explore their environment with the help of an LLM.

Effectance

Python script which implements components for effectance intrinsic motivation, based on detecting and reproducing effects, which enables robots to learn from the consequences of their actions in the environment.

This module includes both internal effectance (detecting P-Node consolidation) and external effectance (detecting and reproducing changes in sensor values).

class cognitive_nodes.effectance.DriveEffectanceExternal(*args: Any, **kwargs: Any)

Bases: Drive, EpisodeSubscription

Drive that detects effects in the environment. In this case, changes from 0 to 1 in a sensor.

This class inherits from the general Drive class and EpisodeSubscription class, which provides helper methods to subscribe to the episodes topic of a cognitive process.

episode_callback(msg)

Callback that processes an episode message.

Parameters:

msg (ROS Message (most cases: cognitive_node_interfaces.msg.Episode)) – Episode message.

evaluate(perception=None)

Calculates drive evaluation. If any new effect has been found, drive evaluation is 1.0.

Returns:

Drive evaluation and its timestamp.

Return type:

cognitive_node_interfaces.Evaluation

find_effects(perception, old_perception)

Checks consecutive perceptions if effects were generated.

Parameters:

• perception (dict) – Current perception.

• old_perception (dict) – Previous perception.

get_effects_callback(request, response: cognitive_node_interfaces.srv.GetEffects.Response)

Callback that provides the effects that have been found.

Parameters:

• request (cognitive_node_interfaces.srv.GetEffects.Request) – Empty request.

• response (cognitive_node_interfaces.srv.GetEffects.Response) – Sensors and attributes for which effects have been found.

Returns:

Sensors and attributes for which effects have been found.

Return type:

cognitive_node_interfaces.srv.GetEffects.Response

class cognitive_nodes.effectance.DriveEffectanceInternal(*args: Any, **kwargs: Any)

Bases: Drive, PNodeSuccess

Drive that detects internal architecture effects. In this case, the consolidation of any P-Node in the architecture.

This class inherits from the general Drive class and PNodeSuccess class, which provides helper methods to subscribe to P-Nodes success rate.

evaluate()

Calculates drive evaluation. If any P-Node is above the minimum confidence, drive evaluation is 1.0.

Returns:

Drive evaluation and its timestamp.

Return type:

cognitive_node_interfaces.Evaluation

pnode_success_callback(msg)

Callback that proccesses a success message from a P-Node.

Parameters:

msg (cognitive_node_interfaces.msg.SuccessRate) – Message with success information.

class cognitive_nodes.effectance.GoalActivatePNode(*args: Any, **kwargs: Any)

Bases: GoalLearnedSpace

Goal that provides reward when the related P-Nodes goes from not activated to activated.

This class inherits from the GoalLearnedSpace class.

calculate_reward(drive_name=None)

This goal does not take into account a drive to obtain reward. This method overrides the default behavior.

async contains_space_callback(request, response)

This method overrides the default behavior of the contains space service. Obtains checks if the space is contained in the P-Node and sends it as response.

Parameters:

• request (cognitive_node_interfaces.srv.ContainsSpace.Request) – Data of the space.

• response (cognitive_node_interfaces.srv.ContainsSpace.Response) – Boolean that indicates if the space is contained inside the goal.

Returns:

Boolean that indicates if the space is contained inside the goal.

Return type:

cognitive_node_interfaces.srv.ContainsSpace.Response

async get_reward(old_perception=None, perception=None)

Method that obtains the reward for the goal. It recieves two consecutive perceptions, calculates the related P-Node activation for each and detects if the P-Node was activated.

Parameters:

• old_perception (dict) – First state perception dictionary.

• perception (dict) – Second state perception dictionary.

Returns:

Reward and current timestamp.

Return type:

Tuple (float, builtin_interfaces.msg.Time)

read_confidence(msg: cognitive_node_interfaces.msg.SuccessRate)

Reads the confidence value from the SuccessRate message and updates the confidence attribute.

Parameters:

msg (SuccessRate) – Message containing the success rate of the P-Node.

async send_goal_space_callback(request, response)

This method overrides the default behavior of the send space service. Obtains the space from the P-Node and sends it as response.

Parameters:

• request (cognitive_node_interfaces.srv.SendSpace.Request) – Empty request.

• response (cognitive_node_interfaces.srv.SendSpace.Response) – Space data.

Returns:

Space data.

Return type:

cognitive_node_interfaces.srv.SendSpace.Response

setup_pnode()

Creates the required service clients and subscriptions.

class cognitive_nodes.effectance.GoalRecreateEffect(*args: Any, **kwargs: Any)

Bases: GoalLearnedSpace

Goal that provides reward when the related P-Nodes goes from not activated to activated.

This class inherits from the GoalLearnedSpace class.

calculate_activation(perception, activation_list)

This method extends the default calculate activation method for goals and provides activation based on the goal’s confidence.

Parameters:

• perception (dict) – The perception for which the activation will be calculated. None can be passed.

• activation_list (dict) – List of activations considered in the node.

get_reward(old_perception, perception)

Method that obtains reward for the goal. It recieves two consecutive perceptions, checks if the effect was recreated (the related attribute went from 0 to 1).

Parameters:

• old_perception (dict) – First state perception dictionary.

• perception (dict) – Second state perception dictionary

Returns:

Reward and current timestamp.

Return type:

Tuple (float, builtin_interfaces.msg.Time)

process_effect(old_perception, perception)

Method that extracts the appropriate reading from the perceptions and returns if effect is found.

Parameters:

• old_perception (dict) – First state perception dictionary.

• perception (dict) – Second state perception dictionary.

Returns:

Tuple with a boolean True if effect is found and the raw readings of the sensor’s attribute for both perceptions.

Return type:

tuple (bool, float, float)

class cognitive_nodes.effectance.PolicyEffectanceExternal(*args: Any, **kwargs: Any)

Bases: Policy

Policy that creates a goal that aims to recreate an effect in the environment.

This class inherits from the general Policy class.

async create_goal(sensor, attribute)

Method that creates a goal related to an effect and registers it in the LTM.

Parameters:

• sensor (str) – Name of the sensor to which the effect is related.

• attribute (str) – Attribute in the sensor to which the effect is related.

async execute_callback(request, response)

Callback that executes the policy.

Parameters:

• request (cognitive_node_interfaces.srv.Execute.Request) – Execution request.

• response (cognitive_node_interfaces.srv.Execute.Response) – Execution response.

Returns:

Execution response.

Return type:

cognitive_node_interfaces.srv.Execute.Response

class cognitive_nodes.effectance.PolicyEffectanceInternal(*args: Any, **kwargs: Any)

Bases: Policy, PNodeSuccess

Policy that creates a goal that aims to reach a consolidated P-Node.

This class inherits from the general Policy class and PNodeSuccess class, which provides helper methods to subscribe to P-Nodes success rate.

changes_in_pnodes(ltm_dump)

Returns True if a P-Node has been added or deleted.

Parameters:

ltm_dump (dict) – Dictionary with the data from the LTM.

Returns:

Changes in P-Nodes.

Return type:

bool

async create_goal(pnode_name)

Method that creates the Goal linked to a P-Node and registers it in the LTM.

Parameters:

pnode_name (str) – P-Node related to goal.

async execute_callback(request, response)

Callback that executes the policy.

Parameters:

• request (cognitive_node_interfaces.srv.Execute.Request) – Execution request.

• response (cognitive_node_interfaces.srv.Execute.Response) – Execution response.

Returns:

Execution response.

Return type:

cognitive_node_interfaces.srv.Execute.Response

find_goals(ltm_dump)

Creates a dictionary with the P-Nodes as keys and a list of the upstream goals as values.

Parameters:

ltm_dump (dict) – Dictionary with the data from the LTM.

Returns:

P-Node-Goal dictionary.

Return type:

dict

pnode_success_callback(msg)

Callback that proccesses a success message from a P-Node.

Parameters:

msg (cognitive_node_interfaces.msg.SuccessRate) – Message with success information.

async process_effectance()

This method proccesses the effectance policy. Selects the higher confidence P-Node. and creates a goal linked to it if the confidence threshold is exceeded.

read_ltm(ltm_dump)

Extracts information from the data provided by the LTM.

Parameters:

ltm_dump (dict) – Dictionary with the data from the LTM.

select_pnode()

Selects the P-Node with the highest confidence.

Prospection

Python script which implements components for prospection intrinsic motivation, focused on discovering hierarchical relationships between goals in the system. This enables robots to build knowledge chains by identifying when achieving one goal can lead to another.

class cognitive_nodes.prospection.PolicyProspection(*args: Any, **kwargs: Any)

Bases: Policy

PolicyProspection class. Implements a policy that links goals to P-Nodes based on prospection.

async execute_callback(request, response)

Callback that executes the policy. It sends a request to the prospection drive to get the knowledge found. Then, it links goals of the relations found.

Parameters:

• request (cognitive_node_interfaces.srv.Execute.Request) – Empty request.

• response (cognitive_node_interfaces.srv.Execute.Response) – Response with the executed policy name.

Returns:

Response with the executed policy name.

Return type:

cognitive_node_interfaces.srv.Execute.Response

class cognitive_nodes.prospection.ProspectionDrive(*args: Any, **kwargs: Any)

Bases: Drive, LTMSubscription

ProspectionDrive class. Represents a drive that searches for new knowledge using forward prospection on the LTM nodes.

changes_in_pnodes(ltm_dump)

Returns True if a P-Node has been added or deleted.

Parameters:

ltm_dump (dict) – The LTM dump to read.

Returns:

True if a P-Node has been added or deleted, False otherwise.

Return type:

bool

configure_prospection_suscriptor(ltm, callback_group)

Setup of the ProspectionDrive class. Subscribes to the LTM nodes and initializes the dictionaries for the P-Nodes, Goals, and knowledge found.

Parameters:

ltm (str) – ID of the LTM to subscribe to.

async do_prospection()

This method executes the prospection process. It iterates over the learned goals and P-Nodes to find relationships between them.

evaluate(perception=None)

Evaluates the drive depending if there is newly discovered knowledge.

Parameters:

perception (dict or NoneType) – Unused perception

find_goals(ltm_dump)

Creates a dictionary with the P-Nodes as keys and a list of the upstream goals as values.

Parameters:

ltm_dump (dict) – The LTM dump to read.

Returns:

Dictionary with P-Nodes as keys and lists of upstream goals as values.

Return type:

dict

async get_knowledge_callback(request, response)

Callback that returns the knowledge found.

Parameters:

• request (cognitive_node_interfaces.srv.GetKnowledge.Request) – Empty request.

• response (cognitive_node_interfaces.srv.GetKnowledge.Response) – Response with the found knowledge.

Returns:

Response with the found knowledge.

Return type:

cognitive_node_interfaces.srv.GetKnowledge.Response

get_neighbor_names(goal_name)

This method returns the names of the neighbors of a goal.

Parameters:

goal_name (str.) – Name of the goal.

Returns:

List of neighbor names.

Return type:

list.

has_loop(upstream_goal, downstream_goal)

Checks if adding downstream_goal to upstream_goal’s neighbors creates a loop.

Parameters:

• upstream_goal (str) – The name of the upstream goal.

• downstream_goal (str) – The name of the downstream goal.

Returns:

True if a loop would be created, False otherwise.

Return type:

bool

has_neighbor(goal_name, neighbor_name)

Checks if a goal has a neighbor.

Parameters:

• goal_name (str) – Name of the goal to check.

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

Returns:

Whether or not the goal has that neighbor.

Return type:

bool

read_ltm(ltm_dump)

Reads a LTM dump and creates the appropriate subscribers for the P-Nodes and Goals.

Parameters:

ltm_dump (dict) – The LTM dump to read.

async success_callback(msg: cognitive_node_interfaces.msg.SuccessRate)

Callback that reads the success rate of a P-Node or a Goal.

Parameters:

msg (cognitive_node_interfaces.msg.SuccessRate) – Message from P-Node or Goal.

Raises:

RuntimeError – If message recieved is not from a P-Node or Goal.

traverse_neighbors(goal_name, downstream_goal, visited)

Traverse the neighbors of a goal to check if the downstream goal is in the chain.

Parameters:

• goal_name (str) – The name of the current goal being checked.

• downstream_goal (str) – The name of the downstream goal to check for loops.

• visited (set) – Set of already visited nodes to prevent revisiting.

Returns:

True if the downstream goal is found in the chain, False otherwise.

Return type:

bool

Model creation

Python module that implements components for autonomous model creation, allowing the cognitive architecture to dynamically generate and integrate new world and utility models based on its interactions with the environment.

class cognitive_nodes.model_creation.ModelCreationDrive(*args: Any, **kwargs: Any)

Bases: Drive, ModelCreationMixin

calculate_activation(perception=None, activation_list=None)

Returns the the activation value of the Drive.

Parameters:

perception (dict) – The given perception.

Returns:

The activation of the instance and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

episode_callback(msg)

Callback that processes the episodes.

Parameters:

msg (ROS2 message. Typically cognitive_process_interfaces.msg.Episode) – Episode message.

Raises:

NotImplementedError – Method must be implemented in the subclass.

evaluate(perception=None)

Get expected valuation for a given perception.

Parameters:

perception (dict) – The given normalized perception.

Raises:

NotImplementedError – Evaluate method has to be implemented in a child class.

class cognitive_nodes.model_creation.ModelCreationMixin

Bases: LTMSubscription, EpisodeSubscription

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

read_ltm(ltm_dump)

Reads the Long-Term Memory (LTM) and populates the LTM cache.

Parameters:

ltm_dump (dict) – LTM dump to be used.

setup_connectors()

Configures the default classes for the cognitive nodes.

class cognitive_nodes.model_creation.ModelCreationPolicy(*args: Any, **kwargs: Any)

Bases: Policy, ModelCreationMixin

calculate_activation(perception=None, activation_list=None)

Calculate the activation level of the policy by obtaining that of its neighboring CNodes As in CNodes, an arbitrary perception can be propagated, calculating the final policy activation for that perception.

Parameters:

• perception (dict) – Arbitrary perception.

• activation_list (list) – List of activations of the neighbors.

Returns:

The activation of the Policy and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

episode_callback(msg)

Callback that processes the episodes.

Parameters:

msg (ROS2 message. Typically cognitive_process_interfaces.msg.Episode) – Episode message.

Raises:

NotImplementedError – Method must be implemented in the subclass.

async execute_callback(request, response)

Callback that executes the policy.

Parameters:

• request (cognitive_node_interfaces.srv.Execute.Request) – Execution request.

• response (cognitive_node_interfaces.srv.Execute.Response) – Execution response.

Returns:

Execution response.

Return type:

cognitive_node_interfaces.srv.Execute.Response