SnoreLab is an excellent iOS application which records audio during sleeptime, and lets you view and listen to your snoring.
This gateway lets you upload SnoreLab data to Fluxtream.org
In [13]:
import glob, os
In [14]:
def exec_ipynb(url):
import json, re, urllib2
nb = (urllib2.urlopen(url) if re.match(r'https?:', url) else open(url)).read()
exec '\n'.join([''.join(cell['input']) for cell in json.loads(nb)['worksheets'][0]['cells'] if cell['cell_type'] == 'code']) in globals()
exec_ipynb('Fluxtream-Library.ipynb')
fluxtream_login()
In [16]:
def epoch_time(dt):
epoch = datetime.datetime(1970, 1, 1, tzinfo=tz.tzutc())
return (dt - epoch).total_seconds()
# Returns 2D array of data suitable for posting to Fluxtream API
def snorelab_history_csv_to_post_data(filename):
reader = csv.reader(open(filename,'rb'), delimiter=',')
# skip 4 rows of headers
for i in range(0,5):
header = reader.next()
rowcount = 0;
data = []
for row in reader:
# In snorelab, the zeroth column is the date of the start of sleep, the first column is the time of start
# of sleep, and the second column is the time of end of sleep. If time of end of sleep < time of start of sleep
# then the date of end of sleep is going to be the following day.
start_date_str= row[0]
start_time_str= row[1]
end_time_str = row[2]
start_dt = datetime.datetime.strptime("%s %s" % (start_date_str, start_time_str), '%d-%b-%Y %H:%M').replace(tzinfo=tz.tzlocal())
end_dt = datetime.datetime.strptime("%s %s" % (start_date_str, end_time_str), '%d-%b-%Y %H:%M').replace(tzinfo=tz.tzlocal())
if(end_dt<start_dt):
end_dt += datetime.timedelta(days=1)
snorescore=float(row[6])
comment = row[9]
data.append([epoch_time(end_dt), snorescore, comment])
return data
def snorelab_parse_report(report_dir):
report_filenames = glob.glob(report_dir + '/SnoreLab Chart Points*csv')
if len(report_filenames) != 1:
raise Exception('Should have found exactly one report in %s but found %d' %
(report_dir, len(report_filenames)))
report_filename = report_filenames[0]
data = []
for row in csv.reader(open(report_filename)):
data.append([float(x) for x in row])
print 'Snore report is %.1f hours long and has %d samples' % ((data[-1][0] - data[0][0]) / 3600.0, len(data))
return data
In [21]:
### Unzip and archive downloaded zipfiles
download_dir = os.path.expanduser('~/Downloads/')
snorelab_reports_dir = 'snorelab-reports'
report_pattern = 'snorelab*[0-9][0-9][0-9][0-9]*'
download_pattern = download_dir + report_pattern + '.zip'
zips = glob.glob(download_pattern)
print 'Looking for files of pattern %s; found %d' % (download_pattern, len(zips))
for zip in zips:
dest = snorelab_reports_dir + '/' + os.path.basename(os.path.splitext(zip)[0])
if os.path.exists(dest):
print ' %s already exists' % dest
else:
subprocess.check_output(['unzip', zip, '-d', dest + '.tmp'])
os.rename(dest + '.tmp', dest)
print ' Created %s (%d files)' % (dest, len(glob.glob(dest + '/*')))
### Upload reports
report_dirs = glob.glob(snorelab_reports_dir + '/' + report_pattern)
print '\nUploading new reports'
for report_dir in report_dirs:
was_uploaded_path = report_dir + '/uploaded-to-fluxtream'
if os.path.exists(was_uploaded_path):
print ' %s already uploaded' % report_dir
else:
print ' Uploading %s' % report_dir
data = snorelab_parse_report(report_dir)
fluxtream_upload('SnoreLab', ['SnoreIntensity'], data)
open(was_uploaded_path, 'w')
In [ ]: