""" EDF/EDF+ binary parser for ResMed CPAP data. Key ResMed deviation from EDF+ spec: - EVE files use \x15 (not \x14) to separate onset from duration in event TALs. - Standard: +onset\x14duration\x14annotation\x14 - ResMed: +onset\x15duration\x14annotation\x14 """ import re import struct from dataclasses import dataclass, field from datetime import datetime @dataclass class EDFSignal: label: str dim: str pmin: float pmax: float dmin: float dmax: float num_samples_per_record: int gain: float = field(init=False) offset: float = field(init=False) def __post_init__(self): drange = self.dmax - self.dmin if drange == 0: self.gain = 1.0 self.offset = 0.0 else: self.gain = (self.pmax - self.pmin) / drange self.offset = self.pmin - self.dmin * self.gain def scale(self, raw: int) -> float: return raw * self.gain + self.offset @dataclass class EDFHeader: start_datetime: datetime num_records: int duration_per_record: float device_serial: str signals: list is_edfplus: bool def _parse_header(data: bytes) -> EDFHeader: """Parse the main 256-byte EDF header and signal headers from raw bytes. Args: data: The raw binary data containing the concatenated EDF headers. Returns: An EDFHeader data structure populated with the parsed configuration. """ startdate = data[168:176].decode("ascii").strip() # DD.MM.YY starttime = data[176:184].decode("ascii").strip() # HH.MM.SS recording = data[88:168].decode("ascii") dd, mm, yy = startdate.split(".") hh, mi, ss = starttime.split(".") year = 2000 + int(yy) start_dt = datetime(year, int(mm), int(dd), int(hh), int(mi), int(ss)) srn_match = re.search(r"SRN=(\S+)", recording) device_serial = srn_match.group(1) if srn_match else "" ns = int(data[252:256].strip()) num_records = int(data[236:244].strip()) duration_str = data[244:252].strip().decode("ascii") duration = float(duration_str) if duration_str else 0.0 reserved = data[192:236].decode("ascii").strip() is_edfplus = reserved.startswith("EDF+") # Signal headers: each field stored as ns-length arrays sig_raw = data[256 : 256 + ns * 256] signals = [] for i in range(ns): label = sig_raw[i * 16 : (i + 1) * 16].decode("ascii").strip() dim = sig_raw[ns * 96 + i * 8 : ns * 96 + (i + 1) * 8].decode("ascii").strip() pmin = float(sig_raw[ns * 104 + i * 8 : ns * 104 + (i + 1) * 8].strip()) pmax = float(sig_raw[ns * 112 + i * 8 : ns * 112 + (i + 1) * 8].strip()) dmin = float(sig_raw[ns * 120 + i * 8 : ns * 120 + (i + 1) * 8].strip()) dmax = float(sig_raw[ns * 128 + i * 8 : ns * 128 + (i + 1) * 8].strip()) spr = int(sig_raw[ns * 216 + i * 8 : ns * 216 + (i + 1) * 8].strip()) signals.append(EDFSignal(label, dim, pmin, pmax, dmin, dmax, spr)) return EDFHeader(start_dt, num_records, duration, device_serial, signals, is_edfplus) def read_header(path: str) -> EDFHeader: """Read and parse only the EDF header block from a file without loading data. Args: path: The absolute filesystem path to the EDF file. Returns: The parsed EDFHeader containing recording metadata and signal layout. """ with open(path, "rb") as f: raw = f.read(256) ns = int(raw[252:256].strip()) sig_bytes = f.read(ns * 256) return _parse_header(raw + sig_bytes) def parse_pld(path: str) -> tuple: """Parse a PLD (processed/derived metrics) EDF file. Excludes the Crc16 channel and scales all raw values to physical units. Args: path: The absolute filesystem path to the PLD EDF file. Returns: A tuple of (EDFHeader, dict[str, list[float]]) containing the parsed header and a mapping of signal labels to physical values. """ with open(path, "rb") as f: raw = f.read(256) ns = int(raw[252:256].strip()) sig_bytes = f.read(ns * 256) data_bytes = f.read() header = _parse_header(raw + sig_bytes) samples_per_record = sum(s.num_samples_per_record for s in header.signals) channels = {s.label: [] for s in header.signals if s.label != "Crc16"} for rec in range(header.num_records): start = rec * samples_per_record * 2 record_bytes = data_bytes[start : start + samples_per_record * 2] if len(record_bytes) < samples_per_record * 2: break all_samples = struct.unpack(f"<{samples_per_record}h", record_bytes) idx = 0 for sig in header.signals: raw_vals = all_samples[idx : idx + sig.num_samples_per_record] idx += sig.num_samples_per_record if sig.label == "Crc16": continue channels[sig.label].extend(sig.scale(v) for v in raw_vals) return header, channels def parse_brp(path: str) -> tuple: """Parse a BRP (breath/pressure waveform) EDF file. ResMed BRP files typically contain Flow.40ms and Press.40ms, sampled 1500 times per 60-second record (25 Hz). Args: path: The absolute filesystem path to the BRP EDF file. Returns: A tuple of (EDFHeader, dict[str, list[float]]) containing the parsed header and a mapping of signal labels to physical waveform values. """ return parse_pld(path) def parse_eve(path: str) -> tuple: """Parse an EVE (events) EDF+ file. Note that ResMed uses a non-standard byte \\x15 to separate onset and duration within event TALs (standard uses \\x14). Args: path: The absolute filesystem path to the EVE EDF+ file. Returns: A tuple of (EDFHeader, list[tuple[float, float, str]]) where the list contains event records in the format (onset_seconds, duration_seconds, event_type). """ with open(path, "rb") as f: raw = f.read(256) ns = int(raw[252:256].strip()) sig_bytes = f.read(ns * 256) data_bytes = f.read() header = _parse_header(raw + sig_bytes) samples_per_record = sum(s.num_samples_per_record for s in header.signals) record_size = samples_per_record * 2 # Annotation signal is always index 0 in EDF+ ann_bytes_per_record = header.signals[0].num_samples_per_record * 2 events = [] for rec in range(header.num_records): record = data_bytes[rec * record_size : (rec + 1) * record_size] if len(record) < record_size: break ann_bytes = record[:ann_bytes_per_record] # TALs are null-terminated within the annotation signal tals = ann_bytes.split(b"\x00") for tal in tals: tal = tal.rstrip(b"\x00") if not tal or tal[0:1] not in (b"+", b"-"): continue if b"\x15" in tal: # ResMed event TAL: +onset\x15duration\x14annotation\x14... sep_idx = tal.index(b"\x15") try: onset = float(tal[1:sep_idx].decode("ascii")) except ValueError: continue rest = tal[sep_idx + 1 :] parts = rest.split(b"\x14") try: duration = float(parts[0].decode("ascii")) if parts[0] else 0.0 except ValueError: duration = 0.0 for ann_part in parts[1:]: txt = ann_part.decode("ascii", errors="ignore").strip("\x00 ") if txt and txt != "Recording starts": events.append((onset, duration, txt)) # TALs with only \x14\x14 are timekeeping markers — skip return header, events def parse_sa2(path: str) -> tuple: """Parse an SA2 (SpO2/pulse oximetry) EDF file. Args: path: The absolute filesystem path to the SA2 EDF file. Returns: A tuple of (EDFHeader, tuple[list[int], list[int]] | None) containing the parsed header and oximetry data arrays (pulse_list, spo2_list), or None if no oximeter was connected. """ with open(path, "rb") as f: raw = f.read(256) ns = int(raw[252:256].strip()) sig_bytes = f.read(ns * 256) data_bytes = f.read() header = _parse_header(raw + sig_bytes) pulse_sig = header.signals[0] # Pulse.1s spo2_sig = header.signals[1] # SpO2.1s samples_per_record = sum(s.num_samples_per_record for s in header.signals) pulse_vals = [] spo2_vals = [] for rec in range(header.num_records): start = rec * samples_per_record * 2 record_bytes = data_bytes[start : start + samples_per_record * 2] if len(record_bytes) < samples_per_record * 2: break all_samples = struct.unpack(f"<{samples_per_record}h", record_bytes) pulse_raw = all_samples[: pulse_sig.num_samples_per_record] spo2_raw = all_samples[ pulse_sig.num_samples_per_record : pulse_sig.num_samples_per_record + spo2_sig.num_samples_per_record ] pulse_vals.extend(pulse_raw) spo2_vals.extend(spo2_raw) if all(v == -1 for v in spo2_vals): return header, None return header, (pulse_vals, spo2_vals)