Cognitive Nodes

Here you can find a description of all the scripts to implement the different cognitive nodes, their specific topics and services, and the documentation of their methods.

The API reference of the base class CognitiveNode can be found in the core API Reference

Perception

Python module that implements the Perception cognitive node, which is responsible for processing sensory inputs and generating perceptions.

Specific topics

/perception/id/value => Anytime a new perception is generated, it is published here.

Specific services

/perception/id/set_activation => Attention mechanisms can modify a perception’s activation.

class cognitive_nodes.perception.DiscreteEventSimulatorPerception(*args: Any, **kwargs: Any)[source]

Bases: Perception

DiscreteEventSimulatorPerception class

process_and_send_reading()[source]: Method that processes the sensor values received.

class cognitive_nodes.perception.FruitShopPerception(*args: Any, **kwargs: Any)[source]

Bases: Perception

Fruit Shop Perception class

process_and_send_reading()[source]: Method that processes the sensor values received.

class cognitive_nodes.perception.OscarLLMPerception(*args: Any, **kwargs: Any)[source]

Bases: Perception

Oscar LLM Perception class

process_and_send_reading()[source]: Method that processes the sensor values received.

class cognitive_nodes.perception.Perception(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

Perception class

calculate_activation(perception=None, activation_list=None)[source]

Returns the the activation value of the instance.

Parameters:

perception (dict) – Perception does not influence the activation of the instance.
activation_list (list) – List of activations. Not used in this case.

Returns:

The activation of the instance and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

process_and_send_reading()[source]: Method that processes the sensor values received. :raise NotImplementedError: This method should be implemented in subclasses.

read_perception_callback(reading)[source]

Callback to process the sensor values.

Parameters:: reading (cognitive_node_interfaces.msg.Perception) – The sensor values.

set_activation_callback(request, response)[source]

Attention mechanisms can modify the activation of a perception instance.

Parameters:

request (cognitive_node_interfaces.srv.SetActivation.Request) – The request that contains the new activation value.
response (cognitive_node_interfaces.srv.SetActivation.Response) – The response indicating if the activation was set.

Returns:

The response indicating if the activation was set.

Return type:

cognitive_node_interfaces.srv.SetActivation.Response

set_inputs_callback(request, response)[source]

Set inputs for the perception. This method is not working yet.

Parameters:

request (cognitive_node_interfaces.SetInputs.Request) – The request that contains the inputs data.
response (cognitive_node_interfaces.SetInputs.Response) – The response that indicates if the inputs were set.

Returns:

The response that indicates if the inputs were set.

Return type:

cognitive_node_interfaces.SetInputs.Response

P-Node

Python module that implements the P-Node cognitive node, which represent perceptual equivalence classes (discretizations of continuous perceptual space) used to operationally group perceptions that lead to the same outcome under identical actions.

Spacific topics

/pnode/id/success_rate => The success rate (percentage of new points that belong to the space) of the P-Node is published here.

Specific services

/pnode/id/add_point => Add a point (or anti-point) to the underlying space.

/pnode/id/send_space => Send the underlying space to the client.

/pnode/id/contais_space => Check if the underlying space contais a given space.

class cognitive_nodes.pnode.PNode(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

P-Node class

add_neighbor_callback(request, response)[source]

Extends the default add_neighbor_callback method to process the neighbors and publish the success rate.

Parameters:

request (cognitive_node_interfaces.srv.AddNeighbor.Request) – Add neighbor request.
response (cognitive_node_interfaces.srv.AddNeighbor.Response) – Response with the result of the add neighbor operation.

Returns:

Response with the result of the add neighbor operation.

Return type:

cognitive_node_interfaces.srv.AddNeighbor.Response

add_point(point, confidence)[source]

Add a new point (or anti-point) to the P-Node.

Parameters:

point (dict) – The point that is added to the P-Node.
confidence (float) – Indicates if the perception added is a point or an antipoint.

add_point_callback(request, response)[source]

DEPRECATED: SEE add_points_callback Callback method for adding a point (or anti-point) to a specific P-Node.

Parameters:

request (cognitive_node_interfaces.srv.AddPoint.Request) – The request that contains the point that is added and its confidence.
response (cognitive_node_interfaces.srv.AddPoint.Response) – The response indicating if the point was added to the P-Node.

Returns:

The response indicating if the point was added to the P-Node.

Return type:

cognitive_node_interfaces.srv.AddPoint.Response

add_points_callback(request, response)[source]

Callback method for adding a point (or anti-point) to a specific P-Node.

Parameters:

request (cognitive_node_interfaces.srv.AddPoints.Request) – The request that contains the list of points that are added and their confidence.
response – The response indicating if the points were added to the P-Node.

Returns:

The response indicating if the points were added to the P-Node.

Return type:

cognitive_node_interfaces.srv.AddPoints.Response

calculate_activation(perception=None, activation_list=None)[source]

Calculate the new activation value for a given perception.

Parameters:

perception (dict) – The perception for which P-Node activation is calculated.
activation_list (list) – The list of activations to be used for the calculation.

Returns:

If there is space, returns the activation of the P-Node. If not, returns 0. It also returs the timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

configure_labels()[source]: Configure the labels of the space.

contains_space_callback(request, response)[source]

Callback that checks if the space contains a given space.

Parameters:

request (cognitive_node_interfaces.srv.ContainsSpace.Request) – Request that contains the space to check.
response (cognitive_node_interfaces.srv.ContainsSpace.Response) – Response that indicates if the space is contained.

Returns:

Response that indicates if the space is contained.

Return type:

cognitive_node_interfaces.srv.ContainsSpace.Response

create_activation_input(node: dict)[source]

Adds perceptions to 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)[source]

Extends the default delete_neighbor_callback method to process the neighbors and publish the success rate.

Parameters:

request (cognitive_node_interfaces.srv.DeleteNeighbor.Request) – Delete neighbor request.
response (cognitive_node_interfaces.srv.DeleteNeighbor.Response) – Response with the result of the delete neighbor operation.

Returns:

Response with the result of the delete neighbor operation.

Return type:

cognitive_node_interfaces.srv.DeleteNeighbor.Response

get_space(perception)[source]

Return the compatible space with perception. (Ugly hack just to see if this works. In that case, everything need to be checked to reduce the number of conversions between sensing, perception and space).

Parameters:: perception (dict) – The perception for which P-Node activation is calculated.
Returns:: If there is space, returns it. If not, returns None.
Return type:: cognitive_nodes.space or None

process_neighbors()[source]: Detects if the P-Node is linked to a Goal node.

publish_success_rate()[source]: Publishes the success rate of the P-Node.

read_activation_callback(msg: cognitive_node_interfaces.msg.PerceptionStamped)[source]

Callback method that reads a perception and stores it in the activation inputs list.

Parameters:: msg (cognitive_node_interfaces.msg.PerceptionStamped) – PerceptionStamped message that contains the perception and its timestamp.

save_model_callback(request, response)[source]

Save the current model to a file.

Parameters:

request (cognitive_node_interfaces.srv.SaveModel.Request) – The request that contains the prefix and suffix for the file name.
response (cognitive_node_interfaces.srv.SaveModel.Response) – The response that contains the saved model path and success status.

Returns:

The response that contains the saved model path and success status.

Return type:

cognitive_node_interfaces.srv.SaveModel.Response

send_pnode_space_callback(request, response)[source]

Callback that sends the space of the P-Node.

Parameters:

request (cognitive_node_interfaces.srv.SendGoalSpace.Request) – Empty request.
response (cognitive_node_interfaces.srv.SendGoalSpace.Response) – Response that contains the space of the P-Node.

Returns:

Response that contains the space of the P-Node.

Return type:

cognitive_node_interfaces.srv.SendGoalSpace.Response

update_history(confidence)[source]

Updates the history of the P-Node with the new confidence value (point or anti-point).

Parameters:: confidence (int) – Confidence value of the new point or anti-point.

RobotPurpose

Python module that implements the Needs and Missions of the cognitive architecture, which represent the desired motivational state of the system. Needs represent intrinsic motivations, while Missions represent extrinsic motivations.

Specific services

/robot_purpose/id/set_activation => Service to set the activation of the robot purpose.

/robot_purpose/id/get_satisfaction => Obtain the satisfaction value of a need (1.0 if satisfied, 0.0 if not).

class cognitive_nodes.robot_purpose.AlignmentMission(*args: Any, **kwargs: Any)[source]

Bases: RobotPurpose

” Need Class for Alignment purposes.

calculate_satisfaction()[source]

Calculate whether the need is satisfied.

Returns:: True if the need is satisfied, False otherwise.
Return type:: bool

class cognitive_nodes.robot_purpose.RobotPurpose(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

” Robot Purpose Class.

calculate_activation(perception=None, activation_list=None)[source]

Returns the the activation value of the robot purpose.

Parameters:

perception (dict.) – Perception does not influence the activation.
activation_list (list) – Activation list does not influence the activation.

Returns:

The activation of the instance and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

calculate_satisfaction()[source]

Calculate whether the robot purpose is satisfied.

Returns:: True if the robot purpose is satisfied, False otherwise.
Return type:: bool

get_satisfaction_callback(request: cognitive_node_interfaces.srv.IsSatisfied.Request, response: cognitive_node_interfaces.srv.IsSatisfied.Response)[source]

Check if the robot purpose has been satisfied.

Parameters:

request (cognitive_node_interfaces.srv.IsSatisfied.Request) – Empty request.
response (cognitive_node_interfaces.srv.IsSatisfied.Response) – Response that indicates if the robot purpose is satisfied or not.

Returns:

Response that indicates if the robot purpose is satisfied or not.

Return type:

cognitive_node_interfaces.srv.IsSatisfied.Response

read_evaluation_callback(msg: cognitive_node_interfaces.msg.Evaluation)[source]

Callback that reads the evaluation of a Drive node. Used to check if the robot purpose is satisfied.

Parameters:: msg (cognitive_node_interfaces.msg.Evaluation) – Message containing the evaluation of the Drive node.

set_activation_callback(request, response)[source]

Service to set the activation of the robot purpose.

Parameters:

request (cognitive_node_interfaces.srv.SetActivation.Request) – The request that contains the new activation value.
response (cognitive_node_interfaces.srv.SetActivation.Response) – The response indicating if the activation was set.

Returns:

The response indicating if the activation was set.

Return type:

cognitive_node_interfaces.srv.SetActivation.Response

Drive

Python module that implements the Drives, which are defined as a fuction that provides a measure (evaluation) of how desirable the satisfaction of a motivational desire (Need or Mission) is.

Specific topics

/drive/id/evaluation => Publish the evaluation of the drive.

Specific services

/drive/id/set_activation => Needs and missions can modify a drive’s activation.

/drive/id/get_sucess_rate => Get a prediction success rate based on a historic of previous predictions.

/drive/id/get_reward => Obtain the reward value. A Drive gives reward if its evaluation value decreases.

/drive/id/get_effects => In case of DriveEffectanceExternal, this service returns the effects that were found in the environment.

/drive/id/get_knowledge => In case of ProspectionDrive, this service returns the knowledge that was found.

class cognitive_nodes.drive.Drive(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

Drive class

calculate_activation(perception=None, activation_list=None)[source]

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

evaluate(perception=None)[source]

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.

get_reward()[source]

Returns the latest reward and its timestamp.

Returns:: The latest reward and its timestamp.
Return type:: Tuple[float, builtin_interfaces.msg.Time]

get_reward_callback(request, response)[source]

Callback method to calculate the reward obtained.

Parameters:

request (cognitive_node_interfaces.srv.GetReward.Request) – Request that includes the new perception to check the reward.
response (cognitive_node_interfaces.srv.GetReward.Response) – Response that contais the reward.

Returns:

Response that contais the reward.

Return type:

cognitive_node_interfaces.srv.GetReward.Response

get_success_rate_callback(request, response)[source]

Get a prediction success rate based on a historic of previous predictions.

Parameters:

request (cognitive_node_interfaces.srv.GetSuccessRate.Request) – Empty request.
response (cognitive_node_interfaces.srv.GetSuccessRate.Response) – The response that contains the predicted success rate.

Returns:

The response that contains the predicted success rate.

Return type:

cognitive_node_interfaces.srv.GetSuccessRate.Response

publish_evaluation_callback()[source]: Callback for publishing evaluation.

set_activation_callback(request, response)[source]

Robot purposes can modify a drive’s activation.

Parameters:

request (cognitive_node_interfaces.srv.SetActivation.Request) – The request that contains the new activation value.
response (cognitive_node_interfaces.srv.SetActivation.Response) – The response indicating if the activation was set.

Returns:

The response indicating if the activation was set.

Return type:

cognitive_node_interfaces.srv.SetActivation.Response

class cognitive_nodes.drive.DriveExponential(*args: Any, **kwargs: Any)[source]

Bases: DriveTopicInput

evaluate(perception=None)[source]

Evaluates the drive value according to an exponential function.

Parameters:: perception (dict) – The given normalized perception.
Returns:: The valuation of the perception and its timestamp.
Return type:: cognitive_node_interfaces.msg.Evaluation

class cognitive_nodes.drive.DriveLLM(*args: Any, **kwargs: Any)[source]

Bases: Drive

calculate_reward()[source]: Calculates the reward depending if the evaluation value increases or decreases.

create_topics()[source]

Creates the list of input topics based on the drive function.

Returns:: The list of input topics.
Return type:: list

evaluate(final_values=None, updated_perceptions=None)[source]

Evaluates the drive value according to the drive function provided by the LLM.

Parameters:

final_values (dict) – The final values extracted from the perceptions.
updated_perceptions (dict) – Dictionary indicating which perceptions have been updated.

Returns:

The valuation of the perception and its timestamp.

Return type:

cognitive_node_interfaces.msg.Evaluation

extract_variables(drive_function)[source]

Extracts the variables (sensors and perceptions) from the drive function string.

Parameters:: drive_function (str) – The drive function as a string.
Returns:: The list of sensors extracted from the drive function.
Return type:: list

get_reward()[source]

Returns the latest reward obtained.

Returns:: Reward and timestamp.
Return type:: Tuple (float, builtin_interfaces.msg.Time)

async publish_activation_callback()[source]: Timed publish of the activation value. This method will calculate the activation based on the evaluation of the drive and the activation of its neighbors, and then publish it in the corresponding topic.

async read_input_callback(msg)[source]

Reads a message from the input topic and updates the evaluation and reward obtained.

Parameters:: msg (Configurable (Typically std_msgs.msg.Float32)) – Input data message.

class cognitive_nodes.drive.DriveTopicInput(*args: Any, **kwargs: Any)[source]

Bases: Drive

Drive class that reads an input topic to obtain its evaluation value.

calculate_reward()[source]: Calculates the reward depending if the evaluation value increases or decreases.

get_reward()[source]

Returns the latest reward obtained.

Returns:: Reward and timestamp.
Return type:: Tuple (float, builtin_interfaces.msg.Time)

async publish_activation_callback()[source]: Timed publish of the activation value. This method will calculate the activation based on the evaluation of the drive and the activation of its neighbors, and then publish it in the corresponding topic.

read_input_callback(msg)[source]

Reads a message from the input topic and updates the evaluation and reward obtained.

Parameters:: msg (Configurable (Typically std_msgs.msg.Float32)) – Input data message.

Goal

Python module that implements the Goal cognitive node, which represents an area in the state space that, when reached, lead to the reduction of at least one of the drives that are part of the robot’s motivational system. That is, it is implicitly a rewarded area.

Spacific topics

/goal/id/confidence => The learning confidence of the goal’s space, if exists, is published here, measured as the success rate of the predictions of received rewards.

Specific services

/goal/id/set_activation => Drives can modify a goal’s activation.

/goal/id/get_reward => Obtains the reward value.

/goal/id/is_reached => Check if the goal has been reached.

/goal/id/send_space => Send the underlying space to the client, if exists.

/goal/id/contains_space => Check if the underlying space contains a given space, if the goal has one.

class cognitive_nodes.goal.Goal(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

Goal class.

add_point(point, confidence)[source]

Placeholder method in base goals. To be implemented in derived classes.

Parameters:

point (dict) – The point that is added to the Goal.
confidence (float) – Indicates if the perception added is a point or an antipoint.

add_point_callback(request, response)[source]

Callback method for adding a point (or anti-point) to a specific Goal.

Parameters:

request (cognitive_node_interfaces.srv.AddPoint.Request) – The request that contains the point that is added and its confidence.
response (cognitive_node_interfaces.srv.AddPoint.Response) – The response indicating if the point was added to the Goal.

Returns:

The response indicating if the point was added to the Goal.

Return type:

cognitive_node_interfaces.srv.AddPoint.Response

async get_reward(old_perception=None, perception=None)[source]

Calculate the reward for the current sensor values.

This is a placeholder for the get reward method that must be implemented according to the required experiment/application. It is a asyncio corrutine so that service calls can be awaited.

Parameters:

old_perception (Any) – The previous perception data. Defaults to None.
perception (Any) – The current perception data. Defaults to None.

Raises:

NotImplementedError – If the method is not overridden in a subclass.

async get_reward_callback(request, response)[source]

Callback method to calculate the reward obtained.

Parameters:

request (cognitive_node_interfaces.srv.GetReward.Request) – Request that includes the new perception to check the reward.
response (cognitive_node_interfaces.srv.GetReward.Response) – Response that contais the reward.

Returns:

Response that contais the reward.

Return type:

cognitive_node_interfaces.srv.GetReward.Response

async is_reached_callback(request, response)[source]

Check if the goal has been reached.

Parameters:

request (cognitive_node_interfaces.srv.IsReached.Request) – Request that includes the new perception to check.
response (cognitive_node_interfaces.srv.IsReached.Response) – Response that indicates if the goal is reached or not.

Returns:

Response that indicates if the goal is reached or not.

Return type:

cognitive_node_interfaces.srv.IsReached.Response

send_goal_space_callback(request, response)[source]

Callback that sends the goal space data.

Parameters:

request (cognitive_node_interfaces.srv.SendSpace.Request) – Empty request.
response (cognitive_node_interfaces.srv.SendSpace.Response) – Response that contains the goal space data.

Returns:

Response containing the goal space data.

Return type:

cognitive_node_interfaces.srv.SendSpace.Response

set_activation_callback(request, response)[source]

Drives can modify a goals’s activation.

Parameters:

request (cognitive_node_interfaces.srv.SetActivation.Request) – The request that contains the new activation value.
response (cognitive_node_interfaces.srv.SetActivation.Response) – The response indicating if the activation was set.

Returns:

The response indicating if the activation was set.

Return type:

cognitive_node_interfaces.srv.SetActivation.Response

class cognitive_nodes.goal.GoalLearnedSpace(*args: Any, **kwargs: Any)[source]

Bases: GoalMotiven

Class that extends the functionality of the GoalMotiven class by adding a space to store goal state space.

add_point(point, confidence)[source]

Add a new point (or anti-point) to the Goal.

Parameters:

point (dict) – The point that is added to the Goal.
confidence (float) – Indicates if the perception added is a point or an antipoint.

configure_labels()[source]: Configure the labels of the space.

contains_space_callback(request, response)[source]

Callback that checks if the space contains a given space.

Parameters:

request (cognitive_node_interfaces.srv.ContainsSpace.Request) – Request that contains the space to check.
response (cognitive_node_interfaces.srv.ContainsSpace.Response) – Response that indicates if the space is contained.

Returns:

Response that indicates if the space is contained.

Return type:

cognitive_node_interfaces.srv.ContainsSpace.Response

get_drive()[source]

Returns the name of the linked drive.

Returns:: Drive name.
Return type:: str

get_expected_reward(perception: dict)[source]

Calculate the expected reward of the goal based on the perception and the goal state.

Parameters:: perception (dict) – Perception to evaluate.
Returns:: Expected reward.
Return type:: float

async get_reward(old_perception=None, perception=None)[source]

Calculate the reward of the goal based on the perception and the reward space or the evaluation of the drive. Updates the space acording to the reward obtained.

Parameters:

old_perception (dict) – First perception. Not used.
perception (dict) – Second perception. Not used.

Returns:

The reward obtained and its timestamp.

Return type:

Tuple (float, builtin_interfaces.msg.Time)

linked_drive()[source]

Check if there is a drived linked to the goal.

Returns:: Value that indicates if there is a linked drive.
Return type:: bool

publish_success_rate()[source]: Publish the success rate of the goal.

send_goal_space_callback(request, response)[source]

Callback that sends the space of the goal.

Parameters:

request (cognitive_node_interfaces.srv.SendGoalSpace.Request) – Empty request
response (cognitive_node_interfaces.srv.SendGoalSpace.Response) – Response that contains the space of the goal.

Returns:

Response that contains the space of the goal.

Return type:

cognitive_node_interfaces.srv.SendGoalSpace.Response

update_space(reward, expected_reward, perception)[source]

Update the space of the goal based on the reward obtained and the expected reward. Adds point or antipoint to the space, updates the confidence score of the goal.

Parameters:

reward (float) – Real reward (from a Drive) obtained.
expected_reward (float) – Reward expected from the state space of the goal.
perception (dict) – Perception that corresponds to the reward obtained.

class cognitive_nodes.goal.GoalMotiven(*args: Any, **kwargs: Any)[source]

Bases: Goal

Class that implements a Goal that aims at minimizing a drive.

add_neighbor_callback(request, response)[source]

This method extends the base add_neighbor_callback by handling the addition of a Drive node.

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

calculate_activation(perception, activation_list)[source]

Calculates the activation of the goal based on the activations of the neighboring drives and goals.

Parameters:

perception (dict) – Unused perception.
activation_list (list) – List of activations of the neighboring nodes.

calculate_reward(drive_name)[source]

Calculates the reward of the goal based on the evaluation of the Drive node.

Parameters:: drive_name (str) – Name of the drive node.

configure_drive_inputs(neighbors)[source]

Reads the neighbors list of the goal and configures inputs for each drive.

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

create_drive_input(drive: dict)[source]

Creates a new drive input if it does not already exist.

Parameters:: drive (dict) – A dictionary containing the drive’s name and node type.
Raises:: KeyError – If the ‘name’ or ‘node_type’ key is missing in the drive dictionary.

delete_drive_input(drive: dict)[source]

Deletes the drive input and its associated subscription.

Parameters:: drive (dict) – The drive input to be deleted.

delete_neighbor_callback(request, response)[source]

This method extends the base delete_neighbor_callback by handling the deletion of a Drive node.

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

get_reward(old_perception=None, perception=None)[source]

Returns the reward of the goal.

Parameters:

old_perception (Any) – Unused perception, defaults to None.
perception (Any.) – Unused perception.

Returns:

Reward of the goal and its timestamp.

Return type:

tuple (float, builtin_interfaces.msg.Time)

read_evaluation_callback(msg: cognitive_node_interfaces.msg.Evaluation)[source]

Callback that reads the evaluation of a Drive node. It updates the reward of the goal.

Parameters:: msg (cognitive_node_interfaces.msg.Evaluation) – Message containing the evaluation of the Drive node

class cognitive_nodes.goal.GoalObjectInBoxStandalone(*args: Any, **kwargs: Any)[source]

Bases: Goal

Goal representing the desire of putting an object in a box.

–DEPRECATED, prefer the use of GoalMotiven class–

ball_and_box_on_the_same_side()[source]

Check if an object and a box are on the same side.

Returns:: A value that indicates if the object is in the same side or not.
Return type:: bool

calculate_activation(perception=None, activation_list=None)[source]

Returns the the activation value of the goal.

Parameters:

perception (dict) – Perception does not influence the activation.
activation_list (list) – List of activations. Not used.

Returns:

The activation of the goal and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

calculate_closest_position(angle)[source]

Calculate the closest position from a given cylinder angle.

Parameters:: angle (float) – The given angle.
Returns:: The closest distance and angle.
Return type:: float, float

denormalize(type, value)[source]

Denormalize a normalized value based on the type of measurement.

Parameters:

type (str) – The type of measurement (e.g., ‘distance’, ‘angle’, ‘diameter’).
value (float) – The normalized value to be denormalized.

Raises:

Exception – If normalization values are not defined.
ValueError – If the type is not recognized.

Returns:

The denormalized value.

Return type:

float

get_iteration_callback(msg: cognitive_processes_interfaces.msg.ControlMsg)[source]

Get the iteration of the experiment, if necessary.

Parameters:: msg (ControlMsg) – The control message containing the iteration information.

async get_reward(old_perception=None, perception=None)[source]

Calculate the reward for the current sensor values.

Parameters:

old_perception (Any) – The previous perception. Not used.
perception (Any) – The current perception. Not used.

Returns:

The reward obtained.

Return type:

float

hand_was_changed()[source]

Check if the held object changed from one hand to another.

Returns:: A value that indicates if the hand changed moved or not
Return type:: bool

new_from_configuration_file(data)[source]

Create attributes from the data configuration dictionary.

Parameters:: data (dict) – The configuration file.

object_held()[source]

Check if an object is held with one hand.

Returns:: A value that indicates if the object is held or not.
Return type:: bool

object_held_before()[source]

Check if an object was held with one hand.

Returns:: A value that indicates if the object was held or not.
Return type:: bool

object_held_with_left_hand()[source]

Check if an object is held with the left hand.

Returns:: A value that indicates if the object is held or not.
Return type:: bool

object_held_with_right_hand()[source]

Check if an object is held with the right hand.

Returns:: A value that indicates if the object is held or not.
Return type:: bool

object_held_with_two_hands()[source]

Check if an object is held with two hands.

Returns:: A value that indicates if the object is held or not.
Return type:: bool

async object_in_close_box()[source]

Check if there is an object inside of a box.

Returns:: A value that indicates if the object is inside or not.
Return type:: bool

async object_in_far_box()[source]

Check if there is an object inside of a box.

Returns:: A value that indicates if the object is inside or not.
Return type:: bool

async object_pickable_with_two_hands()[source]

Check if an object can be hold with two hands.

Returns:: A value that indicates if the object can be hold or not.
Return type:: bool

object_pickable_with_two_hands_request(distance, angle)[source]

Check of an object is pickable with the two hands of the robot.

Parameters:

angle (float) – The distance of the object relative to the robot.
angle – The angle of the object relative to the robot.

Returns:

A value that indicates if the object is pickable or not.

Return type:

bool

object_too_far(distance, angle)[source]

Check is an object is too far.

Parameters:

distance (float) – Distance of the object relative to the robot.
angle (float) – Angle of the object relative to the robot.

Returns:

Value that indicates if the objet is too far or not.

Return type:

bool

async object_was_approximated()[source]

Check if an object was moved towards the robot’s reachable area.

Returns:: A value that indicates if the object can be moved or not.
Return type:: bool

object_with_robot()[source]

Check if there is an object adjacent to the robot.

Returns:: A value that indicates if the object is adjacent or not.
Return type:: bool

sensorial_changes()[source]

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

Returns:: Value that indicates if the percepction values changed or not.
Return type:: bool

class cognitive_nodes.goal.GoalReadPublishedReward(*args: Any, **kwargs: Any)[source]

Bases: Goal

Goal that reads the reward obtained from a topic.

–DEPRECATED, prefer the use of GoalMotiven class–

calculate_activation(perception=None, activation_list=None)[source]

Returns the the activation value of the goal

Parameters:

perception (dict) – Perception does not influence the activation.
activation_list (list) – List of activations. Not used.

Returns:

The activation of the goal and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

get_iteration_callback(msg: cognitive_processes_interfaces.msg.ControlMsg)[source]

Get the iteration of the experiment, if necessary

Returns:: The iteration of the experiment
Return type:: int

get_reward(old_perception=None, perception=None)[source]

Returns the reward obtained.

Parameters:

old_perception (Any) – Unused perception.
perception (Any) – Unused perception.

Returns:

Obtained reward

Return type:

float

new_from_configuration_file(data)[source]

Create attributes from the data configuration dictionary.

Parameters:: data (dict) – The configuration file.

reward_topic_callback(msg)[source]

Callback that reads the reward message. Sets a flag that indicates that reward is updated.

Parameters:: msg (ROS Message (Typically std_msgs.Float32)) – Reward message.

Episodes

Provides a data structure to represent episodes. It also provides methods for its manipulation and conversion between different formats.

class cognitive_nodes.episode.Action(actuation={}, policy_id=None)[source]

Bases: object

Action class used to represent an action in the cognitive architecture.

class cognitive_nodes.episode.Episode(old_perception=None, parent_policy='', action=None, perception=None, reward_list=None)[source]

Bases: object

Episode class that represents a single episode in the cognitive architecture.

cognitive_nodes.episode.action_msg_list_to_obj_list(action_msg_list: list[cognitive_node_interfaces.msg.Action]) → list[Action][source]

Convert a list of ROS2 action messages to a list of Action objects.

Parameters:: action_msg_list (list[cognitive_node_interfaces.msg.Action]) – List of ROS2 action messages.
Returns:: List of Action objects.
Return type:: list[Action]

cognitive_nodes.episode.action_msg_to_obj(action_msg) → Action[source]

Convert a ROS2 action message to an Action object.

Parameters:: action_msg (cognitive_node_interfaces.msg.Action) – The ROS2 action message.
Returns:: An Action object.
Return type:: Action

cognitive_nodes.episode.action_obj_list_to_msg_list(action_list: list[Action]) → list[cognitive_node_interfaces.msg.Action][source]

Convert a list of Action objects to a list of ROS2 action messages.

Parameters:: action_list (list[Action]) – List of Action objects.
Returns:: List of ROS2 action messages.
Return type:: list[cognitive_node_interfaces.msg.Action]

cognitive_nodes.episode.action_obj_to_msg(action: Action)[source]

Convert an Action object to a ROS2 action message.

Parameters:: action (Action) – The Action object.
Returns:: A ROS2 action message.
Return type:: cognitive_node_interfaces.msg.Action

cognitive_nodes.episode.episode_msg_list_to_obj_list(episode_msg_list: list[cognitive_node_interfaces.msg.Episode]) → list[Episode][source]

Convert a list of ROS2 Episode messages to a list of Episode objects.

Parameters:: episode_msg_list (list[cognitive_node_interfaces.msg.Episode]) – List of ROS2 Episode messages.
Returns:: List of Episode objects.
Return type:: list[Episode]

cognitive_nodes.episode.episode_msg_to_obj(episode_msg: cognitive_node_interfaces.msg.Episode) → Episode[source]

Convert a ROS2 Episode message to an Episode object.

Parameters:: episode_msg (cognitive_node_interfaces.msg.Episode) – The ROS2 Episode message.
Returns:: An Episode object.
Return type:: Episode

cognitive_nodes.episode.episode_obj_list_to_msg_list(episode_list: list[Episode]) → list[cognitive_node_interfaces.msg.Episode][source]

Convert a list of Episode objects to a list of ROS2 Episode messages.

Parameters:: episode_list (list[Episode]) – List of Episode objects.
Returns:: List of ROS2 Episode messages.
Return type:: list[cognitive_node_interfaces.msg.Episode]

cognitive_nodes.episode.episode_obj_to_msg(episode: Episode) → cognitive_node_interfaces.msg.Episode[source]

Convert an Episode object to a ROS2 Episode message.

Parameters:: episode (Episode) – The Episode object.
Returns:: A ROS2 Episode message.
Return type:: cognitive_node_interfaces.msg.Episode

cognitive_nodes.episode.reward_dict_to_msg(reward_dict)[source]

Convert a reward dictionary to a ROS2 message format.

Parameters:: reward_dict (dict) – The reward dictionary.
Returns:: A ROS2 message representing the reward.
Return type:: cognitive_node_interfaces.msg.Reward

cognitive_nodes.episode.reward_msg_to_dict(reward_msg: cognitive_processes_interfaces.msg.RewardList) → dict[source]

Convert a ROS2 reward message to a dictionary.

Parameters:: reward_msg (cognitive_node_interfaces.msg.RewardList) – The ROS2 reward message.
Returns:: A dictionary representing the rewards.
Return type:: dict

Episodic Buffer

Python module that implements episodic buffers of different types. These can be instantiated and used by other cognitive nodes to store their short-term memories. It provides methods for the manipulation of the buffer and its contents.

class cognitive_nodes.episodic_buffer.EpisodicBuffer(node: core.cognitive_node.CognitiveNode, main_size, secondary_size, train_split=1.0, inputs=[], outputs=[], random_seed=0, **params)[source]

Bases: object

Class that creates a buffer of episodes to be used as a STM and learn models. It supports a main buffer for training and a secondary buffer for testing.

add_episode(episode: Episode, reward=0.0)[source]

Add an episode to the buffer.

Validates content, updates labels, and routes the episode to the main or secondary buffer based on train_split.

Parameters:

episode (Episode) – Episode instance to add.
reward (float) – Optional reward value (unused in EpisodicBuffer; used by TraceBuffer overrides).

Returns:

None

Return type:

None

Notes: - Empty episodes (w.r.t configured inputs/outputs) are skipped with a warning. - Labels are configured lazily on the first episode and updated when new dimensions appear.

static buffer_to_dataframe(buffer, labels)[source]

Converts a buffer of episodes to a pandas DataFrame using the given labels.

Parameters:

buffer (deque) – Buffer of episodes to convert.
labels (list) – Labels to use for the DataFrame columns.

Returns:

DataFrame containing the episodes in the buffer.

Return type:

pd.DataFrame

static buffer_to_dict_list(buffer, labels)[source]: Converts a buffer of episodes to a list of dicts using the given labels.

static buffer_to_matrix(buffer, labels)[source]

Converts a buffer of episodes to a numpy matrix using the given labels.

Parameters:

buffer (deque) – Buffer of episodes to convert.
labels (list) – Labels to use for the matrix columns.

Returns:

Numpy matrix containing the episodes in the buffer.

Return type:

np.ndarray

clear()[source]: Clears the episodic buffer.

configure_labels(episode: Episode)[source]

Creates the label list.

Parameters:: episode (cognitive_node_interfaces.msg.Episode) – Episode object.

create_dataframes()[source]: Creates pandas DataFrames from the main and secondary buffers.

static empty_episode(episode: Episode, inputs: list, outputs: list)[source]

Checks if an episode is empty.

Parameters:

episode (Episode) – Episode object.
inputs (list) – Input labels to check.
outputs (list) – Output labels to check.

Returns:

True if the episode is empty, False otherwise.

Return type:

bool

static episode_to_flat_dict(episode: Episode, labels)[source]

Converts an episode to a dict representation matching the labels.

Parameters:

episode (Episode) – The episode to convert.
labels (list) – The labels to match in the dict representation.

Returns:

A dict representation of the episode matching the labels.

Return type:

dict

static episode_to_vector(episode: Episode, labels)[source]

Converts an episode to a vector representation ordered according to the given labels.

Parameters:

episode (Episode) – Episode object to convert.
labels (list) – Labels to match in the vector representation.

Returns:

Vector representation of the episode.

Return type:

np.ndarray

get_dataframes()[source]

Returns the DataFrames of the main and secondary buffers.

Returns:: Tuple with the main and secondary DataFrames.
Return type:: tuple

get_dataset(shuffle=False, n_samples=None)[source]

Returns the dataset as numpy arrays.

Parameters:

shuffle (bool, optional) – Option to shuffle the dataset , defaults to False
n_samples (int, optional) – Option to limit the number of samples returned, defaults to None

Returns:

Tuple containing training inputs, training outputs, test inputs, and test outputs as numpy arrays.

Return type:

tuple

get_input_labels()[source]

Returns the input labels of the episodic buffer.

Returns:: List of input labels.
Return type:: list

get_output_labels()[source]

Returns the output labels of the episodic buffer.

Returns:: List of output labels.
Return type:: list

get_sample(index, main=True)[source]

Method to obtain a sample from the buffer.

Parameters:

index (int) – Index of the sample to obtain.
main (bool) – Whether to get the sample from the main buffer or secondary buffer.

Returns:

The requested sample.

Return type:

list

get_test_samples(shuffle=False, n_samples=None)[source]

Returns the test samples as lists of input and output dicts.

Parameters:

shuffle (bool, optional) – Option to shuffle the samples, defaults to False
n_samples (int, optional) – Option to limit the number of samples returned, defaults to None

Returns:

Tuple containing test inputs and test outputs as numpy arrays.

Return type:

tuple

get_train_samples(shuffle=False, n_samples=None)[source]

Returns the training samples as lists of input and output dicts.

Parameters:

shuffle (bool, optional) – Option to shuffle the samples, defaults to False
n_samples (int, optional) – Option to limit the number of samples returned, defaults to None

Returns:

Tuple containing training inputs and training outputs as numpy arrays.

Return type:

tuple

is_compatible(episode: Episode)[source]

Checks if the episode is compatible with the current buffer configuration.

Parameters:: episode (cognitive_node_interfaces.msg.Episode) – Episode object to check compatibility.
Returns:: True if compatible, False otherwise.
Return type:: bool

property main_max_size

Returns the max size of the main buffer.

Returns:: The maximum size of the main buffer.
Return type:: int

property main_size

Returns the current size of the main buffer.

Returns:: The current size of the main buffer.
Return type:: int

static matrix_to_buffer(matrix, labels)[source]

Converts a numpy matrix to a buffer of episodes using the given labels.

Parameters:

matrix (np.ndarray) – Numpy matrix to convert.
labels (list) – Labels to use for the episodes.

Returns:

Buffer of episodes created from the matrix.

Return type:

list

remove_episode(index=None, remove_from_main=True)[source]

Remove an episode from the buffer.

If index is provided, removes the episode at that position; otherwise, removes the oldest (leftmost) episode from the selected buffer.

Parameters:

index (int | None) – Position to remove. If None, removes the oldest entry.
remove_from_main (bool) – True to remove from main_buffer; False for secondary_buffer.

Returns:

None

Return type:

None

reset_new_sample_count(main=True, secondary=True)[source]

Resets the new sample count for the main and/or secondary buffers.

Parameters:

main (bool) – Whether to reset the main buffer count, defaults to True.
secondary (bool) – Whether to reset the secondary buffer count, defaults to True.

property secondary_max_size

Returns the max size of the secondary buffer.

Returns:: The maximum size of the secondary buffer.
Return type:: int

property secondary_size

Returns the current size of the secondary buffer.

Returns:: The current size of the secondary buffer.
Return type:: int

update_labels(episode: Episode)[source]

Updates the label list based on the given episode.

Parameters:: episode (cognitive_node_interfaces.msg.Episode) – Episode object.

static vector_to_episode(vector, labels)[source]

Converts a vector representation to an episode object.

Parameters:

vector (np.ndarray) – Vector representation of the episode.
labels (list) – Labels to match in the vector representation.

Raises:

ValueError – If the length of the vector does not match the number of labels.

Returns:

Episode object created from the vector.

Return type:

Episode

class cognitive_nodes.episodic_buffer.TraceBuffer(node, main_size, secondary_size=0, max_traces=10, min_traces=1, max_antitraces=5, train_split=1.0, inputs=[], outputs=[], evaluation_method='linear', reward_factor=1.0, random_seed=0, **params)[source]

Bases: EpisodicBuffer

Trace Buffer class, a specialized version of the Episodic Buffer that stores traces of episodes.

add_antitrace()[source]: Add an antitrace to the buffer if enough traces exist.

add_episode(episode, reward=0.0)[source]

Add an episode to the trace buffer and complete the trace if reward is positive.

Appends the episode to the main buffer. If a positive reward is provided, evaluates utilities for all episodes in the current trace, stores the trace, and clears the buffer to begin a new trace.

Parameters:

episode (Episode) – Episode instance to add to the current trace.
reward (float, optional) – Reward value for the trace; positive values trigger trace completion, defaults to 0.0

Raises:

ValueError – If episode is not of type Episode.

Returns:

None

Return type:

None

Notes: - Only episodes of type Episode are accepted. - Labels are configured on the first episode if not already set. - Positive rewards trigger utility evaluation and trace storage. - The buffer is cleared after storing a successful trace.

static eval_default(reward, min_val, reward_factor, n, full_length)[source]

Default evaluation method placeholder.

Parameters:

reward (float) – Final reward for the trace.
min_val (float) – Minimum utility value derived from the reward (e.g., reward * min_utility_fraction).
reward_factor (float) – Multiplicative factor applied to the final utility value.
n (int) – Number of episodes in the current trace.
full_length (int) – Maximum possible length of a trace (main buffer capacity).

Raises:

NotImplementedError – Always, as this method is a placeholder.

static eval_exponential(reward, min_val, reward_factor, n, full_length)[source]

Exponential evaluation: grow utilities exponentially towards the final reward.

Uses k = ln(reward/min_val) / (full_length - 1) and evaluates utilities at positions [full_length - n, …, full_length - 2], appending reward * reward_factor as the final value.

Parameters:

reward (float) – Final reward for the trace.
min_val (float) – Minimum utility value derived from the reward; adjusted if non-positive.
reward_factor (float) – Multiplicative factor applied to the final utility value.
n (int) – Number of episodes in the current trace.
full_length (int) – Maximum possible length of a trace (main buffer capacity).

Returns:

List of length n with exponentially increasing utility values.

Return type:

list[float]

static eval_goal_only(reward, min_val, reward_factor, n, full_length)[source]

Goal-only evaluation: zero utility except at the final episode.

Assigns 0.0 to all episodes except the last, which receives reward * reward_factor.

Parameters:

reward (float) – Final reward for the trace.
min_val (float) – Minimum utility value (unused in this method).
reward_factor (float) – Multiplicative factor applied to the final utility value.
n (int) – Number of episodes in the current trace.
full_length (int) – Maximum possible length of a trace (unused in this method).

Returns:

List of length n with zero utilities except the final element.

Return type:

list[float]

static eval_linear(reward, min_val, reward_factor, n, full_length)[source]

Linear evaluation: interpolate utilities from a start value to the final reward.

Computes a start value consistent with the full-length trace and linearly interpolates utilities across episodes, ending at reward * reward_factor.

Parameters:

reward (float) – Final reward for the trace.
min_val (float) – Minimum utility value derived from the reward.
reward_factor (float) – Multiplicative factor applied to the final utility value.
n (int) – Number of episodes in the current trace.
full_length (int) – Maximum possible length of a trace (main buffer capacity).

Returns:

List of length n with linearly increasing utility values.

Return type:

list[float]

evaluate_trace(reward)[source]

Compute utility values for each episode in the current trace.

Uses the configured evaluation method to assign utility values to all episodes in the main buffer based on the final reward. The evaluation method determines how utility is distributed across the episode sequence.

Parameters:: reward (float) – Final reward value for the trace used to compute utilities.
Returns:: List of utility values, one per episode in the trace.
Return type:: list[float]

get_dataset(shuffle=True, n_samples=None)[source]

Return training dataset from flattened traces as numpy arrays.

Flattens stored traces into episode states and utilities, optionally shuffling the resulting dataset.

Parameters:

shuffle (bool, optional) – Whether to shuffle the dataset, defaults to True
n_samples (int | None, optional) – Maximum number of traces to include; if None, uses all traces, defaults to None

Returns:

Tuple of (states, utilities) numpy arrays for training.

Return type:

tuple[np.ndarray, np.ndarray]

get_flattened_traces(n_samples=None)[source]

Flatten stored traces into a single buffer with corresponding utilities.

Optionally samples a subset of traces, then flattens all (episode, utility) pairs from the selected traces into a single sequential buffer.

Parameters:: n_samples (int | None, optional) – Maximum number of traces to sample; if None, uses all traces, defaults to None
Returns:: Tuple of (episode buffer, utilities array) containing flattened trace data.
Return type:: tuple[list, np.ndarray]

property max_antitraces

Maximum capacity of the antitraces buffer.

Returns:: Max number of antitraces retained.
Return type:: int

property max_traces

Maximum capacity of the traces buffer.

Returns:: Max number of traces retained.
Return type:: int

property n_antitraces

Current number of stored antitraces.

Returns:: Count of antitraces in the buffer.
Return type:: int

property n_traces

Current number of stored traces.

Returns:: Count of traces in the buffer.
Return type:: int

reset_new_sample_count(main=True, secondary=True)[source]

Resets the new sample count for the trace buffer.

Parameters:

main (bool) – Unused in this class, defaults to True.
secondary (bool) – Unused in this class, defaults to True.

Deliberative model

Python module that implements the Deliberative Model cognitive node, which is a higher-level cognitive node that integrates the functionalities of World Model, Utility Model.

Specific services

/deliberative_model/id/set_activation => C-nodes can modify a deliberative model’s activation.

/deliberative_model/id/predict => Get predicted values for the input episode.

/deliberative_model/id/get_success_rate => Get a prediction success rate based on a historic of previous predictions.

/deliberative_model/id/is_compatible => Check if the Model is compatible with the current available episode.

/deliberative_model/id/save_model => Save the model to a file if compatible.

class cognitive_nodes.deliberative_model.ANNLearner(node, buffer, batch_size=32, epochs=50, output_activation='sigmoid', hidden_activation='relu', hidden_layers=[128], learning_rate=0.001, loss_function=torch.nn.MSELoss, val_function=torch.nn.L1Loss, model_file=None, device='cpu', **params)[source]

Bases: Learner

Class that implements a Neural Network-based learner using PyTorch.

call(x)[source]

Make predictions with the trained model.

Parameters:: x (np.ndarray or torch.Tensor) – Input data for prediction. Can be numpy array or torch tensor.
Returns:: Model predictions as a numpy array, or None if model is not configured.
Return type:: np.ndarray or None

configure_model(input_length, output_length)[source]

Configure the ANN model architecture and initialize the optimizer.

Parameters:

input_length (int) – Number of input features for the neural network.
output_length (int) – Number of output features/predictions from the neural network.

evaluate(x_test, y_test)[source]

Evaluate the model on test data and return Mean Absolute Error (MAE).

Parameters:

x_test (np.ndarray or torch.Tensor) – Test input data (features).
y_test (np.ndarray or torch.Tensor) – Test target data (labels).

Returns:

Mean Absolute Error between predictions and true values, or None if model is not configured.

Return type:

float or None

get_weights()[source]: Get the current model , uses the state dictionary.

load_model()[source]: Loads model from file.

reset_model_state()[source]: Reset optimizer state while keeping model weights.

save_model(filepath)[source]

Save the trained model to a PyTorch checkpoint file.

Parameters:: filepath (str) – Path where the model checkpoint will be saved. Automatically adds ‘.pth’ extension if not present.
Returns:: Tuple containing success status and the path where the model was saved.
Return type:: tuple(bool, str)

set_weights(weights)[source]

Set the model weights from a state dictionary.

Parameters:: weights (dict) – PyTorch state dictionary containing model weights and biases.
Returns:: True if weights were successfully loaded, False otherwise.
Return type:: bool

train(x_train, y_train, epochs=None, batch_size=None, validation_split=0.0, x_val=None, y_val=None, verbose=1, reset_optimizer=True)[source]

Train the neural network model using PyTorch with optional validation and early stopping.

Parameters:

x_train (np.ndarray or torch.Tensor) – Training input data (features).
y_train (np.ndarray or torch.Tensor) – Training target data (labels).
epochs (int, optional) – Number of training epochs. If None, uses the learner’s default epochs, defaults to None
batch_size (int, optional) – Batch size for training. If None, uses the learner’s default batch size, defaults to None
validation_split (float, optional) – Fraction of training data to use for validation (0.0 to 1.0), defaults to 0.0
x_val (np.ndarray or torch.Tensor, optional) – Validation input data. If provided with y_val, overrides validation_split, defaults to None
y_val (np.ndarray or torch.Tensor, optional) – Validation target data. If provided with x_val, overrides validation_split, defaults to None
verbose (int, optional) – Verbosity level. 0 = silent, 1 = progress logs, defaults to 1
reset_optimizer (bool, optional) – Whether to reset the optimizer state before training, defaults to True

class cognitive_nodes.deliberative_model.ANNModel(*args: Any, **kwargs: Any)[source]

Bases: Module

PyTorch neural network model.

forward(x)[source]

Forward pass through the network.

Parameters:: x (torch.Tensor) – Input tensor.
Returns:: Output tensor after passing through the network.
Return type:: torch.Tensor

class cognitive_nodes.deliberative_model.AsymmetricMSELoss(*args: Any, **kwargs: Any)[source]

Bases: Module

Asymmetric Mean Squared Error Loss that penalizes underestimations and overestimations differently.

forward(y_pred, y_true)[source]

Forward pass to compute the asymmetric mean squared error loss.

Parameters:

y_pred (torch.Tensor) – Predicted values.
y_true (torch.Tensor) – True target values.

Returns:

Computed asymmetric mean squared error loss.

Return type:

torch.Tensor

class cognitive_nodes.deliberative_model.DeliberativeModel(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

Deliberative Model class, this class is a generic model that can be used to implement different types of deliberative models.

calculate_activation(perception=None, activation_list=None)[source]

Returns the activation value of the Model.

Parameters:

perception (dict) – Perception does not influence the activation.
activation_list (list) – List of activation values from other sources, defaults to None.

Returns:

The activation of the instance and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation

get_success_rate_callback(request, response)[source]

Get a prediction success rate based on a historic of previous predictions.

Parameters:

request (cognitive_node_interfaces.srv.GetSuccessRate.Request) – Empty request.
response (cognitive_node_interfaces.srv.GetSuccessRate.Response) – The response that contains the predicted success rate.

Returns:

The response that contains the predicted success rate.

Return type:

cognitive_node_interfaces.srv.GetSuccessRate.Response

is_compatible_callback(request, response)[source]

Check if the Model is compatible with the current available perceptions.

Parameters:

request (cognitive_node_interfaces.srv.IsCompatible.Request) – The request that contains the current available perceptions.
response (cognitive_node_interfaces.srv.IsCompatible.Response) – The response indicating if the Model is compatible or not.

Returns:

The response indicating if the Model is compatible or not.

Return type:

cognitive_node_interfaces.srv.IsCompatible.Response

predict_callback(request, response)[source]

Get predicted perception values for the last perceptions not newer than a given timestamp and for a given policy.

Parameters:

request (cognitive_node_interfaces.srv.Predict.Request) – The request that contains the timestamp and the policy.
response (cognitive_node_interfaces.srv.Predict.Response) – The response that included the obtained perception.

Returns:

The response that included the obtained perception.

Return type:

cognitive_node_interfaces.srv.Predict.Response

save_model_callback(request, response)[source]

Save the current model to a file.

Parameters:

request (cognitive_node_interfaces.srv.SaveModel.Request) – The request that contains the prefix and suffix for the file name.
response (cognitive_node_interfaces.srv.SaveModel.Response) – The response that contains the saved model path and success status.

Returns:

The response that contains the saved model path and success status.

Return type:

cognitive_node_interfaces.srv.SaveModel.Response

set_activation_callback(request, response)[source]

Some processes can modify the activation of a Model.

Parameters:

request (cognitive_node_interfaces.srv.SetActivation.Request) – The request that contains the new activation value.
response (cognitive_node_interfaces.srv.SetActivation.Response) – The response indicating if the activation was set.

Returns:

The response indicating if the activation was set.

Return type:

cognitive_node_interfaces.srv.SetActivation.Response

class cognitive_nodes.deliberative_model.Evaluator(node: core.cognitive_node.CognitiveNode, learner: Learner, buffer: EpisodicBuffer, **params)[source]

Bases: object

Class that evaluates the success rate of a model based on its predictions.

evaluate()[source]

Placeholder method for evaluating the model’s success rate.

Raises:: NotImplementedError – Not implemented yet.

class cognitive_nodes.deliberative_model.Learner(node: core.cognitive_node.CognitiveNode, buffer: EpisodicBuffer, **params)[source]

Bases: object

Class that wraps around a learning model (Linear Classifier, ANN, SVM…)

call(x)[source]

Placeholder method for predicting an outcome.

Parameters:

perception (dict) – Perception dictionary.
action (dict) – Candidate action dictionary.

Raises:

NotImplementedError – Not implemented yet.

train()[source]

Placeholder method for training the model.

Raises:: NotImplementedError – Not implemented yet.

World model

Python module that implements the World Model cognitive node, which is represent the behavior of the domain in which the robot is operating. They are usually instantiated as a predictor of the perceptual situation Pt+1 that will result from the application of an action when in a perceptual state Pt. It inherits all the functionalities of the Deliberative Model, including its services

class cognitive_nodes.world_model.EvaluatorWorldModel(node: WorldModelLearned, learner: ANNLearner, buffer: EpisodicBuffer, **params)[source]

Bases: Evaluator

EvaluatorWorldModel class: Evaluates the success rate of the world model based on its predictions.

evaluate()[source]: Obtain the MSE error of the World Model according to the test samples in the episodic buffer.

publish_prediction_error()[source]: Publishes the prediction error of the world model.

class cognitive_nodes.world_model.Sim2D(node, actuation_config, perception_config, logger: rclpy.impl.rcutils_logger.RcutilsLogger, **params)[source]

Bases: Learner

Sim2D class: A class that mimics a model that learned the dynamics of a 2D simulator. Actually it uses the same simulator as the environment to predict the next perception.

call(perception, action) → cognitive_node_interfaces.msg.Perception[source]

Predicts the next perception according to a perception and an action.

Parameters:

perception (cognitive_node_interfaces.msg.Perception) – The start perception.
action (cognitive_node_interfaces.msg.Actuation) – The action performed.

Returns:

The predicted perception.

Return type:

cognitive_node_interfaces.msg.Perception

denormalize(input_dict, config)[source]

Denormalize the input dictionary according to the configuration

Parameters:

input_dict (dict) – Perception or actuation dictionary.
config (dict) – Configuration of the perception or actuation bounds.

Returns:

Denormalized dictionary.

Return type:

dict

normalize(input_dict, config)[source]

Normalize the input dictionary according to the configuration.

Parameters:

input_dict (dict) – Perception or actuation dictionary.
config (dict) – Configuration of the perception or actuation bounds.

Returns:

Normalized dictionary.

Return type:

dict

class cognitive_nodes.world_model.Sim2DWorldModel(*args: Any, **kwargs: Any)[source]

Bases: WorldModel

Sim2DWorldModel class: A fixed world model of a 2D simulator. It uses the SimpleScenario simulator to predict the next perception.

predict(input_episodes: list[Episode]) → list[Episode][source]

Predict output episodes from input episodes using the world model.

Parameters:: input_episodes (list[Episode]) – List of episodes containing old perceptions and actions.
Returns:: List of predicted episodes with updated perceptions.
Return type:: list[Episode]

class cognitive_nodes.world_model.WorldModel(*args: Any, **kwargs: Any)[source]

Bases: DeliberativeModel

World Model class: A static world model that is always active

predict(input_episodes: list[Episode]) → list[Episode][source]

Predict output episodes from input episodes using the world model.

Parameters:: input_episodes (list[Episode]) – List of episodes containing old perceptions and actions.
Returns:: List of predicted episodes with updated perceptions.
Return type:: list[Episode]

class cognitive_nodes.world_model.WorldModelLearned(*args: Any, **kwargs: Any)[source]

Bases: WorldModel

WorldModelLearned class: A world model that uses episodes to learn the dynamics of the world.

episode_callback(msg: cognitive_node_interfaces.msg.Episode)[source]

Callback for the episode subscription. It receives an episode message and adds it to the episodic buffer.

Parameters:: msg (cognitive_node_interfaces.msg.Episode) – The episode message received.

predict(input_episodes: list[Episode]) → list[Episode][source]

Predict output episodes from input episodes using the world model.

Parameters:: input_episodes (list[Episode]) – List of episodes containing old perceptions and actions.
Returns:: List of predicted episodes with updated perceptions.
Return type:: list[Episode]

Utiliy model

Python module that implements the Utility Model cognitive node, which estimates the expected utility of perceptual states with respect to a goal, based on the probability of achieving it and the potential reward. It inherits all the functionalities of the Deliberative Model, including its services

Specific services

/utility_model/id/execute => Executes the deliberation process with this utility model and its associated world model.

class cognitive_nodes.utility_model.DefaultUtilityEvaluator(node: UtilityModel, learner: DefaultUtilityModelLearner, buffer: None, **params)[source]

Bases: Evaluator

Default Utility Evaluator class, used when no specific utility evaluator is defined. This evaluator does not perform any evaluation, it simply returns a constant value.

evaluate()[source]

Evaluates the input episodes and returns a list of evaluated episodes.

Parameters:: input_episodes (list) – List of input episodes to evaluate.
Returns:: List of evaluated episodes.
Return type:: list

class cognitive_nodes.utility_model.DefaultUtilityModelLearner(node: UtilityModel, buffer, **params)[source]

Bases: Learner

Default Utility Model class, used when no specific utility model is defined. This model does not perform any learning or prediction, it simply returns a constant value.

call(x)[source]

Returns a vector of ones as the utility

Parameters:: x (np.ndarray) – Input data for prediction.
Returns:: Model predictions as a numpy array.
Return type:: np.ndarray

train()[source]

Placeholder method for training the model.

Raises:: NotImplementedError – Not implemented yet.

class cognitive_nodes.utility_model.DummyUtilityModel(*args: Any, **kwargs: Any)[source]

Bases: LearnedUtilityModel

Dummy Utility Model class This model is used as a placeholder and does not perform any actual utility computation. It inherits from the UtilityModel class.

calculate_activation(perception=None, activation_list=None)[source]

Calculate the activation level of the utility model based on perception or activation inputs.

Parameters:

perception (dict, optional) – Perception data to use for activation calculation, defaults to None
activation_list (list, optional) – List of activations from connected nodes to aggregate, defaults to None

Returns:

None

Return type:

None

class cognitive_nodes.utility_model.HardCodedUtilityModel(*args: Any, **kwargs: Any)[source]

Bases: UtilityModel

Hard Coded Utility Model class This model is used to compute the utility of the episodes based on hard coded values. It inherits from the UtilityModel class.

denormalize(input_dict, config)[source]

Denormalize the input dictionary according to the configuration

Parameters:

input_dict (dict) – Perception or actuation dictionary.
config (dict) – Configuration of the perception or actuation bounds.

Returns:

Denormalized dictionary.

Return type:

dict

predict(input_episodes: list[Episode]) → list[float][source]

Predict utilities for input episodes using hard-coded heuristics for ball-to-box task.

Computes utilities based on distances and angles between robot arms, ball, and target box, with penalties for suboptimal configurations and rewards for goal-oriented states.

Parameters:: input_episodes (list[Episode]) – List of Episode objects containing perception data to evaluate.
Returns:: List of normalized utility values (0-1 range) for each input episode.
Return type:: list[float]

setup_model(trace_length, max_iterations, candidate_actions, ltm_id, train_traces=1, max_traces=50, **params)[source]

Sets up the Hard Coded Utility Model by initializing the episodic buffer, learner, and deliberation process.

Parameters:

trace_length (int) – Maximum number of traces to store in the trace buffer.
max_iterations (int) – Maximum number of iterations for the deliberation process.
candidate_actions (int) – Number of candidate actions to generate during deliberation.
ltm_id (str) – Identifier for the Long-Term Memory cache used in deliberation.
train_traces (int, optional) – Minimum number of positive traces required in the buffer, defaults to 1
max_traces (int, optional) – Maximum number of traces to store in the buffer, defaults to 50
params (dict) – Additional keyword parameters reserved for future use.

class cognitive_nodes.utility_model.LearnedUtilityModel(*args: Any, **kwargs: Any)[source]

Bases: UtilityModel

Learned Utility Model class: A utility model that learns from episodic traces to predict utilities.

add_trace_callback(request, response)[source]: Service callback to add traces to the episodic buffer. :param request: The request object containing episodes and rewards to be added. :type request: cognitive_node_interfaces.srv.AddTrace.Request :param response: The response object to be populated with the result of the addition. :type response: cognitive_node_interfaces.srv.AddTrace.Response :return: The response object indicating whether the traces were successfully added. :rtype: cognitive_node_interfaces.srv.AddTrace.Response

calculate_activation(perception=None, activation_list=None)[source]

Calculate the activation level of the utility model based on perception or activation inputs.

Parameters:

perception (dict, optional) – Perception data to use for activation calculation, defaults to None
activation_list (list, optional) – List of activations from connected nodes to aggregate, defaults to None

Returns:

None

Return type:

None

episode_callback(msg)[source]

Handle incoming episode messages and update the episodic buffer and learning model.

This callback processes episode messages from other policies. It handles world resets by clearing the buffer and processes new episodes by extracting rewards for linked goals, adding them to the episodic buffer, and triggering model training if positive rewards are present. Long-term memory is asynchronously updated in a separate thread when new successful traces are added.

Parameters:: msg (cognitive_node_interfaces.msg.Episode) – The episode message received from the subscription, containing perception data, rewards, and parent policy information.

execute_callback(request, response)[source]

Execute the deliberation process and perform a training step if sufficient traces are available.

This method extends the parent class execution callback by adding a training step after deliberation completes. It logs the current state of the episodic buffer including trace counts and triggers model training if the minimum trace threshold is met.

Parameters:

request (cognitive_node_interfaces.srv.Execute.Request) – The request object from the Execute service containing execution parameters.
response (cognitive_node_interfaces.srv.Execute.Response) – The response object to be populated with execution results.

Returns:

The response object containing the policy name and summary episode from deliberation.

Return type:

cognitive_node_interfaces.srv.Execute.Response

predict(input_episodes: list[Episode]) → list[float][source]

Predict utilities for input episodes using the learner model.

Parameters:: input_episodes (list[Episode]) – List of Episode objects to predict utilities for.
Returns:: List of predicted utility values for each input episode.
Return type:: list[float]

setup_model(trace_length, max_iterations, candidate_actions, ltm_id, min_traces=1, max_traces=50, max_antitraces=10, **params)[source]

Sets up the Learned Utility Model by initializing the episodic buffer, learner, and confidence evaluator.

Parameters:

trace_length (int) – Maximum number of traces to store in the trace buffer.
max_iterations (int) – Maximum number of iterations for the deliberation process.
candidate_actions (int) – Number of candidate actions to generate during deliberation.
ltm_id (str) – Identifier for the Long-Term Memory cache used in deliberation.
min_traces (int, optional) – Minimum number of traces required before training can begin, defaults to 1
max_traces (int, optional) – Maximum number of traces to store in the buffer, defaults to 50
max_antitraces (int, optional) – Maximum number of antitraces (negative examples) to store, defaults to 10
params (dict) – Additional keyword parameters reserved for future use.

train_step()[source]: Perform a training step for the utility model if sufficient new traces are available.

update_ltm_and_train(old_perception, perception, policy, reward_list, ltm_cache)[source]: Update the Long-Term Memory cache with new reward basis information. This method updates the Long-Term Memory (LTM) cache with new reward basis information based on the provided perceptions, policy, and reward list. It uses a semaphore to ensure thread-safe access to the LTM during the update process. :param old_perception: The previous perception state before the episode. :type old_perception: dict :param perception: The current perception state after the episode. :type perception: dict :param policy: The policy name associated with the episode. :type policy: str :param reward_list: The list of rewards associated with the episode. :type reward_list: dict :param ltm_cache: The Long-Term Memory cache to be updated. :type ltm_cache: dict

class cognitive_nodes.utility_model.NoveltyUtilityModel(*args: Any, **kwargs: Any)[source]

Bases: UtilityModel

Novelty Utility Model class This model is used to compute the novelty of the episodes. It inherits from the UtilityModel class.

setup_model(trace_length, max_iterations, candidate_actions, ltm_id, train_traces=1, max_traces=50, **params)[source]

Sets up the Novelty Utility Model by initializing the episodic buffer, learner, and deliberation process.

Parameters:

trace_length (int) – Maximum number of traces to store in the episodic buffer.
max_iterations (int) – Maximum number of iterations for the deliberation process.
candidate_actions (int) – Number of candidate actions to generate during deliberation.
ltm_id (str) – Identifier for the Long-Term Memory cache used in deliberation.
train_traces (int, optional) – Minimum number of positive traces required in the buffer, defaults to 1
max_traces (int, optional) – Maximum number of traces to store in the buffer, defaults to 50

class cognitive_nodes.utility_model.NoveltyUtilityModelLearner(node: UtilityModel, buffer, **params)[source]

Bases: Learner

Novelty Utility Model class, used as an exploration method. This model provides higher utility to states not visited previously.

call(x)[source]

Returns the computed novelty for the input data

Parameters:: x (np.ndarray) – Input data for prediction.
Returns:: Model predictions as a numpy array.
Return type:: np.ndarray

compute_novelty(previous_episodes, candidate_episodes)[source]

Compute normalized novelty scores for candidate episodes relative to previously seen episodes.

Parameters:

previous_episodes (np.ndarray | None) – Array of prior episode embeddings with shape (N, D); if empty or None, all candidates are maximally novel.
candidate_episodes (np.ndarray) – Array of candidate episode embeddings with shape (M, D).

Returns:

Novelty scores in [0, 1] for each candidate, where higher values indicate greater novelty.

Return type:

np.ndarray

train()[source]

Placeholder method for training the model.

Raises:: NotImplementedError – Not implemented yet.

class cognitive_nodes.utility_model.QUtilityModel(*args: Any, **kwargs: Any)[source]

Bases: LearnedUtilityModel

Q-Learning Utility Model class: A utility model that uses Q-learning to predict utilities.

This model implements a Q-learning approach with a target network for stable training. It maintains two neural networks: a main learner and a target learner that are periodically synchronized. The model predicts Q-values for state-action pairs and uses these to compute optimal utilities during deliberation.

generate_candidates_matrix(buffer, algorithm='latin')[source]

Generate candidate action matrices from episode buffer states.

For each episode in the buffer, generates multiple candidate actions and creates composite state-action pairs for evaluation.

Parameters:

buffer (list[Episode]) – List of Episode objects representing states.
algorithm (str, optional) – Algorithm for candidate action generation, defaults to “latin”

Returns:

List of candidate episodes with state-action combinations.

Return type:

list[Episode]

obtain_maximum_value(q_values)[source]

Extract maximum Q-value for each state from flattened candidate Q-values.

Given Q-values computed for all candidate actions, returns the maximum Q-value for each state by reshaping and reducing across the candidate actions dimension.

Parameters:: q_values (np.ndarray) – Flattened array of Q-values of shape (n_states * candidate_actions,)
Returns:: Array of maximum Q-values for each state of shape (n_states,)
Return type:: np.ndarray
Raises:: ValueError – If Q-values length is not divisible by candidate_actions

predict(input_episodes: list[Episode], target_learner=False) → list[float][source]

Predict Q-values for input episodes using either main or target learner.

Parameters:

input_episodes (list[Episode]) – List of Episode objects to predict Q-values for.
target_learner (bool, optional) – If True, use target network for predictions; otherwise use main learner, defaults to False

Returns:

List of predicted Q-values for each input episode.

Return type:

list[float]

setup_model(trace_length, max_iterations, candidate_actions, ltm_id, min_traces=1, max_traces=50, max_antitraces=10, **params)[source]

Sets up the Q-Learning Utility Model by initializing dual networks and episodic buffer.

Parameters:

trace_length (int) – Maximum number of traces to store in the trace buffer.
max_iterations (int) – Maximum number of iterations for the deliberation process.
candidate_actions (int) – Number of candidate actions to generate during deliberation.
ltm_id (str) – Identifier for the Long-Term Memory cache used in deliberation.
min_traces (int, optional) – Minimum number of traces required before training can begin, defaults to 1
max_traces (int, optional) – Maximum number of traces to store in the buffer, defaults to 50
max_antitraces (int, optional) – Maximum number of antitraces (negative examples) to store, defaults to 10
params (dict) – Additional keyword parameters reserved for future use.

train_step()[source]

Perform a training step using Q-learning with target network updates.

Initializes both the main and target learners on first training. Computes Q-values for next states using the target network and updates the main learner with computed Q-targets. Periodically synchronizes the target network with the main learner weights.

class cognitive_nodes.utility_model.UtilityModel(*args: Any, **kwargs: Any)[source]

Bases: DeliberativeModel

Utility Model class

calculate_activation(perception=None, activation_list=None)[source]

Calculate the activation level of the utility model based on perception or activation inputs.

Parameters:

perception (dict, optional) – Perception data to use for activation calculation, defaults to None
activation_list (list, optional) – List of activations from connected nodes to aggregate, defaults to None

Returns:

None

Return type:

None

execute_callback(request, response)[source]

Callback for the execute service. Executes the action and returns the response.

Parameters:

request (cognitive_node_interfaces.srv.Execute.Request) – The request from the service.
response (cognitive_node_interfaces.srv.Execute.Response) – The response to be sent back.

Returns:

The response with the executed action.

Return type:

cognitive_node_interfaces.srv.Execute.Response

predict(input_episodes: list[Episode]) → list[float][source]

Predict the expected utilities for a list of input episodes using the learner model.

Parameters:: input_episodes (list[Episode]) – List of Episode objects to predict utilities for.
Returns:: List of predicted utility values, one for each input episode.
Return type:: list[float]

setup_model(trace_length, max_iterations, candidate_actions, ltm_id, **params)[source]

Sets up the Utility Model by initializing the episodic buffer, learner, and confidence evaluator.

Parameters:

trace_length (int) – Maximum number of traces to store in the trace buffer.
max_iterations (int) – Maximum number of iterations for the deliberation process.
candidate_actions (int) – Number of candidate actions to generate during deliberation.
ltm_id (str) – Identifier for the Long-Term Memory cache used in deliberation.
params (dict) – Additional keyword parameters reserved for future use.

spin_deliberation()[source]: Starts a separate thread to spin the deliberation executor.

Policy

Python module that implements the Policy cognitive node, which is a reactive decision structure in the form of a procedural componen that provides the action to apply when at a given perceptual point.

Specific services

/policy/id/set_activation => C-nodes can modify a policy’s activation.

/policy/id/execute => Execute the policy.

class cognitive_nodes.policy.Policy(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

Policy class.

async calculate_activation(perception=None, activation_list=None)[source]

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

execute_callback(request, response)[source]

Placeholder for 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.

Raises:

NotImplementedError – This method should be implemented in subclasses.

set_activation_callback(request, response)[source]

CNodes can modify a policy’s activation.

Parameters:

request (cognitive_node_interfaces.srv.SetActivation.Request) – The request that contains the new activation value.
response (cognitive_node_interfaces.srv.SetActivation.Response) – The response indicating if the activation was set.

Returns:

The response indicating if the activation was set.

Return type:

cognitive_node_interfaces.srv.SetActivation.Response

class cognitive_nodes.policy.PolicyAsync(*args: Any, **kwargs: Any)[source]

Bases: Policy

PolicyAsync class. Represents a policy that does not wait for completion of the execution.

execute_callback(request, response)[source]

Method that publishes the policy that must be exectuted, there should be a node that reads this message and executes the actual policy. It logs the execution and returns the policy name in the response.

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

class cognitive_nodes.policy.PolicyBlocking(*args: Any, **kwargs: Any)[source]

Bases: Policy

PolicyBlocking class. Represents a policy that waits for completion of the execution.

async execute_callback(request, response)[source]

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

C-Node

Python module that implements the C-Node cognitive node, which represens a context. It links a P-Node (initial state), the World Model and a Goal (desired state), with the Policy needed to move from the initial to the desired state.

class cognitive_nodes.cnode.CNode(*args: Any, **kwargs: Any)[source]

Bases: CognitiveNode

C-Node class It represents a context, that is, a link between nodes that were activated together in the past. It is assumed that there is only one element of each type connected to the C-Node.

async calculate_activation(perception=None, activation_list=None)[source]

Calculate the new activation value by multiplying the activation values of its neighbors. When an activation list is passed, this method will multiply the last perceptions of the neighbors. Otherwise, with percerception = None, it will multiply the last activations of its neighbors, but it’s possible to use an arbitrary perception, that will propagate to the neighbors, calculating the final activation of the C-Node for that perception.

Parameters:

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

Returns:

The activation of the C-Node and its timestamp.

Return type:

cognitive_node_interfaces.msg.Activation