API Reference — Data

The data layer handles loading, processing, analysis, and exporting of CGM time series.

High-level

cgmpy.data.GlucoseData

Refactored main class for modular glucose data management.

This class integrates all specialized modules: - DataLoader: Data loading from different sources - DataProcessor: Data processing and validation - DataAnalyzer: Basic data analysis - DataExporter: Data export to different formats

Source code in cgmpy/data/core.py

class GlucoseData:
    """
    Refactored main class for modular glucose data management.

    This class integrates all specialized modules:
    - DataLoader: Data loading from different sources
    - DataProcessor: Data processing and validation
    - DataAnalyzer: Basic data analysis
    - DataExporter: Data export to different formats
    """

    @classmethod
    def from_sources(
        cls,
        sources: list[str | pd.DataFrame | GlucoseData],
        date_col: str = "time",
        glucose_col: str = "glucose",
        delimiter: str | None = None,
        header: int = 0,
        log: bool = False,
        target_type: str = "diabetes",
        unit: str | GlucoseUnit = GlucoseUnit.MG_DL,
        **kwargs,
    ) -> GlucoseData:
        """
        Creates an instance by merging multiple data sources.

        This method handles loading, column renaming, and initial merging.
        The constructor will then handle duplicate removal and sorting.

        Args:
            sources: List of sources (paths, DataFrames, or GlucoseData instances)
            date_col: Default date column name for new sources
            glucose_col: Default glucose column name for new sources
            delimiter: Delimiter for CSV files
            header: Header row for CSV files
            log: Whether to enable logging
            target_type: Type of glucose targets ('diabetes' or 'pregnancy')
            unit: Glucose unit of the data ('mg/dL' or 'mmol/L')
            kwargs: Additional arguments for the constructor

        Returns:
            A new GlucoseData instance containing all merged data
        """
        temp_loader = DataLoader()
        temp_processor = DataProcessor()
        all_dfs = []

        for source in sources:
            if isinstance(source, GlucoseData):
                # If it's another instance, we take its already processed data
                df = source.get_raw_data()
                # These are already named 'time' and 'glucose'
                all_dfs.append(df)

            elif isinstance(source, str | pd.DataFrame):
                # Load the source (if string) or take the DataFrame
                raw_df = temp_loader.load_from_source(
                    source, date_col, glucose_col, delimiter, header
                )
                # Standardize column names before concatenation
                standard_df = temp_processor.rename_columns(raw_df, date_col, glucose_col)
                all_dfs.append(standard_df)

            else:
                raise TypeError(f"Unsupported source type: {type(source)}")

        if not all_dfs:
            raise ValueError("No valid sources provided.")

        # Concatenate all standardized DataFrames
        combined_df = pd.concat(all_dfs, ignore_index=True)

        # Create the final instance.
        # Note: We pass date_col="time" and glucose_col="glucose" because
        # we've already standardized them above.
        return cls(
            combined_df,
            date_col="time",
            glucose_col="glucose",
            log=log,
            target_type=target_type,
            unit=unit,
            **kwargs,
        )

    def __init__(
        self,
        data_source: str | pd.DataFrame,
        date_col: str = "time",
        glucose_col: str = "glucose",
        delimiter: str | None = None,
        header: int = 0,
        start_date: str | datetime.datetime | None = None,
        end_date: str | datetime.datetime | None = None,
        log: bool = False,
        target_type: str = "diabetes",
        unit: str | GlucoseUnit = GlucoseUnit.MG_DL,
        regularize: bool = False,
        regularize_interval: int = 5,
        regularize_max_gap: int = 30,
    ):
        """
        Initializes glucose data with modular architecture.

        Args:
            data_source: CSV/Parquet file or DataFrame with data
            date_col: Name of the date/time column
            glucose_col: Name of the glucose values column
            delimiter: Delimiter for CSV files
            header: Header row for CSV files
            start_date: Start date for filtering data (optional)
            end_date: End date for filtering data (optional)
            log: If True, enables detailed performance logs
            target_type: Type of glucose targets ('diabetes' or 'pregnancy')
            unit: Glucose unit of the data ('mg/dL' or 'mmol/L')
        """
        # Logging configuration
        self.log = log
        self.logger = self._setup_logger()

        # Store parameters
        self.date_col = date_col
        self.glucose_col = glucose_col
        self.target_type = target_type
        # CGMPy computes everything in mg/dL. Remember the unit the data was
        # provided in, but the stored values are normalized to mg/dL (below)
        # so every metric is correct regardless of the input unit.
        if isinstance(unit, str):
            self._source_unit = GlucoseUnit(unit)
        elif isinstance(unit, GlucoseUnit):
            self._source_unit = unit
        else:
            self._source_unit = GlucoseUnit.MG_DL
        self._unit = GlucoseUnit.MG_DL

        # Initialize targets
        self.targets: GlucoseTargets
        if isinstance(target_type, GlucoseTargets):
            self.targets = target_type
        else:
            self.targets = get_targets(target_type)

        # Initialize modules
        self.loader = DataLoader(self.logger)
        self.processor = DataProcessor(self.logger)
        self.analyzer = DataAnalyzer(self.logger)
        self.exporter = DataExporter(self.logger)

        # Process data
        self.data, self.time_diffs, self.typical_interval = self._initialize_data(
            data_source,
            date_col,
            glucose_col,
            delimiter,
            header,
            start_date,
            end_date,
            regularize,
            regularize_interval,
            regularize_max_gap,
        )

        # Normalize glucose to mg/dL (the canonical internal unit) so all
        # downstream metrics and plots are unit-safe.
        if self._source_unit != GlucoseUnit.MG_DL:
            self.data["glucose"] = to_mg_per_dl(self.data["glucose"])
            if self.log:
                self.logger.info("Converted glucose from %s to mg/dL.", self._source_unit.value)

        # Dictionary to store operation logs
        self.logs: dict[str, Any] = {}

    @property
    def glucose(self) -> pd.Series:
        """Glucose values as a pandas Series.

        Convenience accessor for ``self.data["glucose"]``.

        Returns:
            pd.Series: Glucose values in mg/dL.
        """
        return self.data["glucose"]

    @property
    def timestamps(self) -> pd.Series:
        """Timestamps as a pandas Series.

        Convenience accessor for ``self.data["time"]``.

        Returns:
            pd.Series: Timestamps of glucose readings.
        """
        return self.data["time"]

    @property
    def unit(self) -> GlucoseUnit:
        """Internal storage unit of the glucose values (always mg/dL).

        Data is normalized to mg/dL on load so every metric is unit-safe.
        Use :meth:`glucose_in_unit` to obtain values in another unit, and
        :attr:`source_unit` to see the unit the data was provided in.

        Returns:
            GlucoseUnit: Always ``GlucoseUnit.MG_DL``.
        """
        return self._unit

    @property
    def source_unit(self) -> GlucoseUnit:
        """The unit the data was originally provided in.

        Returns:
            GlucoseUnit: The input unit (``mg/dL`` unless ``mmol/L`` was given).
        """
        return self._source_unit

    def glucose_in_unit(self, unit: GlucoseUnit | str = GlucoseUnit.MG_DL) -> pd.Series:
        """Return glucose values converted to the specified unit.

        Internally the data is stored in mg/dL; this converts that canonical
        series to the requested unit.

        Args:
            unit: Target unit.

        Returns:
            Glucose values in the requested unit.
        """
        target = GlucoseUnit(unit) if isinstance(unit, str) else unit
        if target == self._unit:
            return self.glucose
        return convert(self.glucose, self._unit, target)

    def _setup_logger(self) -> logging.Logger:
        """
        Configures the logger for the class.

        Returns:
            Configured logger
        """
        logger = logging.getLogger(__name__)

        if self.log and not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter("%(name)s - %(levelname)s - %(message)s")
            handler.setFormatter(formatter)
            logger.addHandler(handler)
            logger.setLevel(logging.DEBUG)

        return logger

    def _initialize_data(
        self,
        data_source: str | pd.DataFrame,
        date_col: str,
        glucose_col: str,
        delimiter: str | None,
        header: int,
        start_date: str | datetime.datetime | None,
        end_date: str | datetime.datetime | None,
        regularize: bool = False,
        regularize_interval: int = 5,
        regularize_max_gap: int = 30,
    ) -> tuple[pd.DataFrame, pd.Series, float]:
        """
        Initializes and processes data using modules.

        Returns:
            Tuple with (data, time_diffs, typical_interval)
        """
        t_start = time.time()

        # Step 1: Load data
        t0 = time.time()
        raw_data = self.loader.load_from_source(
            data_source, date_col, glucose_col, delimiter, header
        )
        t1 = time.time()

        # Step 2: Rename columns
        t2 = time.time()
        data = self.processor.rename_columns(raw_data, date_col, glucose_col)
        # After renaming, the columns are always "time" and "glucose".
        # Update the instance attributes so downstream consumers (e.g.
        # the AGATA adapter) reference the correct column names.
        self.date_col = "time"
        self.glucose_col = "glucose"
        t3 = time.time()

        # Step 3: Filter by dates
        t4 = time.time()
        if start_date is not None or end_date is not None:
            data = self.processor.filter_by_dates(data, start_date, end_date)
        t5 = time.time()

        # Step 4: Process data
        t6 = time.time()
        processed_data, time_diffs = self.processor.process_data(data, "time", "glucose", self.log)
        t7 = time.time()

        # Step 5: Calculate typical interval
        t8 = time.time()
        typical_interval = self.analyzer.calculate_typical_interval(time_diffs, self.log)
        t9 = time.time()

        # Step 6: Optional Regularization
        if regularize:
            t10 = time.time()
            processed_data = self.processor.regularize(
                processed_data,
                interval_mins=regularize_interval,
                max_gap_mins=regularize_max_gap,
                date_col="time",
                glucose_col="glucose",
            )
            time_diffs = processed_data["time"].diff()
            typical_interval = self.analyzer.calculate_typical_interval(time_diffs, self.log)
            t11 = time.time()
            if self.log:
                self.logger.debug(f"Regularization applied: {t11 - t10:.3f}s")

        t_end = time.time()

        # Performance logging
        if self.log:
            self.logger.debug(f"""
            Modular initialization timing:
            Data loading: {t1 - t0:.3f}s
            Column renaming: {t3 - t2:.3f}s
            Date filtering: {t5 - t4:.3f}s
            Data processing: {t7 - t6:.3f}s
            Typical interval calculation: {t9 - t8:.3f}s
            Total time: {t_end - t_start:.3f}s
            """)

        if self.log:
            self.logger.info(f"Data source loaded and processed in {t_end - t_start:.3f} seconds.")

        return processed_data, time_diffs, typical_interval

    def get_typical_interval(self) -> float:
        """
        Returns the typical interval between measurements in minutes.

        Returns:
            Typical interval in minutes
        """
        return self.typical_interval

    def info(self, include_disconnections: bool = False) -> dict[str, Any]:
        """
        Shows basic file information.

        Args:
            include_disconnections: Whether to include disconnection details

        Returns:
            Dictionary with basic information
        """
        return self.analyzer.get_basic_info(
            self.data, self.time_diffs, self.typical_interval, include_disconnections
        )

    def __str__(self) -> str:
        """
        String representation of the object with basic information.

        Returns:
            String with information summary
        """
        info = self.info()
        return self.analyzer.get_summary_string(info)

    def get_data_quality_metrics(self) -> dict[str, Any]:
        """
        Calculates data quality metrics.

        Returns:
            Dictionary with quality metrics
        """
        return self.analyzer.get_data_quality_metrics(
            self.data, self.time_diffs, self.typical_interval
        )

    def get_logs(self) -> dict[str, Any]:
        """
        Returns all stored logs.

        Returns:
            Dictionary with generated logs
        """
        if not self.log:
            self.logger.warning(
                "Logging is not enabled. Initialize the class with log=True to generate logs."
            )
            return {}
        return self.logs

    # Export methods - Delegate to DataExporter
    def to_parquet(self, file_path: str, compression: str = "snappy", sort: bool = True):
        """Delegates to DataExporter to save in Parquet format."""
        self.exporter.to_parquet(self.data, file_path, compression, sort)

    def append_to_parquet(
        self,
        file_path: str,
        compression: str = "snappy",
        handle_duplicates: str = "keep_new",
    ) -> int:
        """Delegates to DataExporter to append to an existing Parquet file."""
        return self.exporter.append_to_parquet(self.data, file_path, compression, handle_duplicates)

    def to_csv(self, file_path: str, separator: str = ",", include_index: bool = False):
        """Delegates to DataExporter to save in CSV format."""
        self.exporter.to_csv(self.data, file_path, separator, include_index)

    def to_excel(self, file_path: str, sheet_name: str = "glucose_data"):
        """Delegates to DataExporter to save in Excel format."""
        self.exporter.to_excel(self.data, file_path, sheet_name)

    # Data access methods
    def get_raw_data(self) -> pd.DataFrame:
        """
        Returns the DataFrame with processed data.

        Returns:
            DataFrame with glucose data
        """
        return self.data.copy()

    def get_glucose_values(self) -> pd.Series:
        """
        Returns only glucose values.

        Returns:
            Series with glucose values
        """
        return self.data["glucose"].copy()

    def get_timestamps(self) -> pd.Series:
        """
        Returns only timestamps.

        Returns:
            Series with timestamps
        """
        return self.data["time"].copy()

    def get_time_differences(self) -> pd.Series:
        """
        Returns time differences between measurements.

        Returns:
            Series with time differences
        """
        return self.time_diffs.copy()

    # Filtering methods
    def filter_by_date_range(
        self,
        start_date: str | datetime.datetime,
        end_date: str | datetime.datetime,
    ) -> GlucoseData:
        """
        Creates a new instance filtered by date range.

        Args:
            start_date: Start date
            end_date: End date

        Returns:
            New instance with filtered data
        """
        # Use the processor to filter data
        filtered_data = self.processor.filter_by_dates(self.data, start_date, end_date)

        # Create new instance using the constructor
        return self._create_filtered_instance(filtered_data)

    def filter_by_glucose_range(self, min_glucose: float, max_glucose: float) -> GlucoseData:
        """
        Creates a new instance filtered by glucose range.

        Args:
            min_glucose: Minimum glucose value
            max_glucose: Maximum glucose value

        Returns:
            New instance with filtered data
        """
        mask = (self.data["glucose"] >= min_glucose) & (self.data["glucose"] <= max_glucose)
        filtered_data = self.data[mask].copy()

        if len(filtered_data) == 0:
            raise ValueError("No data in the specified glucose range.")

        return self._create_filtered_instance(filtered_data)

    def regularize(self, interval_mins: int = 5, max_gap_mins: int = 30) -> GlucoseData:
        """
        Regularizes the current instance's data in-place.
        Returns self for method chaining.

        Args:
            interval_mins: Target interval in minutes.
            max_gap_mins: Maximum gap to interpolate.

        Returns:
            GlucoseData: The current instance with uniform frequency.
        """
        # 1. Use processor to regularize data
        self.data = self.processor.regularize(
            self.data,
            interval_mins=interval_mins,
            max_gap_mins=max_gap_mins,
            date_col="time",
            glucose_col="glucose",
        )

        # 2. Update time differences and internal metrics
        self.time_diffs = self.data["time"].diff()
        self.typical_interval = self.analyzer.calculate_typical_interval(self.time_diffs)

        # 3. Refresh trimesters if specialized for pregnancy
        if hasattr(self, "_split_trimesters"):
            self.trimesters = self._split_trimesters()

        if self.log:
            self.logger.info(f"Data regularized in-place to {interval_mins}min.")

        return self

    def _create_filtered_instance(self, filtered_data: pd.DataFrame) -> GlucoseData:
        """
        Creates a new instance (clone) with modified data.

        Args:
            filtered_data: DataFrame with filtered or modified data

        Returns:
            New GlucoseData instance
        """
        new_instance = copy.copy(self)
        new_instance.logs = {}

        # Refresh dependent data
        new_instance.data = filtered_data.copy()
        new_instance.time_diffs = new_instance.data["time"].diff()
        new_instance.typical_interval = self.analyzer.calculate_typical_interval(
            new_instance.time_diffs
        )

        # Refresh trimesters if PregnancyData
        if hasattr(self, "_split_trimesters"):
            new_instance.trimesters = new_instance._split_trimesters()  # type: ignore[attr-defined]

        return new_instance

