Skip to content

API Reference — AGATA

The AGATA integration layer wraps the AGATA reference library.

Note: AGATA is an optional dependency. Install it with pip install cgmpy[agata].

Wrapper

cgmpy.agata.metrics.AgataAnalysis

Bases: GlucoseData

Object-oriented bridge to the py_agata analysis pipeline.

Inherits from :class:~cgmpy.data.core.GlucoseData so it can be constructed in exactly the same way (file path, DataFrame, etc.), and exposes :meth:run / :meth:analyze_one_arm to drive the py_agata analysis.

Raises:

Type Description
AgataNotInstalledError

If py_agata is not installed.
EmptyDataError

If :meth:run (or :meth:analyze_one_arm) is called on empty data, raised by the adapter during data preparation.
Source code in cgmpy/agata/metrics.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
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
182
183
184
185
186
187
188
189
190
191
192
193
class AgataAnalysis(GlucoseData):
    """Object-oriented bridge to the ``py_agata`` analysis pipeline.

    Inherits from :class:`~cgmpy.data.core.GlucoseData` so it can
    be constructed in exactly the same way (file path, DataFrame, etc.),
    and exposes :meth:`run` / :meth:`analyze_one_arm` to drive the
    py_agata analysis.

    Raises:
        AgataNotInstalledError: If ``py_agata`` is not installed.
        EmptyDataError: If :meth:`run` (or :meth:`analyze_one_arm`) is
            called on empty data, raised by the adapter during data
            preparation.
    """

    def __init__(self, *args, glycemic_target: str = "diabetes", **kwargs: Any):
        """
        Initializes the Agata analysis.

        Args:
            *args: Positional arguments for GlucoseData.
            glycemic_target (str): The glycemic target ('diabetes' or 'pregnancy').
            **kwargs: Other arguments for GlucoseData.
        """
        super().__init__(*args, **kwargs)
        self.glycemic_target = glycemic_target

    def run(self, glycemic_target: str = "diabetes", summary: bool = False, **kwargs: Any) -> dict:
        """
        Runs the full py_agata analysis pipeline on the loaded data.

        Args:
            glycemic_target (str): The glycemic target to use ('diabetes', 'pregnancy', etc.).
                                   If None, uses the value defined at class initialization.
            summary (bool): If True, returns a flat dictionary with summary metrics.
                            If False, returns the complete nested dictionary.
            **kwargs: Additional arguments for future use.

        Returns:
            dict: The results dictionary.

        Raises:
            EmptyDataError: If the underlying data is empty.
            AgataNotInstalledError: If ``py_agata`` is not installed.
        """
        # Use the defined target if no specific one is passed in the call
        target = glycemic_target or self.glycemic_target

        # 1. Get raw results
        results = analyze_with_agata(self, glycemic_target=target, **kwargs)

        # 2. Summarize if requested
        if summary:
            return summarize_agata_results(results)

        return results

    @classmethod
    def analyze_one_arm(
        cls,
        data_list: list[GlucoseData],
        glycemic_target: str = "diabetes",
        summary: bool = False,
        **kwargs,
    ) -> dict:
        """
        Analyzes a group (arm) of GlucoseData objects.

        Args:
            data_list (list[GlucoseData]): List of objects to analyze.
            glycemic_target (str): The glycemic target to use.
            summary (bool): If True, returns a flat summary.

        Raises:
            EmptyDataError: If any element of ``data_list`` is empty.
            AgataNotInstalledError: If ``py_agata`` is not installed.
        """
        results = analyze_one_arm(data_list, glycemic_target=glycemic_target, **kwargs)

        if summary:
            # Note: py_agata.analyze_one_arm usually returns a similar format
            return summarize_agata_results(results)

        return results

run 

run(glycemic_target: str = 'diabetes', summary: bool = False, **kwargs: Any) -> dict

Runs the full py_agata analysis pipeline on the loaded data.

Parameters:

Name Type Description Default
glycemic_target str

The glycemic target to use ('diabetes', 'pregnancy', etc.). If None, uses the value defined at class initialization.

 'diabetes'
summary bool

If True, returns a flat dictionary with summary metrics. If False, returns the complete nested dictionary.

 False
**kwargs Any

Additional arguments for future use.

 {}

Returns:

Name Type Description
dict dict

The results dictionary.

Raises:

Type Description
EmptyDataError

If the underlying data is empty.
AgataNotInstalledError

If py_agata is not installed.
Source code in cgmpy/agata/metrics.py 
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
def run(self, glycemic_target: str = "diabetes", summary: bool = False, **kwargs: Any) -> dict:
    """
    Runs the full py_agata analysis pipeline on the loaded data.

    Args:
        glycemic_target (str): The glycemic target to use ('diabetes', 'pregnancy', etc.).
                               If None, uses the value defined at class initialization.
        summary (bool): If True, returns a flat dictionary with summary metrics.
                        If False, returns the complete nested dictionary.
        **kwargs: Additional arguments for future use.

    Returns:
        dict: The results dictionary.

    Raises:
        EmptyDataError: If the underlying data is empty.
        AgataNotInstalledError: If ``py_agata`` is not installed.
    """
    # Use the defined target if no specific one is passed in the call
    target = glycemic_target or self.glycemic_target

    # 1. Get raw results
    results = analyze_with_agata(self, glycemic_target=target, **kwargs)

    # 2. Summarize if requested
    if summary:
        return summarize_agata_results(results)

    return results

