Skip to content

Sensors

The output-agnostic sensor interface and the default CGM error model. See the CGM error model guide.

Sensor

Bases: ABC

Output-agnostic sensor that turns a model output into a measurement.

A sensor observes the model's output trace during replay() and produces a (typically noisy, sub-sampled) measurement of it. It is deliberately generic: it knows nothing about glucose or any specific model. Concrete subclasses (e.g. :class:~sensors.vettoretti19_cgm.Vettoretti19CGM) implement the actual error model in :meth:measure.

...

Attributes:

Name Type Description
ts int

Sample time of the sensor, in integration steps. The replay loop takes a measurement when k % ts == 0.

t_offset int

Extra age of the current sensor, in integration steps (0 when new). Added to the time elapsed since connected_at to give the sensor's age, e.g. to account for a gap between two replay runs that share the sensor.

max_lifetime int

Maximum lifetime of the sensor, in integration steps (10 days at ts=1). When the sensor's age reaches it, the replay loop reconnects a fresh sensor realization.

connected_at int

Integration step, in the current replay's step frame, at which the current sensor was connected. replay() rebases it at the end of a run (it subtracts tsteps), so a sensor carried into the next run with continue_sensor=True keeps aging contiguously. It is negative for a sensor connected during an earlier run.

Methods:

Name Description
connect_new

Connects a fresh sensor by (re)sampling its error realization.

age

Age of the current sensor at integration step k, in integration steps.

measure

Produces a measurement from the current and past model outputs.

add_offset

Ages the current sensor by to_add integration steps.

Source code in src/py_replay_bg/sensors/sensor.py
class Sensor(ABC):
    """Output-agnostic sensor that turns a model output into a measurement.

    A sensor observes the model's output trace during ``replay()`` and produces a
    (typically noisy, sub-sampled) measurement of it. It is deliberately generic:
    it knows nothing about glucose or any specific model. Concrete subclasses (e.g.
    :class:`~sensors.vettoretti19_cgm.Vettoretti19CGM`) implement the actual error
    model in :meth:`measure`.

    ...
    Attributes
    ----------
    ts : int
        Sample time of the sensor, in integration steps. The replay loop takes a
        measurement when ``k % ts == 0``.
    t_offset : int
        Extra age of the current sensor, in integration steps (0 when new). Added
        to the time elapsed since ``connected_at`` to give the sensor's age, e.g.
        to account for a gap between two replay runs that share the sensor.
    max_lifetime : int
        Maximum lifetime of the sensor, in integration steps (10 days at ``ts=1``).
        When the sensor's age reaches it, the replay loop reconnects a fresh
        sensor realization.
    connected_at : int
        Integration step, in the current replay's step frame, at which the current
        sensor was connected. ``replay()`` rebases it at the end of a run (it
        subtracts ``tsteps``), so a sensor carried into the next run with
        ``continue_sensor=True`` keeps aging contiguously. It is negative for a
        sensor connected during an earlier run.

    Methods
    -------
    connect_new(connected_at):
        Connects a fresh sensor by (re)sampling its error realization.
    age(k):
        Age of the current sensor at integration step ``k``, in integration steps.
    measure(value, past_values, t):
        Produces a measurement from the current and past model outputs.
    add_offset(to_add):
        Ages the current sensor by ``to_add`` integration steps.
    """

    def __init__(self):
        """Constructs all the necessary attributes for the Sensor object."""
        self.ts = 5
        self.t_offset = 0

        self.connect_new()

        # Set max lifetime (integration steps): 10 days at ts=1
        self.max_lifetime = 1440 * 10
        self.connected_at = 0

    def connect_new(self, connected_at=0):
        """Connects a fresh sensor by (re)sampling its error realization.

        Parameters
        ----------
        connected_at : int, optional, default : 0
            Integration step at which the sensor is connected.
        """
        # Set the offset
        self.t_offset = 0

        # Set the connection time
        self.connected_at = connected_at

    def age(self, k: int) -> int:
        """Age of the current sensor at integration step ``k``.

        Parameters
        ----------
        k : int
            Integration step, in the current replay's step frame.

        Returns
        -------
        int
            The sensor's age in integration steps:
            ``k - connected_at + t_offset``.
        """
        return k - self.connected_at + self.t_offset

    @abstractmethod
    def measure(self, value: float, past_values, t: float) -> float:
        """Produces a measurement from the current and past model outputs.

        Parameters
        ----------
        value : float
            The current model output. This is the value being measured.
        past_values : numpy.ndarray
            The series of past model outputs (``out[:k]``).
        t : float
            The sensor's age, in days (``t_offset`` included).

        Returns
        -------
        float
            The measurement at the current time.
        """
        pass

    def add_offset(self, to_add: int) -> None:
        """Ages the current sensor by ``to_add`` integration steps.

        Only meaningful for a sensor carried into a replay with
        ``continue_sensor=True`` (a fresh replay reconnects the sensor, which
        resets ``t_offset``). Use it, for example, to account for a gap between
        two replay runs that share the sensor. Back-to-back runs need no offset:
        ``replay()`` already rebases the sensor's age at the end of a run.

        Parameters
        ----------
        to_add : int
            The number of integration steps to add to the sensor's age.
        """
        self.t_offset += to_add

