Skip to content

Callbacks

Closed-loop control policies for replay(). Subclass ReplayCallback and read the per-step state from ReplayContext. See the Replay guide.

ReplayCallback

Base class for replay control policies.

A callback is invoked once per integration minute, before the model steps, so it can inspect the simulation so far and modify the inputs the model is about to consume. Subclasses:

  • store hyperparameters in __init__ (supplied at the replay() call site when the instance is created);
  • keep memory in instance attributes, which persist across all timesteps of a single replay() call and remain inspectable on the instance after the run finishes;
  • share state with other callbacks through ctx.shared, a free-form dict that persists across every step of the run; prefer it to holding references to other callback objects;
  • override :meth:action to read the simulation state from the context and modify the current-step inputs via ctx.set_input / ctx.add_input.

Callbacks run in the order they are passed to replay(); each mutates the same input vector in place, so later callbacks observe earlier edits. The return value of :meth:action is ignored.

...

Attributes:

Name Type Description
name str or None

Identifies this callback in the action log. Defaults to None, in which case replay() uses the class name. Set a distinct name when using two instances of the same class.
rbg_data object

The replay rbg_data, injected before the run starts so callbacks can access data-specific quantities (e.g. rbg_data.body_weight). None until replay() is called.

Methods:

Name Description
action

Inspects the context and modifies the current-step inputs in place.
Source code in src/py_replay_bg/callbacks/callback.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
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
class ReplayCallback:
    """Base class for replay control policies.

    A callback is invoked once per integration minute, *before* the model steps,
    so it can inspect the simulation so far and modify the inputs the model is
    about to consume. Subclasses:

    - store **hyperparameters** in ``__init__`` (supplied at the ``replay()`` call
      site when the instance is created);
    - keep **memory** in instance attributes, which persist across all timesteps
      of a single ``replay()`` call and remain inspectable on the instance after
      the run finishes;
    - share **state with other callbacks** through ``ctx.shared``, a free-form
      dict that persists across every step of the run; prefer it to holding
      references to other callback objects;
    - override :meth:`action` to read the simulation state from the context and
      modify the current-step inputs via ``ctx.set_input`` / ``ctx.add_input``.

    Callbacks run in the order they are passed to ``replay()``; each mutates the
    same input vector in place, so later callbacks observe earlier edits. The
    return value of :meth:`action` is ignored.

    ...
    Attributes
    ----------
    name : str or None
        Identifies this callback in the action log. Defaults to ``None``, in
        which case ``replay()`` uses the class name.
        Set a distinct ``name`` when using two instances of the same class.
    rbg_data : object
        The replay ``rbg_data``, injected before the run starts so callbacks can
        access data-specific quantities (e.g. ``rbg_data.body_weight``). ``None``
        until ``replay()`` is called.

    Methods
    -------
    action(ctx):
        Inspects the context and modifies the current-step inputs in place.
    """

    #: Identifies this callback in the action log. When ``None`` (the default),
    #: ``replay()`` uses the class name. Set a distinct ``name`` when using two
    #: instances of the same class.
    name: str = None

    #: The replay ``rbg_data`` is injected here before the run starts, so
    #: callbacks can access data-specific quantities (e.g. ``rbg_data.body_weight``)
    #: for their own unit conversions. ``None`` until ``replay()`` is called.
    rbg_data = None

    def action(self, ctx) -> None:
        """Inspects ``ctx`` and modifies the current-step inputs in place.

        Parameters
        ----------
        ctx : ReplayContext
            The :class:`~callbacks.context.ReplayContext` for the current step.
        """
        raise NotImplementedError

name = None class-attribute instance-attribute

rbg_data = None class-attribute instance-attribute

action(ctx)

Inspects ctx and modifies the current-step inputs in place.

Parameters:

Name Type Description Default
ctx ReplayContext

The :class:~callbacks.context.ReplayContext for the current step.

 required
