Convert Dreem CSV to OSCAR-friendly Zeo CSV

dreem_csv_to_oscar_zeo.py

from pytz import timezone
from datetime import datetime
from io import open
from sys import argv, version_info

# Dreem is timezone-aware, but neither the Zeo CSV format nor OSCAR are. So we set a default
# timezone, so we can adjust deviations accordingly. Set to the timezone you're machine time
# is set for.
# Q: what about DST?

PST = timezone('US/Pacific')
MST = timezone('US/Mountain')
EST = timezone('US/Eastern')

default_tz = EST

def hms_to_m(value):
    try:
        h, m, s = map(int, value.split(':'))
    except (AttributeError, ValueError) as e:
        return
    return int(timedelta(hours=h, minutes=m, seconds=s).total_seconds() / 60)

# this is the function we need to make tz-aware
def iso_8601_to_local(value):
    # Zeo aligns to five minute boundary, however OSCAR doesn't care
    try:
        date_time_obj = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z")
        timezone_date_time_obj = date_time_obj.astimezone(default_tz)
        return timezone_date_time_obj.strftime("%m/%d/%Y %H:%M")
    except (TypeError, ValueError):
        return

def calc_rise_time(stage_series, start_time, sample_t=30):
    """Returns rise_t by offsetting start_t with product of last non-wake index and sampling time."""
    # Time of day at the end of the last 5 minute block of sleep in the sleep graph.
    # However, OSCAR doesn't care about the 5-min alignment, so just use minute precision.
    try:
        last_sleep_idx = max(idx for idx, val in enumerate(stage_series) if val in ("2", "3", "4"))
    except ValueError:
        return

    try:
        start_time_dt = datetime.strptime(start_time, "%Y-%m-%dT%H:%M:%S%z")
    except (TypeError, ValueError):
        return
    return (start_time_dt + (last_sleep_idx * timedelta(seconds=sample_t))).strftime("%m/%d/%Y %H:%M")

def hypnogram_to_stages(hypnogram):
    stage_map = {
        "n/a": "0",
        "wake": "1",
        "rem": "2",
        "light": "3",
        "deep": "4"
    }
    try:
        return [stage_map.get(value.lower()) for value in hypnogram[1:-1].split(',')]
    except TypeError:
        return []

def translate_row(row=None):
    row = row if row is not None else {}
    stages = hypnogram_to_stages(row.get("Hypnogram"))

    # OSCAR performs indexOf on keys, so order does not need to ve maintained on py2
    return {
        "ZQ": 0,
        "Total Z": hms_to_m(row.get("Sleep Duration")),
        "Time to Z": hms_to_m(row.get("Sleep Onset Duration")),
        "Time in Wake": hms_to_m(row.get("Wake After Sleep Onset Duration")),
        "Time in REM": hms_to_m(row.get("REM Duration")),
        "Time in Light": hms_to_m(row.get("Light Sleep Duration")),
        "Time in Deep": hms_to_m(row.get("Deep Sleep Duration")),
        "Awakenings": row.get("Number of awakenings"),
        "Sleep Graph": "",  # Appears to be unused in OSCAR
        "Detailed Sleep Graph": " ".join(stages),
        "Start of Night": iso_8601_to_local(row.get("Start Time")),
        "End of Night": iso_8601_to_local(row.get("Stop Time")),
        "Rise Time": calc_rise_time(stages, row.get("Start Time")),
        "Alarm Reason": None,
        "Snooze Time": None,
        "Wake Zone": None,
        "Wake Window": None,
        "Alarm Type": None,
        "First Alarm Ring": None,
        "Last Alarm Ring": None,
        "First Snooze Time": None,
        "Last Snooze Time": None,
        "Set Alarm Time": None,
        "Morning Feel": None,
        "Firmware Version": None,
        "My ZEO Version": None
    }

def convert(dreem_csv, zeo_csv):
    with open(dreem_csv, newline='') as fh_r:
        reader = DictReader(filter(lambda row: row[0] != '#', fh_r), delimiter=';')

        # Py2/3 compatibility
        args = {"mode": 'wb'} if version_info.major < 3 else {"mode": 'w', "newline": ''}
        with open(zeo_csv, **args) as fh_w:
            writer = DictWriter(fh_w, delimiter=',', fieldnames=translate_row())
            writer.writeheader()
            for in_row in reader:
                out_row = translate_row(in_row)
                print("IN: {}".format(dict(in_row)))
                print("OUT: {}".format(out_row))
                writer.writerow(out_row)

convert(argv[1], argv[2])