Skip to content

swarmrl.tasks.task Module API Reference

Module for the parent class of the tasks.

Notes

The reward classes handle the computation of the reward from an environment and compute the loss for the models to train on.

Task

Parent class for the reinforcement learning tasks.

Source code in swarmrl/tasks/task.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class Task:
    """
    Parent class for the reinforcement learning tasks.
    """

    def __init__(self, particle_type: int = 0):
        """
        Constructor for the reward class.

        Parameters
        ----------
        particle_type : int (default=0)
                Particle type to compute the reward for.
        """
        self.particle_type = particle_type

        self._kill_switch = False

    @property
    def kill_switch(self):
        """
        Kill switch property of the task
        """
        return self._kill_switch

    @kill_switch.setter
    def kill_switch(self, value: bool):
        """
        Set the kill switch property.

        Parameters
        ----------
        value : bool
            Value to set the kill switch to.
        """
        self._kill_switch = value

    def initialize(self, colloids: List[Colloid]):
        """
        Initialize the task with starting positions of the colloids.

        The parent method will just pass. This is because some observables
        might not need to be initialized. Those that do need to be initialized
        will override this method.

        Parameters
        ----------
        colloids : List[Colloid]
                List of colloids with which to initialize the observable.

        Returns
        -------
        Updates the class state.
        """
        pass

    def get_colloid_indices(self, colloids: List[Colloid], p_type: int = None):
        """
        Get the indices of the colloids in the observable of a specific type.

        Parameters
        ----------
        colloids : List[Colloid]
                List of colloids from which to get the indices.
        p_type : int (default=None)
                Type of the colloids to get the indices for. If None, the
                particle_type attribute of the class is used.


        Returns
        -------
        indices : List[int]
                List of indices for the colloids of a particular type.
        """
        if p_type is None:
            p_type = self.particle_type

        indices = []
        for i, colloid in enumerate(colloids):
            if colloid.type == p_type:
                indices.append(i)

        return indices

    def __call__(self, colloids: List[Colloid]) -> float:
        """
        Compute the reward on the whole group of particles.

        Parameters
        ----------
        colloids : List[Colloid] (n_colloids, dimension)
                List of colloid objects in the system.

        Returns
        -------
        Reward : float
                Reward for the current state.

        Examples
        --------
        my_task = Task()
        reward = my_task(state)
        """
        raise NotImplementedError("Implemented in child class.")

kill_switch property writable

Kill switch property of the task

__call__(colloids)

Compute the reward on the whole group of particles.

Parameters

colloids : List[Colloid] (n_colloids, dimension) List of colloid objects in the system.

Returns

Reward : float Reward for the current state.

Examples

my_task = Task() reward = my_task(state)

Source code in swarmrl/tasks/task.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def __call__(self, colloids: List[Colloid]) -> float:
    """
    Compute the reward on the whole group of particles.

    Parameters
    ----------
    colloids : List[Colloid] (n_colloids, dimension)
            List of colloid objects in the system.

    Returns
    -------
    Reward : float
            Reward for the current state.

    Examples
    --------
    my_task = Task()
    reward = my_task(state)
    """
    raise NotImplementedError("Implemented in child class.")

__init__(particle_type=0)

Constructor for the reward class.

Parameters

particle_type : int (default=0) Particle type to compute the reward for.

Source code in swarmrl/tasks/task.py 
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, particle_type: int = 0):
    """
    Constructor for the reward class.

    Parameters
    ----------
    particle_type : int (default=0)
            Particle type to compute the reward for.
    """
    self.particle_type = particle_type

    self._kill_switch = False

get_colloid_indices(colloids, p_type=None)

Get the indices of the colloids in the observable of a specific type.

Parameters

colloids : List[Colloid] List of colloids from which to get the indices. p_type : int (default=None) Type of the colloids to get the indices for. If None, the particle_type attribute of the class is used.

Returns

indices : List[int] List of indices for the colloids of a particular type.

Source code in swarmrl/tasks/task.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def get_colloid_indices(self, colloids: List[Colloid], p_type: int = None):
    """
    Get the indices of the colloids in the observable of a specific type.

    Parameters
    ----------
    colloids : List[Colloid]
            List of colloids from which to get the indices.
    p_type : int (default=None)
            Type of the colloids to get the indices for. If None, the
            particle_type attribute of the class is used.


    Returns
    -------
    indices : List[int]
            List of indices for the colloids of a particular type.
    """
    if p_type is None:
        p_type = self.particle_type

    indices = []
    for i, colloid in enumerate(colloids):
        if colloid.type == p_type:
            indices.append(i)

    return indices

initialize(colloids)

Initialize the task with starting positions of the colloids.

The parent method will just pass. This is because some observables might not need to be initialized. Those that do need to be initialized will override this method.

Parameters

colloids : List[Colloid] List of colloids with which to initialize the observable.

Returns

Updates the class state.

Source code in swarmrl/tasks/task.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def initialize(self, colloids: List[Colloid]):
    """
    Initialize the task with starting positions of the colloids.

    The parent method will just pass. This is because some observables
    might not need to be initialized. Those that do need to be initialized
    will override this method.

    Parameters
    ----------
    colloids : List[Colloid]
            List of colloids with which to initialize the observable.

    Returns
    -------
    Updates the class state.
    """
    pass