ts = 5 instance-attribute

t_offset = 0 instance-attribute

max_lifetime = 1440 * 10 instance-attribute

connected_at = 0 instance-attribute

__init__()

Constructs all the necessary attributes for the Sensor object.

Source code in src/py_replay_bg/sensors/sensor.py
def __init__(self):
    """Constructs all the necessary attributes for the Sensor object."""
    self.ts = 5
    self.t_offset = 0

    self.connect_new()

    # Set max lifetime (integration steps): 10 days at ts=1
    self.max_lifetime = 1440 * 10
    self.connected_at = 0

connect_new(connected_at=0)

Connects a fresh sensor by (re)sampling its error realization.

Parameters:

Name Type Description Default
connected_at int

Integration step at which the sensor is connected.

: 0
Source code in src/py_replay_bg/sensors/sensor.py
def connect_new(self, connected_at=0):
    """Connects a fresh sensor by (re)sampling its error realization.

    Parameters
    ----------
    connected_at : int, optional, default : 0
        Integration step at which the sensor is connected.
    """
    # Set the offset
    self.t_offset = 0

    # Set the connection time
    self.connected_at = connected_at

age(k)

Age of the current sensor at integration step k.

Parameters:

Name Type Description Default
k int

Integration step, in the current replay's step frame.

required

Returns:

Type Description
int

The sensor's age in integration steps: k - connected_at + t_offset.

Source code in src/py_replay_bg/sensors/sensor.py
def age(self, k: int) -> int:
    """Age of the current sensor at integration step ``k``.

    Parameters
    ----------
    k : int
        Integration step, in the current replay's step frame.

    Returns
    -------
    int
        The sensor's age in integration steps:
        ``k - connected_at + t_offset``.
    """
    return k - self.connected_at + self.t_offset

measure(value, past_values, t) abstractmethod

Produces a measurement from the current and past model outputs.

Parameters:

Name Type Description Default
value float

The current model output. This is the value being measured.

required
past_values ndarray

The series of past model outputs (out[:k]).

required
t float

The sensor's age, in days (t_offset included).

required

Returns:

Type Description
float

The measurement at the current time.

Source code in src/py_replay_bg/sensors/sensor.py
@abstractmethod
def measure(self, value: float, past_values, t: float) -> float:
    """Produces a measurement from the current and past model outputs.

    Parameters
    ----------
    value : float
        The current model output. This is the value being measured.
    past_values : numpy.ndarray
        The series of past model outputs (``out[:k]``).
    t : float
        The sensor's age, in days (``t_offset`` included).

    Returns
    -------
    float
        The measurement at the current time.
    """
    pass