Loaders

cgmpy.data.loader.DataLoader

Class responsible for loading data from different sources (CSV, Parquet, DataFrame).

Source code in cgmpy/data/loader.py

class DataLoader:
    """
    Class responsible for loading data from different sources (CSV, Parquet, DataFrame).
    """

    def __init__(self, logger: logging.Logger | None = None):
        """
        Initializes the DataLoader.

        Args:
            logger: Logger to record operations
        """
        self.logger = logger or logging.getLogger(__name__)

    def load_from_source(
        self,
        data_source: str | pd.DataFrame,
        date_col: str,
        glucose_col: str,
        delimiter: str | None = None,
        header: int = 0,
    ) -> pd.DataFrame:
        """
        Loads data from different sources.

        Args:
            data_source: CSV/Parquet file or DataFrame
            date_col: Name of the date column
            glucose_col: Name of the glucose column
            delimiter: Delimiter for CSV files
            header: Header row

        Returns:
            DataFrame with loaded data
        """
        if isinstance(data_source, str):
            return self._load_from_file(data_source, date_col, glucose_col, delimiter, header)
        elif isinstance(data_source, pd.DataFrame):
            return self._load_from_dataframe(data_source)
        else:
            raise ValueError("data_source must be a CSV file, Parquet file, or a DataFrame")

    def _load_from_file(
        self,
        file_path: str,
        date_col: str,
        glucose_col: str,
        delimiter: str | None,
        header: int,
    ) -> pd.DataFrame:
        """
        Loads data from a file.

        Args:
            file_path: Path to the file
            date_col: Name of the date column
            glucose_col: Name of the glucose column
            delimiter: Delimiter for CSV
            header: Header row

        Returns:
            DataFrame with the data
        """
        if not Path(file_path).is_file():
            raise FileNotFoundError(f"File not found: {file_path}")

        is_parquet = file_path.lower().endswith(".parquet")

        if is_parquet:
            return self._load_parquet(file_path, date_col, glucose_col)
        else:
            return self._load_csv(file_path, date_col, glucose_col, delimiter, header)

    def _load_parquet(self, file_path: str, date_col: str, glucose_col: str) -> pd.DataFrame:
        """
        Loads data from a Parquet file.

        Args:
            file_path: Path to the Parquet file
            date_col: Name of the date column
            glucose_col: Name of the glucose column

        Returns:
            DataFrame with the data

        Raises:
            InvalidCSVFormatError: If pyarrow/pandas fails to read the
                file (corrupt, missing columns, schema mismatch, ...).
        """
        try:
            return pd.read_parquet(file_path, columns=[date_col, glucose_col])
        except Exception as e:
            raise InvalidCSVFormatError(file_path, reason=str(e)) from e

    def _load_csv(
        self,
        file_path: str,
        date_col: str,
        glucose_col: str,
        delimiter: str | None,
        header: int,
    ) -> pd.DataFrame:
        """
        Loads data from a CSV file.

        Args:
            file_path: Path to the CSV file
            date_col: Name of the date column
            glucose_col: Name of the glucose column
            delimiter: Delimiter
            header: Header row

        Returns:
            DataFrame with the data

        Raises:
            InvalidCSVFormatError: If the file cannot be parsed by pandas
                using either ``,`` or ``;`` as the delimiter.
        """
        if delimiter is None:
            delimiter = ","

        try:
            return pd.read_csv(
                file_path,
                delimiter=delimiter,
                header=header,
                usecols=[date_col, glucose_col],
            )
        except Exception as e:
            # Try with alternative delimiter if it fails
            if delimiter == ",":
                try:
                    return pd.read_csv(
                        file_path,
                        delimiter=";",
                        header=header,
                        usecols=[date_col, glucose_col],
                    )
                except Exception as inner_e:
                    raise InvalidCSVFormatError(
                        file_path,
                        reason=str(inner_e),
                        hint=("Try specifying the delimiter with the 'delimiter' parameter."),
                    ) from inner_e
            else:
                raise InvalidCSVFormatError(
                    file_path,
                    reason=(f"Error reading CSV file with delimiter '{delimiter}': {e!s}"),
                ) from e

    def _load_from_dataframe(self, dataframe: pd.DataFrame) -> pd.DataFrame:
        """
        Loads data from an existing DataFrame.

        Args:
            dataframe: DataFrame with data

        Returns:
            Copy of the DataFrame
        """
        return dataframe.copy()

Processors

cgmpy.data.processor.DataProcessor

Class responsible for glucose data processing, validation, and cleaning.

Source code in cgmpy/data/processor.py

