cpap-py 1.0.0

A comprehensive Python library for parsing and analyzing ResMed CPAP machine data

cpap-py

A lightweight Python library for parsing ResMed CPAP (Continuous Positive Airway Pressure) device data files. This library provides complete access to all data stored on CPAP devices including device identification, summary statistics, detailed waveform data, and configuration changes.

Features

• Zero Dependencies: Pure Python implementation using only the standard library
• Complete Data Extraction: Parses all available CPAP data including pressure settings, delivered pressures, leak rates, respiratory metrics, and event indices
• Device Identification: Parse both .tgt (text) and .json format identification files
• Summary Data: Extract daily statistics from STR.edf files including AHI, leak, pressure, respiratory rate
• Session Data: Parse detailed waveform data from DATALOG EDF files (BRP, PLD, SAD, EVE, CSL, AEV)
• Settings & Configuration: Extract device settings including pressure ranges (min/max), comfort settings, humidification
• Pure Python EDF Parser: Full implementation of European Data Format (EDF/EDF+) parser
• Multiple Device Support: Works with ResMed S9, AirSense 10, AirSense 11, and AirCurve series
• BiLevel Support: Full support for BiLevel therapy modes (S, ST, T, PAC, ASV)
• Comprehensive Testing: 97% code coverage with 188 automated tests
• Python 3.9+: Supports Python 3.9, 3.10, 3.11, and 3.12

Installation

pip install cpap-py

For development installation with test dependencies:

git clone https://github.com/dynacylabs/cpap-py.git
cd cpap-py
pip install -e ".[dev]"

See INSTALL.md for detailed installation instructions.

Quick Start

Load All CPAP Data

from cpap_py import CPAPLoader

# Load all data from a CPAP data directory
loader = CPAPLoader("path/to/cpap/data")
data = loader.load_all()

# Access device information
print(f"Device: {data.machine_info.model}")
print(f"Serial: {data.machine_info.serial}")
print(f"Model Series: {data.machine_info.series}")

# Access daily summary records
for record in data.summary_records:
    print(f"Date: {record.date}")
    print(f"  AHI: {record.ahi:.1f} events/hour")
    print(f"  Duration: {record.mask_duration/3600:.1f} hours")
    print(f"  Leak (median): {record.leak_50:.1f} L/min")
    print(f"  Pressure (95th): {record.mp_95:.1f} cmH2O")

# Access detailed session data
for session in data.sessions:
    print(f"Session: {session.start_time}")
    print(f"  Duration: {session.duration/3600:.1f} hours")
    print(f"  Flow rate samples: {len(session.flow_rate)}")
    print(f"  Pressure samples: {len(session.pressure)}")
    print(f"  Events recorded: {len(session.events)}")
    print(f"  File type: {session.file_type}")

Usage Examples

Parse Device Identification

from cpap_py import IdentificationParser

parser = IdentificationParser("path/to/data")
info = parser.parse()

if info:
    print(f"Model: {info.model}")
    print(f"Serial: {info.serial}")
    print(f"Series: {info.series}")
    print(f"Model Number: {info.model_number}")

Parse Summary Data (STR.edf)

from cpap_py import STRParser

parser = STRParser("path/to/STR.edf")
if parser.parse():
    for record in parser.records:
        if record.date:
            print(f"{record.date}: AHI={record.ahi:.1f}, Hours={record.mask_duration/3600:.1f}")
            print(f"  Leak (median/95th): {record.leak_50:.1f}/{record.leak_95:.1f} L/min")
            print(f"  Pressure (median/95th): {record.mp_50:.1f}/{record.mp_95:.1f} cmH2O")
            
    # Filter by date range
    from datetime import date
    filtered = parser.get_records_by_date_range(
        date(2025, 12, 1),
        date(2025, 12, 15)
    )

Parse Session Data (DATALOG)

from cpap_py import DatalogParser

parser = DatalogParser("path/to/DATALOG")

# Parse all sessions
sessions = parser.parse_all_sessions()