add_offset(to_add)

Ages the current sensor by to_add integration steps.

Only meaningful for a sensor carried into a replay with continue_sensor=True (a fresh replay reconnects the sensor, which resets t_offset). Use it, for example, to account for a gap between two replay runs that share the sensor. Back-to-back runs need no offset: replay() already rebases the sensor's age at the end of a run.

Parameters:

Name Type Description Default
to_add int

The number of integration steps to add to the sensor's age.

required
Source code in src/py_replay_bg/sensors/sensor.py
def add_offset(self, to_add: int) -> None:
    """Ages the current sensor by ``to_add`` integration steps.

    Only meaningful for a sensor carried into a replay with
    ``continue_sensor=True`` (a fresh replay reconnects the sensor, which
    resets ``t_offset``). Use it, for example, to account for a gap between
    two replay runs that share the sensor. Back-to-back runs need no offset:
    ``replay()`` already rebases the sensor's age at the end of a run.

    Parameters
    ----------
    to_add : int
        The number of integration steps to add to the sensor's age.
    """
    self.t_offset += to_add

Vettoretti19CGM

Bases: Sensor

A CGM sensor based on the error model in Vettoretti et al., Sensors, 2019.

Concrete :class:~sensors.sensor.Sensor whose measurement is a continuous glucose monitoring (CGM) reading of the model output (interstitial glucose). The error model adds a time-varying calibration error plus colored noise generated by an AR(2) process whose parameters are sampled per sensor connection.

...

Attributes:

Name Type Description
cgm_error_parameters ndarray

Array of the sampled CGM error-model parameters.

output_noise_sd float

Output standard deviation of the CGM error model (mg/dl).

ekm1 float

Memory term for the colored noise, e(k-1).

ekm2 float

Memory term for the colored noise, e(k-2).

Methods:

Name Description
connect_new

Connects a new CGM sensor by sampling new error parameters.

measure

Produces a CGM measurement of the model output.