class DataProcessor:
    """
    Class responsible for glucose data processing, validation, and cleaning.
    """

    def __init__(self, logger: logging.Logger | None = None):
        """
        Initializes the DataProcessor.

        Args:
            logger: Logger to record operations
        """
        self.logger = logger or logging.getLogger(__name__)
        self._last_validation_report: ValidationReport | None = None

    def process_data(
        self,
        data: pd.DataFrame,
        date_col: str,
        glucose_col: str,
        log_performance: bool = False,
        strict_glucose_range: bool = False,
    ) -> tuple[pd.DataFrame, pd.Series]:
        """
        Processes glucose data in an optimized way.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column
            log_performance: If True, records performance metrics
            strict_glucose_range: If ``True``, raise
                :class:`GlucoseRangeError` when any glucose value falls outside
                the physiologically plausible range (39-600 mg/dL). Defaults to
                ``False`` to preserve the legacy warn-only behaviour.

        Returns:
            Tuple with processed DataFrame and time differences Series

        Raises:
            GlucoseRangeError: Only when ``strict_glucose_range=True``
                and the validation report contains out-of-range values.
        """
        t_start = time.time()

        if log_performance:
            self.logger.debug("\n--- DETAILED PERFORMANCE ANALYSIS ---")

        # Validate columns
        self._validate_columns(data, date_col, glucose_col)

        # Determine if data comes from optimized Parquet
        from_parquet = self._is_optimized_parquet(data, date_col, glucose_col)

        if from_parquet:
            processed_data, time_diffs = self._process_parquet_optimized(
                data, date_col, glucose_col, log_performance
            )
        else:
            processed_data, time_diffs = self._process_standard(
                data, date_col, glucose_col, log_performance
            )

        # Final validation
        if not pd.api.types.is_datetime64_any_dtype(processed_data[date_col]):
            raise ValueError("Error in date conversion")

        # Optional strict gate on glucose range (does not change the
        # internal validation flow: _convert_data_types still runs
        # validate_glucose_range(warn=True) so the report is always fresh
        # and inspectable via `last_validation_report`).
        if strict_glucose_range and self._last_validation_report is not None:
            report = self._last_validation_report
            n_invalid = report.n_below + report.n_above
            if n_invalid > 0:
                # Handle NaN min/max for fully-empty series gracefully.
                if (
                    report.n_total == 0
                    or report.min_glucose != report.min_glucose
                    or report.max_glucose != report.max_glucose
                ):
                    min_value, max_value = 0.0, 0.0
                else:
                    min_value = float(report.min_glucose)
                    max_value = float(report.max_glucose)
                raise GlucoseRangeError(
                    n_invalid=n_invalid,
                    total=report.n_total,
                    min_value=min_value,
                    max_value=max_value,
                    bounds=(report.low_threshold, report.high_threshold),
                )

        if log_performance:
            t_end = time.time()
            memory_bytes = processed_data.memory_usage(deep=True).sum()
            memory_mb = memory_bytes / (1024 * 1024)
            self.logger.debug(f"DataFrame memory usage: {memory_mb:.2f} MB")
            self.logger.debug(f"Total processing time: {t_end - t_start:.3f}s")
            self.logger.debug("--- END OF ANALYSIS ---\n")

        return processed_data, time_diffs

    def _validate_columns(self, data: pd.DataFrame, date_col: str, glucose_col: str):
        """
        Validates that the specified columns exist in the DataFrame.

        Args:
            data: DataFrame to validate
            date_col: Name of the date column
            glucose_col: Name of the glucose column

        Raises:
            ColumnNotFoundError: If ``date_col`` or ``glucose_col`` is
                not present in ``data.columns``. A separate exception is raised
                for each missing column, with the list of available columns
                attached for diagnostics.
        """
        available = list(data.columns)
        if date_col not in data.columns:
            raise ColumnNotFoundError(
                date_col,
                available=available,
                message=(
                    f"Required date column '{date_col}' not found in the "
                    f"DataFrame. Available columns: {available}."
                ),
            )
        if glucose_col not in data.columns:
            raise ColumnNotFoundError(
                glucose_col,
                available=available,
                message=(
                    f"Required glucose column '{glucose_col}' not found in the "
                    f"DataFrame. Available columns: {available}."
                ),
            )

    def _is_optimized_parquet(self, data: pd.DataFrame, date_col: str, glucose_col: str) -> bool:
        """
        Determines if the data comes from an optimized Parquet file.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column

        Returns:
            True if it is optimized Parquet
        """
        return (
            pd.api.types.is_datetime64_any_dtype(data[date_col])
            and data[glucose_col].dtype == "int16"
        )

    def _process_parquet_optimized(
        self,
        data: pd.DataFrame,
        date_col: str,
        glucose_col: str,
        log_performance: bool = False,
    ) -> tuple[pd.DataFrame, pd.Series]:
        """
        Processes optimized Parquet data with a fast path.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column
            log_performance: If True, records performance metrics

        Returns:
            Tuple with processed DataFrame and time differences Series
        """
        if log_performance:
            self.logger.debug("Detected Parquet source data with optimized types.")
            self.logger.debug("Applying fast path for Parquet data...")

        processed_data = data.copy()

        # Check and remove nulls
        t_nulls = time.time()
        if processed_data.isna().any().any():
            processed_data = processed_data.dropna(subset=[date_col, glucose_col])
            if log_performance:
                self.logger.debug(f"  - Removed null values: {time.time() - t_nulls:.3f}s")
        elif log_performance:
            self.logger.debug(f"  - No null values found: {time.time() - t_nulls:.3f}s")

        # Check and sort
        t_sort = time.time()
        if not processed_data[date_col].is_monotonic_increasing:
            if log_performance:
                self.logger.debug("  - Sorting data...")
            processed_data = processed_data.sort_values(date_col, ignore_index=True)
        elif log_performance:
            self.logger.debug("  - Data already sorted")

        if log_performance:
            self.logger.debug(f"  - Order verification: {time.time() - t_sort:.3f}s")

        # Optimized calculation of time differences
        t_diff = time.time()
        time_values = processed_data[date_col].values
        time_diffs_ns = np.diff(time_values.astype("datetime64[ns]"))
        time_diffs_ns = np.insert(time_diffs_ns, 0, np.timedelta64(0, "ns"))
        time_diffs = pd.Series(pd.TimedeltaIndex(time_diffs_ns), index=processed_data.index)

        if log_performance:
            self.logger.debug(f"  - Optimized difference calculation: {time.time() - t_diff:.3f}s")

        return processed_data, time_diffs

    def _process_standard(
        self,
        data: pd.DataFrame,
        date_col: str,
        glucose_col: str,
        log_performance: bool = False,
    ) -> tuple[pd.DataFrame, pd.Series]:
        """
        Processes data with full validations for CSV and other formats.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column
            log_performance: If True, records performance metrics

        Returns:
            Tuple with processed DataFrame and time differences Series
        """
        if log_performance:
            self.logger.debug("Processing data with type conversion and full validations.")

        processed_data = data.copy()

        # Handle nulls
        t_nulls = time.time()
        processed_data = self._handle_nulls(processed_data, date_col, glucose_col)
        if log_performance:
            self.logger.debug(f"2. Null removal: {time.time() - t_nulls:.3f}s")

        # Type conversion
        t_types = time.time()
        processed_data = self._convert_data_types(processed_data, date_col, glucose_col)
        if log_performance:
            self.logger.debug(f"3. Type conversion: {time.time() - t_types:.3f}s")

        # Handle duplicates
        t_dups = time.time()
        processed_data = self._handle_duplicates(processed_data, date_col, glucose_col)
        if log_performance:
            self.logger.debug(f"4. Duplicate processing: {time.time() - t_dups:.3f}s")

        # Sorting
        t_sort = time.time()
        processed_data = self._sort_data(processed_data, date_col)
        if log_performance:
            self.logger.debug(f"5. Sorting: {time.time() - t_sort:.3f}s")

        # Calculate differences
        t_diff = time.time()
        time_diffs = processed_data[date_col].diff()
        if log_performance:
            self.logger.debug(f"6. Difference calculation: {time.time() - t_diff:.3f}s")

        return processed_data, time_diffs

    def _handle_nulls(self, data: pd.DataFrame, date_col: str, glucose_col: str) -> pd.DataFrame:
        """
        Removes rows with null values in key columns.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column

        Returns:
            DataFrame without nulls
        """
        rows_before = len(data)
        data_cleaned = data.dropna(subset=[date_col, glucose_col])
        rows_after = len(data_cleaned)

        if rows_before > rows_after:
            self.logger.debug(f"  - Removed {rows_before - rows_after} rows with null values.")

        return data_cleaned

    def _convert_data_types(
        self, data: pd.DataFrame, date_col: str, glucose_col: str
    ) -> pd.DataFrame:
        """
        Convert date and glucose columns to correct types.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column

        Returns:
            DataFrame with correct types
        """
        data_converted = data.copy()

        # Convert date column
        if not pd.api.types.is_datetime64_any_dtype(data_converted[date_col]):
            self.logger.debug(f"  - Converting column '{date_col}' to datetime...")
            if pd.api.types.is_numeric_dtype(data_converted[date_col]):
                unit = "ms" if data_converted[date_col].iloc[0] > 1e10 else "s"
                data_converted[date_col] = pd.to_datetime(data_converted[date_col], unit=unit)
            else:
                data_converted[date_col] = pd.to_datetime(
                    data_converted[date_col], errors="coerce", format="mixed"
                )

        # Convert glucose column
        if not pd.api.types.is_numeric_dtype(data_converted[glucose_col]):
            self.logger.debug(f"  - Converting column '{glucose_col}' to numeric...")

            # If the column is object (string), handle possible decimal commas
            if data_converted[glucose_col].dtype == "object":
                # Replace commas with dots to support European formats
                data_converted[glucose_col] = (
                    data_converted[glucose_col].astype(str).str.replace(",", ".", regex=False)
                )

            data_converted[glucose_col] = pd.to_numeric(
                data_converted[glucose_col], errors="coerce"
            )
            data_converted = data_converted.dropna(subset=[glucose_col])

        # Optimize glucose type to int16 if possible
        if data_converted[glucose_col].dtype != "int16":
            self.logger.debug(f"  - Optimizing column '{glucose_col}'...")
            min_val, max_val = (
                data_converted[glucose_col].min(),
                data_converted[glucose_col].max(),
            )
            if pd.notna(min_val) and pd.notna(max_val) and min_val >= -32768 and max_val <= 32767:
                # Only convert to int16 if values are already integers to avoid precision loss
                is_integer = np.all(np.mod(data_converted[glucose_col].dropna(), 1) == 0)
                if is_integer:
                    data_converted[glucose_col] = data_converted[glucose_col].astype("int16")
                else:
                    data_converted[glucose_col] = data_converted[glucose_col].astype("float32")
            else:
                data_converted[glucose_col] = pd.to_numeric(
                    data_converted[glucose_col], errors="coerce", downcast="float"
                )

        # Validate that glucose values are within a physiologically plausible range.
        # `validate_glucose_range` expects a column named "glucose"; rename
        # locally on a copy so the caller's column name is preserved.
        if glucose_col in data_converted.columns and not data_converted.empty:
            validation_view = data_converted.rename(columns={glucose_col: "glucose"})
            self._last_validation_report = validate_glucose_range(validation_view, warn=True)

        return data_converted

    def _handle_duplicates(
        self, data: pd.DataFrame, date_col: str, glucose_col: str
    ) -> pd.DataFrame:
        """
        Finds and resolves duplicates in the date column.

        Args:
            data: DataFrame with data
            date_col: Name of the date column
            glucose_col: Name of the glucose column

        Returns:
            DataFrame without duplicates
        """
        mask_duplicates = data.duplicated(subset=[date_col], keep=False)
        n_duplicates = mask_duplicates.sum()

        if n_duplicates > 0:
            self.logger.debug(f"  - Found {n_duplicates // 2} duplicate timestamps. Resolving...")

            df_dups = data[mask_duplicates].copy()
            df_dups["diff"] = df_dups.groupby(date_col)[glucose_col].transform(
                lambda x: (x - x.mean()).abs()
            )

            idx_to_keep = df_dups.groupby(date_col)["diff"].idxmin()

            return pd.concat(
                [data[~mask_duplicates], df_dups.loc[idx_to_keep].drop(columns="diff")]
            )

        return data

    def _sort_data(self, data: pd.DataFrame, date_col: str) -> pd.DataFrame:
        """
        Sorts the DataFrame by the date column if it is not already sorted.

        Args:
            data: DataFrame with data
            date_col: Name of the date column

        Returns:
            Sorted DataFrame
        """
        if not data[date_col].is_monotonic_increasing:
            self.logger.debug("  - Sorting data by timestamp...")
            return data.sort_values(date_col, ignore_index=True)

        self.logger.debug("  - Data already sorted, skipping sorting.")
        return data

    def rename_columns(self, data: pd.DataFrame, date_col: str, glucose_col: str) -> pd.DataFrame:
        """
        Renames columns to standard names.

        Args:
            data: DataFrame with data
            date_col: Current name of the date column
            glucose_col: Current name of the glucose column

        Returns:
            DataFrame with renamed columns
        """
        renamed_data = data.copy()

        if date_col != "time":
            renamed_data = renamed_data.rename(columns={date_col: "time"})
        if glucose_col != "glucose":
            renamed_data = renamed_data.rename(columns={glucose_col: "glucose"})

        return renamed_data

    def filter_by_dates(
        self,
        data: pd.DataFrame,
        start_date: str | pd.Timestamp | None = None,
        end_date: str | pd.Timestamp | None = None,
        date_col: str = "time",
    ) -> pd.DataFrame:
        """
        Filters data by date range.

        Args:
            data: DataFrame with data
            start_date: Start date
            end_date: End date
            date_col: Name of the date column

        Returns:
            Filtered DataFrame
        """
        filtered_data = data.copy()

        if start_date is not None:
            if isinstance(start_date, str):
                start_date = pd.to_datetime(start_date)
            filtered_data = filtered_data[filtered_data[date_col] >= start_date]

        if end_date is not None:
            if isinstance(end_date, str):
                end_date = pd.to_datetime(end_date)
            filtered_data = filtered_data[filtered_data[date_col] <= end_date]

        if len(filtered_data) == 0:
            raise ValueError("No data available in the specified date range.")

        return filtered_data

    def regularize(
        self,
        data: pd.DataFrame,
        interval_mins: int = 5,
        max_gap_mins: int = 30,
        date_col: str = "time",
        glucose_col: str = "glucose",
    ) -> pd.DataFrame:
        """
        Regularizes glucose data to a constant sampling frequency.

        This method is critical when combining data from sensors with different
        frequencies (e.g., 1 min vs 15 min) to avoid statistical bias.

        Args:
            data: The original glucose DataFrame.
            interval_mins: Target interval in minutes (default 5).
            max_gap_mins: Maximum gap to interpolate (default 30).
                          Gaps larger than this are kept as NaNs to mark disconnections.
            date_col: Name of the time column.
            glucose_col: Name of the glucose column.

        Returns:
            pd.DataFrame: A new DataFrame with uniform frequency.
        """
        if len(data) < 2:
            return data.copy()

        # 1. Create a copy and set the index to the time column for resampling
        df = data.copy()
        df = df.set_index(date_col).sort_index()

        # 2. Resample to the target frequency
        freq = f"{interval_mins}min"
        # We use mean() for points that collapse into the same bin (if any)
        resampled = df[glucose_col].resample(freq).mean()

        # 3. Calculate the interpolation limit
        # If interval is 5min and max_gap is 30min, we can interpolate up to 6 points (30/5)
        limit = max_gap_mins // interval_mins

        # 4. Interpolate with limit
        # This fills small gaps but keeps large ones as NaN (disconnections)
        regularized_glucose = resampled.interpolate(
            method="linear", limit=limit, limit_area="inside"
        )

        # 5. Reconstruct the DataFrame
        regularized_df = pd.DataFrame(
            {date_col: regularized_glucose.index, glucose_col: regularized_glucose.values}
        )

        # 6. Drop NaNs created by resampling that weren't interpolated
        # (These are the actual long disconnections)
        regularized_df = regularized_df.dropna(subset=[glucose_col])

        # 7. Ensure correct types (glucose as int16 if appropriate)
        regularized_df = self._convert_data_types(regularized_df, date_col, glucose_col)

        self.logger.info(
            f"Data regularized to {interval_mins}min. Rows changed from {len(data)} to {len(regularized_df)}."
        )

        return regularized_df

