Skip to content

swarmrl.observables.concentration_field Module API Reference

Historical position observable computer.

Notes

Observable for sensing changes in some field value, or, the gradient.

ConcentrationField

Bases: Observable, ABC

Position in box observable.

Attributes

historic_positions : dict A dictionary of past positions of the colloid to be used in the gradient computation.

Source code in swarmrl/observables/concentration_field.py 
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class ConcentrationField(Observable, ABC):
    """
    Position in box observable.

    Attributes
    ----------
    historic_positions : dict
            A dictionary of past positions of the colloid to be used in the gradient
            computation.
    """

    def __init__(
        self,
        source: np.ndarray,
        decay_fn: callable,
        box_length: np.ndarray,
        scale_factor: int = 100,
        particle_type: int = 0,
    ):
        """
        Constructor for the observable.

        Parameters
        ----------
        source : np.ndarray
                Source of the field.
        decay_fn : callable
                Decay function of the field.
        box_size : np.ndarray
                Array for scaling of the distances.
        scale_factor : int (default=100)
                Scaling factor for the observable.
        particle_type : int (default=0)
                Particle type to compute the observable for.
        """
        super().__init__(particle_type=particle_type)

        self.source = source / box_length
        self.decay_fn = decay_fn
        self._historic_positions = {}
        self.box_length = box_length
        self.scale_factor = scale_factor
        self._observable_shape = (3,)

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

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

        Returns
        -------
        Updates the class state.
        """
        for item in colloids:
            index = onp.copy(item.id)
            position = onp.copy(item.pos) / self.box_length
            self._historic_positions[str(index)] = position

    def compute_single_observable(self, index: int, colloids: List[Colloid]) -> float:
        """
        Compute the observable for a single colloid.

        Parameters
        ----------
        index : int
                Index of the colloid to compute the observable for.
        colloids : List[Colloid]
                List of colloids in the system.
        """
        reference_colloid = colloids[index]
        position = onp.copy(reference_colloid.pos) / self.box_length
        index = onp.copy(reference_colloid.id)
        previous_position = self._historic_positions[str(index)]

        # Update historic position.
        self._historic_positions[str(index)] = position

        current_distance = np.linalg.norm(self.source - position)
        historic_distance = np.linalg.norm(self.source - previous_position)

        delta = self.decay_fn(current_distance) - self.decay_fn(historic_distance)

        return self.scale_factor * delta

    def compute_observable(self, colloids: List[Colloid]):
        """
        Compute the position of the colloid.

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

        Returns
        -------
        observables : List[float] (n_colloids, dimension)
                List of observables, one for each colloid. In this case,
                current field value minus to previous field value.
        """
        reference_ids = self.get_colloid_indices(colloids)

        if self._historic_positions == {}:
            msg = (
                f"{type(self).__name__} requires initialization. Please set the "
                "initialize attribute of the gym to true and try again."
            )
            raise ValueError(msg)

        observables = [
            self.compute_single_observable(index, colloids) for index in reference_ids
        ]

        return np.array(observables).reshape(-1, 1)

__init__(source, decay_fn, box_length, scale_factor=100, particle_type=0)

Constructor for the observable.

Parameters

source : np.ndarray Source of the field. decay_fn : callable Decay function of the field. box_size : np.ndarray Array for scaling of the distances. scale_factor : int (default=100) Scaling factor for the observable. particle_type : int (default=0) Particle type to compute the observable for.

Source code in swarmrl/observables/concentration_field.py 
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
def __init__(
    self,
    source: np.ndarray,
    decay_fn: callable,
    box_length: np.ndarray,
    scale_factor: int = 100,
    particle_type: int = 0,
):
    """
    Constructor for the observable.

    Parameters
    ----------
    source : np.ndarray
            Source of the field.
    decay_fn : callable
            Decay function of the field.
    box_size : np.ndarray
            Array for scaling of the distances.
    scale_factor : int (default=100)
            Scaling factor for the observable.
    particle_type : int (default=0)
            Particle type to compute the observable for.
    """
    super().__init__(particle_type=particle_type)

    self.source = source / box_length
    self.decay_fn = decay_fn
    self._historic_positions = {}
    self.box_length = box_length
    self.scale_factor = scale_factor
    self._observable_shape = (3,)

compute_observable(colloids)

Compute the position of the colloid.

Parameters

colloids : List[Colloid] (n_colloids, ) List of all colloids in the system.

Returns

observables : List[float] (n_colloids, dimension) List of observables, one for each colloid. In this case, current field value minus to previous field value.

Source code in swarmrl/observables/concentration_field.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def compute_observable(self, colloids: List[Colloid]):
    """
    Compute the position of the colloid.

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

    Returns
    -------
    observables : List[float] (n_colloids, dimension)
            List of observables, one for each colloid. In this case,
            current field value minus to previous field value.
    """
    reference_ids = self.get_colloid_indices(colloids)

    if self._historic_positions == {}:
        msg = (
            f"{type(self).__name__} requires initialization. Please set the "
            "initialize attribute of the gym to true and try again."
        )
        raise ValueError(msg)

    observables = [
        self.compute_single_observable(index, colloids) for index in reference_ids
    ]

    return np.array(observables).reshape(-1, 1)

compute_single_observable(index, colloids)

Compute the observable for a single colloid.

Parameters

index : int Index of the colloid to compute the observable for. colloids : List[Colloid] List of colloids in the system.

Source code in swarmrl/observables/concentration_field.py 
 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
def compute_single_observable(self, index: int, colloids: List[Colloid]) -> float:
    """
    Compute the observable for a single colloid.

    Parameters
    ----------
    index : int
            Index of the colloid to compute the observable for.
    colloids : List[Colloid]
            List of colloids in the system.
    """
    reference_colloid = colloids[index]
    position = onp.copy(reference_colloid.pos) / self.box_length
    index = onp.copy(reference_colloid.id)
    previous_position = self._historic_positions[str(index)]

    # Update historic position.
    self._historic_positions[str(index)] = position

    current_distance = np.linalg.norm(self.source - position)
    historic_distance = np.linalg.norm(self.source - previous_position)

    delta = self.decay_fn(current_distance) - self.decay_fn(historic_distance)

    return self.scale_factor * delta

initialize(colloids)

Initialize the observable with starting positions of the colloids.

Parameters

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

Returns

Updates the class state.

Source code in swarmrl/observables/concentration_field.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def initialize(self, colloids: List[Colloid]):
    """
    Initialize the observable with starting positions of the colloids.

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

    Returns
    -------
    Updates the class state.
    """
    for item in colloids:
        index = onp.copy(item.id)
        position = onp.copy(item.pos) / self.box_length
        self._historic_positions[str(index)] = position