I almost forgot to mention:

When I tried to run both scripts at the same time it seemed to create a collision in the serial code:

Capture error: Mismatched lengths.
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/mnt/sharedhd/downloads/zeo/ZeoRawDataLib/Distribute/NapTrainer/ZeoRawData/BaseLink.py", line 81, in run
    c = self.ser.read(1)
  File "/usr/lib64/python2.6/site-packages/serial/serialposix.py", line 449, in read
    buf = os.read(self.fd, size-len(read))
OSError: [Errno 11] Resource temporarily unavailable

I'm probably just going to make one script that combines the functionality of both of the scripts I just made.  Nonetheless, this makes me wonder if it's worthwhile to have a system-wide deamon that proxies all of the serial communications for all instances of BaseLink on a given port.  It looks easy enough to just register more event handlers within a script, but that requires the scripts to know about each other and share BaseLink instances.

Here's the program that both records data and wakes someone in light stage sleep:

#!/usr/bin/python -d
# -*- coding: utf-8 -*-

# NapTrainer.pyw
# This program will attempt to wake the sleeper whenever they are in light sleep
#   for too long.  This could potentially be used to acquire better polyphasic
#   naps very early in an adaptation and possibly train the user to nap
#   efficiently, though this is all guesswork so far.  This is definitely not
#   recommended for normal sleep!
# You will need to provide a program for this script to run as a way to wake
#   you up.  The default setup calls mplayer and tells it to play the
#   "alarm.wav" file that should be found in the same directory as this script.

import sys
import time
import csv
import subprocess
import os
from serial import *
from glob import glob
from ZeoRawData import BaseLink, Parser

def scanPorts():
    portList = []
    
    # Helps find USB>Serial converters on linux
    for p in glob('/dev/ttyUSB*'):
        portList.append(p)
    #Linux and Windows
    for i in range(256):
        try:
            ser = Serial(i)
            if ser.isOpen():
                #Check that the port is actually valid
                #Otherwise invalid /dev/ttyS* ports may be added
                portList.append(ser.portstr)
            ser.close()
        except SerialException:
            pass

    return portList
    

class ZeoToCSV:
    def __init__(self, parent=None):
        samplesFileName = 'raw_samples.csv'
        sgramFileName = 'spectrogram.csv'
        hgramFileName = 'hypnogram.csv'
        eventsFileName = 'events.csv'
        
        self.hypToHeight = {'Undefined' : 0,
                            'Deep'      : 1,
                            'Light'     : 2,
                            'REM'       : 3,
                            'Awake'     : 4}
        
        # Only create headers when the files are being created for the first time.
        # After that, all new data should be appended to the existing files.
        samplesNeedHeader = False
        sgramNeedHeader = False
        hgramNeedHeader = False
        eventsNeedHeader = False
        
        if not os.path.isfile(samplesFileName):
            samplesNeedHeader = True
        
        if not os.path.isfile(sgramFileName):
            sgramNeedHeader = True
        
        if not os.path.isfile(hgramFileName):
            hgramNeedHeader = True
    
        if not os.path.isfile(eventsFileName):
            eventsNeedHeader = True
        
        self.rawSamples = csv.writer(open(samplesFileName, 'a+b'), delimiter=',',
                                quotechar='"', quoting=csv.QUOTE_MINIMAL)
        self.spectrogram = csv.writer(open(sgramFileName, 'a+b'), delimiter=',',
                                quotechar='"', quoting=csv.QUOTE_MINIMAL)
        self.hypnogram = csv.writer(open(hgramFileName, 'a+b'), delimiter=',',
                                quotechar='"', quoting=csv.QUOTE_MINIMAL)
        self.eventsOut = csv.writer(open(eventsFileName, 'a+b'), delimiter=',',
                                quotechar='"', quoting=csv.QUOTE_MINIMAL)
        
        if samplesNeedHeader:
            self.rawSamples.writerow(["Time Stamp","Version","SQI","Impedance","Bad Signal (Y/N)","Voltage (uV)"])
        
        if sgramNeedHeader:
            self.spectrogram.writerow(["Time Stamp","Version","SQI","Impedance","Bad Signal (Y/N)",
                                    "2-4 Hz","4-8 Hz","8-13 Hz","11-14 Hz","13-18 Hz","18-21 Hz","30-50 Hz"])
        
        if hgramNeedHeader:
            self.hypnogram.writerow(["Time Stamp","Version","SQI","Impedance","Bad Signal (Y/N)","State (0-4)","State (named)"])
        
        if eventsNeedHeader:
            self.eventsOut.writerow(["Time Stamp","Version","Event"])
    
    def updateSlice(self, slice):
        
        timestamp = slice['ZeoTimestamp']
        ver = slice['Version']
                                        
        if not slice['SQI'] == None:
            sqi = str(slice['SQI'])
        else:
            sqi = '--'
                                        
        if not slice['Impedance'] == None:
            imp = str(int(slice['Impedance']))
        else:
            imp = '--'

        if slice['BadSignal']:
            badSignal = 'Y'
        else:
            badSignal = 'N'

        if not slice['Waveform'] == []:
            self.rawSamples.writerow([timestamp,ver,sqi,imp,badSignal] + slice['Waveform'])

        if len(slice['FrequencyBins'].values()) == 7:
            f = slice['FrequencyBins']
            bins = [f['2-4'],f['4-8'],f['8-13'],f['11-14'],f['13-18'],f['18-21'],f['30-50']]
            self.spectrogram.writerow([timestamp,ver,sqi,imp,badSignal] + bins)

        if not slice['SleepStage'] == None:
            stage = slice['SleepStage']
            self.hypnogram.writerow([timestamp,ver,sqi,imp,badSignal] +
                                     [self.hypToHeight[stage],str(stage)])
    
    def updateEvent(self, timestamp, version, event):
        self.eventsOut.writerow([timestamp,version,event])

