Skip to content

Task Module

task

ChangePointTask

Change-point task: Implementation of the basic functions.

ChangePointTask

Specifies attributes and methods of the Task object that models the change-point task.

Source code in rbmpy/task/ChangePointTask.py 
 10
 11
 12
 13
 14
 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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class ChangePointTask:
    """Specifies attributes and methods of the Task object that models the change-point task."""

    def __init__(self, task_vars: TaskVars):
        """Creates the Task object based on the initialization object.

        Parameters
        ----------
        task_vars : TaskVars
            Object instance with task parameters.
        """

        self.sigma = task_vars.sigma
        self.kappa = task_vars.kappa
        self.h = task_vars.h
        self.min_x = task_vars.min_x
        self.max_x = task_vars.max_x
        self.min_mu = task_vars.min_mu
        self.max_mu = task_vars.max_mu
        self.new_block = task_vars.new_block
        self.variable_shield = task_vars.variable_shield
        self.shield_min = task_vars.shield_min
        self.shield_max = task_vars.shield_max
        self.shield_mu = task_vars.shield_mu
        self.safe = task_vars.safe
        self.s = self.safe
        self.circular = task_vars.circular
        self.catch_trial_prob = task_vars.catch_trial_prob

        # Initialize other variables
        self.x_t = np.nan
        self.mu = np.nan
        self.cp = np.nan
        self.shield_size = np.nan
        self.catch_trial = np.nan

    def sample_cp(self) -> None:
        """Samples change points.

        The function takes into account the hazard rate h and the safe criterion s.

        Returns
        -------
        None
            This function does not return any value.
        """

        if self.new_block == 1:
            self.cp = 1
        elif self.s == 0:
            self.cp = np.random.binomial(1, self.h)
        else:
            self.cp = 0

        # Update safe criterion
        if self.cp:
            self.s = self.safe
        else:
            self.s = max([self.s - 1, 0])

    def sample_mu(self) -> None:
        """Samples the mean of the outcome-generating distribution conditional on a change point.

        Returns
        -------
        None
            This function does not return any value.
        """

        if self.cp == 1:
            self.mu = np.random.uniform(self.min_mu, self.max_mu)

    def sample_outcome(self) -> None:
        """Samples the outcome conditional on the outcome-generating mean.

        The function works for normal and circular outcome spaces.

        Returns
        -------
        None
            This function does not return any value.
        """

        if not self.circular:

            self.x_t = round(np.random.normal(self.mu, self.sigma))
            if self.x_t <= self.min_x:
                self.x_t = self.min_x
            elif self.x_t >= self.max_x:
                self.x_t = self.max_x

        elif self.circular:

            # Sample outcome from von Mises distribution
            self.x_t = np.random.vonmises(self.mu, self.kappa) % (2 * np.pi)

        else:

            sys.exit("Invalid option for outcome space")

    def sample_shield(self) -> None:
        """Samples the size of the shield.

        Returns
        -------
        None
            This function does not return any value.
        """

        if self.variable_shield:

            # Sample shield from exponential distribution
            self.shield_size = np.nan
            while (
                np.isnan(self.shield_size)
                or self.shield_size < self.shield_min
                or self.shield_size > self.shield_max
            ):
                self.shield_size = np.random.exponential(self.shield_mu)

        else:

            self.shield_size = self.shield_mu

    def sample_catch_trial(self) -> None:
        """Samples catch trials.

        Returns
        -------
        None
            This function does not return any value.
        """

        if self.cp == 0:
            self.catch_trial = np.random.binomial(1, self.catch_trial_prob)
        else:
            self.catch_trial = 0
__init__(task_vars)

Creates the Task object based on the initialization object.

Parameters:

Name Type Description Default
task_vars TaskVars

Object instance with task parameters.

 required
Source code in rbmpy/task/ChangePointTask.py 
13
14
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
def __init__(self, task_vars: TaskVars):
    """Creates the Task object based on the initialization object.

    Parameters
    ----------
    task_vars : TaskVars
        Object instance with task parameters.
    """

    self.sigma = task_vars.sigma
    self.kappa = task_vars.kappa
    self.h = task_vars.h
    self.min_x = task_vars.min_x
    self.max_x = task_vars.max_x
    self.min_mu = task_vars.min_mu
    self.max_mu = task_vars.max_mu
    self.new_block = task_vars.new_block
    self.variable_shield = task_vars.variable_shield
    self.shield_min = task_vars.shield_min
    self.shield_max = task_vars.shield_max
    self.shield_mu = task_vars.shield_mu
    self.safe = task_vars.safe
    self.s = self.safe
    self.circular = task_vars.circular
    self.catch_trial_prob = task_vars.catch_trial_prob

    # Initialize other variables
    self.x_t = np.nan
    self.mu = np.nan
    self.cp = np.nan
    self.shield_size = np.nan
    self.catch_trial = np.nan