for session in sessions:
    print(f"{session.date} - {session.file_type}")
    print(f"  Start: {session.start_time}")
    print(f"  Duration: {session.duration/3600:.2f} hours")
    print(f"  Sample rate: {session.sample_rate} Hz")
    
    # Access waveform data
    if session.flow_rate:
        print(f"  Flow rate: {len(session.flow_rate)} samples")
    if session.pressure:
        print(f"  Pressure: {len(session.pressure)} samples")
    if session.spo2:
        print(f"  SpO2: {len(session.spo2)} samples")
        
    # Access events
    for event in session.events:
        print(f"  Event: {event.event_type} at {event.timestamp}s")

# Get sessions for specific date
from datetime import date
date_sessions = parser.get_sessions_by_date(date(2025, 12, 15))

Parse Settings Files

from cpap_py import SettingsParser

parser = SettingsParser("path/to/SETTINGS")
changes = parser.parse_all()

for change in changes:
    print(f"{change.timestamp}: {change.setting} = {change.value}")

Parse Individual EDF Files

from cpap_py import EDFParser

edf = EDFParser("path/to/file.edf")
if edf.parse():
    # Access header information
    print(f"Recording date: {edf.header.start_date}")
    print(f"Duration: {edf.header.duration} seconds")
    print(f"Signals: {len(edf.signals)}")
    
    # Access signals
    for signal in edf.signals:
        print(f"  {signal.label}: {len(signal.data)} samples")
        print(f"    Unit: {signal.physical_dimension}")
        print(f"    Range: {signal.physical_min} to {signal.physical_max}")
        
    # Find specific signal
    flow_signal = edf.get_signal("Flow")
    if flow_signal:
        # Get physical (scaled) values
        physical_values = edf.get_physical_values(flow_signal)
        print(f"Flow range: {min(physical_values):.1f} to {max(physical_values):.1f} L/min")

Data Directory Structure

The library expects data in the ResMed CPAP format:

data_directory/
├── Identification.tgt     # or Identification.json
├── STR.edf               # Daily summary data
├── DATALOG/
│   ├── 20251126/        # Date folders (YYYYMMDD)
│   │   ├── BRP00001.edf # Breathing data
│   │   ├── PLD00001.edf # Pressure/leak data
│   │   └── EVE00001.edf # Events
│   └── 20251127/
└── SETTINGS/
    ├── CGL.tgt          # Clinical settings
    └── UGL.tgt          # User settings

File Types

Identification Files

• .tgt: Text format with #KEY value pairs
• .json: JSON format (AirSense 11)

EDF Files

• STR.edf: Daily summary statistics
• BRP.edf: Breathing waveforms (flow, tidal volume, etc.)
• PLD.edf: Pressure and leak data
• SAD.edf: Summary/advanced data
• EVE.edf: Event markers (apneas, hypopneas)
• CSL.edf: Clinical settings log
• AEV.edf: Advanced events

API Reference

CPAPLoader

High-level interface for loading all CPAP data.

loader = CPAPLoader(data_path: str)

Methods:

• load_all() → CPAPData: Load all data (identification, summary, sessions, settings)
• load_identification_only() → MachineInfo | None: Load only device identification
• load_summary_only() → List[STRRecord]: Load only STR.edf summary data
• load_sessions_for_date(date) → List[SessionData]: Load sessions for specific date
• get_date_range() → tuple[date, date] | None: Get (start_date, end_date) of available data

IdentificationParser

Parse device identification files (.tgt or .json).

parser = IdentificationParser(data_path: str)

Methods:

• parse() → MachineInfo | None: Parse identification file and return device info

Supported Files:

• Identification.tgt - Text format with #KEY value pairs
• Identification.json - JSON format (AirSense 11)

STRParser

Parse STR.edf daily summary files.

parser = STRParser(filepath: str, serial_number: str = None)

Methods:

• parse() → bool: Parse file and populate records list
• get_records_by_date_range(start: date, end: date) → List[STRRecord]: Filter records by date

Mode Constants:

• MODE_UNKNOWN = 0
• MODE_CPAP = 1
• MODE_APAP = 2
• MODE_BILEVEL_FIXED = 3
• MODE_BILEVEL_AUTO = 4
• MODE_BILEVEL_S = 5
• MODE_BILEVEL_ST = 6
• MODE_BILEVEL_T = 7
• MODE_BILEVEL_PAC = 8
• MODE_ASV = 9