Source code in src/py_replay_bg/callbacks/callback.py 
51
52
53
54
55
56
57
58
59
def action(self, ctx) -> None:
    """Inspects ``ctx`` and modifies the current-step inputs in place.

    Parameters
    ----------
    ctx : ReplayContext
        The :class:`~callbacks.context.ReplayContext` for the current step.
    """
    raise NotImplementedError

ReplayContext

Per-step view of a replay simulation passed to every callback.

A single instance is created once per replay() call and refreshed in place each minute (no per-step allocation). Callbacks address input channels by name via get_input / set_input / add_input and read history directly from output_history / input_history using k.

The context is model-agnostic: inputs are read and written in the model's own units. Any human-unit conversion (e.g. insulin U, carbs grams) is the callback's responsibility, using data it carries itself (e.g. body weight).

Note

The data layer holds a bolus/meal value across the whole yts window. A callback that sets an input for a single minute therefore delivers a one-minute pulse, which is an acceptable approximation for correction boluses; spread the value over several steps yourself if it matters.

Note

shared records only values, not when they were written. Because callbacks run in the order they are passed to replay(), a reader placed before a writer sees the value that writer published at an earlier step. When freshness matters, store the step alongside the value (e.g. ctx.shared["last_bolus"] = {"u": cb_u, "k": ctx.k}) and compare against ctx.k on read — see CorrectionBolus / HypoTreatment for a worked example.

...

Attributes:

Name Type Description
k int

Current integration minute (the step being decided).
t_hour float

Hour-of-day at step k.
yts int

Data cadence in integration steps (rbg_data.yts, i.e. 5 minutes when ts=1); use it to self-gate. A sensor's own cadence is sensor.ts.
u ndarray

Live, mutable input vector for this step (model units). Callbacks edit it through the helpers; later callbacks see earlier edits.
output_history ndarray

Interstitial glucose outputs; valid up to index k-1.
input_history ndarray

Applied input rows; valid up to index k-1.
measurement float

Latest sensor measurement available at step k (held between sensor samples). When replay() is called without a sensor this carries the true model output, so policies reading it behave as if perfectly observed. Prefer this over output_history in control policies.
measurement_history ndarray

Sensor measurement at integration resolution (zero-order held between samples); valid up to index k-1. Mirrors output_history but reflects the noisy sensor signal when a sensor is supplied.
data_to_input dict

Mapping from channel index to channel name.
model object

The live model (read-only by convention) for advanced state such as the plasma insulin trajectory (model.Ip[k-1]) to derive insulin-on-board.
shared dict

Free-form store for callbacks to exchange data by name. Empty at the start of the run; entries persist until overwritten and are never cleared between steps. Keys are a contract between cooperating callbacks: any callback able to produce a quantity may publish it, and readers need not know which class did.

Methods:

Name Description
get_input

Returns the current value of an input channel.
set_input

Sets an input channel to a value.
add_input

Adds a value to an input channel.
log

