Parsers to fetch data out of an exported Apple Health log

cda_parser.py

import csv
import datetime
from lxml import etree

# Constants
INPUT_FILE = 'export_cda.xml'
HR_OUTPUT = 'heartRate_sorted.csv'
O2_OUTPUT = 'bloodOxygen_sorted.csv'

# LOINC Codes from your snippets
CODE_HEART_RATE = "8867-4"
CODE_OXYGEN_SAT = "2710-2"

def format_date(date_str):
    """Converts 20260410153251-0400 to 4/10/26 3:32:51 PM"""
    try:
        # Strip timezone offset
        clean_date = date_str.split('-')[0].split('+')[0]
        dt = datetime.datetime.strptime(clean_date, "%Y%m%d%H%M%S")
        return dt, dt.strftime("%-m/%-d/%y %-I:%M:%S %p")
    except Exception:
        return datetime.datetime.min, date_str

def process_health_data():
    hr_points = []
    o2_points = []
    ns = {'n': 'urn:hl7-org:v3'}

    print(f"Streaming {INPUT_FILE}...")

    # Using iterparse to keep memory usage low for the 300MB+ file
    context = etree.iterparse(INPUT_FILE, events=('end',), tag='{urn:hl7-org:v3}observation')

    for event, elem in context:
        try:
            code_elem = elem.find('n:code', ns)
            if code_elem is None:
                continue
            
            current_code = code_elem.get('code')
            
            value_elem = elem.find('n:value', ns)
            raw_val = value_elem.get('value') if value_elem is not None else None
            
            time_elem = elem.find('n:effectiveTime/n:low', ns)
            raw_time = time_elem.get('value') if time_elem is not None else None

            if not raw_val or not raw_time:
                continue

            dt_obj, human_time = format_date(raw_time)

            # 1. Handle Heart Rate
            if current_code == CODE_HEART_RATE:
                hr_points.append({
                    'sort_key': dt_obj,
                    'time': human_time,
                    'val': raw_val
                })

            # 2. Handle Blood Oxygen
            elif current_code == CODE_OXYGEN_SAT:
                # Convert 0.97 to 97 for readability
                try:
                    pct_val = float(raw_val) * 100
                    display_val = f"{pct_val:.1f}"
                except ValueError:
                    display_val = raw_val

                o2_points.append({
                    'sort_key': dt_obj,
                    'time': human_time,
                    'val': display_val
                })

        finally:
            # Memory cleanup
            elem.clear()
            while elem.getprevious() is not None:
                del elem.getparent()[0]

    # Sorting and Writing Heart Rate
    if hr_points:
        print(f"Sorting {len(hr_points)} Heart Rate entries...")
        hr_points.sort(key=lambda x: x['sort_key'], reverse=True)
        with open(HR_OUTPUT, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['DateTime', 'HeartRate_BPM'])
            for p in hr_points:
                writer.writerow([p['time'], p['val']])
        print(f"Saved to {HR_OUTPUT}")

    # Sorting and Writing Blood Oxygen
    if o2_points:
        print(f"Sorting {len(o2_points)} Blood Oxygen entries...")
        o2_points.sort(key=lambda x: x['sort_key'], reverse=True)
        with open(O2_OUTPUT, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['DateTime', 'Oxygen_Percentage'])
            for p in o2_points:
                writer.writerow([p['time'], p['val']])
        print(f"Saved to {O2_OUTPUT}")

if __name__ == "__main__":
    process_health_data()

gpx_parser.py

import os
import csv
import xml.etree.ElementTree as ET
from datetime import datetime

input_dir = 'workout-routes'
output_file = 'workout_summary.csv'
NS = {'gpx': 'http://www.topografix.com/GPX/1/1'}

def parse_gpx(file_path):
    try:
        tree = ET.parse(file_path)
        root = tree.getroot()
        track_points = root.findall('.//gpx:trkpt', NS)
        
        if not track_points:
            return None

        times = []
        elevations = []
        speeds = []

        for pt in track_points:
            # Time
            t_str = pt.find('gpx:time', NS).text
            times.append(datetime.fromisoformat(t_str.replace('Z', '+00:00')))
            
            # Elevation
            ele = pt.find('gpx:ele', NS)
            if ele is not None:
                elevations.append(float(ele.text))
                
            # Speed
            speed = pt.find('.//gpx:speed', NS)
            if speed is not None:
                speeds.append(float(speed.text))

        duration = (max(times) - min(times)).total_seconds()
        avg_speed = sum(speeds) / len(speeds) if speeds else 0
        ele_delta = max(elevations) - min(elevations) if elevations else 0

        return {
            'filename': os.path.basename(file_path),
            'start_time': min(times).strftime('%Y-%m-%d %H:%M:%S'),
            'duration_sec': int(duration),
            'avg_speed_mps': round(avg_speed, 3),
            'elevation_delta_m': round(ele_delta, 2)
        }
    except Exception as e:
        print(f"Skipping {file_path}: {e}")
        return None

def main():
    results = []
    
    if not os.path.exists(input_dir):
        print(f"Directory '{input_dir}' not found.")
        return

    for filename in os.listdir(input_dir):
        if filename.endswith('.gpx'):
            data = parse_gpx(os.path.join(input_dir, filename))
            if data:
                results.append(data)

    # SORTING: Newest entries first based on the workout start time
    results.sort(key=lambda x: x['start_time'], reverse=True)

    if results:
        with open(output_file, 'w', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=results[0].keys())
            writer.writeheader()
            writer.writerows(results)
        print(f"Done! Created {output_file} with {len(results)} rows.")

if __name__ == "__main__":
    main()