Merge pull request #45 from lcossu/snack_overrides

Manually override absorption rates and delay of snacks and hypo treatments

Author: gcappon

docs/documentation/replaying.md

@@ -50,6 +50,9 @@ rbg.replay(data: pd.DataFrame,
    n_replay: int = 1000,
    sensors: list | None = None,
    sensor_cgm: CGM = Vettoretti19CGM,
+   snack_absorption=snack_absorption,
+   snack_absorption_delay=snack_absorption_delay,
+   hypotreatment_absorption=hypotreatment_absorption,
 ) -> Dict:
 ```
 ### Input parameters
@@ -119,6 +122,9 @@ replay simulations. Ignored if twinning_method is 'map'.
 - `sensors`: , optional, default: `None`: A `list[Sensors]` to be used in each of the replay simulations. Its length 
 must coincide with the selected `n_replay`. Used when working with intervals. If `None` new sensors will be used.
 - `sensor_cgm`, optional, default: `Vettoretti19CGM`: The class of the CGM error model to be used during the replay simulation.
+- `snack_absorption`, optional, default: `None`: A value to override the identified snack absorption rate.
+- `snack_absorption_delay`, optional, default: `None`: A value to override the identified snack absorption delay (between 0 and 60 minutes)
+- `hypotreatment_absorption`, optional, default: `None`: A value to override the identified hypotreatment absorption rate.
 
 ::: tip REMEMBER
 The total length of the simulation, `simulation_length`, is defined in minutes and determined by ReplayBG automatically 

py_replay_bg/example/code/replay_map_custom_snack.py

@@ -0,0 +1,47 @@
+import os
+import numpy as np
+
+from utils import load_test_data, load_patient_info
+
+from py_replay_bg.py_replay_bg import ReplayBG
+from py_replay_bg.visualizer import Visualizer
+from py_replay_bg.analyzer import Analyzer
+
+# Set verbosity
+verbose = True
+plot_mode = False
+
+# Set other parameters for twinning
+blueprint = 'multi-meal'
+save_folder = os.path.join(os.path.abspath(''), '..', '..', '..')
+
+# load patient_info
+patient_info = load_patient_info()
+p = np.where(patient_info['patient'] == 1)[0][0]
+# Set bw and u2ss
+bw = float(patient_info.bw.values[p])
+
+# Instantiate ReplayBG
+rbg = ReplayBG(blueprint=blueprint, save_folder=save_folder,
+                         yts=5, exercise=False,
+                         seed=1,
+                         verbose=verbose, plot_mode=plot_mode)
+
+# Load data and set save_name
+data = load_test_data(day=1)
+save_name = 'data_day_' + str(1)
+
+print("Replaying " + save_name)
+
+# Replay the twin with the same input data used for twinning
+replay_results = rbg.replay(data=data, bw=bw, save_name=save_name,
+                            twinning_method='map',
+                            save_workspace=True,
+                            save_suffix='_replay_map', snack_absorption=.5, snack_absorption_delay=60)
+
+# Visualize and analyze results
+Visualizer.plot_replay_results(replay_results, data=data)
+analysis = Analyzer.analyze_replay_results(replay_results, data=data)
+print('Mean glucose: %.2f mg/dl' % analysis['median']['glucose']['variability']['mean_glucose'])
+print('TIR: %.2f %%' % analysis['median']['glucose']['time_in_ranges']['time_in_target'])
+print('N Days: %.2f days' % analysis['median']['glucose']['data_quality']['number_days_of_observation'])

py_replay_bg/input_validation/__init__.py

@@ -609,3 +609,51 @@ def __init__(self, yts):
     def validate(self):
         if not isinstance(self.yts, int):
             raise Exception("'yts' input must be an integer.'")
+
+
+class SnackAbsorptionValidator:
+    """
+    Validates the 'snack_absorption' parameter.
+    """
+
+    def __init__(self, snack_absorption):
+        self.snack_absorption = snack_absorption
+
+    def validate(self):
+        if self.snack_absorption is not None:
+            if not isinstance(self.snack_absorption, float):
+                raise Exception("'snack_absorption' input must be a float.'")
+            if not (0 <= self.snack_absorption <= 1):
+                raise Exception("'snack_absorption' input must be between 0 and 1.'")
+
+
+class SnackAbsorptionDelayValidator:
+    """
+    Validates the 'snack_absorption_delay' parameter.
+    """
+
+    def __init__(self, snack_absorption_delay):
+        self.snack_absorption_delay = snack_absorption_delay
+
+    def validate(self):
+        if self.snack_absorption_delay is not None:
+            if not isinstance(self.snack_absorption_delay, int):
+                raise Exception("'snack_absorption_delay' input must be a integer.'")
+            if not (0 <= self.snack_absorption_delay <= 60):
+                raise Exception("'snack_absorption_delay' input must be between 0 and 60.'")
+
+
+class HypotreatmentAbsorptionValidator:
+    """
+    Validates the 'hypotreatment_absorption' parameter.
+    """
+
+    def __init__(self, hypotreatment_absorption):
+        self.hypotreatment_absorption = hypotreatment_absorption
+
+    def validate(self):
+        if self.hypotreatment_absorption is not None:
+            if not isinstance(self.hypotreatment_absorption, float):
+                raise Exception("'hypotreatment_absorption' input must be a float.'")
+            if not (0 <= self.hypotreatment_absorption <= 1):
+                raise Exception("'hypotreatment_absorption' input must be between 0 and 1.'")

py_replay_bg/input_validation/input_validator_replay.py

@@ -119,6 +119,9 @@ def __init__(self,
                  sensors: list,
                  blueprint: str,
                  exercise: bool,
+                 snack_absorption: float,
+                 snack_absorption_delay: int,
+                 hypotreatment_absorption: float,
                  ):
         self.data = data
         self.bw = bw
@@ -147,7 +150,9 @@ def __init__(self,
         self.sensors = sensors
         self.blueprint = blueprint
         self.exercise = exercise
-
+        self.snack_absorption = snack_absorption
+        self.snack_absorption_delay = snack_absorption_delay
+        self.hypotreatment_absorption = hypotreatment_absorption
 
     def validate(self):
         """
@@ -156,7 +161,8 @@ def validate(self):
 
         # Validate the 'data' input
         DataValidator(modality='replay', data=self.data, blueprint=self.blueprint, exercise=self.exercise,
-                      bolus_source=self.bolus_source, basal_source=self.basal_source, cho_source=self.cho_source).validate()
+                      bolus_source=self.bolus_source, basal_source=self.basal_source,
+                      cho_source=self.cho_source).validate()
 
         # Validate the 'bw' input
         BWValidator(bw=self.bw).validate()
@@ -192,7 +198,8 @@ def validate(self):
         BolusCalculatorHandlerValidator(bolus_calculator_handler=self.bolus_calculator_handler).validate()
 
         # Validate the 'bolus_calculator_handler_params' input
-        BolusCalculatorHandlerParamsValidator(bolus_calculator_handler_params=self.bolus_calculator_handler_params).validate()
+        BolusCalculatorHandlerParamsValidator(
+            bolus_calculator_handler_params=self.bolus_calculator_handler_params).validate()
 
         # Validate the 'basal_handler' input
         BasalHandlerValidator(basal_handler=self.basal_handler).validate()
@@ -207,7 +214,8 @@ def validate(self):
         HypotreatmentsHandlerValidator(hypotreatments_handler=self.hypotreatments_handler).validate()
 
         # Validate the 'hypotreatments_handler_params' input
-        HypotreatmentsHandlerParamsValidator(hypotreatments_handler_params=self.hypotreatments_handler_params).validate()
+        HypotreatmentsHandlerParamsValidator(
+            hypotreatments_handler_params=self.hypotreatments_handler_params).validate()
 
         # Validate the 'enable_correction_boluses' input
         EnableCorrectionBolusesValidator(enable_correction_boluses=self.enable_correction_boluses).validate()
@@ -216,7 +224,8 @@ def validate(self):
         CorrectionBolusesHandlerValidator(correction_boluses_handler=self.correction_boluses_handler).validate()
 
         # Validate the 'correction_boluses_handler_params' input
-        CorrectionBolusesHandlerParamsValidator(correction_boluses_handler_params=self.correction_boluses_handler_params).validate()
+        CorrectionBolusesHandlerParamsValidator(
+            correction_boluses_handler_params=self.correction_boluses_handler_params).validate()
 
         # Validate the 'save_suffix' input
         SaveSuffixValidator(save_suffix=self.save_suffix).validate()
@@ -229,3 +238,12 @@ def validate(self):
 
         # Validate the 'sensors' input
         SensorsValidator(sensors=self.sensors).validate()
+
+        # Validate the 'snack_absorption' input
+        SnackAbsorptionValidator(snack_absorption=self.snack_absorption).validate()
+
+        # Validate the 'snack_absorption_delay' input
+        SnackAbsorptionDelayValidator(snack_absorption_delay=self.snack_absorption_delay).validate()
+
+        # Validate the 'hypotreatment_absorption' input
+        HypotreatmentAbsorptionValidator(hypotreatment_absorption=self.hypotreatment_absorption).validate()

py_replay_bg/py_replay_bg.py

@@ -300,6 +300,9 @@ def replay(self,
                n_replay: int = 1000,
                sensors: list | None = None,
                sensor_cgm: CGM = Vettoretti19CGM,
+               snack_absorption: float = None,
+               snack_absorption_delay: int = None,
+               hypotreatment_absorption: float = None,
                ) -> Dict:
         """
         Runs ReplayBG according to the chosen modality.
@@ -375,6 +378,13 @@ def replay(self,
         sensor_cgm: CGM, optional, default: Vettoretti19CGM
             The class representing the sensors to be used in each of the replay simulations.
 
+        snack_absorption: float, optional, default: None
+            The absorption rate to be used for snacks (kabs_S in the model). If None, the value identified during twinning is used.
+        snack_absorption_delay: int, optional, default: None
+            The absorption delay in minutes to be used for snacks (beta_S in the model). If None, the value identified during twinning is used.
+        hypotreatment_absorption: float, optional, default: None
+            The absorption rate to be used for hypotreatments (kabs_H in the model). If None, the value identified during twinning is used.
+
         Returns
         -------
         replay_results: dict
@@ -445,6 +455,9 @@ def replay(self,
             sensors=sensors,
             exercise=self.environment.exercise,
             blueprint=self.environment.blueprint,
+            snack_absorption=snack_absorption,
+            snack_absorption_delay=snack_absorption_delay,
+            hypotreatment_absorption=hypotreatment_absorption,
         ).validate()
 
         if self.environment.verbose:
@@ -460,6 +473,13 @@ def replay(self,
         draws = twinning_results['draws']
         u2ss = twinning_results['u2ss']
 
+        if snack_absorption is not None:
+            draws['kabs_S'] = snack_absorption
+        if snack_absorption_delay is not None:
+            draws['beta_S'] = snack_absorption_delay
+        if hypotreatment_absorption is not None:
+            draws['kabs_H'] = hypotreatment_absorption
+
         if self.environment.blueprint == 'single-meal':
             model = T1DModelSingleMeal(data=data, bw=bw, u2ss=u2ss, x0=x0,
                                        previous_data_name=previous_data_name,