jeremyblow/dreem_csv_to_oscar_zeo.py

## dreem_csv_to_oscar_zeo.py
"""Convert Dreem CSV to OSCAR-friendly Zeo CSV

Liberty taken with the ZQ column, pinning it to 0. Other non-common fields are nulled.
Tested with Python 3.7.3 and 2.7.10.

Usage:
    python dreem_csv_to_oscar_zeo.py data_in.csv data_out.csv
"""
from csv import DictReader, DictWriter
from datetime import datetime, timedelta
from io import open
from sys import argv, version_info


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)

def iso_8601_to_local(value):
    # Zeo aligns to five minute boundary, however OSCAR doesn't care
    try:
        return datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z").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])
	"""Convert Dreem CSV to OSCAR-friendly Zeo CSV

	Liberty taken with the ZQ column, pinning it to 0. Other non-common fields are nulled.
	Tested with Python 3.7.3 and 2.7.10.

	Usage:
	python dreem_csv_to_oscar_zeo.py data_in.csv data_out.csv
	"""
	from csv import DictReader, DictWriter
	from datetime import datetime, timedelta
	from io import open
	from sys import argv, version_info


	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)

	def iso_8601_to_local(value):
	# Zeo aligns to five minute boundary, however OSCAR doesn't care
	try:
	return datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z").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])