Appends an action record to the action log.
Source code in src/py_replay_bg/callbacks/context.py 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
class ReplayContext:
    """Per-step view of a replay simulation passed to every callback.

    A single instance is created once per ``replay()`` call and refreshed in
    place each minute (no per-step allocation). Callbacks address input channels
    by *name* via ``get_input`` / ``set_input`` / ``add_input`` and read history
    directly from ``output_history`` / ``input_history`` using ``k``.

    The context is model-agnostic: inputs are read and written in the model's own
    units. Any human-unit conversion (e.g. insulin U, carbs grams) is the
    callback's responsibility, using data it carries itself (e.g. body weight).

    Note
    ----
    The data layer holds a bolus/meal value across the whole ``yts`` window.
    A callback that sets an input for a single minute therefore delivers a
    one-minute pulse, which is an acceptable approximation for correction
    boluses; spread the value over several steps yourself if it matters.

    Note
    ----
    ``shared`` records only values, not when they were written. Because
    callbacks run in the order they are passed to ``replay()``, a reader placed
    *before* a writer sees the value that writer published at an *earlier* step.
    When freshness matters, store the step alongside the value (e.g.
    ``ctx.shared["last_bolus"] = {"u": cb_u, "k": ctx.k}``) and compare against
    ``ctx.k`` on read — see ``CorrectionBolus`` / ``HypoTreatment`` for a worked
    example.

    ...
    Attributes
    ----------
    k : int
        Current integration minute (the step being decided).
    t_hour : float
        Hour-of-day at step ``k``.
    yts : int
        Data cadence in integration steps (``rbg_data.yts``, i.e. 5 minutes when
        ``ts=1``); use it to self-gate. A sensor's own cadence is ``sensor.ts``.
    u : numpy.ndarray
        Live, mutable input vector for this step (model units). Callbacks edit it
        through the helpers; later callbacks see earlier edits.
    output_history : numpy.ndarray
        Interstitial glucose outputs; valid up to index ``k-1``.
    input_history : numpy.ndarray
        Applied input rows; valid up to index ``k-1``.
    measurement : float
        Latest sensor measurement available at step ``k`` (held between sensor
        samples). When ``replay()`` is called without a sensor this carries the
        true model output, so policies reading it behave as if perfectly
        observed. Prefer this over ``output_history`` in control policies.
    measurement_history : numpy.ndarray
        Sensor measurement at integration resolution (zero-order held between
        samples); valid up to index ``k-1``. Mirrors ``output_history`` but
        reflects the noisy sensor signal when a sensor is supplied.
    data_to_input : dict
        Mapping from channel index to channel name.
    model : object
        The live model (read-only by convention) for advanced state such as
        the plasma insulin trajectory (``model.Ip[k-1]``) to derive
        insulin-on-board.
    shared : dict
        Free-form store for callbacks to exchange data by name. Empty at the
        start of the run; entries persist until overwritten and are never
        cleared between steps. Keys are a contract between cooperating
        callbacks: any callback able to produce a quantity may publish it, and
        readers need not know which class did.

    Methods
    -------
    get_input(name):
        Returns the current value of an input channel.
    set_input(name, value):
        Sets an input channel to a value.
    add_input(name, value):
        Adds a value to an input channel.
    log(**fields):
        Appends an action record to the action log.
    """

    def __init__(self, rbg_data, model, output_history, input_history):
        """Constructs all the necessary attributes for the ReplayContext object.

        Parameters
        ----------
        rbg_data : object
            The replay data object providing the inputs, time grid and channel
            mapping.
        model : object
            The live model being replayed.
        output_history : numpy.ndarray
            The output trajectory buffer (filled as the replay progresses).
        input_history : numpy.ndarray
            The applied-input trajectory buffer (filled as the replay progresses).
        """
        self.k = 0
        self._t_hour = rbg_data.t_hour
        self.t_hour = self._t_hour[0]
        self.yts = rbg_data.yts
        self.u = rbg_data.u[0].copy()
        self.output_history = output_history
        self.input_history = input_history
        self.measurement = output_history[0]
        self.measurement_history = output_history
        self.data_to_input = rbg_data.data_to_input
        self.model = model
        self.shared = {}

        self._name_to_idx = {name: idx for idx, name in rbg_data.data_to_input.items()}
        self._actions = []
        self._active_cb = None

    # -- lifecycle (called by the replay loop) --------------------------------

    def _advance(self, k, u):
        """Refresh the context for integration minute ``k`` with baseline ``u``."""
        self.k = k
        self.t_hour = self._t_hour[k]
        self.u = u

    # -- input accessors (name-based) -----------------------------------------

    def get_input(self, name) -> float:
        """Returns the current value of an input channel (model units).

        Parameters
        ----------
        name : str
            The channel name (as in ``data_to_input``).

        Returns
        -------
        float
            The current value of input channel ``name``.
        """
        return float(self.u[self._name_to_idx[name]])

    def set_input(self, name, value) -> None:
        """Sets an input channel to a value (model units).

        Parameters
        ----------
        name : str
            The channel name (as in ``data_to_input``).
        value : float
            The value to assign to the channel.
        """
        self.u[self._name_to_idx[name]] = value

    def add_input(self, name, value) -> None:
        """Adds a value (model units) to an input channel.

        Parameters
        ----------
        name : str
            The channel name (as in ``data_to_input``).
        value : float
            The value to add to the channel.
        """
        self.u[self._name_to_idx[name]] += value

    # -- action logging -------------------------------------------------------

    def log(self, **fields) -> None:
        """Appends an action record to the action log (``k``, ``callback`` auto-filled).

        Pass whatever fields make the action reconstructable later, e.g. the
        previous and new input values if you want to track what changed.

        Parameters
        ----------
        **fields
            Arbitrary keyword fields stored on the record alongside the
            auto-filled ``k`` (current step) and ``callback`` (active callback).
        """
        record = {
            "k": self.k,
            "callback": self._active_cb,
        }
        record.update(fields)
        self._actions.append(record)