DatalogParser

Parse DATALOG session files (BRP, PLD, SAD, EVE, CSL, AEV).

parser = DatalogParser(datalog_path: str)

Methods:

• scan_files() → Dict[date, List[Path]]: Scan directory and return files by date
• parse_session_file(filepath: str) → SessionData | None: Parse single session file
• parse_all_sessions() → List[SessionData]: Parse all session files in directory
• get_sessions_by_date(target_date: date) → List[SessionData]: Get sessions for specific date
• get_sessions_by_date_range(start: date, end: date) → List[SessionData]: Get sessions in range

File Types:

• BRP - Breathing waveforms (flow, tidal volume, minute ventilation, respiratory rate)
• PLD - Pressure and leak data
• SAD - Summary/advanced data
• EVE - Event markers (apneas, hypopneas, flow limitations)
• CSL - Clinical settings log
• AEV - Advanced events

SettingsParser

Parse SETTINGS files (.tgt format).

parser = SettingsParser(settings_path: str)

Methods:

• parse_all() → List[SettingChange]: Parse all settings files
• parse_file(filepath: str) → List[SettingChange]: Parse single settings file

EDFParser

Low-level European Data Format (EDF/EDF+) parser.

parser = EDFParser(filepath: str)

Methods:

• open() → bool: Open EDF file
• close(): Close file
• parse() → bool: Parse entire file (header + signals + data)
• parse_header() → bool: Parse only file header
• parse_signal_headers() → bool: Parse signal definitions
• parse_data() → bool: Parse signal data
• get_signal(label: str, index: int = 0) → EDFSignal | None: Find signal by label
• get_physical_values(signal: EDFSignal) → List[float]: Convert digital to physical values

Data Classes

CPAPData

Complete CPAP data container.

Fields:

• machine_info: MachineInfo | None - Device information
• summary_records: List[STRRecord] - Daily summary statistics
• sessions: List[SessionData] - Detailed session waveform data
• settings_changes: List[SettingChange] - Configuration changes

MachineInfo

Device identification information.

Fields:

• serial: str - Device serial number
• model: str - Model name (e.g., "AirSense 10 AutoSet")
• model_number: str - Model number
• series: str - Device series (e.g., "AirSense 11", "AirSense 10", "S9")
• properties: Dict[str, str] - Additional device properties

STRRecord

Daily summary record from STR.edf.

Key Fields:

• date: date - Record date
• ahi: float - Apnea-Hypopnea Index (events/hour)
• ai: float - Apnea Index
• hi: float - Hypopnea Index
• cai: float - Central Apnea Index
• oai: float - Obstructive Apnea Index
• mask_duration: float - Therapy duration (seconds)
• mask_on: List[int] - Mask-on timestamps
• mask_off: List[int] - Mask-off timestamps
• leak_50: float - Median leak (L/min)
• leak_95: float - 95th percentile leak (L/min)
• mp_50: float - Median mask pressure (cmH2O)
• mp_95: float - 95th percentile mask pressure (cmH2O)
• mode: int - Therapy mode
• min_pressure: float - Minimum pressure setting
• max_pressure: float - Maximum pressure setting
• ipap: float - IPAP setting (BiLevel)
• epap: float - EPAP setting (BiLevel)
• epr: int - EPR setting
• And 70+ more statistics fields...

SessionData

Detailed session data from DATALOG files.

Fields:

• date: date - Session date
• start_time: datetime - Session start timestamp
• duration: float - Session duration (seconds)
• file_type: str - File type (BRP, PLD, SAD, EVE, etc.)
• sample_rate: float - Sample rate (Hz)
• flow_rate: List[float] - Flow rate waveform (L/min)
• pressure: List[float] - Pressure waveform (cmH2O)
• mask_pressure: List[float] - Mask pressure (cmH2O)
• leak: List[float] - Leak rate (L/min)
• tidal_volume: List[float] - Tidal volume (mL)
• minute_vent: List[float] - Minute ventilation (L/min)
• resp_rate: List[float] - Respiratory rate (breaths/min)
• target_ipap: List[float] - Target IPAP (cmH2O)
• target_epap: List[float] - Target EPAP (cmH2O)
• spo2: List[float] - SpO2 (%)
• pulse: List[float] - Pulse rate (bpm)
• events: List[SessionEvent] - Respiratory events