process_data

process_data(data: DataFrame, date_col: str, glucose_col: str, log_performance: bool = False, strict_glucose_range: bool = False) -> tuple[DataFrame, Series]

Processes glucose data in an optimized way.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with data	required
date_col	str	Name of the date column	required
glucose_col	str	Name of the glucose column	required
log_performance	bool	If True, records performance metrics	False
strict_glucose_range	bool	If True, raise :class:GlucoseRangeError when any glucose value falls outside the physiologically plausible range (39-600 mg/dL). Defaults to False to preserve the legacy warn-only behaviour.	False

Returns:

Type	Description
tuple[DataFrame, Series]	Tuple with processed DataFrame and time differences Series

Raises:

Type	Description
GlucoseRangeError	Only when strict_glucose_range=True and the validation report contains out-of-range values.

Source code in cgmpy/data/processor.py

def process_data(
    self,
    data: pd.DataFrame,
    date_col: str,
    glucose_col: str,
    log_performance: bool = False,
    strict_glucose_range: bool = False,
) -> tuple[pd.DataFrame, pd.Series]:
    """
    Processes glucose data in an optimized way.

    Args:
        data: DataFrame with data
        date_col: Name of the date column
        glucose_col: Name of the glucose column
        log_performance: If True, records performance metrics
        strict_glucose_range: If ``True``, raise
            :class:`GlucoseRangeError` when any glucose value falls outside
            the physiologically plausible range (39-600 mg/dL). Defaults to
            ``False`` to preserve the legacy warn-only behaviour.

    Returns:
        Tuple with processed DataFrame and time differences Series

    Raises:
        GlucoseRangeError: Only when ``strict_glucose_range=True``
            and the validation report contains out-of-range values.
    """
    t_start = time.time()

    if log_performance:
        self.logger.debug("\n--- DETAILED PERFORMANCE ANALYSIS ---")

    # Validate columns
    self._validate_columns(data, date_col, glucose_col)

    # Determine if data comes from optimized Parquet
    from_parquet = self._is_optimized_parquet(data, date_col, glucose_col)

    if from_parquet:
        processed_data, time_diffs = self._process_parquet_optimized(
            data, date_col, glucose_col, log_performance
        )
    else:
        processed_data, time_diffs = self._process_standard(
            data, date_col, glucose_col, log_performance
        )

    # Final validation
    if not pd.api.types.is_datetime64_any_dtype(processed_data[date_col]):
        raise ValueError("Error in date conversion")

    # Optional strict gate on glucose range (does not change the
    # internal validation flow: _convert_data_types still runs
    # validate_glucose_range(warn=True) so the report is always fresh
    # and inspectable via `last_validation_report`).
    if strict_glucose_range and self._last_validation_report is not None:
        report = self._last_validation_report
        n_invalid = report.n_below + report.n_above
        if n_invalid > 0:
            # Handle NaN min/max for fully-empty series gracefully.
            if (
                report.n_total == 0
                or report.min_glucose != report.min_glucose
                or report.max_glucose != report.max_glucose
            ):
                min_value, max_value = 0.0, 0.0
            else:
                min_value = float(report.min_glucose)
                max_value = float(report.max_glucose)
            raise GlucoseRangeError(
                n_invalid=n_invalid,
                total=report.n_total,
                min_value=min_value,
                max_value=max_value,
                bounds=(report.low_threshold, report.high_threshold),
            )

    if log_performance:
        t_end = time.time()
        memory_bytes = processed_data.memory_usage(deep=True).sum()
        memory_mb = memory_bytes / (1024 * 1024)
        self.logger.debug(f"DataFrame memory usage: {memory_mb:.2f} MB")
        self.logger.debug(f"Total processing time: {t_end - t_start:.3f}s")
        self.logger.debug("--- END OF ANALYSIS ---\n")

    return processed_data, time_diffs

Analyzers

cgmpy.data.analyzer.DataAnalyzer

Class responsible for basic analysis of glucose data.

Source code in cgmpy/data/analyzer.py