class NapTrainer:
    def __init__(self, parent=None):
        
        self.lightCount = 0
        self.hypToHeight = {'Undefined' : 0,
                            'Deep'      : 1,
                            'Light'     : 2,
                            'REM'       : 3,
                            'Awake'     : 4}
        
    
    def updateSlice(self, slice):
        
        timestamp = slice['ZeoTimestamp']

        if not slice['BadSignal'] and not slice['SleepStage'] == None:
            stage = slice['SleepStage']
            print "Sleep stage: "+ stage
            #if stage == 'Awake':
            if stage == 'Light':
                self.lightCount = self.lightCount + 1
            else:
                self.lightCount = 0
        
        # This is about 2 minutes of light sleep.
        #if self.lightCount >= 1:
        if self.lightCount >= 4:
            subprocess.call(["mplayer","alarm.wav"])
            self.lightCount = 0
    
    def updateEvent(self, timestamp, version, event):
        """Stub: not needed."""

if __name__ == '__main__':
    # Find ports.
    # TODO: offer a command line option for selecting ports.
    ports = scanPorts()
    if len(ports) > 0 :
        print "Found the following ports:"
        for port in ports:
            print port
        print ""
        print "Using port "+ports[0]
        print ""
        portStr = ports[0]
    else:
        sys.exit("No serial ports found.")
    
    # Initialize
    trainer = NapTrainer()
    output = ZeoToCSV()
    link = BaseLink.BaseLink(portStr)
    parser1 = Parser.Parser()
    parser2 = Parser.Parser()
    # Add callbacks
    link.addCallback(parser1.update)
    parser1.addEventCallback(trainer.updateEvent)
    parser1.addSliceCallback(trainer.updateSlice)
    link.addCallback(parser2.update)
    parser2.addEventCallback(output.updateEvent)
    parser2.addSliceCallback(output.updateSlice)
    # Start Link
    link.start()
    
    # TODO: perhaps use a more forgiving key?  This would require polling the keyboard without blocking.
    print "Hit ctrl-C at any time to stop."
    while True:
        time.sleep(5)
    
    sys.exit()

Daniel said:

Here's an idea: a program that causes the alarm to go off after twenty total minutes of non-wake. Why? Because some people take longer to fall asleep during naps while attempting to adapt to polyphasic. It would be nice if this equipment could compensate for how long it takes to fall asleep automatically when determining alarm time, rather than requiring us to get up and manually set it after failing to fall asleep for ten minutes. That would really help, not only for adapting, but for nappers in general. I know that I would use it.

 

Daniel