k = 0 instance-attribute

_t_hour = rbg_data.t_hour instance-attribute

t_hour = self._t_hour[0] instance-attribute

yts = rbg_data.yts instance-attribute

u = rbg_data.u[0].copy() instance-attribute

output_history = output_history instance-attribute

input_history = input_history instance-attribute

measurement = output_history[0] instance-attribute

measurement_history = output_history instance-attribute

data_to_input = rbg_data.data_to_input instance-attribute

model = model instance-attribute

shared = {} instance-attribute

_name_to_idx = {name: idx for idx, name in (rbg_data.data_to_input.items())} instance-attribute

_actions = [] instance-attribute

_active_cb = None instance-attribute

__init__(rbg_data, model, output_history, input_history)

Constructs all the necessary attributes for the ReplayContext object.

Parameters:

Name Type Description Default
rbg_data object

The replay data object providing the inputs, time grid and channel mapping.

 required
model object

The live model being replayed.

 required
output_history ndarray

The output trajectory buffer (filled as the replay progresses).

 required
input_history ndarray

The applied-input trajectory buffer (filled as the replay progresses).

 required
Source code in src/py_replay_bg/callbacks/context.py 
 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
def __init__(self, rbg_data, model, output_history, input_history):
    """Constructs all the necessary attributes for the ReplayContext object.

    Parameters
    ----------
    rbg_data : object
        The replay data object providing the inputs, time grid and channel
        mapping.
    model : object
        The live model being replayed.
    output_history : numpy.ndarray
        The output trajectory buffer (filled as the replay progresses).
    input_history : numpy.ndarray
        The applied-input trajectory buffer (filled as the replay progresses).
    """
    self.k = 0
    self._t_hour = rbg_data.t_hour
    self.t_hour = self._t_hour[0]
    self.yts = rbg_data.yts
    self.u = rbg_data.u[0].copy()
    self.output_history = output_history
    self.input_history = input_history
    self.measurement = output_history[0]
    self.measurement_history = output_history
    self.data_to_input = rbg_data.data_to_input
    self.model = model
    self.shared = {}

    self._name_to_idx = {name: idx for idx, name in rbg_data.data_to_input.items()}
    self._actions = []
    self._active_cb = None

_advance(k, u)

Refresh the context for integration minute k with baseline u.

Source code in src/py_replay_bg/callbacks/context.py 
115
116
117
118
119
def _advance(self, k, u):
    """Refresh the context for integration minute ``k`` with baseline ``u``."""
    self.k = k
    self.t_hour = self._t_hour[k]
    self.u = u

get_input(name)

Returns the current value of an input channel (model units).

Parameters:

Name Type Description Default
name str

The channel name (as in data_to_input).

 required

Returns:

Type Description
float

The current value of input channel name.
Source code in src/py_replay_bg/callbacks/context.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def get_input(self, name) -> float:
    """Returns the current value of an input channel (model units).

    Parameters
    ----------
    name : str
        The channel name (as in ``data_to_input``).

    Returns
    -------
    float
        The current value of input channel ``name``.
    """
    return float(self.u[self._name_to_idx[name]])

set_input(name, value)

Sets an input channel to a value (model units).

Parameters:

Name Type Description Default
name str

The channel name (as in data_to_input).

 required
value float

The value to assign to the channel.

 required