SessionEvent

Event recorded during CPAP session.

Fields:

• timestamp: float - Time since session start (seconds)
• event_type: str - Event type (e.g., "Obstructive Apnea", "Hypopnea")
• duration: float - Event duration (seconds)
• data: Dict[str, float] - Additional event data

SettingChange

Device setting change record.

Fields:

• timestamp: datetime - When setting was changed
• setting: str - Setting name
• value: str - New value
• source: str - Source file

EDFHeader

EDF file header information.

Fields:

• version: str - EDF version
• patient_id: str - Patient identifier
• recording_id: str - Recording identifier
• start_date: datetime - Recording start date/time
• header_bytes: int - Header size
• reserved: str - Reserved field (may contain "EDF+C" for EDF+)
• num_records: int - Number of data records
• duration: float - Duration of data record (seconds)
• num_signals: int - Number of signals

EDFSignal

EDF signal descriptor.

Fields:

• label: str - Signal label
• transducer_type: str - Transducer type
• physical_dimension: str - Physical unit (e.g., "L/min", "cmH2O")
• physical_min: float - Physical minimum
• physical_max: float - Physical maximum
• digital_min: int - Digital minimum
• digital_max: int - Digital maximum
• prefiltering: str - Prefiltering info
• sample_count: int - Samples per data record
• reserved: str - Reserved field
• data: List[int] - Raw digital samples
• gain: float - Calculated gain for conversion
• offset: float - Calculated offset for conversion

Project Structure

cpap-py/
├── src/
│   └── cpap_py/                    # Main library package
│       ├── __init__.py             # Package initialization and exports
│       ├── edf_parser.py           # EDF/EDF+ file parser (pure Python)
│       ├── identification.py       # Device identification parser
│       ├── str_parser.py           # STR.edf summary data parser
│       ├── datalog_parser.py       # DATALOG session data parser
│       ├── settings_parser.py      # Settings file parser
│       ├── loader.py               # High-level data loader
│       └── utils.py                # Utility functions
├── tests/                          # Comprehensive test suite (97% coverage)
│   ├── conftest.py                 # Pytest fixtures
│   ├── test_init.py                # Package initialization tests
│   ├── test_identification.py      # ID parser tests
│   ├── test_edf_parser.py          # EDF parser tests
│   ├── test_utils.py               # Utility function tests
│   ├── test_parser_core.py         # Core parser functionality
│   ├── test_integration.py         # Integration tests
│   ├── test_mock_scenarios.py      # Mock-based tests
│   ├── test_realistic_edf_data.py  # Realistic EDF data tests
│   ├── test_signal_combinations.py # Signal combination tests
│   ├── test_bilevel_modes.py       # BiLevel mode tests
│   ├── test_settings_alternative_signals.py  # Alternative signal tests
│   └── test_optional_signals_errors.py       # Error handling tests
├── setup.py                        # Setup configuration
├── pyproject.toml                  # Project metadata and build config
├── requirements.txt                # Python dependencies
├── README.md                       # This file
├── INSTALL.md                      # Installation guide
├── USAGE.md                        # Detailed usage guide
├── DEVELOPMENT.md                  # Development guide
├── CONTRIBUTING.md                 # Contribution guidelines
├── TEST_SUITE.md                   # Test suite documentation
├── TESTING_COMPLETE.md             # Test coverage report
└── LICENSE                         # MIT License

Utility Functions

The utils module provides helper functions for CPAP data analysis:

from cpap_py.utils import (
    split_sessions_by_noon,
    format_duration,
    calculate_ahi,
    therapy_mode_name,
    downsample_signal,
    calculate_percentile
)

# Split session timestamps by noon boundary
sessions = split_sessions_by_noon(timestamps)

