fragile.core.walkers#

Module Contents#

Classes#

SimpleWalkers

The WalkersAPI class defines the base functionality for managing walkers in a swarm simulation.

ScoreMetric

A base class for score metrics in a swarm of walkers.

RewardScore

A class representing the reward score metric for a swarm of walkers.

SonicScore

A base class for score metrics in a swarm of walkers.

MarioScore

A base class for score metrics in a swarm of walkers.

DiversityMetric

A base class for diversity metrics in a swarm of walkers.

RandomDistance

A class representing the random distance diversity metric for a swarm of walkers.

Walkers

Walkers class handles the walkers' dynamics, including scoring, diversity, cloning, and resetting.

ExplorationWalkers

Walkers class handles the walkers' dynamics, including scoring, diversity, cloning, and resetting.

NoBalance

A class representing walkers with no balancing behavior in a swarm.

Functions#

l2_norm(x, y)

Compute the L2 norm between two tensors.

class fragile.core.walkers.SimpleWalkers(accumulate_reward=True, score_scale=1.0, diversity_scale=1.0, minimize=False, **kwargs)[source]#

Bases: fragile.core.api_classes.WalkersAPI

The WalkersAPI class defines the base functionality for managing walkers in a swarm simulation. This class inherits from the SwarmComponent class.

Parameters
  • accumulate_reward (bool) –

  • score_scale (float) –

  • diversity_scale (float) –

  • minimize (bool) –

default_inputs#
default_param_dict#
default_outputs#
run_epoch(observs, rewards, scores, oobs=None, inplace=True, **kwargs)[source]#

Implement the functionality for running an epoch in the derived class. This method is called during the balance operation.

Parameters
  • inplace (bool, optional) – If True, updates the swarm state with the data generated during the epoch. If False, returns the data. Defaults to True.

  • **kwargs – Additional keyword arguments required for running the epoch.

Returns

A dictionary containing the data generated during the epoch if inplace is

False. Otherwise, returns None.

Return type

StateData

fragile.core.walkers.l2_norm(x, y)[source]#

Compute the L2 norm between two tensors.

Parameters
  • x (Tensor) – The first tensor.

  • y (Tensor) – The second tensor.

Returns

The L2 norm between the two input tensors.

Return type

Tensor