Source code in src/py_replay_bg/sensors/vettoretti19_cgm.py
class Vettoretti19CGM(Sensor):
    """A CGM sensor based on the error model in Vettoretti et al., Sensors, 2019.

    Concrete :class:`~sensors.sensor.Sensor` whose measurement is a continuous glucose
    monitoring (CGM) reading of the model output (interstitial glucose). The error
    model adds a time-varying calibration error plus colored noise generated by an
    AR(2) process whose parameters are sampled per sensor connection.

    ...
    Attributes
    ----------
    cgm_error_parameters : numpy.ndarray
        Array of the sampled CGM error-model parameters.
    output_noise_sd : float
        Output standard deviation of the CGM error model (mg/dl).
    ekm1 : float
        Memory term for the colored noise, e(k-1).
    ekm2 : float
        Memory term for the colored noise, e(k-2).

    Methods
    -------
    connect_new(connected_at):
        Connects a new CGM sensor by sampling new error parameters.
    measure(value, past_values, t):
        Produces a CGM measurement of the model output.
    """

    def __init__(self):
        """Constructs the CGM sensor. The sample time ``ts`` (5) is inherited."""
        # Declare the CGM-specific attributes before connecting: ``super().__init__``
        # calls ``connect_new``, which populates them with a sampled error realization.
        self.cgm_error_parameters = []
        self.output_noise_sd = []
        self.ekm1 = None
        self.ekm2 = None

        super().__init__()

    def connect_new(self, connected_at=0):
        """Connects a new CGM sensor by sampling new error parameters.

        Parameters
        ----------
        connected_at : int, optional, default : 0
            Integration step at which the sensor is connected.
        """
        super().connect_new(connected_at)

        # Load mean vector and covariance matrix of the parameter vector as
        # defined in Vettoretti et al., Sensors, 2019
        mu = np.array([0.94228767821000314341972625697963, 0.0049398821141803427384187052950892,
                       -0.0005848748565491275084801681138913, 6.382602204050874306062723917421,
                       1.2604070417357611244568715846981, -0.4022228938823663169088717950217,
                       3.2516360856114072674927228945307])
        sigma = np.array([[0.013245827952891258902368143424155, -0.0039513025350735725416129184850433,
                           0.00031276743283791636970891936186945, 0.15717912467153988265167186000326,
                           0.0026876560011614997885986966252858, -0.0028904633825263671524641306831427,
                           -0.0031801707001874032418320403792222],
                          [-0.0039513025350735725416129184850433, 0.0018527975980744701509778105119608,
                           -0.00015580332205794781403294935184789, -0.10288007693621757654423021222101,
                           -0.0013902327543057948350258001823931, 0.0011591852212130876378232136048041,
                           -0.0027284927011686846420879248853453],
                          [0.00031276743283791636970891936186945, -0.00015580332205794781403294935184789,
                           0.000013745962164724157000697882247131, 0.0080685863688738888865881193623864,
                           0.00012074974710011031125631020266553, -0.00010042135441622312822841645019167,
                           0.00011130290033867137325027107941366],
                          [0.15717912467153988265167186000326, -0.10288007693621757654423021222101,
                           0.0080685863688738888865881193623864, 29.005838188852990811028575990349,
                           0.12408344051778112671069465022811, -0.10193644943826736526393261783596,
                           0.60075381294204155402383094042307],
                          [0.0026876560011614997885986966252858, -0.0013902327543057948350258001823931,
                           0.00012074974710011031125631020266553, 0.12408344051778112671069465022811,
                           0.02079352674233487727195601735275, -0.018431109170459980539646949182497,
                           -0.015721846813032722134373386779771],
                          [-0.0028904633825263671524641306831427, 0.0011591852212130876378232136048041,
                           -0.00010042135441622312822841645019167, -0.10193644943826736526393261783596,
                           -0.018431109170459980539646949182497, 0.018700867933453400870913441167431,
                           0.01552333576829629385729347745837],
                          [-0.0031801707001874032418320403792222, -0.0027284927011686846420879248853453,
                           0.00011130290033867137325027107941366, 0.60075381294204155402383094042307,
                           -0.015721846813032722134373386779771, 0.01552333576829629385729347745837,
                           0.72356838038463477946748980684788]])

        # Modulation factor of the covariance of the parameter vector not to generate
        # too extreme realizations of parameter vector
        f = 0.90
        # Modulate the covariance matrix
        sigma = sigma * f
        # Maximum allowed output noise SD (mg/dl)
        max_output_noise_sd = 10
        # Tolerance for model stability check
        toll = 0.02

        # Flag for stability of the AR(2) model
        stable = 0
        output_noise_sd = np.inf
        cgm_error_parameters = []
        while (~ stable) or (output_noise_sd > max_output_noise_sd):
            # Sample CGM error parameters
            cgm_error_parameters = np.random.multivariate_normal(mu, sigma)

            # Check the stability of the resulting AR(2) model
            stable = (cgm_error_parameters[5] >= -1) and (
                    cgm_error_parameters[5] <= (1 - np.abs(cgm_error_parameters[4]) - toll))

            output_noise_var = cgm_error_parameters[6] ** 2 / (
                    1 - (cgm_error_parameters[4] ** 2) / (1 - cgm_error_parameters[5]) - cgm_error_parameters[5] * (
                    (cgm_error_parameters[4] ** 2) / (1 - cgm_error_parameters[5]) + cgm_error_parameters[5]))

            if output_noise_var < 0:
                continue

            # Compute the output noise standard deviation
            output_noise_sd = np.sqrt(output_noise_var)

        # Set the parameters of the CGM
        self.cgm_error_parameters = cgm_error_parameters
        self.output_noise_sd = output_noise_sd

        # Set memory terms
        self.ekm1 = 0
        self.ekm2 = 0

    def measure(self, value, past_values, t):
        """Provides a CGM measure using the model of Vettoretti et al., Sensors, 2019.

        Parameters
        ----------
        value : float
            The current interstitial glucose concentration (mg/dl).
        past_values : numpy.ndarray
            The series of past interstitial glucose concentrations (mg/dl).
        t : float
            The sensor's age, in days (``t_offset`` included).

        Returns
        -------
        float
            The CGM measurement at the current time (mg/dl).
        """
        # Apply calibration error
        ig_s = (self.cgm_error_parameters[0] + self.cgm_error_parameters[1] * t +
                self.cgm_error_parameters[2] * (t ** 2)) * value + self.cgm_error_parameters[3]

        # Generate noise
        z = np.random.normal(0, 1)
        u = self.cgm_error_parameters[6] * z
        e = u + self.cgm_error_parameters[4] * self.ekm1 + self.cgm_error_parameters[5] * self.ekm2

        # Update memory terms
        self.ekm2 = self.ekm1
        self.ekm1 = e

        # Get final CGM
        return ig_s + e