I agree that this would be neat.  The reason I didn't write it is because I (personally) haven't needed it.  I actually thought I would, but didn't.  At some level I'm not just being rigid about when I take naps, but about when I wake up from them as well.  So far I've managed it quite well, which is even a bit surprising to me.  Note that in preparation for going into polyphasic sleep I trained myself to wake up at exactly the same time every morning, regardless of how much sleep I had obtained.  I suspect this is why it's very easy for me to wake up from my core 3 hours, which end at exactly that time.  Of course, it hasn't helped at all with the nasty task of staying awake after the core; that still sucks ;) 

Sorry, I'm rambling.  I can tell how this would be useful to others.

chadjoan said:

Daniel said:

Here's an idea: a program that causes the alarm to go off after twenty total minutes of non-wake. Why? Because some people take longer to fall asleep during naps while attempting to adapt to polyphasic. It would be nice if this equipment could compensate for how long it takes to fall asleep automatically when determining alarm time, rather than requiring us to get up and manually set it after failing to fall asleep for ten minutes. That would really help, not only for adapting, but for nappers in general. I know that I would use it.

 

Daniel

---

I agree that this would be neat.  The reason I didn't write it is because I (personally) haven't needed it.  I actually thought I would, but didn't.  At some level I'm not just being rigid about when I take naps, but about when I wake up from them as well.  So far I've managed it quite well, which is even a bit surprising to me.  Note that in preparation for going into polyphasic sleep I trained myself to wake up at exactly the same time every morning, regardless of how much sleep I had obtained.  I suspect this is why it's very easy for me to wake up from my core 3 hours, which end at exactly that time.  Of course, it hasn't helped at all with the nasty task of staying awake after the core; that still sucks ;) 

Sorry, I'm rambling.  I can tell how this would be useful to others.

Here it is.  This is only lightly tested (I ran it to tease out any obvious crashes, but have not napped with it).  This doesn't have any intelligence in it besides the obvious "wake me up 20 after the onset", so JxA you'll be looking for something else. This is just an easy change to the light-sleep inhibitor. 

#!/usr/bin/python -d
# -*- coding: utf-8 -*-

# NapAlarm.pyw
# This program will attempt to wake the sleeper when they have been asleep for
#   a fixed amount of time.  By default this is set to the typical polyphasic
#   nap length of 20 minutes.
# You will need to provide a program for this script to run as a way to wake
#   you up.  The default setup calls mplayer and tells it to play the
#   "alarm.wav" file that should be found in the same directory as this script.

import sys
import time
import subprocess
import os
from serial import *
from glob import glob
from ZeoRawData import BaseLink, Parser

def scanPorts():
    portList = []
    
    # Helps find USB>Serial converters on linux
    for p in glob('/dev/ttyUSB*'):
        portList.append(p)
    #Linux and Windows
    for i in range(256):
        try:
            ser = Serial(i)
            if ser.isOpen():
                #Check that the port is actually valid
                #Otherwise invalid /dev/ttyS* ports may be added
                portList.append(ser.portstr)
            ser.close()
        except SerialException:
            pass

    return portList

class NapAlarm:
    def __init__(self, parent=None):
        
        # The time, in minutes, to wait until waking the sleeper.
        # This starts to count down after sleep onset.
        #self.napLength = 20.0
        self.napLength = 0.25  #Testing
        
        #
        self.sleepStarted = False
        self.sleepCount = 0
        self.hypToHeight = {'Undefined' : 0,
                            'Deep'      : 1,
                            'Light'     : 2,
                            'REM'       : 3,
                            'Awake'     : 4}
        
    
    def updateSlice(self, slice):
        
        timestamp = slice['ZeoTimestamp']

        if not slice['BadSignal'] and not slice['SleepStage'] == None:
            stage = slice['SleepStage']
            print "Sleep stage: "+ stage
            if stage == 'REM' or stage == 'Light' or stage == 'Deep':
                self.sleepStarted = True
            if self.sleepStarted:
                self.sleepCount = self.sleepCount + 1
            ## I originally considered doing the below, but decided against it
            ##   because the sleeper could wake up in the middle of a 20 second
            ##   nap and reset the timer, thus allowing themselves to oversleep.
            #if stage == 'Awake':
            #    self.sleepCount = 0
            #elif stage == 'REM' or stage == 'Light' or stage == 'Deep':
            #    self.sleepCount = self.sleepCount + 1
            # 'Undefined' sleep doesn't count either way.
        
        # Multiply napLength by 2 to get the number of 30-second sleep epochs
        #   that are recorded by the Zeo.
        if self.sleepCount >= self.napLength * 2:
            # Run a separate program to wake up the sleeper.
            subprocess.call(["mplayer","alarm.wav"])
            
            # TODO: there are probably more intelligent things that could be
            #   done after sounding the alarm.
            self.sleepStarted = False
            self.sleepCount = 0
    
    def updateEvent(self, timestamp, version, event):
        """Stub: not needed."""