Source code in src/py_replay_bg/callbacks/context.py 
138
139
140
141
142
143
144
145
146
147
148
def set_input(self, name, value) -> None:
    """Sets an input channel to a value (model units).

    Parameters
    ----------
    name : str
        The channel name (as in ``data_to_input``).
    value : float
        The value to assign to the channel.
    """
    self.u[self._name_to_idx[name]] = value

add_input(name, value)

Adds a value (model units) to an input channel.

Parameters:

Name Type Description Default
name str

The channel name (as in data_to_input).

 required
value float

The value to add to the channel.

 required
Source code in src/py_replay_bg/callbacks/context.py 
150
151
152
153
154
155
156
157
158
159
160
def add_input(self, name, value) -> None:
    """Adds a value (model units) to an input channel.

    Parameters
    ----------
    name : str
        The channel name (as in ``data_to_input``).
    value : float
        The value to add to the channel.
    """
    self.u[self._name_to_idx[name]] += value

log(**fields)

Appends an action record to the action log (k, callback auto-filled).

Pass whatever fields make the action reconstructable later, e.g. the previous and new input values if you want to track what changed.

Parameters:

Name Type Description Default
**fields

Arbitrary keyword fields stored on the record alongside the auto-filled k (current step) and callback (active callback).

 {}
Source code in src/py_replay_bg/callbacks/context.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def log(self, **fields) -> None:
    """Appends an action record to the action log (``k``, ``callback`` auto-filled).

    Pass whatever fields make the action reconstructable later, e.g. the
    previous and new input values if you want to track what changed.

    Parameters
    ----------
    **fields
        Arbitrary keyword fields stored on the record alongside the
        auto-filled ``k`` (current step) and ``callback`` (active callback).
    """
    record = {
        "k": self.k,
        "callback": self._active_cb,
    }
    record.update(fields)
    self._actions.append(record)

Built-in callbacks

Bases: ReplayCallback

Reactive correction-bolus policy with a refractory window.

If the most recent glucose reading is above threshold and no correction has been issued within the last lockout_min minutes, deliver a correction bolus of (glucose - target) / cf units.

...

Attributes:

Name Type Description
threshold float

Glucose level (mg/dL) above which a correction is considered.
target float

Target glucose level (mg/dL) used to size the correction.
cf float

Correction factor (mg/dL per unit of insulin).
lockout_min int

Minimum number of minutes between consecutive boluses.
last_correction_k int

Integration minute of the last issued correction (memory across steps).
Notes

Publishes every issued bolus on the shared store under "last_bolus" as {"u": <units>, "k": <step>}, so other callbacks can react to recent insulin (e.g. size up a hypo treatment). See HypoTreatment.

Methods:

Name Description
action

Issues a correction bolus when the policy conditions are met.
Source code in src/py_replay_bg/callbacks/default_callbacks.py 
 4
 5
 6
 7
 8
 9
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
class CorrectionBolus(ReplayCallback):
    """Reactive correction-bolus policy with a refractory window.

    If the most recent glucose reading is above ``threshold`` and no correction
    has been issued within the last ``lockout_min`` minutes, deliver a correction
    bolus of ``(glucose - target) / cf`` units.

    ...
    Attributes
    ----------
    threshold : float
        Glucose level (mg/dL) above which a correction is considered.
    target : float
        Target glucose level (mg/dL) used to size the correction.
    cf : float
        Correction factor (mg/dL per unit of insulin).
    lockout_min : int
        Minimum number of minutes between consecutive boluses.
    last_correction_k : int
        Integration minute of the last issued correction (memory across steps).

    Notes
    -----
    Publishes every issued bolus on the shared store under ``"last_bolus"`` as
    ``{"u": <units>, "k": <step>}``, so other callbacks can react to recent
    insulin (e.g. size up a hypo treatment). See ``HypoTreatment``.

    Methods
    -------
    action(ctx):
        Issues a correction bolus when the policy conditions are met.
    """

    def __init__(self, threshold=180.0, target=120.0, cf=40.0, lockout_min=60):
        """Constructs all the necessary attributes for the CorrectionBolus object.

        Parameters
        ----------
        threshold : float, optional, default : 180.0
            Glucose level (mg/dL) above which a correction is considered.
        target : float, optional, default : 120.0
            Target glucose level (mg/dL) used to size the correction.
        cf : float, optional, default : 40.0
            Correction factor (mg/dL per unit of insulin).
        lockout_min : int, optional, default : 60
            Minimum number of minutes between consecutive boluses.
        """
        self.threshold = threshold
        self.target = target
        self.cf = cf
        self.lockout_min = lockout_min
        self.last_correction_k = -10 ** 9

    def action(self, ctx):
        """Issues a correction bolus when glucose is high and the lockout elapsed.

        Parameters
        ----------
        ctx : ReplayContext
            The replay context for the current step.
        """
        g = ctx.measurement
        if g > self.threshold and (ctx.k - self.last_correction_k) >= self.lockout_min:
            cb_u = (g - self.target) / self.cf
            cb_mu_kgmin = cb_u * (1000.0 / self.rbg_data.body_weight)
            bolus_prev = ctx.get_input("bolus")
            ctx.add_input("bolus", cb_mu_kgmin)
            ctx.log(cb_u=cb_u, prev_u=bolus_prev, new_u=ctx.get_input("bolus"))
            self.last_correction_k = ctx.k
            ctx.shared["last_bolus"] = {"u": cb_u, "k": ctx.k}