cgm_error_parameters = [] instance-attribute

output_noise_sd = [] instance-attribute

ekm1 = None instance-attribute

ekm2 = None instance-attribute

__init__()

Constructs the CGM sensor. The sample time ts (5) is inherited.

Source code in src/py_replay_bg/sensors/vettoretti19_cgm.py
def __init__(self):
    """Constructs the CGM sensor. The sample time ``ts`` (5) is inherited."""
    # Declare the CGM-specific attributes before connecting: ``super().__init__``
    # calls ``connect_new``, which populates them with a sampled error realization.
    self.cgm_error_parameters = []
    self.output_noise_sd = []
    self.ekm1 = None
    self.ekm2 = None

    super().__init__()

connect_new(connected_at=0)

Connects a new CGM sensor by sampling new error parameters.

Parameters:

Name Type Description Default
connected_at int

Integration step at which the sensor is connected.

: 0
Source code in src/py_replay_bg/sensors/vettoretti19_cgm.py
def connect_new(self, connected_at=0):
    """Connects a new CGM sensor by sampling new error parameters.

    Parameters
    ----------
    connected_at : int, optional, default : 0
        Integration step at which the sensor is connected.
    """
    super().connect_new(connected_at)

    # Load mean vector and covariance matrix of the parameter vector as
    # defined in Vettoretti et al., Sensors, 2019
    mu = np.array([0.94228767821000314341972625697963, 0.0049398821141803427384187052950892,
                   -0.0005848748565491275084801681138913, 6.382602204050874306062723917421,
                   1.2604070417357611244568715846981, -0.4022228938823663169088717950217,
                   3.2516360856114072674927228945307])
    sigma = np.array([[0.013245827952891258902368143424155, -0.0039513025350735725416129184850433,
                       0.00031276743283791636970891936186945, 0.15717912467153988265167186000326,
                       0.0026876560011614997885986966252858, -0.0028904633825263671524641306831427,
                       -0.0031801707001874032418320403792222],
                      [-0.0039513025350735725416129184850433, 0.0018527975980744701509778105119608,
                       -0.00015580332205794781403294935184789, -0.10288007693621757654423021222101,
                       -0.0013902327543057948350258001823931, 0.0011591852212130876378232136048041,
                       -0.0027284927011686846420879248853453],
                      [0.00031276743283791636970891936186945, -0.00015580332205794781403294935184789,
                       0.000013745962164724157000697882247131, 0.0080685863688738888865881193623864,
                       0.00012074974710011031125631020266553, -0.00010042135441622312822841645019167,
                       0.00011130290033867137325027107941366],
                      [0.15717912467153988265167186000326, -0.10288007693621757654423021222101,
                       0.0080685863688738888865881193623864, 29.005838188852990811028575990349,
                       0.12408344051778112671069465022811, -0.10193644943826736526393261783596,
                       0.60075381294204155402383094042307],
                      [0.0026876560011614997885986966252858, -0.0013902327543057948350258001823931,
                       0.00012074974710011031125631020266553, 0.12408344051778112671069465022811,
                       0.02079352674233487727195601735275, -0.018431109170459980539646949182497,
                       -0.015721846813032722134373386779771],
                      [-0.0028904633825263671524641306831427, 0.0011591852212130876378232136048041,
                       -0.00010042135441622312822841645019167, -0.10193644943826736526393261783596,
                       -0.018431109170459980539646949182497, 0.018700867933453400870913441167431,
                       0.01552333576829629385729347745837],
                      [-0.0031801707001874032418320403792222, -0.0027284927011686846420879248853453,
                       0.00011130290033867137325027107941366, 0.60075381294204155402383094042307,
                       -0.015721846813032722134373386779771, 0.01552333576829629385729347745837,
                       0.72356838038463477946748980684788]])

    # Modulation factor of the covariance of the parameter vector not to generate
    # too extreme realizations of parameter vector
    f = 0.90
    # Modulate the covariance matrix
    sigma = sigma * f
    # Maximum allowed output noise SD (mg/dl)
    max_output_noise_sd = 10
    # Tolerance for model stability check
    toll = 0.02

    # Flag for stability of the AR(2) model
    stable = 0
    output_noise_sd = np.inf
    cgm_error_parameters = []
    while (~ stable) or (output_noise_sd > max_output_noise_sd):
        # Sample CGM error parameters
        cgm_error_parameters = np.random.multivariate_normal(mu, sigma)

        # Check the stability of the resulting AR(2) model
        stable = (cgm_error_parameters[5] >= -1) and (
                cgm_error_parameters[5] <= (1 - np.abs(cgm_error_parameters[4]) - toll))

        output_noise_var = cgm_error_parameters[6] ** 2 / (
                1 - (cgm_error_parameters[4] ** 2) / (1 - cgm_error_parameters[5]) - cgm_error_parameters[5] * (
                (cgm_error_parameters[4] ** 2) / (1 - cgm_error_parameters[5]) + cgm_error_parameters[5]))

        if output_noise_var < 0:
            continue

        # Compute the output noise standard deviation
        output_noise_sd = np.sqrt(output_noise_var)

    # Set the parameters of the CGM
    self.cgm_error_parameters = cgm_error_parameters
    self.output_noise_sd = output_noise_sd

    # Set memory terms
    self.ekm1 = 0
    self.ekm2 = 0