if __name__ == '__main__':
    # Find ports.
    # TODO: offer a command line option for selecting ports.
    ports = scanPorts()
    if len(ports) > 0 :
        print "Found the following ports:"
        for port in ports:
            print port
        print ""
        print "Using port "+ports[0]
        print ""
        portStr = ports[0]
    else:
        sys.exit("No serial ports found.")
    
    # Initialize
    trainer = NapAlarm()
    link = BaseLink.BaseLink(portStr)
    parser = Parser.Parser()
    # Add callbacks
    link.addCallback(parser.update)
    parser.addEventCallback(trainer.updateEvent)
    parser.addSliceCallback(trainer.updateSlice)
    # Start Link
    link.start()
    
    # TODO: perhaps use a more forgiving key?  This would require polling the keyboard without blocking.
    print "Hit ctrl-C at any time to stop."
    while True:
        time.sleep(5)
    
    sys.exit()

JxA said:

 Chad,

I was just working with my friend who has been doing opensource for years and he asked me to ask for your permission to use your code, modify it and put it on github.

He'll give you attribution and post the code as open source for all to build upon.

 

Thanks,

John

Oh jeeze, I thought I responded to this like three days ago.  Sorry for the delay!

I'm entirely fine with sharing!

At some level it is my wish that these things be boost licensed.  I didn't do that before because I was being lazy and I didn't reread the RDL's terms and conditions to see if that would even make sense.  These are very simple scripts, so I wasn't too worried about licensing and such.

JxA said:

Chad,

After one successful night I've had a couple problems getting this running the last couple nights.

The error I'm getting is: "Capture Error: Read timeout in sync."

Is anyone else seeing this?

 

I can't yet determine the root cause of the error - python, windows7, Zeo, USB-Serial Cable/driver. (I suspect it is windows7).

I think I'll make a USB Ubuntu build and boot off of there to rule out my current OS.

 

Thanks,

 

John

Note that this is the error message I get when my Zeo is off or the serial cable is not seated in the Zeo correctly. 

The serial cable has some circuitry in the USB side of the cable that converts the serial transmissions into USB protocol.  That USB conversion cirtcuitry is, in and of itself, a USB device.  What this means is that you can probably plug the cable into the computer with nothing connected on the serial end and you will still get a USB device registered.  This is how it works on my Linux system at least.  Of course, even though the serial port is there, the scripts will still be unable to use it until something starts sending actual meaningful data on it.

I've hit this one a lot because I take my Zeo and laptop to work with me and use them to take naps in the car.  I have to check my serial connector every time because it will very easily sit in the Zeo's serial socket yet be slightly twisted away such that a pin or two don't make contact.  It doesn't like that ;)

How to get it going:

Make sure you have python 2.7+ installed.

Make sure you have pyserial installed.

Place the code in a .pyw file.  I call it "DataRecorder.pyw".  Put that file in a directory next to the "ZeoRawData" folder.

Run "python DataRecorder.pyw".  Receive bacon- err, data.

The "ZeoRawData" folder is found in the following place in the raw data library:

Distribute/ZeoRawData-2.0/ZeoRawData

where Distribute is (or is in) the root folder of the zip file you download.

I usually take the optional step of closing it down between runs and moving the *.csv files to separate directories so that I can keep track of stuff better.  What I end up with looks like the more recent directories found here.  (Yes that's my data, and I'll update it later)

I'm actually working heavily on a much improved version of this script that brings it all together under a nicer interface.  I'm trying to avoid distractions for the time being, but I'll look into releasing the new code with executables on the big 3 platforms. 

Mac might be a bit glitchy though because I have no machine to test it on.  Actually Windows too because I was never able to run PySerial on my 64-bit WinXP.  Maybe it'll run under Wine.

- Chad