class fragile.core.walkers.ScoreMetric(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: fragile.core.api_classes.WalkersMetric

A base class for score metrics in a swarm of walkers.

Parameters
  • swarm (Optional[SwarmAPI]) –

  • param_dict (Optional[fragile.core.typing.StateDict]) –

  • inputs (Optional[fragile.core.typing.InputDict]) –

  • outputs (Optional[Tuple[str]]) –

default_param_dict#

Default parameters dictionary for the score metric.

Type

dict

default_outputs#

Tuple containing the default output keys.

Type

tuple

default_param_dict#
default_outputs#
class fragile.core.walkers.RewardScore(accumulate_reward=True, keep_max_reward=False, **kwargs)[source]#

Bases: ScoreMetric

A class representing the reward score metric for a swarm of walkers.

Parameters
  • accumulate_reward (bool) –

  • keep_max_reward (bool) –

default_inputs#

Default inputs dictionary for the reward score metric.

Type

dict

default_inputs#
property inputs#

Return a dictionary containing the data that this component needs to function.

Return type

fragile.core.typing.InputDict

calculate(rewards=None, scores=None, **kwargs)[source]#

Calculate the scores for the walkers based on rewards.

Parameters
  • rewards (array) – Array of walkers’ rewards.

  • scores (array, optional) – Array of walkers’ scores.

  • **kwargs – Additional keyword arguments.

Returns

A dictionary containing the scores.

Return type

dict

reset(inplace=True, root_walker=None, states=None, **kwargs)[source]#

Reset the reward score metric.

Parameters
  • inplace (bool, optional) – Whether to reset in-place. Defaults to True.

  • root_walker (StateData, optional) – The root walker to reset. Defaults to None.

  • states (Optional[StateData], optional) – The state data to reset. Defaults to None.

  • **kwargs – Additional keyword arguments.

Return type

None

class fragile.core.walkers.SonicScore(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: ScoreMetric

A base class for score metrics in a swarm of walkers.

Parameters
  • swarm (Optional[SwarmAPI]) –

  • param_dict (Optional[fragile.core.typing.StateDict]) –

  • inputs (Optional[fragile.core.typing.InputDict]) –

  • outputs (Optional[Tuple[str]]) –

default_param_dict#

Default parameters dictionary for the score metric.

Type

dict

default_outputs#

Tuple containing the default output keys.

Type

tuple

accumulate_reward = False#
default_inputs#
static score_from_info(info)[source]#
calculate(rewards, scores=None, **kwargs)[source]#
class fragile.core.walkers.MarioScore(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: ScoreMetric

A base class for score metrics in a swarm of walkers.

Parameters
  • swarm (Optional[SwarmAPI]) –

  • param_dict (Optional[fragile.core.typing.StateDict]) –

  • inputs (Optional[fragile.core.typing.InputDict]) –

  • outputs (Optional[Tuple[str]]) –

default_param_dict#

Default parameters dictionary for the score metric.

Type

dict

default_outputs#

Tuple containing the default output keys.

Type

tuple

name = MarioScore#
accumulate_reward = False#
default_inputs#
static score_from_info(info)[source]#
calculate(rewards=None, scores=None, **kwargs)[source]#
class fragile.core.walkers.DiversityMetric(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: fragile.core.api_classes.WalkersMetric

A base class for diversity metrics in a swarm of walkers.

Parameters
  • swarm (Optional[SwarmAPI]) –

  • param_dict (Optional[fragile.core.typing.StateDict]) –

  • inputs (Optional[fragile.core.typing.InputDict]) –

  • outputs (Optional[Tuple[str]]) –

default_param_dict#

Default parameters dictionary for the diversity metric.

Type

dict

default_outputs#

Tuple containing the default output keys.

Type

tuple

default_param_dict#
default_outputs#
class fragile.core.walkers.RandomDistance(swarm=None, param_dict=None, inputs=None, outputs=None)[source]#

Bases: DiversityMetric

A class representing the random distance diversity metric for a swarm of walkers.

Parameters
  • swarm (Optional[SwarmAPI]) –

  • param_dict (Optional[fragile.core.typing.StateDict]) –

  • inputs (Optional[fragile.core.typing.InputDict]) –

  • outputs (Optional[Tuple[str]]) –

default_inputs#

Default inputs dictionary for the random distance diversity metric.

Type

dict

use_pbc = False#
default_inputs#
calculate(observs, oobs, **kwargs)[source]#

Calculate the diversities for the walkers based on the L2 distance to another walker chosen at random.

Parameters
  • observs (array) – Array of walkers’ observations.

  • oobs (array) – Array of out-of-bounds flags for walkers.

  • **kwargs – Additional keyword arguments.

Returns

A dictionary containing the diversities.

Return type

dict

class fragile.core.walkers.Walkers(score=None, diversity=None, minimize=False, score_scale=1.0, diversity_scale=1.0, track_data=None, accumulate_reward=True, keep_max_reward=False, clone_period=1, freeze_walkers=True, **kwargs)[source]#

Bases: fragile.core.api_classes.WalkersAPI

Walkers class handles the walkers’ dynamics, including scoring, diversity, cloning, and resetting.

Inherits from WalkersAPI.

Parameters
default_param_dict#
default_outputs = ['compas_clone', 'virtual_rewards', 'clone_probs', 'will_clone']#
default_inputs#
property param_dict#

Return the dictionary defining all the data attributes that the component requires.

Return type

fragile.core.typing.StateDict

property inputs#

Return a dictionary containing the data that this component needs to function.

Return type

fragile.core.typing.InputDict

property outputs#

Return a tuple containing the names of the data attribute that the component outputs.

Return type

Tuple[str, Ellipsis]

setup(swarm)[source]#

Set up the Walkers object by initializing its swarm.

Parameters

swarm – Swarm object to be initialized.

balance(inplace=True, **kwargs)[source]#

Balance the walkers.

Parameters
  • inplace (bool, optional) – Whether to perform the operation inplace. Defaults to True.

  • **kwargs – Additional keyword arguments.

Returns

None if inplace=True, otherwise a StateData object.

Return type

Union[None, StateData]

run_epoch(inplace=True, oobs=None, **kwargs)[source]#

Execute an epoch in the walkers’ dynamics.

Parameters
  • inplace (bool, optional) – Whether to perform the operation inplace. Defaults to True.

  • oobs – Out of bounds flags.

  • **kwargs – Additional keyword arguments.

Returns

State data after the epoch execution.

Return type

StateData

calculate_virtual_reward(scores, diversities, **kwargs)[source]#

Calculate the virtual rewards for walkers based on scores and diversities.

Parameters
  • scores (array) – Array of walkers’ scores.

  • diversities (array) – Array of walkers’ diversities.

  • **kwargs – Additional keyword arguments.

Returns

A dictionary containing the virtual rewards.

Return type

dict

calculate_clones(virtual_rewards, oobs=None)[source]#

Calculate the walkers that will clone and their target companions.

Parameters
  • virtual_rewards (array) – Array of walkers’ virtual rewards.

  • oobs (array, optional) – Array of out-of-bounds flags for walkers. Defaults to None.

Returns

A dictionary containing clone probabilities, will_clone flags, companion clones, and active flags.

Return type

dict

reset(inplace=True, **kwargs)[source]#

Reset the Walkers object and its score and diversity components.

Parameters
  • inplace (bool, optional) – Whether to perform the operation inplace. Defaults to True.

  • **kwargs – Additional keyword arguments.

Return type

None

class fragile.core.walkers.ExplorationWalkers(exploration_scale=1.0, **kwargs)[source]#

Bases: Walkers

Walkers class handles the walkers’ dynamics, including scoring, diversity, cloning, and resetting.

Inherits from WalkersAPI.

Parameters

exploration_scale (float) –

property explore_counts#
Return type

Dict[Any, int]

calculate_virtual_reward(scores, diversities, **kwargs)[source]#

Apply the virtual reward formula to account for all the different goal scores.

get_explore_rewards()[source]#
get_coords_keys()[source]#
class fragile.core.walkers.NoBalance(score=None, diversity=None, minimize=False, score_scale=1.0, diversity_scale=1.0, track_data=None, accumulate_reward=True, keep_max_reward=False, clone_period=1, freeze_walkers=True, **kwargs)[source]#

Bases: Walkers

A class representing walkers with no balancing behavior in a swarm.

Inherits from the Walkers class and modifies the properties and methods to disable balancing.

Parameters
property param_dict#

Return the dictionary defining all the data attributes that the component requires.

Return type

fragile.core.typing.StateDict

property inputs#

Return a dictionary containing the data that this component needs to function.

Return type

fragile.core.typing.InputDict

property outputs#

Return a tuple containing the names of the data attribute that the component outputs.

Return type

Tuple[str, Ellipsis]

run_epoch(inplace=True, oobs=None, **kwargs)[source]#

Run an epoch for the walkers without balancing.

Parameters
  • inplace (bool, optional) – Whether to run the epoch in-place. Defaults to True.

  • oobs (array, optional) – Array of out-of-bounds walkers. Defaults to None.

  • **kwargs – Additional keyword arguments.

Returns

A dictionary containing the scores.

Return type

dict