threshold = threshold instance-attribute

target = target instance-attribute

cf = cf instance-attribute

lockout_min = lockout_min instance-attribute

last_correction_k = -10 ** 9 instance-attribute

__init__(threshold=180.0, target=120.0, cf=40.0, lockout_min=60)

Constructs all the necessary attributes for the CorrectionBolus object.

Parameters:

Name Type Description Default
threshold float

Glucose level (mg/dL) above which a correction is considered.

 : 180.0
target float

Target glucose level (mg/dL) used to size the correction.

 : 120.0
cf float

Correction factor (mg/dL per unit of insulin).

 : 40.0
lockout_min int

Minimum number of minutes between consecutive boluses.

 : 60
Source code in src/py_replay_bg/callbacks/default_callbacks.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(self, threshold=180.0, target=120.0, cf=40.0, lockout_min=60):
    """Constructs all the necessary attributes for the CorrectionBolus object.

    Parameters
    ----------
    threshold : float, optional, default : 180.0
        Glucose level (mg/dL) above which a correction is considered.
    target : float, optional, default : 120.0
        Target glucose level (mg/dL) used to size the correction.
    cf : float, optional, default : 40.0
        Correction factor (mg/dL per unit of insulin).
    lockout_min : int, optional, default : 60
        Minimum number of minutes between consecutive boluses.
    """
    self.threshold = threshold
    self.target = target
    self.cf = cf
    self.lockout_min = lockout_min
    self.last_correction_k = -10 ** 9

action(ctx)

Issues a correction bolus when glucose is high and the lockout elapsed.

Parameters:

Name Type Description Default
ctx ReplayContext

The replay context for the current step.

 required
Source code in src/py_replay_bg/callbacks/default_callbacks.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def action(self, ctx):
    """Issues a correction bolus when glucose is high and the lockout elapsed.

    Parameters
    ----------
    ctx : ReplayContext
        The replay context for the current step.
    """
    g = ctx.measurement
    if g > self.threshold and (ctx.k - self.last_correction_k) >= self.lockout_min:
        cb_u = (g - self.target) / self.cf
        cb_mu_kgmin = cb_u * (1000.0 / self.rbg_data.body_weight)
        bolus_prev = ctx.get_input("bolus")
        ctx.add_input("bolus", cb_mu_kgmin)
        ctx.log(cb_u=cb_u, prev_u=bolus_prev, new_u=ctx.get_input("bolus"))
        self.last_correction_k = ctx.k
        ctx.shared["last_bolus"] = {"u": cb_u, "k": ctx.k}

Bases: ReplayCallback

Reactive hypo-treatment policy with a refractory window.

If the most recent glucose reading is below threshold and no treatment has been issued within the last lockout_min minutes, deliver a fixed amount of fast-acting carbohydrates (the "rule of 15": 15 g, wait 15 min).