Adapter

cgmpy.agata.adapter

Adapter to prepare cgmpy data for the py_agata library.

prepare_data_for_agata 

prepare_data_for_agata(glucose_data: GlucoseData, resample_freq: str = '5min') -> DataFrame

Prepare a :class:~cgmpy.data.core.GlucoseData for py_agata.

The function aligns the timestamp grid by flooring every measurement to resample_freq (default "5min"), then re-emits the data on a homogeneous, gap-aware time grid that py_agata can consume.

Parameters:

Name Type Description Default
glucose_data GlucoseData

The cgmpy data object to adapt.

 required
resample_freq str

Pandas frequency string for the target grid (e.g. "5min", "15min"). Must be a pandas-recognised frequency.

 '5min'

Returns:

Type Description
DataFrame

pd.DataFrame: A two-column DataFrame with columns:

  • t (datetime64[ns]) -- the regular time index.
  • glucose (float, mg/dL) -- the glucose values, with NaN for timestamps that had no original measurement.

Raises:

Type Description
EmptyDataError

If the input is empty, if the timestamp normalization drops every row, if the cleaned data covers a zero-duration interval, or if every glucose value on the regular grid is NaN.
Notes

The function does not convert mmol/L to mg/dL; the input is assumed to already be in mg/dL. Unit conversion is a deferred feature targeted at v0.9.0 (see ROADMAP.md).

Source code in cgmpy/agata/adapter.py 
 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
def prepare_data_for_agata(glucose_data: GlucoseData, resample_freq: str = "5min") -> pd.DataFrame:
    """Prepare a :class:`~cgmpy.data.core.GlucoseData` for py_agata.

    The function aligns the timestamp grid by flooring every measurement to
    ``resample_freq`` (default ``"5min"``), then re-emits the data on a
    homogeneous, gap-aware time grid that py_agata can consume.

    Args:
        glucose_data: The cgmpy data object to adapt.
        resample_freq: Pandas frequency string for the target grid
            (e.g. ``"5min"``, ``"15min"``). Must be a pandas-recognised
            frequency.

    Returns:
        pd.DataFrame: A two-column DataFrame with columns:

            * ``t`` (datetime64[ns]) -- the regular time index.
            * ``glucose`` (float, mg/dL) -- the glucose values, with NaN for
              timestamps that had no original measurement.

    Raises:
        EmptyDataError: If the input is empty, if the timestamp
            normalization drops every row, if the cleaned data covers a
            zero-duration interval, or if every glucose value on the
            regular grid is NaN.

    Notes:
        The function does **not** convert mmol/L to mg/dL; the input is
        assumed to already be in mg/dL. Unit conversion is a deferred
        feature targeted at v0.9.0 (see ``ROADMAP.md``).
    """
    df = glucose_data.data.copy()
    date_col = glucose_data.date_col
    glucose_col = glucose_data.glucose_col

    # Guard 1: nothing to do if the input has zero rows.
    if len(df) == 0:
        raise EmptyDataError(context="input GlucoseData")

    # Ensure the date column is of datetime type
    df[date_col] = pd.to_datetime(df[date_col])

    # Sort the data chronologically first
    df_clean = df.sort_values(date_col)

    # -- KEY STEP: STANDARDIZE THE TIME COLUMN --
    # Round each time down to the nearest 5-minute interval.
    df_clean[date_col] = df_clean[date_col].dt.floor(resample_freq)

    # Now, remove duplicates. If 00:01 and 00:03 rounded to 00:00,
    # we keep the first one that appeared.
    df_clean = df_clean.drop_duplicates(subset=[date_col], keep="first")

    # Guard 2: the timestamp normalization step may have dropped every row
    # (e.g. all timestamps were NaT and floor() left them as NaT, or
    # duplicate removal wiped the dataset).
    if df_clean.empty:
        raise EmptyDataError(context="after timestamp normalization")

    # The rest of the process already works perfectly
    start_time = df_clean[date_col].min()
    end_time = df_clean[date_col].max()

    # Guard 3: a zero-duration interval (start == end is allowed and yields
    # a single-point grid, but a NaT boundary or otherwise invalid range
    # must be surfaced as EmptyDataError, not a pandas ValueError).
    if pd.isna(start_time) or pd.isna(end_time):
        raise EmptyDataError(context="the cleaned data covers a zero-duration interval")
    time_range = pd.date_range(start=start_time, end=end_time, freq=resample_freq)
    if len(time_range) == 0:
        raise EmptyDataError(context="the cleaned data covers a zero-duration interval")

    df_homogeneous = pd.DataFrame({date_col: time_range})

    # Merge with the already standardized data
    df_final = df_homogeneous.merge(df_clean, on=date_col, how="left")

    # Rename columns
    df_final = df_final.rename(columns={date_col: "t", glucose_col: "glucose"})

    # Guard 4: py_agata cannot work with a fully empty glucose series.
    # Reaching this point with all-NaN glucose means every original sample
    # was dropped by the merge (e.g. they fell outside the resample grid,
    # or the original glucose column was empty after processing).
    if df_final["glucose"].isna().all():
        raise EmptyDataError(
            context="all glucose values became NaN after merging to the regular grid"
        )

    return df_final[["t", "glucose"]]