Added new machinery.

goneuro.go

@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"github.com/jbrukh/atomic"
 )
 
 // Approx the number of data points to be
@@ -97,67 +98,16 @@ type ThinkGearListener struct {
 	EEGPower       EEGPowerListener
 }
 
-// Connect to the device over the serial port
-// and start parsing data; the serial port
-// is typically a string of the form
-// 
-//   /dev/tty.MindBand
-//
-// or whatever you set up in your systems Bluetooth
-// options for the device. Note that the various
-// portions of the ThinkGearListener will be triggered
-// synchronously to parsing, so it may be desirable
-// in certain situations for the user to throw data
-// onto channels for serial, asynchronous processing.
-// If you do use a channel, make sure that this channel
-// is asynchronous, or you can still hold up processing.
-//
-// This method will return a send-only channel
-// for the purposes of ceasing the connection. In
-// order to close the connection, send true to
-// the disconnect channel.
-func Connect(serialPort string, listener *ThinkGearListener) (disconnect chan<- bool, err error) {
-	var device io.ReadCloser
-	device, err = os.Open(serialPort)
-	if err != nil {
-		str := fmt.Sprintf("device problem: %s", err)
-		return nil, errors.New(str)
-	}
-	println("connected: ", serialPort)
-
-	// create the disconnect channel
-	ch := make(chan bool)
-
-	// go and process this this stream asynchronously
-	// until the user sends a signal to disconnect
-	go thinkGearParse(device, listener, ch)
+// payloadParser decides what to do with the
+// payload of a checksum-verified packet
+type payloadParser func(*[]byte)
 
-	disconnect = ch // cast to send-only
-	return
-}
-
-// ConnectRaw streams just the raw data on a channel;
-// this is provided as a convenience method
-func ConnectRaw(serialPort string) (disconnect chan<- bool, data <-chan float64, err error) {
-    ch := make(chan float64, WINDOW_SIZE)
-    listener := &ThinkGearListener{
-        RawSignal: func(a, b byte) {
-            ch <- float64(int16(a)<<8 | int16(b))
-        },
-    }
-    disconnect, err = Connect(serialPort, listener)
-    if err != nil {
-        return
-    }
-    data = ch
-    return
-}
-
-// thinkGearParse parses the TG byte stream
-func thinkGearParse(device io.ReadCloser, listener *ThinkGearListener, disconnect <-chan bool) {
+// parseByteStream parses the TG byte stream
+func parseByteStream(device io.ReadCloser, pparser payloadParser, conn <-chan bool) {
 	reader := bufio.NewReader(device)
 	defer device.Close()
-
+	engaged := false
+
 	// function that reads the stream
 	// one byte at a time
 	next := func() byte {
@@ -172,13 +122,18 @@ func thinkGearParse(device io.ReadCloser, listener *ThinkGearListener, disconnec
 	for {
 		// check for exit
 		select {
-		case v, ok := (<-disconnect):
-			println("v, ok:", v, ok)
-			if ok && v == true {
-				println("disconnecting from device")
-				return
+		case v, ok := <-conn:
+			if ok & !v {
+				return // disconnect when "false" sent
+			} else if ok && v {
+				engaged = true
+				continue // engage when user sends "true"
 			}
 		default:
+			if !engaged {
+				next()
+				continue // no parsing until engaged
+			}
 		}
 
 		// sync up
@@ -216,15 +171,34 @@ func thinkGearParse(device io.ReadCloser, listener *ThinkGearListener, disconnec
 			println("checksum has failed: ", checksum, "expected: ", stated)
 			continue
 		}
-		parsePayload(&payload, listener)
+		if payloadParser != nil {
+			payloadParser(&payload)
+		}
 	}
 	println("done with parsing")
 }
 
+// fullPayloadParser delivers a payload parser with the
+// given listener
+func fullPayloadParser(listener *ThinkGearListener) {
+	return func(payloadPtr *[]byte) {
+		parseFullPayload(payloadPtr, listener)
+	}
+}
+
+// rawPayloadParser delivers a payload parser that only
+// parses raw signal and ignores everything else, and then
+// delivers the raw signal on a channel
+func rawPayloadParser(output chan float64) {
+	return func(payloadPtr *[]byte) {
+		parseRawPayload(payloadPtr, output)
+	}
+}
+
 // parsePayload will parse the payload buffer and trigger
 // the appropriate listeners in the provided listener
 // object
-func parsePayload(payloadPtr *[]byte, listener *ThinkGearListener) {
+func parseFullPayload(payloadPtr *[]byte, listener *ThinkGearListener) {
 	payload := *payloadPtr
 	inx := 0
 	var codeLevel int
@@ -289,3 +263,171 @@ func parsePayload(payloadPtr *[]byte, listener *ThinkGearListener) {
 		}
 	}
 }
+
+// parseRawPayload will parse the payload buffer for
+// raw signal only, and deliver that signal on the
+// giveb channel
+func parseRawPayload(payloadPtr *[]byte, output chan<- float64) {
+	payload := *payloadPtr
+	inx := 0
+	var codeLevel int
+	nextRow := func(k int) {
+		inx += k
+		codeLevel = 0
+	}
+	for inx < len(payload) {
+		switch payload[inx] {
+		case EXCODE:
+			// not used in the current protocol
+			// but provided here for completeness
+			codeLevel++
+		case CODE_RAW_VALUE:
+			if payload[inx+1] == 2 {
+				// get the data
+			 	output <- float64(int16(payload[inx+2])<<8 | int16(payload[inx+3]))
+			} else {
+					println("raw signal did not have 2 bytes")
+					break
+			}
+			nextRow(4)
+		default:
+			break
+		}
+	}
+}
+
+// Device represents a NeuroSky/ThinkGear device
+type Device {
+	conn chan<- bool
+	Port string
+	connected *atomic.AtomicValue
+}
+
+func (d *Device) New(serialPort string) *Device {
+	return &Device{
+		conn: make(chan bool)
+		Port: serialPort,
+		connected: atomic.NewWithValue(false)
+	}
+}
+
+// Egage will engage the processing of the byte
+// stream. No listeners will be triggered or raw
+// data will stream until this call.
+// If the device is not connected, then this call
+// will have no effect.
+func (d *Device) Engage() {
+	if (d.connected.Get().(bool)) {
+		conn <- true
+	}
+}
+
+// Disconnect will disconnect from the device and
+// close the serial port. If the device is not
+// connected, this call will have no effect.
+func (d *Device) Disconnect() {
+	if (d.connected.Get().(bool)) {
+		conn <- false
+	}
+}
+
+// Connect allows you to initiate a connection with the
+// device and to provide an event handler for all the
+// different types of data simultaneously.
+// 
+// More precisely, Connect will initialize the Bluetooth
+// conection and will begin receiving data through the
+// serial port. However, all this data will be quietly
+// dropped until a call to Engage().
+//
+// If the device is already connected, then this call
+// will have no effect.
+func (d *Device) Connect(listener *ThinkGearListener) (err error) {
+	if (d.connected.Get().(bool)) {
+		return errors.New("device is already connected")
+	}
+	var device io.ReadCloser
+	device, err = os.Open(d.Port)
+	if err != nil {
+		str := fmt.Sprintf("device problem: %s", err)
+		return errors.New(str)
+	}
+	d.connected.Set(true)
+	println("connected: ", d.Port)
+
+	// start spinning the data stream on another thread 
+	// and wait for Engage() call
+	go parseByteStream(device, fullPayloadParser(listener), d.conn)
+	return
+}
+
+// ConnectRaw will first initialize the Bluetooth
+// connection and begin receiving data through the
+// serial port. However, the data will not be parsed
+// until a call to Engage(), at which point the
+// devices raw signal will be streamed to the 
+// provided channel.
+//
+// If the device is already connected, then this call
+// will have no effect.
+func (d *Device) ConnectRaw(output chan<- float64) {
+	if (d.connected.Get().(bool)) {
+		return
+	}
+}
+
+// Connect to the device over the serial port
+// and start parsing data; the serial port
+// is typically a string of the form
+// 
+//   /dev/tty.MindBand
+//
+// or whatever you set up in your systems Bluetooth
+// options for the device. Note that the various
+// portions of the ThinkGearListener will be triggered
+// synchronously to parsing, so it may be desirable
+// in certain situations for the user to throw data
+// onto channels for serial, asynchronous processing.
+// If you do use a channel, make sure that this channel
+// is asynchronous, or you can still hold up processing.
+//
+// This method will return a send-only channel
+// for the purposes of ceasing the connection. In
+// order to close the connection, send true to
+// the disconnect channel.
+func Connect(serialPort string, listener *ThinkGearListener) (conn chan<- bool, err error) {
+	var device io.ReadCloser
+	device, err = os.Open(serialPort)
+	if err != nil {
+		str := fmt.Sprintf("device problem: %s", err)
+		return nil, errors.New(str)
+	}
+	println("connected: ", serialPort)
+
+	// create the disconnect channel
+	ch := make(chan bool)
+
+	// go and process this this stream asynchronously
+	// until the user sends a signal to disconnect
+	go parseByteStream(device, fullPayloadParser(listener), ch)
+
+	disconnect = ch // cast to send-only
+	return
+}
+
+// ConnectRaw streams just the raw data on a channel;
+// this is provided as a convenience method
+func ConnectRaw(serialPort string) (disconnect chan<- bool, data <-chan float64, err error) {
+    ch := make(chan float64, WINDOW_SIZE)
+    listener := &ThinkGearListener{
+        RawSignal: func(a, b byte) {
+            ch <- float64(int16(a)<<8 | int16(b))
+        },
+    }
+    disconnect, err = Connect(serialPort, listener)
+    if err != nil {
+        return
+    }
+    data = ch
+    return
+}