# Format duration as HH:MM:SS
duration_str = format_duration(seconds)

# Calculate AHI from event counts and hours
ahi = calculate_ahi(apneas=10, hypopneas=5, hours=7.5)

# Get therapy mode name from mode code
mode_name = therapy_mode_name(mode_code)  # "CPAP", "APAP", "BiLevel S", etc.

# Downsample signal data
downsampled = downsample_signal(data, factor=10)

# Calculate percentile (50th, 95th, etc.)
p95 = calculate_percentile(data, 95)

Testing

The library includes a comprehensive test suite with 97% code coverage and 188 automated tests.

Run Tests

# Run all tests with coverage
./run_tests.sh

# Run specific test file
pytest tests/test_identification.py -v

# Run with detailed coverage
pytest tests/ --cov=cpap_py --cov-report=term-missing

See TEST_SUITE.md for detailed testing documentation.

Documentation

• README.md - This file - Quick start and API reference
• INSTALL.md - Detailed installation instructions
• USAGE.md - Comprehensive usage guide with examples
• DEVELOPMENT.md - Development setup and workflows
• CONTRIBUTING.md - How to contribute to the project
• TEST_SUITE.md - Test suite documentation
• TESTING_COMPLETE.md - Test coverage report

Requirements

• Python: 3.9 or higher
• Dependencies: None (pure Python standard library implementation)
• Development Dependencies (optional):
  • pytest >= 7.0.0
  • pytest-cov >= 4.0.0
  • pytest-mock >= 3.10.0
  • coverage >= 7.0.0
  • black >= 23.0.0 (code formatting)
  • ruff >= 0.1.0 (linting)

Supported Devices

• ResMed AirSense 11 - Latest generation with JSON identification
• ResMed AirSense 10 - AutoSet, Elite, CPAP, For Her
• ResMed AirCurve 10 - S, ST, VAuto, ASV
• ResMed S9 - AutoSet, Elite, CPAP, VPAP series
• Other ResMed devices using EDF format

Supported Data Files

Identification Files

• Identification.tgt - Text format (#KEY value)
• Identification.json - JSON format (AirSense 11)

Summary Files

• STR.edf - Daily summary statistics (AHI, leak, pressure, etc.)

Session Data Files (DATALOG/)

• BRP*.edf - Breathing waveforms (flow rate, tidal volume, minute ventilation, respiratory rate)
• PLD*.edf - Pressure and leak data
• SAD*.edf - Summary/advanced data
• EVE*.edf - Respiratory events (apneas, hypopneas, flow limitations)
• CSL*.edf - Clinical settings log
• AEV*.edf - Advanced events

Settings Files (SETTINGS/)

• *.tgt - Settings files (CGL, UGL, etc.)

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

Development Workflow

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Add tests for new functionality
5. Run the test suite (./run_tests.sh)
6. Commit your changes (git commit -m 'Add amazing feature')
7. Push to the branch (git push origin feature/amazing-feature)
8. Open a Pull Request

License

MIT License - see LICENSE file for details.

Acknowledgments

This library is based on the file format specifications from OSCAR (Open Source CPAP Analysis Reporter), an excellent open-source CPAP data analysis application.

Special thanks to the OSCAR development team for their comprehensive documentation of ResMed file formats.

Support

• Issues: GitHub Issues
• Documentation: GitHub Repository
• Repository: https://github.com/dynacylabs/cpap-py

Related Projects

• OSCAR - Open Source CPAP Analysis Reporter (Qt/C++ desktop application)
• SleepHQ - Online CPAP data analysis platform
• pyedflib - Python library for reading/writing EDF files

Changelog

Version 0.1.0 (Current)

• Initial release
• Complete EDF/EDF+ parser (pure Python)
• Support for all ResMed device generations (S9, AirSense 10, AirSense 11)
• Parse identification files (.tgt, .json)
• Parse STR.edf summary data
• Parse DATALOG session data (BRP, PLD, SAD, EVE, CSL, AEV)
• Parse SETTINGS files
• High-level CPAPLoader interface
• Comprehensive test suite (97% coverage, 188 tests)
• Zero external dependencies
• Python 3.9+ support