class DataAnalyzer:
    """
    Class responsible for basic analysis of glucose data.
    """

    def __init__(self, logger: logging.Logger | None = None):
        """
        Initializes the DataAnalyzer.

        Args:
            logger: Logger to record operations
        """
        self.logger = logger or logging.getLogger(__name__)

    def calculate_typical_interval(
        self, time_diffs: pd.Series, log_performance: bool = False
    ) -> float:
        """
        Calculates the typical interval between measurements in minutes.

        Args:
            time_diffs: Series with the time differences
            log_performance: If True, records performance metrics

        Returns:
            Typical interval in minutes
        """
        if log_performance:
            t_start = time.time()
            self.logger.debug("\n--- TYPICAL INTERVAL CALCULATION ANALYSIS ---")

        # Convert to NumPy array for faster operations
        time_diffs_seconds = time_diffs.dt.total_seconds().values
        # Filter valid values (greater than 0)
        valid_diffs = time_diffs_seconds[time_diffs_seconds > 0]

        if len(valid_diffs) > 0:
            # Use NumPy to calculate the median (faster)
            interval = np.median(valid_diffs) / 60
        else:
            # Default value if no valid differences
            interval = 5.0

        if log_performance:
            t_end = time.time()
            self.logger.debug(f"Optimized median calculation: {t_end - t_start:.3f}s")
            self.logger.debug(f"Total interval calculation time: {t_end - t_start:.3f}s")
            self.logger.debug("--- END OF ANALYSIS ---\n")

        return abs(interval)

    def get_basic_info(
        self,
        data: pd.DataFrame,
        time_diffs: pd.Series,
        typical_interval: float,
        include_disconnections: bool = False,
    ) -> dict[str, Any]:
        """
        Generates basic information about the glucose data.

        Args:
            data: DataFrame with the glucose data
            time_diffs: Series with the time differences
            typical_interval: Typical interval between measurements
            include_disconnections: Whether to include disconnection details

        Returns:
            Dictionary with basic information
        """
        # Basic information
        n_records = len(data)
        start_date = data["time"].min()
        end_date = data["time"].max()

        # Disconnection analysis
        disconnection_threshold = pd.Timedelta(minutes=typical_interval + 10)
        disconnections = time_diffs[time_diffs > disconnection_threshold]
        n_disconnections = len(disconnections)

        # Total disconnection time
        total_disconnection_time = disconnections.sum()
        disconnection_hours = total_disconnection_time.total_seconds() / 3600

        # Memory usage
        memory_bytes = data.memory_usage(deep=True).sum()
        memory_mb = memory_bytes / (1024 * 1024)

        # Expected theoretical data
        total_time = (data["time"].max() - data["time"].min()).total_seconds() / 60

        # Avoid errors if total_time or typical_interval are invalid
        if pd.isna(total_time) or typical_interval <= 0:
            expected_data = 0
        else:
            expected_data = int(total_time / typical_interval)

        completeness = (n_records / expected_data * 100) if expected_data > 0 else 0.0

        # Create summary dictionary
        summary = {
            "n_records": n_records,
            "start_date": start_date,
            "end_date": end_date,
            "typical_interval": typical_interval,
            "expected_data": expected_data,
            "completeness": completeness,
            "n_disconnections": (
                f"{n_disconnections} disconnections (For more info, "
                "use info(include_disconnections=True))"
            ),
            "total_disconnection_time": disconnection_hours,
            "memory_usage_mb": memory_mb,
        }

        if include_disconnections:
            summary["disconnection_list"] = self._get_disconnection_details(data, disconnections)

        return summary

    def _get_disconnection_details(self, data: pd.DataFrame, disconnections: pd.Series) -> list:
        """
        Gets details of disconnections.

        Args:
            data: DataFrame with data
            disconnections: Series with disconnections

        Returns:
            List with disconnection details
        """
        disconnection_list = []

        if len(disconnections) > 0:
            for idx, index in enumerate(disconnections.index, 1):
                try:
                    current_pos = data.index.get_loc(index)
                    if current_pos > 0:
                        disconnection_end = data.iloc[current_pos]["time"]
                        disconnection_start = data.iloc[current_pos - 1]["time"]
                        duration_minutes = (
                            disconnection_end - disconnection_start
                        ).total_seconds() / 60
                        hours = int(duration_minutes // 60)
                        minutes = int(duration_minutes % 60)
                        disconnection_list.append(
                            {
                                "start": disconnection_start.strftime("%d/%m/%Y %H:%M"),
                                "end": disconnection_end.strftime("%d/%m/%Y %H:%M"),
                                "duration": f"{hours:02d} hours and {minutes:02d} minutes",
                            }
                        )
                except (KeyError, ValueError, TypeError, AttributeError) as e:
                    self.logger.warning(f"Error processing disconnection {idx}: {e}")

        return disconnection_list

    def get_summary_string(self, info: dict[str, Any]) -> str:
        """
        Generates a string representation of basic information.

        Args:
            info: Dictionary with basic information

        Returns:
            String with summary
        """
        return (
            f"File contains {info['n_records']} records between {info['start_date']} and {info['end_date']}.\n"
            f"Typical interval between measurements: {info['typical_interval']:.1f} minutes.\n"
            f"Expected theoretical data: {info['expected_data']}\n"
            f"Data availability percentage: {info['completeness']:.1f}%\n"
            f"Detected {info['n_disconnections']}\n"
            f"Total disconnection time: {info['total_disconnection_time']:.1f} hours.\n"
            f"DataFrame memory usage: {info['memory_usage_mb']:.2f} MB"
        )

    def get_data_quality_metrics(
        self, data: pd.DataFrame, time_diffs: pd.Series, typical_interval: float
    ) -> dict[str, Any]:
        """
        Calculates data quality metrics.

        Args:
            data: DataFrame with data
            time_diffs: Series with time differences
            typical_interval: Typical interval between measurements

        Returns:
            Dictionary with quality metrics
        """
        # Calculate gaps in the data
        gap_threshold = pd.Timedelta(minutes=typical_interval * 2)
        gaps = time_diffs[time_diffs > gap_threshold]

        # Calculate interval statistics
        valid_intervals = time_diffs[time_diffs > pd.Timedelta(0)].dt.total_seconds() / 60

        return {
            "total_gaps": len(gaps),
            "max_gap_hours": gaps.max().total_seconds() / 3600 if len(gaps) > 0 else 0,
            "mean_interval": valid_intervals.mean() if len(valid_intervals) > 0 else 0,
            "std_interval": valid_intervals.std() if len(valid_intervals) > 0 else 0,
            "min_glucose": data["glucose"].min(),
            "max_glucose": data["glucose"].max(),
            "mean_glucose": data["glucose"].mean(),
            "std_glucose": data["glucose"].std(),
        }

calculate_typical_interval

calculate_typical_interval(time_diffs: Series, log_performance: bool = False) -> float

Calculates the typical interval between measurements in minutes.

Parameters:

Name	Type	Description	Default
time_diffs	Series	Series with the time differences	required
log_performance	bool	If True, records performance metrics	False

Returns:

Type	Description
float	Typical interval in minutes

Source code in cgmpy/data/analyzer.py

def calculate_typical_interval(
    self, time_diffs: pd.Series, log_performance: bool = False
) -> float:
    """
    Calculates the typical interval between measurements in minutes.

    Args:
        time_diffs: Series with the time differences
        log_performance: If True, records performance metrics

    Returns:
        Typical interval in minutes
    """
    if log_performance:
        t_start = time.time()
        self.logger.debug("\n--- TYPICAL INTERVAL CALCULATION ANALYSIS ---")

    # Convert to NumPy array for faster operations
    time_diffs_seconds = time_diffs.dt.total_seconds().values
    # Filter valid values (greater than 0)
    valid_diffs = time_diffs_seconds[time_diffs_seconds > 0]

    if len(valid_diffs) > 0:
        # Use NumPy to calculate the median (faster)
        interval = np.median(valid_diffs) / 60
    else:
        # Default value if no valid differences
        interval = 5.0

    if log_performance:
        t_end = time.time()
        self.logger.debug(f"Optimized median calculation: {t_end - t_start:.3f}s")
        self.logger.debug(f"Total interval calculation time: {t_end - t_start:.3f}s")
        self.logger.debug("--- END OF ANALYSIS ---\n")

    return abs(interval)

get_data_quality_metrics

get_data_quality_metrics(data: DataFrame, time_diffs: Series, typical_interval: float) -> dict[str, Any]

Calculates data quality metrics.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with data	required
time_diffs	Series	Series with time differences	required
typical_interval	float	Typical interval between measurements	required

Returns:

Type	Description
dict[str, Any]	Dictionary with quality metrics

Source code in cgmpy/data/analyzer.py

def get_data_quality_metrics(
    self, data: pd.DataFrame, time_diffs: pd.Series, typical_interval: float
) -> dict[str, Any]:
    """
    Calculates data quality metrics.

    Args:
        data: DataFrame with data
        time_diffs: Series with time differences
        typical_interval: Typical interval between measurements

    Returns:
        Dictionary with quality metrics
    """
    # Calculate gaps in the data
    gap_threshold = pd.Timedelta(minutes=typical_interval * 2)
    gaps = time_diffs[time_diffs > gap_threshold]

    # Calculate interval statistics
    valid_intervals = time_diffs[time_diffs > pd.Timedelta(0)].dt.total_seconds() / 60

    return {
        "total_gaps": len(gaps),
        "max_gap_hours": gaps.max().total_seconds() / 3600 if len(gaps) > 0 else 0,
        "mean_interval": valid_intervals.mean() if len(valid_intervals) > 0 else 0,
        "std_interval": valid_intervals.std() if len(valid_intervals) > 0 else 0,
        "min_glucose": data["glucose"].min(),
        "max_glucose": data["glucose"].max(),
        "mean_glucose": data["glucose"].mean(),
        "std_glucose": data["glucose"].std(),
    }

Exporters

cgmpy.data.exporter.DataExporter

Class responsible for exporting glucose data in different formats.

Source code in cgmpy/data/exporter.py

class DataExporter:
    """
    Class responsible for exporting glucose data in different formats.
    """

    def __init__(self, logger: logging.Logger | None = None):
        """
        Initializes the DataExporter.

        Args:
            logger: Logger to record operations
        """
        self.logger = logger or logging.getLogger(__name__)

    def to_parquet(
        self,
        data: pd.DataFrame,
        file_path: str,
        compression: str = "snappy",
        sort: bool = True,
    ):
        """
        Saves data in optimized Parquet format.

        Args:
            data: DataFrame with data
            file_path: Path where to save the file
            compression: Compression algorithm
            sort: Whether to sort data before saving
        """
        self.logger.info("Preparing data to save in optimized Parquet format...")

        # Prepare data for saving
        df_to_save = data.copy()

        # Sort by time if requested
        if sort and not df_to_save["time"].is_monotonic_increasing:
            self.logger.info("  - Sorting data by timestamp...")
            df_to_save = df_to_save.sort_values("time", ignore_index=True)

        # Optimize data types
        df_to_save = self._optimize_data_types(df_to_save)

        # Remove duplicates if they exist
        df_to_save = self._remove_duplicates(df_to_save)

        # Save in Parquet format
        t_start = time.time()
        df_to_save.to_parquet(file_path, compression=compression, index=False, engine="pyarrow")
        t_end = time.time()

        # Final information
        self._log_save_info(file_path, df_to_save, t_end - t_start)

    def append_to_parquet(
        self,
        data: pd.DataFrame,
        file_path: str,
        compression: str = "snappy",
        handle_duplicates: str = "keep_new",
    ) -> int:
        """
        Appends data to an existing Parquet file.

        Args:
            data: DataFrame with new data
            file_path: Path to the Parquet file
            compression: Compression algorithm
            handle_duplicates: Strategy for handling duplicates

        Returns:
            Number of records added
        """
        if not Path(file_path).exists():
            self.logger.info(f"File {file_path} does not exist. Creating new file...")
            self.to_parquet(data, file_path, compression=compression)
            return len(data)

        self.logger.info(f"Appending data to existing Parquet file: {file_path}")

        # Load existing data
        t_start = time.time()
        existing_data = pd.read_parquet(file_path)
        t_load = time.time()
        self.logger.info(f"  - Existing file loaded in {t_load - t_start:.3f}s")
        self.logger.info(f"  - Existing records: {len(existing_data):,}")

        # Prepare new data
        new_data = self._prepare_new_data(data)

        # Handle duplicates
        existing_data, new_data = self._handle_duplicates(
            existing_data, new_data, handle_duplicates
        )

        # Combine and sort data
        t_combine = time.time()
        final_data = pd.concat([existing_data, new_data])
        final_data = final_data.sort_values("time", ignore_index=True)
        t_sort = time.time()
        self.logger.info(f"  - Data combined and sorted in {t_sort - t_combine:.3f}s")

        # Save result
        final_data.to_parquet(file_path, compression=compression, index=False, engine="pyarrow")
        t_save = time.time()

        # Final information
        records_added = len(final_data) - len(existing_data)
        file_size = Path(file_path).stat().st_size / 1024 / 1024
        self.logger.info("Data added successfully:")
        self.logger.info(f"  - Records added: {records_added:,}")
        self.logger.info(f"  - Total records: {len(final_data):,}")
        self.logger.info(f"  - File size: {file_size:.2f} MB")
        self.logger.info(f"  - Total time: {t_save - t_start:.3f}s")

        return records_added

    def _optimize_data_types(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Optimizes data types for efficient storage.

        Args:
            data: DataFrame with data

        Returns:
            DataFrame with optimized types
        """
        df_optimized = data.copy()

        # Optimize glucose column
        if not pd.api.types.is_integer_dtype(df_optimized["glucose"]):
            self.logger.info("  - Converting 'glucose' to numeric format...")
            df_optimized["glucose"] = pd.to_numeric(df_optimized["glucose"], errors="coerce")

        # Try to convert to int16 if possible
        min_val = df_optimized["glucose"].min()
        max_val = df_optimized["glucose"].max()

        if pd.notna(min_val) and pd.notna(max_val) and min_val >= -32768 and max_val <= 32767:
            self.logger.info(
                f"  - Optimizing 'glucose' to int16 (range: {min_val} to {max_val})..."
            )
            df_optimized["glucose"] = df_optimized["glucose"].astype("int16")
        else:
            self.logger.warning(
                f"  - Glucose values outside int16 range ({min_val} to {max_val}). Using int32."
            )
            df_optimized["glucose"] = df_optimized["glucose"].astype("int32")

        # Verify that time is datetime
        if not pd.api.types.is_datetime64_any_dtype(df_optimized["time"]):
            self.logger.info("  - Converting 'time' to datetime...")
            df_optimized["time"] = pd.to_datetime(df_optimized["time"], errors="coerce")

        return df_optimized

    def _remove_duplicates(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Removes duplicates based on the time column.

        Args:
            data: DataFrame with data

        Returns:
            DataFrame without duplicates
        """
        duplicates = data.duplicated(subset=["time"], keep="first")
        if duplicates.any():
            n_duplicates = duplicates.sum()
            self.logger.info(f"  - Removing {n_duplicates} duplicate timestamps...")
            return data.drop_duplicates(subset=["time"], keep="first")
        return data

    def _prepare_new_data(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Prepares new data for insertion.

        Args:
            data: DataFrame with new data

        Returns:
            Prepared DataFrame
        """
        new_data = data.copy()

        # Ensure correct types
        if not pd.api.types.is_datetime64_any_dtype(new_data["time"]):
            new_data["time"] = pd.to_datetime(new_data["time"], errors="coerce")

        if not pd.api.types.is_integer_dtype(new_data["glucose"]):
            new_data["glucose"] = pd.to_numeric(new_data["glucose"], errors="coerce")

            # Convert to int16 if possible
            min_val = new_data["glucose"].min()
            max_val = new_data["glucose"].max()
            if pd.notna(min_val) and pd.notna(max_val) and min_val >= -32768 and max_val <= 32767:
                new_data["glucose"] = new_data["glucose"].astype("int16")

        return new_data

    def _handle_duplicates(
        self, existing_data: pd.DataFrame, new_data: pd.DataFrame, strategy: str
    ) -> tuple:
        """
        Handles duplicates between existing and new data.

        Args:
            existing_data: DataFrame with existing data
            new_data: DataFrame with new data
            strategy: Strategy for handling duplicates

        Returns:
            Tuple with processed DataFrames
        """
        # Identify duplicates
        combined = pd.concat([existing_data, new_data])
        duplicated_times = combined["time"].duplicated(keep=False)
        num_duplicates = duplicated_times.sum() // 2

        if num_duplicates > 0:
            self.logger.info(f"  - Found {num_duplicates} duplicate timestamps")

            if strategy == "keep_new":
                self.logger.info("  - Strategy: Keep new data in case of duplicates")
                existing_times = set(existing_data["time"])
                new_times = set(new_data["time"])
                common_times = existing_times.intersection(new_times)

                if common_times:
                    existing_data = existing_data[~existing_data["time"].isin(common_times)]

            elif strategy == "keep_old":
                self.logger.info("  - Strategy: Keep existing data in case of duplicates")
                existing_times = set(existing_data["time"])
                new_data = new_data[~new_data["time"].isin(existing_times)]

        return existing_data, new_data

    def _log_save_info(self, file_path: str, data: pd.DataFrame, save_time: float):
        """
        Logs information about the save operation.

        Args:
            file_path: Path of the saved file
            data: Saved DataFrame
            save_time: Save time
        """
        file_size = Path(file_path).stat().st_size / 1024 / 1024
        self.logger.info(f"Data saved in Parquet format at: {file_path}")
        self.logger.info(f"  - File size: {file_size:.2f} MB")
        self.logger.info(f"  - Save time: {save_time:.3f}s")
        self.logger.info(f"  - Records saved: {len(data):,}")
        self.logger.info(f"  - Date range: {data['time'].min()} to {data['time'].max()}")
        self.logger.info("  - Format ready for fast loading")

    def to_csv(
        self,
        data: pd.DataFrame,
        file_path: str,
        separator: str = ",",
        include_index: bool = False,
    ):
        """
        Saves data in CSV format.

        Args:
            data: DataFrame with data
            file_path: Path where to save the file
            separator: Field separator
            include_index: Whether to include the index
        """
        self.logger.info(f"Saving data in CSV format: {file_path}")

        t_start = time.time()
        data.to_csv(file_path, sep=separator, index=include_index)
        t_end = time.time()

        file_size = Path(file_path).stat().st_size / 1024 / 1024
        self.logger.info(f"Data saved in CSV format at: {file_path}")
        self.logger.info(f"  - File size: {file_size:.2f} MB")
        self.logger.info(f"  - Save time: {t_end - t_start:.3f}s")
        self.logger.info(f"  - Records saved: {len(data):,}")

    def to_excel(self, data: pd.DataFrame, file_path: str, sheet_name: str = "glucose_data"):
        """
        Saves data in Excel format.

        Args:
            data: DataFrame with data
            file_path: Path where to save the file
            sheet_name: Sheet name
        """
        self.logger.info(f"Saving data in Excel format: {file_path}")

        t_start = time.time()
        data.to_excel(file_path, sheet_name=sheet_name, index=False)
        t_end = time.time()

        file_size = Path(file_path).stat().st_size / 1024 / 1024
        self.logger.info(f"Data saved in Excel format at: {file_path}")
        self.logger.info(f"  - File size: {file_size:.2f} MB")
        self.logger.info(f"  - Save time: {t_end - t_start:.3f}s")
        self.logger.info(f"  - Records saved: {len(data):,}")

to_parquet

to_parquet(data: DataFrame, file_path: str, compression: str = 'snappy', sort: bool = True)

Saves data in optimized Parquet format.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with data	required
file_path	str	Path where to save the file	required
compression	str	Compression algorithm	'snappy'
sort	bool	Whether to sort data before saving	True

Source code in cgmpy/data/exporter.py

def to_parquet(
    self,
    data: pd.DataFrame,
    file_path: str,
    compression: str = "snappy",
    sort: bool = True,
):
    """
    Saves data in optimized Parquet format.

    Args:
        data: DataFrame with data
        file_path: Path where to save the file
        compression: Compression algorithm
        sort: Whether to sort data before saving
    """
    self.logger.info("Preparing data to save in optimized Parquet format...")

    # Prepare data for saving
    df_to_save = data.copy()

    # Sort by time if requested
    if sort and not df_to_save["time"].is_monotonic_increasing:
        self.logger.info("  - Sorting data by timestamp...")
        df_to_save = df_to_save.sort_values("time", ignore_index=True)

    # Optimize data types
    df_to_save = self._optimize_data_types(df_to_save)

    # Remove duplicates if they exist
    df_to_save = self._remove_duplicates(df_to_save)

    # Save in Parquet format
    t_start = time.time()
    df_to_save.to_parquet(file_path, compression=compression, index=False, engine="pyarrow")
    t_end = time.time()

    # Final information
    self._log_save_info(file_path, df_to_save, t_end - t_start)

to_csv

to_csv(data: DataFrame, file_path: str, separator: str = ',', include_index: bool = False)

Saves data in CSV format.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with data	required
file_path	str	Path where to save the file	required
separator	str	Field separator	','
include_index	bool	Whether to include the index	False

Source code in cgmpy/data/exporter.py

def to_csv(
    self,
    data: pd.DataFrame,
    file_path: str,
    separator: str = ",",
    include_index: bool = False,
):
    """
    Saves data in CSV format.

    Args:
        data: DataFrame with data
        file_path: Path where to save the file
        separator: Field separator
        include_index: Whether to include the index
    """
    self.logger.info(f"Saving data in CSV format: {file_path}")

    t_start = time.time()
    data.to_csv(file_path, sep=separator, index=include_index)
    t_end = time.time()

    file_size = Path(file_path).stat().st_size / 1024 / 1024
    self.logger.info(f"Data saved in CSV format at: {file_path}")
    self.logger.info(f"  - File size: {file_size:.2f} MB")
    self.logger.info(f"  - Save time: {t_end - t_start:.3f}s")
    self.logger.info(f"  - Records saved: {len(data):,}")

to_excel

to_excel(data: DataFrame, file_path: str, sheet_name: str = 'glucose_data')

Saves data in Excel format.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with data	required
file_path	str	Path where to save the file	required
sheet_name	str	Sheet name	'glucose_data'

Source code in cgmpy/data/exporter.py

def to_excel(self, data: pd.DataFrame, file_path: str, sheet_name: str = "glucose_data"):
    """
    Saves data in Excel format.

    Args:
        data: DataFrame with data
        file_path: Path where to save the file
        sheet_name: Sheet name
    """
    self.logger.info(f"Saving data in Excel format: {file_path}")

    t_start = time.time()
    data.to_excel(file_path, sheet_name=sheet_name, index=False)
    t_end = time.time()

    file_size = Path(file_path).stat().st_size / 1024 / 1024
    self.logger.info(f"Data saved in Excel format at: {file_path}")
    self.logger.info(f"  - File size: {file_size:.2f} MB")
    self.logger.info(f"  - Save time: {t_end - t_start:.3f}s")
    self.logger.info(f"  - Records saved: {len(data):,}")

Device-specific loaders

cgmpy.data.specialized.Dexcom

Bases: GlucoseData

Specialized class for Dexcom device data.

This class inherits from GlucoseData and automatically configures the specific column names for files exported from Dexcom Clarity.

Source code in cgmpy/data/specialized.py

class Dexcom(GlucoseData):
    """
    Specialized class for Dexcom device data.

    This class inherits from GlucoseData and automatically configures
    the specific column names for files exported from Dexcom Clarity.
    """

    def __init__(
        self,
        file_path: str,
        start_date: str | datetime.datetime | None = None,
        end_date: str | datetime.datetime | None = None,
        log: bool = False,
    ):
        """
        Initializes Dexcom data.

        Args:
            file_path: Path to the exported Clarity CSV file
            start_date: Optional start date filter (YYYY-MM-DD)
            end_date: Optional end date filter (YYYY-MM-DD)
            log: If True, enables detailed performance logs

        Usage example:
        >>> dexcom = Dexcom("dexcom_data.csv")
        >>> print(dexcom.info())
        """
        super().__init__(
            data_source=file_path,
            date_col="Marca temporal (AAAA-MM-DDThh:mm:ss)",
            glucose_col="Nivel de glucosa (mg/dL)",
            start_date=start_date,
            end_date=end_date,
            log=log,
        )

    def __str__(self) -> str:
        """Custom representation for Dexcom."""
        return _device_str("Dexcom", self.info())

__init__

__init__(file_path: str, start_date: str | datetime | None = None, end_date: str | datetime | None = None, log: bool = False)

Initializes Dexcom data.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the exported Clarity CSV file	required
start_date	str | datetime | None	Optional start date filter (YYYY-MM-DD)	None
end_date	str | datetime | None	Optional end date filter (YYYY-MM-DD)	None
log	bool	If True, enables detailed performance logs	False

Usage example:

dexcom = Dexcom("dexcom_data.csv") print(dexcom.info())

Source code in cgmpy/data/specialized.py

def __init__(
    self,
    file_path: str,
    start_date: str | datetime.datetime | None = None,
    end_date: str | datetime.datetime | None = None,
    log: bool = False,
):
    """
    Initializes Dexcom data.

    Args:
        file_path: Path to the exported Clarity CSV file
        start_date: Optional start date filter (YYYY-MM-DD)
        end_date: Optional end date filter (YYYY-MM-DD)
        log: If True, enables detailed performance logs

    Usage example:
    >>> dexcom = Dexcom("dexcom_data.csv")
    >>> print(dexcom.info())
    """
    super().__init__(
        data_source=file_path,
        date_col="Marca temporal (AAAA-MM-DDThh:mm:ss)",
        glucose_col="Nivel de glucosa (mg/dL)",
        start_date=start_date,
        end_date=end_date,
        log=log,
    )

__str__

__str__() -> str

Custom representation for Dexcom.

Source code in cgmpy/data/specialized.py

def __str__(self) -> str:
    """Custom representation for Dexcom."""
    return _device_str("Dexcom", self.info())

cgmpy.data.specialized.Libreview

Bases: GlucoseData

Specialized class for Libreview device data.

This class inherits from GlucoseData and automatically configures the specific column names for files exported from Libreview.

Source code in cgmpy/data/specialized.py

class Libreview(GlucoseData):
    """
    Specialized class for Libreview device data.

    This class inherits from GlucoseData and automatically configures
    the specific column names for files exported from Libreview.
    """

    def __init__(
        self,
        file_path: str,
        header: int = 2,
        start_date: str | datetime.datetime | None = None,
        end_date: str | datetime.datetime | None = None,
        log: bool = False,
    ):
        """
        Initializes Libreview data.

        Args:
            file_path: Path to the exported Libreview CSV file
            header: Header row (usually 2 for Libreview)
            start_date: Optional start date filter (YYYY-MM-DD)
            end_date: Optional end date filter (YYYY-MM-DD)
            log: If True, enables detailed performance logs

        Usage example:
        >>> libreview = Libreview("libreview_data.csv")
        >>> print(libreview.info())
        """
        super().__init__(
            data_source=file_path,
            date_col="Sello de tiempo del dispositivo",
            glucose_col="Historial de glucosa mg/dL",
            header=header,
            start_date=start_date,
            end_date=end_date,
            log=log,
        )

    def __str__(self) -> str:
        """Custom representation for Libreview."""
        return _device_str("Libreview", self.info())

__init__

__init__(file_path: str, header: int = 2, start_date: str | datetime | None = None, end_date: str | datetime | None = None, log: bool = False)

Initializes Libreview data.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the exported Libreview CSV file	required
header	int	Header row (usually 2 for Libreview)	2
start_date	str | datetime | None	Optional start date filter (YYYY-MM-DD)	None
end_date	str | datetime | None	Optional end date filter (YYYY-MM-DD)	None
log	bool	If True, enables detailed performance logs	False

Usage example:

libreview = Libreview("libreview_data.csv") print(libreview.info())

Source code in cgmpy/data/specialized.py

def __init__(
    self,
    file_path: str,
    header: int = 2,
    start_date: str | datetime.datetime | None = None,
    end_date: str | datetime.datetime | None = None,
    log: bool = False,
):
    """
    Initializes Libreview data.

    Args:
        file_path: Path to the exported Libreview CSV file
        header: Header row (usually 2 for Libreview)
        start_date: Optional start date filter (YYYY-MM-DD)
        end_date: Optional end date filter (YYYY-MM-DD)
        log: If True, enables detailed performance logs

    Usage example:
    >>> libreview = Libreview("libreview_data.csv")
    >>> print(libreview.info())
    """
    super().__init__(
        data_source=file_path,
        date_col="Sello de tiempo del dispositivo",
        glucose_col="Historial de glucosa mg/dL",
        header=header,
        start_date=start_date,
        end_date=end_date,
        log=log,
    )

__str__

__str__() -> str

Custom representation for Libreview.

Source code in cgmpy/data/specialized.py

def __str__(self) -> str:
    """Custom representation for Libreview."""
    return _device_str("Libreview", self.info())

cgmpy.data.specialized.MedtronicCarelink

Bases: GlucoseData

Specialized class for Medtronic CareLink device data.

This class inherits from GlucoseData and automatically configures the specific column names for files exported from CareLink.

Source code in cgmpy/data/specialized.py

class MedtronicCarelink(GlucoseData):
    """
    Specialized class for Medtronic CareLink device data.

    This class inherits from GlucoseData and automatically configures
    the specific column names for files exported from CareLink.
    """

    def __init__(
        self,
        file_path: str,
        start_date: str | datetime.datetime | None = None,
        end_date: str | datetime.datetime | None = None,
        log: bool = False,
    ):
        """
        Initializes Medtronic CareLink data.

        Args:
            file_path: Path to the exported CareLink CSV file
            start_date: Optional start date filter (YYYY-MM-DD)
            end_date: Optional end date filter (YYYY-MM-DD)
            log: If True, enables detailed performance logs

        Usage example:
        >>> carelink = MedtronicCarelink("carelink_data.csv")
        >>> print(carelink.info())
        """
        super().__init__(
            data_source=file_path,
            date_col="Fecha y hora",
            glucose_col="Valor del sensor (mg/dL)",
            start_date=start_date,
            end_date=end_date,
            log=log,
        )

    def __str__(self) -> str:
        """Custom representation for Medtronic CareLink."""
        return _device_str("MedtronicCarelink", self.info())

__init__

__init__(file_path: str, start_date: str | datetime | None = None, end_date: str | datetime | None = None, log: bool = False)

Initializes Medtronic CareLink data.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the exported CareLink CSV file	required
start_date	str | datetime | None	Optional start date filter (YYYY-MM-DD)	None
end_date	str | datetime | None	Optional end date filter (YYYY-MM-DD)	None
log	bool	If True, enables detailed performance logs	False

Usage example:

carelink = MedtronicCarelink("carelink_data.csv") print(carelink.info())

Source code in cgmpy/data/specialized.py

def __init__(
    self,
    file_path: str,
    start_date: str | datetime.datetime | None = None,
    end_date: str | datetime.datetime | None = None,
    log: bool = False,
):
    """
    Initializes Medtronic CareLink data.

    Args:
        file_path: Path to the exported CareLink CSV file
        start_date: Optional start date filter (YYYY-MM-DD)
        end_date: Optional end date filter (YYYY-MM-DD)
        log: If True, enables detailed performance logs

    Usage example:
    >>> carelink = MedtronicCarelink("carelink_data.csv")
    >>> print(carelink.info())
    """
    super().__init__(
        data_source=file_path,
        date_col="Fecha y hora",
        glucose_col="Valor del sensor (mg/dL)",
        start_date=start_date,
        end_date=end_date,
        log=log,
    )

__str__

__str__() -> str

Custom representation for Medtronic CareLink.

Source code in cgmpy/data/specialized.py

def __str__(self) -> str:
    """Custom representation for Medtronic CareLink."""
    return _device_str("MedtronicCarelink", self.info())

cgmpy.data.specialized.TandemDiabetes

Bases: GlucoseData

Specialized class for Tandem Diabetes device data.

This class inherits from GlucoseData and automatically configures the specific column names for files exported from Tandem.

Source code in cgmpy/data/specialized.py

class TandemDiabetes(GlucoseData):
    """
    Specialized class for Tandem Diabetes device data.

    This class inherits from GlucoseData and automatically configures
    the specific column names for files exported from Tandem.
    """

    def __init__(
        self,
        file_path: str,
        start_date: str | datetime.datetime | None = None,
        end_date: str | datetime.datetime | None = None,
        log: bool = False,
    ):
        """
        Initializes Tandem Diabetes data.

        Args:
            file_path: Path to the exported Tandem CSV file
            start_date: Optional start date filter (YYYY-MM-DD)
            end_date: Optional end date filter (YYYY-MM-DD)
            log: If True, enables detailed performance logs

        Usage example:
        >>> tandem = TandemDiabetes("tandem_data.csv")
        >>> print(tandem.info())
        """
        super().__init__(
            data_source=file_path,
            date_col="Timestamp",
            glucose_col="CGM Glucose Value (mg/dL)",
            start_date=start_date,
            end_date=end_date,
            log=log,
        )

    def __str__(self) -> str:
        """Custom representation for Tandem Diabetes."""
        return _device_str("TandemDiabetes", self.info())

__init__

__init__(file_path: str, start_date: str | datetime | None = None, end_date: str | datetime | None = None, log: bool = False)

Initializes Tandem Diabetes data.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the exported Tandem CSV file	required
start_date	str | datetime | None	Optional start date filter (YYYY-MM-DD)	None
end_date	str | datetime | None	Optional end date filter (YYYY-MM-DD)	None
log	bool	If True, enables detailed performance logs	False

Usage example:

tandem = TandemDiabetes("tandem_data.csv") print(tandem.info())

Source code in cgmpy/data/specialized.py

def __init__(
    self,
    file_path: str,
    start_date: str | datetime.datetime | None = None,
    end_date: str | datetime.datetime | None = None,
    log: bool = False,
):
    """
    Initializes Tandem Diabetes data.

    Args:
        file_path: Path to the exported Tandem CSV file
        start_date: Optional start date filter (YYYY-MM-DD)
        end_date: Optional end date filter (YYYY-MM-DD)
        log: If True, enables detailed performance logs

    Usage example:
    >>> tandem = TandemDiabetes("tandem_data.csv")
    >>> print(tandem.info())
    """
    super().__init__(
        data_source=file_path,
        date_col="Timestamp",
        glucose_col="CGM Glucose Value (mg/dL)",
        start_date=start_date,
        end_date=end_date,
        log=log,
    )

__str__

__str__() -> str

Custom representation for Tandem Diabetes.

Source code in cgmpy/data/specialized.py

def __str__(self) -> str:
    """Custom representation for Tandem Diabetes."""
    return _device_str("TandemDiabetes", self.info())

Exceptions

All data-loading errors raise a subclass of cgmpy.errors.DataError, which inherits from both CGMPyError and ValueError. Catch CGMPyError to handle any CGMPy-raised error. Common ones:

• ColumnNotFoundError — required column missing (has .column, .available)
• InvalidCSVFormatError — CSV cannot be parsed (has .file_path, .reason)
• DeviceDetectionError — auto-detect failed (has .file_path, .columns_found)
• GlucoseRangeError — values outside 39-600 mg/dL (has .n_invalid, .total)
• EmptyDataError — no rows left after filtering

For the full hierarchy (including MetricError, AgataIntegrationError, AgataNotInstalledError, ConfigurationError), see cgmpy.errors in the source tree.

Pregnancy

cgmpy.data.pregnancy_data.PregnancyData

Bases: GlucoseData

Class for managing pregnancy-specific data and dates. Inherits from GlucoseData to integrate base loading and processing, adding trimester segmentation.

Source code in cgmpy/data/pregnancy_data.py

class PregnancyData(GlucoseData):
    """
    Class for managing pregnancy-specific data and dates.
    Inherits from GlucoseData to integrate base loading and processing,
    adding trimester segmentation.
    """

    def __init__(
        self,
        data_source: str | pd.DataFrame,
        delivery_date: str,
        week: int,
        day: int = 0,
        date_col: str = "time",
        glucose_col: str = "glucose",
        delimiter: str | None = None,
        header: int = 0,
        start_date: str | datetime.datetime | None = None,
        end_date: str | datetime.datetime | None = None,
        log: bool = False,
        target_type: str = "pregnancy",
        regularize: bool = False,
        regularize_interval: int = 5,
        regularize_max_gap: int = 30,
    ):
        """
        Initializes pregnancy data.
        """
        # Base initialization of GlucoseData
        super().__init__(
            data_source=data_source,
            date_col=date_col,
            glucose_col=glucose_col,
            delimiter=delimiter,
            header=header,
            start_date=start_date,
            end_date=end_date,
            log=log,
            target_type=target_type,
            regularize=regularize,
            regularize_interval=regularize_interval,
            regularize_max_gap=regularize_max_gap,
        )

        # Specific pregnancy configuration
        self.gestational_info = self.calculate_dates(delivery_date, week, day)

        # 1. Attribute mapping for easy access
        self.delivery_date = self.gestational_info["delivery_date"]
        self.conception_date = self.gestational_info["conception_date"]
        self.gestation_week = self.gestational_info["gestation_week_decimal"]
        self.first_trimester_end = self.gestational_info["first_trimester_end"]
        self.second_trimester_end = self.gestational_info["second_trimester_end"]

        # 2. Filter the main dataframe to ensure "Overall" metrics only cover the pregnancy period
        mask = (self.data["time"] >= self.conception_date) & (
            self.data["time"] <= self.delivery_date
        )
        self.data = self.data[mask].copy()

        # 3. Recalculate time differences and typical interval after filtering
        self.time_diffs = self.data["time"].diff()
        self.typical_interval = self.analyzer.calculate_typical_interval(self.time_diffs)

        # 4. Creation of dataframes by trimester
        self.trimesters = self._split_trimesters()

    @staticmethod
    def calculate_dates(delivery_date: str, week: int, day: int = 0) -> dict:
        """
        Calculates key pregnancy dates.
        """
        delivery_date_dt = pd.to_datetime(delivery_date)
        if pd.isna(delivery_date_dt):
            raise ValueError("Invalid delivery date")

        gestation_week = week + (day / 7)
        conception_date = delivery_date_dt - pd.Timedelta(weeks=gestation_week)

        if pd.isna(conception_date):
            raise ValueError("Invalid conception date")

        first_trimester_end = conception_date + pd.Timedelta(weeks=13, days=6)
        second_trimester_end = conception_date + pd.Timedelta(weeks=27, days=6)

        return {
            "delivery_date": delivery_date_dt,
            "conception_date": conception_date,
            "first_trimester_end": first_trimester_end,
            "second_trimester_end": second_trimester_end,
            "gestation_week_decimal": gestation_week,
        }

    def _split_trimesters(self) -> dict[str, pd.DataFrame]:
        """
        Splits the main dataframe into three dataframes, one per trimester.
        """
        return {
            "first_trimester": self.get_trimester_data(
                self.conception_date, self.first_trimester_end
            ),
            "second_trimester": self.get_trimester_data(
                self.first_trimester_end, self.second_trimester_end
            ),
            "third_trimester": self.get_trimester_data(
                self.second_trimester_end, self.delivery_date
            ),
        }

    def get_trimester_data(self, start_date: pd.Timestamp, end_date: pd.Timestamp) -> pd.DataFrame:
        """
        Filters the DataFrame for a date range.
        """
        return self.data[(self.data["time"] >= start_date) & (self.data["time"] < end_date)].copy()

    @staticmethod
    def decimal_to_weeks_days(weeks_decimal: float) -> tuple[int, int]:
        """
        Converts decimal weeks to (weeks, days).
        """
        weeks = int(weeks_decimal)
        days = round((weeks_decimal - weeks) * 7)
        return weeks, days

    def get_weeks_days(self) -> tuple[int, int]:
        """
        Returns gestation weeks and days in traditional format.
        """
        return self.decimal_to_weeks_days(self.gestation_week)

    def __str__(self) -> str:
        """
        Overwrites the string representation to include pregnancy-specific information.
        """
        weeks, days = self.get_weeks_days()

        # Trimester definitions for calculation
        periods = [
            ("First Trimester", self.conception_date, self.first_trimester_end, "first_trimester"),
            (
                "Second Trimester",
                self.first_trimester_end,
                self.second_trimester_end,
                "second_trimester",
            ),
            ("Third Trimester", self.second_trimester_end, self.delivery_date, "third_trimester"),
        ]

        trimester_details = []
        for label, start, end, key in periods:
            count = len(self.trimesters[key])

            # Calculate expected records
            duration_mins = (end - start).total_seconds() / 60

            if pd.isna(duration_mins) or self.typical_interval <= 0:
                expected = 0
            else:
                expected = int(duration_mins / self.typical_interval)

            coverage = (count / expected * 100) if expected > 0 else 0

            trimester_details.append(f"  * {label}: {count} records ({coverage:.1f}% coverage)")

        # Base summary from GlucoseData
        base_summary = super().__str__()

        summary = [
            "Pregnancy Data Summary",
            "=====================",
            f"Gestation: {weeks} weeks and {days} days",
            f"Conception Date: {self.conception_date.strftime('%Y-%m-%d')}",
            f"Estimated Delivery: {self.delivery_date.strftime('%Y-%m-%d')}",
            "",
            "Records by Trimester:",
            *trimester_details,
            "",
            "General Data Info:",
            "-----------------",
            base_summary,
        ]

        return "\n".join(summary)

__init__

__init__(data_source: str | DataFrame, delivery_date: str, week: int, day: int = 0, date_col: str = 'time', glucose_col: str = 'glucose', delimiter: str | None = None, header: int = 0, start_date: str | datetime | None = None, end_date: str | datetime | None = None, log: bool = False, target_type: str = 'pregnancy', regularize: bool = False, regularize_interval: int = 5, regularize_max_gap: int = 30)

Initializes pregnancy data.

Source code in cgmpy/data/pregnancy_data.py

def __init__(
    self,
    data_source: str | pd.DataFrame,
    delivery_date: str,
    week: int,
    day: int = 0,
    date_col: str = "time",
    glucose_col: str = "glucose",
    delimiter: str | None = None,
    header: int = 0,
    start_date: str | datetime.datetime | None = None,
    end_date: str | datetime.datetime | None = None,
    log: bool = False,
    target_type: str = "pregnancy",
    regularize: bool = False,
    regularize_interval: int = 5,
    regularize_max_gap: int = 30,
):
    """
    Initializes pregnancy data.
    """
    # Base initialization of GlucoseData
    super().__init__(
        data_source=data_source,
        date_col=date_col,
        glucose_col=glucose_col,
        delimiter=delimiter,
        header=header,
        start_date=start_date,
        end_date=end_date,
        log=log,
        target_type=target_type,
        regularize=regularize,
        regularize_interval=regularize_interval,
        regularize_max_gap=regularize_max_gap,
    )

    # Specific pregnancy configuration
    self.gestational_info = self.calculate_dates(delivery_date, week, day)

    # 1. Attribute mapping for easy access
    self.delivery_date = self.gestational_info["delivery_date"]
    self.conception_date = self.gestational_info["conception_date"]
    self.gestation_week = self.gestational_info["gestation_week_decimal"]
    self.first_trimester_end = self.gestational_info["first_trimester_end"]
    self.second_trimester_end = self.gestational_info["second_trimester_end"]

    # 2. Filter the main dataframe to ensure "Overall" metrics only cover the pregnancy period
    mask = (self.data["time"] >= self.conception_date) & (
        self.data["time"] <= self.delivery_date
    )
    self.data = self.data[mask].copy()

    # 3. Recalculate time differences and typical interval after filtering
    self.time_diffs = self.data["time"].diff()
    self.typical_interval = self.analyzer.calculate_typical_interval(self.time_diffs)

    # 4. Creation of dataframes by trimester
    self.trimesters = self._split_trimesters()

calculate_dates staticmethod

calculate_dates(delivery_date: str, week: int, day: int = 0) -> dict

Calculates key pregnancy dates.

Source code in cgmpy/data/pregnancy_data.py

@staticmethod
def calculate_dates(delivery_date: str, week: int, day: int = 0) -> dict:
    """
    Calculates key pregnancy dates.
    """
    delivery_date_dt = pd.to_datetime(delivery_date)
    if pd.isna(delivery_date_dt):
        raise ValueError("Invalid delivery date")

    gestation_week = week + (day / 7)
    conception_date = delivery_date_dt - pd.Timedelta(weeks=gestation_week)

    if pd.isna(conception_date):
        raise ValueError("Invalid conception date")

    first_trimester_end = conception_date + pd.Timedelta(weeks=13, days=6)
    second_trimester_end = conception_date + pd.Timedelta(weeks=27, days=6)

    return {
        "delivery_date": delivery_date_dt,
        "conception_date": conception_date,
        "first_trimester_end": first_trimester_end,
        "second_trimester_end": second_trimester_end,
        "gestation_week_decimal": gestation_week,
    }

get_trimester_data

get_trimester_data(start_date: Timestamp, end_date: Timestamp) -> DataFrame

Filters the DataFrame for a date range.

Source code in cgmpy/data/pregnancy_data.py

def get_trimester_data(self, start_date: pd.Timestamp, end_date: pd.Timestamp) -> pd.DataFrame:
    """
    Filters the DataFrame for a date range.
    """
    return self.data[(self.data["time"] >= start_date) & (self.data["time"] < end_date)].copy()

decimal_to_weeks_days staticmethod

decimal_to_weeks_days(weeks_decimal: float) -> tuple[int, int]

Converts decimal weeks to (weeks, days).

Source code in cgmpy/data/pregnancy_data.py

@staticmethod
def decimal_to_weeks_days(weeks_decimal: float) -> tuple[int, int]:
    """
    Converts decimal weeks to (weeks, days).
    """
    weeks = int(weeks_decimal)
    days = round((weeks_decimal - weeks) * 7)
    return weeks, days

get_weeks_days

get_weeks_days() -> tuple[int, int]

Returns gestation weeks and days in traditional format.

Source code in cgmpy/data/pregnancy_data.py

def get_weeks_days(self) -> tuple[int, int]:
    """
    Returns gestation weeks and days in traditional format.
    """
    return self.decimal_to_weeks_days(self.gestation_week)

__str__

__str__() -> str

Overwrites the string representation to include pregnancy-specific information.

Source code in cgmpy/data/pregnancy_data.py

def __str__(self) -> str:
    """
    Overwrites the string representation to include pregnancy-specific information.
    """
    weeks, days = self.get_weeks_days()

    # Trimester definitions for calculation
    periods = [
        ("First Trimester", self.conception_date, self.first_trimester_end, "first_trimester"),
        (
            "Second Trimester",
            self.first_trimester_end,
            self.second_trimester_end,
            "second_trimester",
        ),
        ("Third Trimester", self.second_trimester_end, self.delivery_date, "third_trimester"),
    ]

    trimester_details = []
    for label, start, end, key in periods:
        count = len(self.trimesters[key])

        # Calculate expected records
        duration_mins = (end - start).total_seconds() / 60

        if pd.isna(duration_mins) or self.typical_interval <= 0:
            expected = 0
        else:
            expected = int(duration_mins / self.typical_interval)

        coverage = (count / expected * 100) if expected > 0 else 0

        trimester_details.append(f"  * {label}: {count} records ({coverage:.1f}% coverage)")

    # Base summary from GlucoseData
    base_summary = super().__str__()

    summary = [
        "Pregnancy Data Summary",
        "=====================",
        f"Gestation: {weeks} weeks and {days} days",
        f"Conception Date: {self.conception_date.strftime('%Y-%m-%d')}",
        f"Estimated Delivery: {self.delivery_date.strftime('%Y-%m-%d')}",
        "",
        "Records by Trimester:",
        *trimester_details,
        "",
        "General Data Info:",
        "-----------------",
        base_summary,
    ]

    return "\n".join(summary)

cgmpy.data.pregnancy_data.PregnancyDataHandler module-attribute

PregnancyDataHandler = PregnancyData