measure(value, past_values, t)

Provides a CGM measure using the model of Vettoretti et al., Sensors, 2019.

Parameters:

Name Type Description Default
value float

The current interstitial glucose concentration (mg/dl).

required
past_values ndarray

The series of past interstitial glucose concentrations (mg/dl).

required
t float

The sensor's age, in days (t_offset included).

required

Returns:

Type Description
float

The CGM measurement at the current time (mg/dl).

Source code in src/py_replay_bg/sensors/vettoretti19_cgm.py
def measure(self, value, past_values, t):
    """Provides a CGM measure using the model of Vettoretti et al., Sensors, 2019.

    Parameters
    ----------
    value : float
        The current interstitial glucose concentration (mg/dl).
    past_values : numpy.ndarray
        The series of past interstitial glucose concentrations (mg/dl).
    t : float
        The sensor's age, in days (``t_offset`` included).

    Returns
    -------
    float
        The CGM measurement at the current time (mg/dl).
    """
    # Apply calibration error
    ig_s = (self.cgm_error_parameters[0] + self.cgm_error_parameters[1] * t +
            self.cgm_error_parameters[2] * (t ** 2)) * value + self.cgm_error_parameters[3]

    # Generate noise
    z = np.random.normal(0, 1)
    u = self.cgm_error_parameters[6] * z
    e = u + self.cgm_error_parameters[4] * self.ekm1 + self.cgm_error_parameters[5] * self.ekm2

    # Update memory terms
    self.ekm2 = self.ekm1
    self.ekm1 = e

    # Get final CGM
    return ig_s + e