sample_cp()

Samples change points.

The function takes into account the hazard rate h and the safe criterion s.

Returns:

Type Description
None

This function does not return any value.
Source code in rbmpy/task/ChangePointTask.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def sample_cp(self) -> None:
    """Samples change points.

    The function takes into account the hazard rate h and the safe criterion s.

    Returns
    -------
    None
        This function does not return any value.
    """

    if self.new_block == 1:
        self.cp = 1
    elif self.s == 0:
        self.cp = np.random.binomial(1, self.h)
    else:
        self.cp = 0

    # Update safe criterion
    if self.cp:
        self.s = self.safe
    else:
        self.s = max([self.s - 1, 0])
sample_mu()

Samples the mean of the outcome-generating distribution conditional on a change point.

Returns:

Type Description
None

This function does not return any value.
Source code in rbmpy/task/ChangePointTask.py 
70
71
72
73
74
75
76
77
78
79
80
def sample_mu(self) -> None:
    """Samples the mean of the outcome-generating distribution conditional on a change point.

    Returns
    -------
    None
        This function does not return any value.
    """

    if self.cp == 1:
        self.mu = np.random.uniform(self.min_mu, self.max_mu)
sample_outcome()

Samples the outcome conditional on the outcome-generating mean.

The function works for normal and circular outcome spaces.

Returns:

Type Description
None

This function does not return any value.
Source code in rbmpy/task/ChangePointTask.py 
 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
def sample_outcome(self) -> None:
    """Samples the outcome conditional on the outcome-generating mean.

    The function works for normal and circular outcome spaces.

    Returns
    -------
    None
        This function does not return any value.
    """

    if not self.circular:

        self.x_t = round(np.random.normal(self.mu, self.sigma))
        if self.x_t <= self.min_x:
            self.x_t = self.min_x
        elif self.x_t >= self.max_x:
            self.x_t = self.max_x

    elif self.circular:

        # Sample outcome from von Mises distribution
        self.x_t = np.random.vonmises(self.mu, self.kappa) % (2 * np.pi)

    else:

        sys.exit("Invalid option for outcome space")
sample_shield()

Samples the size of the shield.

Returns:

Type Description
None

This function does not return any value.
Source code in rbmpy/task/ChangePointTask.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def sample_shield(self) -> None:
    """Samples the size of the shield.

    Returns
    -------
    None
        This function does not return any value.
    """

    if self.variable_shield:

        # Sample shield from exponential distribution
        self.shield_size = np.nan
        while (
            np.isnan(self.shield_size)
            or self.shield_size < self.shield_min
            or self.shield_size > self.shield_max
        ):
            self.shield_size = np.random.exponential(self.shield_mu)

    else:

        self.shield_size = self.shield_mu
sample_catch_trial()

Samples catch trials.

Returns:

Type Description
None

This function does not return any value.
Source code in rbmpy/task/ChangePointTask.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
def sample_catch_trial(self) -> None:
    """Samples catch trials.

    Returns
    -------
    None
        This function does not return any value.
    """

    if self.cp == 0:
        self.catch_trial = np.random.binomial(1, self.catch_trial_prob)
    else:
        self.catch_trial = 0

TaskVars

TaskVars: Initialization of the change-point task.

TaskVars

Specifies attributes of the TaskVars object that are used for the change-point task.

Source code in rbmpy/task/TaskVars.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class TaskVars:
    """Specifies attributes of the TaskVars object that are used for the change-point task."""

    def __init__(self):
        """Determines the default task variables."""

        self.kappa = 16
        self.sigma = np.sqrt(1 / self.kappa)
        self.h = 0.125
        self.min_x = 0
        self.max_x = 2 * np.pi
        self.min_mu = 0
        self.max_mu = 2 * np.pi
        self.new_block = 1
        self.variable_shield = True
        self.shield_min = np.deg2rad(10)
        self.shield_max = np.pi
        self.shield_mu = np.deg2rad(10)
        self.safe = 3
        self.circular = True
        self.catch_trial_prob = 0.1
__init__()

Determines the default task variables.

Source code in rbmpy/task/TaskVars.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(self):
    """Determines the default task variables."""

    self.kappa = 16
    self.sigma = np.sqrt(1 / self.kappa)
    self.h = 0.125
    self.min_x = 0
    self.max_x = 2 * np.pi
    self.min_mu = 0
    self.max_mu = 2 * np.pi
    self.new_block = 1
    self.variable_shield = True
    self.shield_min = np.deg2rad(10)
    self.shield_max = np.pi
    self.shield_mu = np.deg2rad(10)
    self.safe = 3
    self.circular = True
    self.catch_trial_prob = 0.1