...

Attributes:

Name Type Description
threshold float

Glucose level (mg/dL) below which a treatment is considered.
carbs float

Amount of rescue carbohydrates to deliver (grams).
lockout_min int

Minimum number of minutes between consecutive treatments.
stacking_window_min int

If a bolus was published on ctx.shared["last_bolus"] within this many minutes, that insulin is still working, so stacking_carbs extra grams are added to the rescue. 0 disables the behaviour.
stacking_carbs float

Extra rescue carbohydrates (grams) added when a recent bolus is detected.
last_treatment_k int

Integration minute of the last issued treatment (memory across steps).

Methods:

Name Description
action

Issues a hypo-treatment when the policy conditions are met.
Source code in src/py_replay_bg/callbacks/default_callbacks.py 
 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
147
148
149
150
151
152
153
154
155
156
157
class HypoTreatment(ReplayCallback):
    """Reactive hypo-treatment policy with a refractory window.

    If the most recent glucose reading is below ``threshold`` and no treatment
    has been issued within the last ``lockout_min`` minutes, deliver a fixed
    amount of fast-acting carbohydrates (the "rule of 15": 15 g, wait 15 min).

    ...
    Attributes
    ----------
    threshold : float
        Glucose level (mg/dL) below which a treatment is considered.
    carbs : float
        Amount of rescue carbohydrates to deliver (grams).
    lockout_min : int
        Minimum number of minutes between consecutive treatments.
    stacking_window_min : int
        If a bolus was published on ``ctx.shared["last_bolus"]`` within this many
        minutes, that insulin is still working, so ``stacking_carbs`` extra grams
        are added to the rescue. ``0`` disables the behaviour.
    stacking_carbs : float
        Extra rescue carbohydrates (grams) added when a recent bolus is detected.
    last_treatment_k : int
        Integration minute of the last issued treatment (memory across steps).

    Methods
    -------
    action(ctx):
        Issues a hypo-treatment when the policy conditions are met.
    """

    def __init__(self, threshold=70.0, carbs=15.0, lockout_min=15,
                 stacking_window_min=0, stacking_carbs=0.0):
        """Constructs all the necessary attributes for the HypoTreatment object.

        Parameters
        ----------
        threshold : float, optional, default : 70.0
            Glucose level (mg/dL) below which a treatment is considered.
        carbs : float, optional, default : 15.0
            Amount of rescue carbohydrates to deliver (grams).
        lockout_min : int, optional, default : 15
            Minimum number of minutes between consecutive treatments.
        stacking_window_min : int, optional, default : 0
            If a bolus was published on ``ctx.shared["last_bolus"]`` within this
            many minutes, add ``stacking_carbs`` extra grams to the rescue.
            ``0`` disables the behaviour.
        stacking_carbs : float, optional, default : 0.0
            Extra rescue carbohydrates (grams) added when a recent bolus is
            detected within ``stacking_window_min``.
        """
        self.threshold = threshold
        self.carbs = carbs
        self.lockout_min = lockout_min
        self.stacking_window_min = stacking_window_min
        self.stacking_carbs = stacking_carbs
        self.last_treatment_k = -10 ** 9

    def action(self, ctx):
        """Issues a hypo-treatment when glucose is low and the lockout elapsed.

        Parameters
        ----------
        ctx : ReplayContext
            The replay context for the current step.
        """
        g = ctx.measurement
        if g < self.threshold and (ctx.k - self.last_treatment_k) >= self.lockout_min:
            carbs = self.carbs
            recent_bolus_u = 0.0
            if self.stacking_window_min > 0:
                last_bolus = ctx.shared.get("last_bolus")
                # The store is not timestamped, so check the step the writer recorded.
                if last_bolus is not None and (ctx.k - last_bolus["k"]) <= self.stacking_window_min:
                    recent_bolus_u = last_bolus["u"]
                    carbs += self.stacking_carbs
            carbs_mg_kgmin = carbs * (1000.0 / self.rbg_data.body_weight)
            meal_prev = ctx.get_input("meal_H")
            ctx.add_input("meal_H", carbs_mg_kgmin)
            ctx.log(ht_g=carbs, recent_bolus_u=recent_bolus_u,
                    prev_u=meal_prev, new_u=ctx.get_input("meal_H"))
            self.last_treatment_k = ctx.k

