:py:mod:`fragile.core.api_classes` ================================== .. py:module:: fragile.core.api_classes Module Contents --------------- Classes ~~~~~~~ .. autoapisummary:: fragile.core.api_classes.SwarmComponent fragile.core.api_classes.EnvironmentAPI fragile.core.api_classes.PolicyAPI fragile.core.api_classes.Callback fragile.core.api_classes.WalkersMetric fragile.core.api_classes.WalkersAPI fragile.core.api_classes.SwarmAPI .. py:class:: SwarmComponent(swarm = None, param_dict = None, inputs = None, outputs = None) A component of a swarm simulation. Every class that stores its data in :class:`SwarmState` inherits from this class. :param swarm: Reference to the :class:`Swarm` that incorporates this :class:`SwarmComponent`. Defaults to None. :param param_dict: Describes the multi-dimensional arrays that define the internal state of the component. Defaults to None. :param inputs: Describes the data that this components processes as inputs. This data is present in the internal state of the :class:`Swarm`, and generated by other components. Defaults to None. :param outputs: Name of the data attributes that this component makes available as part of its internal state. Defaults to None. .. attribute:: swarm The parent :class:`Swarm` instance. :type: Swarm .. attribute:: n_walkers Number of walkers in the :class:`Swarm`. :type: int .. attribute:: inputs A dictionary mapping input names to their corresponding type and default values. :type: Dict[str, Dict[str, Any]] .. attribute:: outputs A tuple containing the names of the data attributes. :type: Tuple[str] .. attribute:: param_dict A dictionary mapping parameter names. :type: Dict[str, Dict[str, Any]] .. method:: get(...) .. method:: get_input_data(...) .. method:: update(...) .. method:: reset(...) .. method:: _prepare_tensors(...) .. py:attribute:: default_inputs .. py:attribute:: default_outputs .. py:attribute:: default_param_dict .. py:method:: swarm() :property: Return a reference to the :class:`Swarm` that includes the current component. .. py:method:: n_walkers() :property: Return the number of walkers of the :class:`Swarm`. .. py:method:: inputs() :property: Return a dictionary containing the data that this component needs to function. .. py:method:: outputs() :property: Return a tuple containing the names of the data attribute that the component outputs. .. py:method:: param_dict() :property: Return the dictionary defining all the data attributes that the component requires. .. py:method:: setup(swarm) Prepare the component during the setup phase of the :class:`Swarm`. .. py:method:: get(name, default = None, raise_error = False, inactives = False) Access attributes of the :class:`Swarm` and its children. .. py:method:: get_input_data() Return a dictionary with all the data that this component requires as inputs. :returns: Dictionary containing all the data required by the :class:`SwarmComponent`, where each key is the name of a data attribute and its value a multi-dimensional array. .. py:method:: update(other = None, inactives = False, **kwargs) Modify the data stored in the SwarmState instance. Existing attributes will be updated, and new attributes will be created if needed. :param other: State class that will be copied upon update. :param inactives: Whether to update the walkers marked as inactive. :param \*\*kwargs: It is possible to specify the update as name value attributes, where name is the name of the attribute to be updated, and value is the new value for the attribute. :returns: None .. py:method:: _prepare_tensors(**kwargs) .. py:method:: reset(inplace = True, root_walker = None, states = None, **kwargs) Reset the internal state of the :class:`SwarmComponent`. :param inplace: Unused. Defaults to True. :type inplace: bool, optional :param root_walker: Set the internal state of the :class:`SwarmComponent` to this value. Defaults to None. :type root_walker: Optional[StateData], optional :param states: Set the internal state of the :class:`SwarmComponent` to this value. Defaults to None. :type states: Optional[StateData], optional :param kwargs: Other parameters required to reset the component. .. py:class:: EnvironmentAPI(action_shape, action_dtype, observs_shape, observs_dtype, swarm = None) Bases: :py:obj:`SwarmComponent` The Environment is in charge of stepping the walkers, acting as a state transition function. For every different problem, a new Environment needs to be implemented following the :class:`EnvironmentAPI` interface. .. py:attribute:: default_inputs .. py:attribute:: default_outputs :annotation: = ['observs', 'rewards', 'oobs'] .. py:method:: observs_shape() :property: Return the shape of the observations. .. py:method:: observs_dtype() :property: Return the dtype of the observations. .. py:method:: action_shape() :property: Return the shape of the actions. .. py:method:: action_dtype() :property: Return the dtype of the actions. .. py:method:: param_dict() :property: Return the dictionary defining all the data attributes that the component requires. .. py:method:: step(**kwargs) :abstractmethod: Return the data corresponding to the new state of the environment after using the input data to make the corresponding state transition. .. py:method:: make_transitions(inplace = True, inactives = False, **kwargs) Return the data corresponding to the new state of the environment after using the input data to make the corresponding state transition. :param inplace: If ``False`` return the new data. If ``True``, update the state of the Swarm. :param inactives: Whether to update the walkers marked as inactive. :param \*\*kwargs: Keyword arguments passed if the returned value from the ``states_to_data`` function of the class was a dictionary. :returns: Dictionary containing the data representing the state of the environment after the state transition. The keys of the dictionary are the names of the data attributes and its values are arrays representing a batch of new values for that attribute. The :class:`StatesEnv` returned by ``step`` will contain the returned data. .. py:method:: reset(inplace = True, root_walker = None, states = None, inactives = True, **kwargs) Reset the internal state of the :class:`SwarmComponent`. :param inplace: If ``False`` return the new data. If ``True``, update the state of the Swarm. Defaults to ``True``. :type inplace: bool, optional :param root_walker: _description_. Defaults to None. :type root_walker: Optional[StateData], optional :param states: _description_. Defaults to None. :type states: Optional[StateData], optional :param inactives: Whether to update the walkers marked as inactive. Defaults to ``True``. :type inactives: bool, optional :param kwargs: Other arguments passed to make_transitions. :returns: Dictionary containing the data representing the state of the environment after the state transition. The keys of the dictionary are the names of the data attributes and its values are arrays representing a batch of new values for that attribute. The :class:`StatesEnv` returned by ``step`` will contain the returned data. :rtype: Union[None, StateData] .. py:class:: PolicyAPI(swarm = None, param_dict = None, inputs = None, outputs = None) Bases: :py:obj:`SwarmComponent` The policy is in charge of calculating the interactions with the :class:`Environment`. The PolicyAPI class is responsible for defining the policy that determines the actions for interacting with the :class:`Environment` in a swarm simulation. This is an abstract base class, and specific policy implementations should inherit from this class and implement the 'select_actions' method. .. py:attribute:: default_outputs .. py:method:: select_actions(**kwargs) :abstractmethod: Select actions for each walker in the swarm based on the current state. This method must be implemented by subclasses. :param \*\*kwargs: Additional keyword arguments required for selecting actions. :returns: The selected actions as a Tensor or a StateData dictionary. :rtype: Union[Tensor, StateData] .. py:method:: act(inplace = True, **kwargs) Calculate SwarmState containing the data needed to interact with the environment. :param inplace: If True, updates the swarm state with the selected actions. If False, returns the selected actions. Defaults to True. :type inplace: bool, optional :param \*\*kwargs: Additional keyword arguments required for selecting actions. :returns: None if inplace is True. Otherwise, a StateData dictionary containing the selected actions. :rtype: Union[None, StateData] .. py:method:: reset(inplace = True, root_walker = None, states = None, **kwargs) Reset the internal state of the :class:`PolicyAPI`. :param inplace: If True, updates the swarm state with the selected actions. If False, returns the selected actions. Defaults to True. :type inplace: bool, optional :param root_walker: Set the internal state of the PolicyAPI to this value. Defaults to None. :type root_walker: Optional[StateData], optional :param states: Set the internal state of the PolicyAPI to this value. Defaults to None. :type states: Optional[StateData], optional :param \*\*kwargs: Other parameters required to reset the component. :returns: None if inplace is True. Otherwise, a StateData dictionary containing the selected actions. :rtype: Union[None, StateData] .. py:class:: Callback(swarm = None, param_dict = None, inputs = None, outputs = None) Bases: :py:obj:`SwarmComponent` The :class:`Walkers` is a data structure that takes care of all the data involved in making a Swarm evolve. .. py:attribute:: name .. py:method:: before_reset() .. py:method:: after_reset() .. py:method:: before_evolve() .. py:method:: after_evolve() .. py:method:: before_policy() .. py:method:: after_policy() .. py:method:: before_env() .. py:method:: after_env() .. py:method:: before_walkers() .. py:method:: after_walkers() .. py:method:: reset(inplace = True, root_walker = None, states = None, **kwargs) Reset the internal state of the :class:`SwarmComponent`. :param inplace: Unused. Defaults to True. :type inplace: bool, optional :param root_walker: Set the internal state of the :class:`SwarmComponent` to this value. Defaults to None. :type root_walker: Optional[StateData], optional :param states: Set the internal state of the :class:`SwarmComponent` to this value. Defaults to None. :type states: Optional[StateData], optional :param kwargs: Other parameters required to reset the component. .. py:method:: evolution_end() .. py:method:: run_end() .. py:class:: WalkersMetric(swarm = None, param_dict = None, inputs = None, outputs = None) Bases: :py:obj:`SwarmComponent` A component of a swarm simulation. Every class that stores its data in :class:`SwarmState` inherits from this class. :param swarm: Reference to the :class:`Swarm` that incorporates this :class:`SwarmComponent`. Defaults to None. :param param_dict: Describes the multi-dimensional arrays that define the internal state of the component. Defaults to None. :param inputs: Describes the data that this components processes as inputs. This data is present in the internal state of the :class:`Swarm`, and generated by other components. Defaults to None. :param outputs: Name of the data attributes that this component makes available as part of its internal state. Defaults to None. .. attribute:: swarm The parent :class:`Swarm` instance. :type: Swarm .. attribute:: n_walkers Number of walkers in the :class:`Swarm`. :type: int .. attribute:: inputs A dictionary mapping input names to their corresponding type and default values. :type: Dict[str, Dict[str, Any]] .. attribute:: outputs A tuple containing the names of the data attributes. :type: Tuple[str] .. attribute:: param_dict A dictionary mapping parameter names. :type: Dict[str, Dict[str, Any]] .. method:: get(...) .. method:: get_input_data(...) .. method:: update(...) .. method:: reset(...) .. method:: _prepare_tensors(...) .. py:method:: __call__(inplace = True, **kwargs) .. py:method:: calculate(**kwargs) :abstractmethod: .. py:method:: reset(inplace = True, root_walker = None, states = None, **kwargs) Reset the internal state of the :class:`SwarmComponent`. :param inplace: Unused. Defaults to True. :type inplace: bool, optional :param root_walker: Set the internal state of the :class:`SwarmComponent` to this value. Defaults to None. :type root_walker: Optional[StateData], optional :param states: Set the internal state of the :class:`SwarmComponent` to this value. Defaults to None. :type states: Optional[StateData], optional :param kwargs: Other parameters required to reset the component. .. py:class:: WalkersAPI(swarm = None, param_dict = None, inputs = None, outputs = None) Bases: :py:obj:`SwarmComponent` The WalkersAPI class defines the base functionality for managing walkers in a swarm simulation. This class inherits from the SwarmComponent class. .. py:method:: get_input_data() Return a dictionary with all the data that this component requires as inputs, including the data for inactive walkers. :returns: A dictionary containing all the required data for the WalkersAPI component, where each key is the name of a data attribute and its value is a multi-dimensional array. :rtype: StateData .. py:method:: update(other = None, inactives = True, **kwargs) Modify the data stored in the SwarmState instance. Existing attributes will be updated, and new attributes will be created if needed. :param other: State class that will be copied upon update. :param inactives: Whether to update the walkers marked as inactive. :param \*\*kwargs: It is possible to specify the update as name value attributes, where name is the name of the attribute to be updated, and value is the new value for the attribute. :returns: None .. py:method:: balance(inplace = True, **kwargs) Perform a balance operation on the swarm state. :param inplace: If True, updates the swarm state with the balanced data. If False, returns the balanced data. Defaults to True. :type inplace: bool, optional :param \*\*kwargs: Additional keyword arguments required for running the balance operation. :returns: None if inplace is True. Otherwise, a StateData dictionary containing the balanced data. :rtype: Union[None, StateData] .. py:method:: run_epoch(inplace = True, **kwargs) :abstractmethod: Implement the functionality for running an epoch in the derived class. This method is called during the balance operation. :param inplace: If `True`, updates the swarm state with the data generated during the epoch. If False, returns the data. Defaults to `True`. :type inplace: bool, optional :param \*\*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. :rtype: StateData .. py:method:: reset(inplace = True, **kwargs) Reset the internal state of the Walkers class. This method should be implemented in the derived class. :param inplace: Unused. Defaults to True. :type inplace: bool, optional :param \*\*kwargs: Additional keyword arguments required for resetting the component. :returns: None .. py:method:: get_in_bounds_compas(oobs=None) Return an array of indexes corresponding to alive walkers chosen at random. :param oobs: An optional boolean array indicating out-of-bounds walkers. If not provided, all walkers are considered alive. Defaults to None. :type oobs: Optional[np.ndarray], optional :returns: An array of indexes corresponding to randomly chosen alive walkers. :rtype: Tensor .. py:method:: clone_walkers(will_clone=None, compas_clone=None, **kwargs) Sample the clone probability distribution and clone the walkers accordingly. :param will_clone: A boolean array indicating which walkers will be cloned. If not provided, the clone operation will be performed on all walkers. Defaults to None. :type will_clone: Optional[np.ndarray], optional :param compas_clone: An array of indexes indicating the walkers to be cloned. If not provided, random alive walkers will be chosen. Defaults to None. :type compas_clone: Optional[np.ndarray], optional :param \*\*kwargs: Additional keyword arguments required for cloning the walkers. :returns: None .. py:class:: SwarmAPI(n_walkers, env, policy, walkers, callbacks = None, minimize = False, max_epochs = 1e+100) The Swarm implements the iteration logic to make the :class:`Walkers` evolve. It contains the necessary logic to use an Environment, a Model, and a Walkers instance to create the algorithm execution loop. This class defines a method called run() that receives two optional arguments, root_walker and state, and has no return value. This method runs the fractal AI Swarm evolution process until a stop condition is met. In its implementation, it calls several other methods: (before_reset(), reset(), after_reset(), evolve(), before_evolve(), after_evolve(), evolution_end(), before_env(), after_env(), before_policy(), after_policy(), before_walkers(), after_walkers(), run_end()) defined within the same class, which are mainly used to manage different aspects of the search process or to invoke user-defined callbacks. The evolve() method updates the states of the search environment and model, makes the walkers undergo a perturbation process, and balances them. It also invokes several other methods that trigger user-defined callbacks. The evolution_end() method returns ``True`` if any of the following conditions is met: 1. The current epoch exceeds the maximum allowed epochs. 2. All walkers are out of the problem domain. 3. Any callback of the class has set the ``evolution_end`` flag to ``True``. .. attribute:: walkers_last If `True` indicates that the :class:`~fragile.Walkers` class runs after acting on the environment. If `Fase`, the walkers run before acting on the environment. :type: bool :param n_walkers: The number of walkers in the swarm. :type n_walkers: int :param env: An environment that simulates the objective function. :type env: :class:`EnvironmentAPI` :param policy: A policy that defines how the individuals evolve. :type policy: :class:`PolicyAPI` :param walkers: A set of motion rules to control a population of walkers. :type walkers: :class:`WalkersAPI` :param callbacks: A list of functions to call at each iteration. :type callbacks: Optional[Iterable[Callback]] :param minimize: If ``True``, take the minimum value of fitness, else take the maximum. :type minimize: bool :param max_epochs: Maximum number of epochs allowed before the swarm search is stopped. :type max_epochs: int .. py:attribute:: walkers_last :annotation: = True .. py:method:: n_walkers() :property: Return the number of walkers in the swarm. .. py:method:: n_actives() :property: Returns the number of active walkers in the swarm. .. py:method:: epoch() :property: Return the current epoch of the search algorithm. .. py:method:: state() :property: Returns the state instance describing the walkers of the swarm. .. py:method:: env() :property: All the simulation code (problem specific) will be handled here. .. py:method:: policy() :property: All the policy and random perturbation code (problem specific) will be handled here. .. py:method:: walkers() :property: Access the :class:`Walkers` in charge of implementing the FAI evolution process. .. py:method:: callbacks() :property: Return the dictionary containing all the user-defined callbacks. .. py:method:: param_dict() :property: Return the copy of parameters dictionary describing the attributes of the walkers. .. py:method:: clone_names() :property: Return the set of all the attributes that are cloned when iterating the Swarm. .. py:method:: inputs() :property: Return the dictionary containing all the inputs of the search algorithm. .. py:method:: __len__() Return the the number of walkers in the swarm. .. py:method:: __getattr__(item) .. py:method:: setup_state(param_dict, n_walkers = None) Set up :class:`SwarmState` instance for the search. :param param_dict: Initial dictionary of parameters. :param n_walkers: Number of :class:`Walker` instances. If not set, uses the previous number. :returns: None. .. py:method:: setup() Prepare :class:`Swarm` and internal components. :returns: None. .. py:method:: register_callback(callback, setup = True) Register a ``Callback`` object with the :class:`Swarm`. When `setup=True`, calls setup() on the callback. Adds the Callback's attribute dictionary to the instance's :class:`State`. This allows the state to manage all the data the callback needs. :param callback: An instance of :class:`Callback`. :param setup: Indicates whether we should call `setup()` on the given ``callback``. :returns: None. .. py:method:: get(name, default = None, raise_error = False) Access attributes of the :class:`Swarm` and its children. :param name: Name of the attribute :param default: Default value to return if attribute doesn't exist. :param raise_error: Raise an error if the named attribute cannot be found. :returns: The attribute if it exists, else the default. .. py:method:: reset(root_walker = None, state = None) Reset a :class:`~fragile.Swarm` and clear the internal data to start a new search process. Upon being called, ``reset`` clears all internal data from previous runs and sets up the Swarm to begin a new search process. :param root_walker: A walker representing the initial state of the search. The walkers will be reset to thiswalker and it will be added to the root of the :class:`~fragile.StateTree`. :type root_walker: :class:`~fragile.OneWalker`, optional :param state: Defines the initial state of the `Swarm`. Defaults are loaded from the current setup given during instantiation of the `Swarm()` object. :type state: :class:`~fragile.SwarmState` .. py:method:: run(root_walker = None, state = None) Run a new search process until the stop condition is met. This method runs a new search process by resetting the walkers to the initial state given in the ``root_walker`` argument. If not provided, it will use the previously set root walker. The initial swarm state can be defined using the `state` dictionary. During each epoch of the search process, the swarm undergoes an evolution process in the environment, and the search continues until the stop condition is met. The stop condition is defined by either of the following conditions: - Maximum number of epochs are exceeded (defined in ``max_epochs`` attribute). - All the walkers are out of bounds (defined in ``oobs`` attribute of the ``walkers``). - Any user-defined callback returns True for its ``evolution_end`` method. After the stop condition is met, the ``run_end`` method is called and all callbacks' ``run_end`` methods are executed. :param root_walker: Walker representing the initial state of the search. The walkers will be reset to this walker, and it will be added to the root of the :class:`StateTree` if any. :type root_walker: Optional[StateData] :param state: StateData dictionary that define the initial state of the Swarm. :type state: Optional[StateData] :returns: None. .. py:method:: evolution_end() Check if the :class:`Swarm`'s evolution process should stop. Checks whether the maximum number of epochs has been reached, all walkers are out of bounds or any of the callbacks have ended their evolution. :returns: A boolean value indicating whether the evolution process has to stop. :rtype: bool .. py:method:: evolve() Make the walkers undergo a perturbation process in the swarm :class:`Environment`. The evolution algorithm function undergoes the following steps: 2. If the ``self.walkers_last`` is ``False`` then it calls the callback functions declared as 'before_walkers'. 3. If the ``self.walkers_last`` is ``False`` It balances the walkers according to the clone probabilities by calling the balance function on the :class:`Walkers` instance that is a component of the :class:`Swarm` class. 4. If the ``self.walkers_last`` is ``False`` After the balancing is complete, the function calls more callback functions declared as 'after_walkers'. 5. It then calls more callback functions defined under the 'before_policy' tag. 6. Makes each walker select an action by calling the act() method of the policy object. 7. Performs transition based on the new state created after taking the given action. 8. Calls more callback functions defined under the 'after_env' tag. 9. If the ``self.walkers_last`` is ``True`` runs the walkers balance process and related callbacks. This function updates the :class:`StatesEnv` and the :class:`StatesModel` after each step. .. py:method:: before_reset() "Called before resetting the search process back to its initial state. .. py:method:: after_reset() Called after resetting the search process back to its initial state. .. py:method:: before_policy() Called before using the policy to sample an action. .. py:method:: after_policy() Called after using the policy to sample an action. .. py:method:: before_env() Called before the environment has been updated. .. py:method:: after_env() Called after the environment has been updated. .. py:method:: before_walkers() Called before the walkers have been balanced. .. py:method:: after_walkers() Called after the walkers have been balanced. .. py:method:: before_evolve() Called before the evolve step has started. .. py:method:: after_evolve() Called after the evolve step has ended. .. py:method:: run_end() Called after the evolution process has ended.