threshold = threshold instance-attribute

carbs = carbs instance-attribute

lockout_min = lockout_min instance-attribute

stacking_window_min = stacking_window_min instance-attribute

stacking_carbs = stacking_carbs instance-attribute

last_treatment_k = -10 ** 9 instance-attribute

__init__(threshold=70.0, carbs=15.0, lockout_min=15, stacking_window_min=0, stacking_carbs=0.0)

Constructs all the necessary attributes for the HypoTreatment object.

Parameters:

Name Type Description Default
threshold float

Glucose level (mg/dL) below which a treatment is considered.

 : 70.0
carbs float

Amount of rescue carbohydrates to deliver (grams).

 : 15.0
lockout_min int

Minimum number of minutes between consecutive treatments.

 : 15
stacking_window_min int

If a bolus was published on ctx.shared["last_bolus"] within this many minutes, add stacking_carbs extra grams to the rescue. 0 disables the behaviour.

 : 0
stacking_carbs float

Extra rescue carbohydrates (grams) added when a recent bolus is detected within stacking_window_min.

 : 0.0
Source code in src/py_replay_bg/callbacks/default_callbacks.py 
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
def __init__(self, threshold=70.0, carbs=15.0, lockout_min=15,
             stacking_window_min=0, stacking_carbs=0.0):
    """Constructs all the necessary attributes for the HypoTreatment object.

    Parameters
    ----------
    threshold : float, optional, default : 70.0
        Glucose level (mg/dL) below which a treatment is considered.
    carbs : float, optional, default : 15.0
        Amount of rescue carbohydrates to deliver (grams).
    lockout_min : int, optional, default : 15
        Minimum number of minutes between consecutive treatments.
    stacking_window_min : int, optional, default : 0
        If a bolus was published on ``ctx.shared["last_bolus"]`` within this
        many minutes, add ``stacking_carbs`` extra grams to the rescue.
        ``0`` disables the behaviour.
    stacking_carbs : float, optional, default : 0.0
        Extra rescue carbohydrates (grams) added when a recent bolus is
        detected within ``stacking_window_min``.
    """
    self.threshold = threshold
    self.carbs = carbs
    self.lockout_min = lockout_min
    self.stacking_window_min = stacking_window_min
    self.stacking_carbs = stacking_carbs
    self.last_treatment_k = -10 ** 9

action(ctx)

Issues a hypo-treatment when glucose is low and the lockout elapsed.

Parameters:

Name Type Description Default
ctx ReplayContext

The replay context for the current step.

 required
Source code in src/py_replay_bg/callbacks/default_callbacks.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def action(self, ctx):
    """Issues a hypo-treatment when glucose is low and the lockout elapsed.

    Parameters
    ----------
    ctx : ReplayContext
        The replay context for the current step.
    """
    g = ctx.measurement
    if g < self.threshold and (ctx.k - self.last_treatment_k) >= self.lockout_min:
        carbs = self.carbs
        recent_bolus_u = 0.0
        if self.stacking_window_min > 0:
            last_bolus = ctx.shared.get("last_bolus")
            # The store is not timestamped, so check the step the writer recorded.
            if last_bolus is not None and (ctx.k - last_bolus["k"]) <= self.stacking_window_min:
                recent_bolus_u = last_bolus["u"]
                carbs += self.stacking_carbs
        carbs_mg_kgmin = carbs * (1000.0 / self.rbg_data.body_weight)
        meal_prev = ctx.get_input("meal_H")
        ctx.add_input("meal_H", carbs_mg_kgmin)
        ctx.log(ht_g=carbs, recent_bolus_u=recent_bolus_u,
                prev_u=meal_prev, new_u=ctx.get_input("meal_H"))
        self.last_treatment_k = ctx.k