Significant changes to DataThread classes.

DataThreads (e.g. IntanThread, FileReaderThread, FPGAThread) are now created at
the same time as the source node, rather than at the start of data acquisition.
New methods for starting/stopping individual threads are required, although
only the appropriate methods for the IntanThread have been written.

Another important change is that the SourceNode now periodically checks for
an appropriate input source every few seconds while acquisition is not in
progress. It's the responsibility of the individual DataThreads to notify
the SourceNode if their input has disappears. In the case of the IntanThread,
this involves attempting to change the baud rate. If an error code returns,
it informs the SourceNode that the input is missing. This, in turn, informs
the FilterViewport that the source is no longer enabled, thus deactivating
that particular signal chain.

Builds/Linux/build/savedState.xml

@@ -2,7 +2,7 @@
 
 <PROCESSORGRAPH>
   <SIGNALCHAIN>
-    <PROCESSOR name="Sources/Signal Generator" insertionPoint="0"/>
+    <PROCESSOR name="Sources/Intan Demo Board" insertionPoint="0"/>
     <PROCESSOR name="Sinks/LFP Viewer" insertionPoint="1"/>
   </SIGNALCHAIN>
 </PROCESSORGRAPH>

Builds/Linux/build/windowState.xml

 <?xml version="1.0" encoding="UTF-8"?>
 
 <MAINWINDOW>
-  <BOUNDS x="1723" y="79" w="1113" h="793" fullscreen="0"/>
+  <BOUNDS x="1729" y="52" w="1161" h="968" fullscreen="0"/>
 </MAINWINDOW>

Source/Processors/DataThreads/DataThread.cpp

@@ -21,7 +21,8 @@ void DataThread::run() {
 		const MessageManagerLock mml (Thread::getCurrentThread());
 		if (! mml.lockWasGained())
 			return;
-		updateBuffer();
+		if (!updateBuffer())
+			signalThreadShouldExit();
 	}
 }
 
@@ -29,3 +30,4 @@ DataBuffer* DataThread::getBufferAddress() {
 	return dataBuffer;
 }
 
+

Source/Processors/DataThreads/DataThread.h

@@ -27,11 +27,15 @@ public:
 
 	DataBuffer* getBufferAddress();
 
-	virtual void updateBuffer() = 0;
+	virtual bool updateBuffer() = 0;
 
 	DataBuffer* dataBuffer;
 
-	virtual bool threadStarted() {return true;}
+	virtual bool foundInputSource() = 0;
+	virtual bool startAcquisition() = 0;
+	virtual bool stopAcquisition() = 0;
+	virtual int getNumChannels() = 0;
+	virtual float getSampleRate() = 0;
 
 };
 

Source/Processors/DataThreads/FPGAThread.cpp

@@ -63,7 +63,7 @@ FPGAThread::~FPGAThread() {
 
 }
 
-void FPGAThread::updateBuffer() {
+bool FPGAThread::updateBuffer() {
 	
 	dev->ReadFromPipeOut(0xA0, sizeof(pBuffer), pBuffer);
 
@@ -101,6 +101,8 @@ void FPGAThread::updateBuffer() {
 			}
 		j++; // keep scanning for timecodes
 	}
+
+	return true;
 }
 
 

Source/Processors/DataThreads/FPGAThread.h

@@ -28,6 +28,12 @@ class FPGAThread : public DataThread
 public:
 	FPGAThread();
 	~FPGAThread();
+
+	bool foundInputSource() {return true;}
+	bool startAcquisition() {return true;}
+	bool stopAcquisition() {return true;}
+	int getNumChannels() {return 32;}
+	float getSampleRate() {return 25000.0;}
 	
 private:
 
@@ -49,7 +55,7 @@ private:
 	int numchannels;
 	int Ndatabytes;
 
-	void updateBuffer();
+	bool updateBuffer();
 	bool initializeFPGA(okCFrontPanel*, char*);
 
 	JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (FPGAThread);

Source/Processors/DataThreads/FileReaderThread.cpp

@@ -24,23 +24,29 @@ FileReaderThread::FileReaderThread() : DataThread(),
 
 	std::cout << "File Reader Thread initialized." << std::endl;
 
-	startThread();
 }
 
 FileReaderThread::~FileReaderThread() {
 
-	stopThread(500);
+	deleteAndZero(input);
 
-	std::cout << "File reader received disable signal." << std::endl;
+	deleteAndZero(dataBuffer);
 
-	deleteAndZero(input);
+}
+
+bool FileReaderThread::startAcquisition()
+{
+	startThread();
 
-	delete dataBuffer;
-	dataBuffer = 0;
 }
 
+bool FileReaderThread::stopAcquisition()
+{
+	stopThread(500);
+	std::cout << "File reader received disable signal." << std::endl;
+}
 
-void FileReaderThread::updateBuffer()
+bool FileReaderThread::updateBuffer()
 {
 
 	while (dataBuffer->getNumSamples() < 4096)
@@ -59,4 +65,5 @@ void FileReaderThread::updateBuffer()
 
     }
 
+    return true;
 }

Source/Processors/DataThreads/FileReaderThread.h

@@ -23,6 +23,12 @@ class FileReaderThread : public DataThread
 public:
 	FileReaderThread();
 	~FileReaderThread();
+
+	bool foundInputSource() {return true;}
+	bool startAcquisition();
+	bool stopAcquisition();
+	int getNumChannels() {return 16;}
+	float getSampleRate() {return 40000.0;}
 	
 private:
 
@@ -34,7 +40,7 @@ private:
 
 	float thisSample[16];
 
-	void updateBuffer();
+	bool updateBuffer();
 
 	JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (FileReaderThread);
 };

Source/Processors/DataThreads/IntanThread.cpp

@@ -23,45 +23,80 @@ IntanThread::IntanThread() : DataThread(),
 
 	 dataBuffer = new DataBuffer(16,4096);
 
-	if (initializeUSB())
-	{
-    	std::cout << "FTDI interface initialized." << std::endl;
-		ftdi_write_data(&ftdic, &startCode, 1);
-		startThread();
-		isTransmitting = true;
-	}
-
-	
+     deviceFound = initializeUSB(true);
+
 }
 
-IntanThread::~IntanThread() {
+IntanThread::~IntanThread() 
+{
+	closeUSB();
+	deleteAndZero(dataBuffer);
+}
 
-	stopThread(500);
+int IntanThread::getNumChannels()
+{
+    return 16;
+}
 
-	if (isTransmitting) {
-		ftdi_write_data(&ftdic, &stopCode, 1);
-		unsigned char buf[4097]; // has to be bigger than the on-chip buffer
-		ftdi_read_data(&ftdic, buf, sizeof(buf));
+float IntanThread::getSampleRate()
+{
+    return 25000.0;
+}
 
-		ftdi_usb_close(&ftdic);
-	    ftdi_deinit(&ftdic);
-		std::cout << "FTDI interface destroyed." << std::endl;
-	}
+bool IntanThread::foundInputSource()
+{
+    if (deviceFound)
+    {
+        int return_value;
+        // try to update the baud rate
+        if ((return_value = ftdi_set_baudrate(&ftdic, baudrate)) < 0)
+        {
+            deviceFound = false;
+            return false;
+        }
+    } else {
+        // try to initialize USB
+        if (!initializeUSB(false))
+        {
+            return false;
+        } else {
+            deviceFound = true;
+        }
+    }
 
-	deleteAndZero(dataBuffer);
+    return true;
+
+}
+
+bool IntanThread::startAcquisition()
+{
+    ftdi_write_data(&ftdic, &startCode, 1);
+    startThread();
 }
 
-bool IntanThread::initializeUSB()
+bool IntanThread::stopAcquisition()
+{
+    stopThread(500);
+
+    ftdi_write_data(&ftdic, &stopCode, 1);
+    unsigned char buf[4097]; // has to be bigger than the on-chip buffer
+    ftdi_read_data(&ftdic, buf, sizeof(buf));
+}
+
+
+bool IntanThread::initializeUSB(bool verbose)
 {
 	int return_value;
 
 	 // Step 1: initialise the ftdi_context:
 	if (ftdi_init(&ftdic) < 0) {// -1 = couldn't allocate read buffer
                                // -2 = couldn't allocate struct buffer
-        fprintf(stderr, "ftdi_init failed\n");
+        if (verbose)
+            fprintf(stderr, "ftdi_init failed\n");
         return false;
     } else {
-        std::cout << "FTDI context initialized." << std::endl;
+        if (verbose)
+            std::cout << "FTDI context initialized." << std::endl;
     }
 
 	// Step 2: open USB device
@@ -69,7 +104,8 @@ bool IntanThread::initializeUSB()
     // -8 = wrong permissions
    	if ((return_value = ftdi_usb_open(&ftdic, vendorID, productID)) < 0)
     {
-        fprintf(stderr, "unable to open FTDI device: %d (%s)\n",
+        if (verbose)
+            fprintf(stderr, "unable to open FTDI device: %d (%s)\n",
                         return_value, 
                         ftdi_get_error_string(&ftdic));
         return false;
@@ -80,7 +116,8 @@ bool IntanThread::initializeUSB()
 	// Step 3: set the baud rate
 	if ((return_value = ftdi_set_baudrate(&ftdic, baudrate)) < 0)
     {
-        fprintf(stderr, "unable to set baud rate: %d (%s)\n",
+        if (verbose)
+            fprintf(stderr, "unable to set baud rate: %d (%s)\n",
                         return_value, 
                         ftdi_get_error_string(&ftdic));
         return false;
@@ -88,19 +125,33 @@ bool IntanThread::initializeUSB()
         std::cout << "Baud rate set to 115200" << std::endl;
     }
 
-
-
 	return true;
 
 }
 
+bool IntanThread::closeUSB()
+{
+    ftdi_usb_close(&ftdic);
+    ftdi_deinit(&ftdic);
+    std::cout << "FTDI interface destroyed." << std::endl;
+}
+
 
-void IntanThread::updateBuffer()
+bool IntanThread::updateBuffer()
 {
 
-	// Step 1: update buffer
-	ftdi_read_data(&ftdic, buffer, sizeof(buffer));
+    int bytes_read;
 
+    // Step 1: update buffer
+    // error codes:
+    //  -666: USB device unavailable
+    //  <0  : error code from libusb_bulk_transfer()
+    //  0   : no data available
+    //  >0  : number of bytes read
+    if ((bytes_read = ftdi_read_data(&ftdic, buffer, sizeof(buffer))) < 0)
+    {
+        return false;
+    }
 
 	// Step 2: sort data
 	int TTLval, channelVal;
@@ -117,7 +168,6 @@ void IntanThread::updateBuffer()
 
          }
   
-
          TTLval = (buffer[index+2] & 4) >> 2; // extract TTL value (bit 3)
          channelVal = buffer[index+2] & 60;   // extract channel value
 
@@ -127,5 +177,8 @@ void IntanThread::updateBuffer()
          }
 
     }
+
+    return true;
+
 }
 

Source/Processors/DataThreads/IntanThread.h

@@ -23,7 +23,9 @@ public:
 	IntanThread();
 	~IntanThread();
 
-	bool threadStarted() {return isTransmitting;}
+	bool foundInputSource();
+	int getNumChannels();
+	float getSampleRate();
 	
 private:
 
@@ -31,18 +33,22 @@ private:
 	int vendorID, productID;
 	int baudrate;
 	bool isTransmitting;
+	bool deviceFound;
 
-	bool initializeUSB();
+	bool initializeUSB(bool);
+	bool closeUSB();
+
+	bool startAcquisition();
+	bool stopAcquisition();
 	
 	unsigned char startCode, stopCode;
 	unsigned char buffer[240]; // should be 5 samples per channel
 
 	float thisSample[16];
-	//float thisSample[64];
 
 	int ch;
 
-	void updateBuffer();
+	bool updateBuffer();
 
 	JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (IntanThread);
 };

Source/Processors/DataThreads/NetworkThread.cpp

@@ -38,7 +38,7 @@ NetworkThread::~NetworkThread() {
 }
 
 
-void NetworkThread::updateBuffer(){
+bool NetworkThread::updateBuffer(){
 		
 	 NetCom::rxWave (my_netcomdat, &lfp);
 
@@ -48,4 +48,6 @@ void NetworkThread::updateBuffer(){
 	 	}
 	 	dataBuffer->addToBuffer(thisSample,1);
 	 }
+
+	 return true;
 }
\ No newline at end of file

Source/Processors/DataThreads/NetworkThread.h

@@ -28,6 +28,12 @@ public:
 	NetworkThread();
 	~NetworkThread();
 
+	bool foundInputSource() {return true;}
+	bool startAcquisition() {return true;}
+	bool stopAcquisition() {return true;}
+	int getNumChannels() {return 16;}
+	float getSampleRate() {return 40000.0;}
+
 private:
 
 	NetCom my_netcom;
@@ -39,7 +45,7 @@ private:
 
 	float thisSample[8];
 
-	void updateBuffer();
+	bool updateBuffer();
 
 	JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR (NetworkThread);
 

Source/Processors/Editors/GenericEditor.cpp

@@ -107,14 +107,16 @@ void GenericEditor::disable()
 
 bool GenericEditor::getEnabledState()
 {
-	return isEnabled;
+	GenericProcessor* p = (GenericProcessor*) getProcessor();
+	return p->enabledState();
 }
 
 void GenericEditor::setEnabledState(bool t)
 {
-	isEnabled = t;
+	
 	GenericProcessor* p = (GenericProcessor*) getProcessor();
-	p->enabledState(isEnabled);
+	p->enabledState(t);
+	isEnabled = p->enabledState();
 }
 
 void GenericEditor::paint (Graphics& g)

Source/Processors/GenericProcessor.h

@@ -133,8 +133,8 @@ public:
 	virtual bool enable() {return true;}
 	virtual bool disable() {return true;}
 
-	bool enabledState() {return isEnabled;}
-	void enabledState(bool t) {isEnabled = t;}
+	virtual bool enabledState() {return isEnabled;}
+	virtual void enabledState(bool t) {isEnabled = t;}
 
 	virtual AudioSampleBuffer* getContinuousBuffer() {return 0;}
 	virtual MidiBuffer* getEventBuffer() {return 0;}
@@ -155,13 +155,15 @@ public:
 	void checkForMidiEvents(MidiBuffer& mb);
 	void addMidiEvent(MidiBuffer& mb, int a);
 
+	bool isEnabled;
+
 private:
 
 	void processBlock (AudioSampleBuffer &buffer, MidiBuffer &midiMessages);
 
 	
 
-	bool isEnabled;
+	
 	
 	int getNumSamples(MidiBuffer&);
 	void setNumSamples(MidiBuffer&, int);

Source/Processors/SourceNode.cpp

@@ -15,50 +15,71 @@
 
 SourceNode::SourceNode(const String& name_)
 	: GenericProcessor(name_),
-	  dataThread(0)
+	  dataThread(0),
+	  sourceCheckInterval(1500)
 {
 	if (getName().equalsIgnoreCase("Intan Demo Board")) {
-		setNumOutputs(16);
-		setNumInputs(0);
+		dataThread = new IntanThread();
 	} else if (getName().equalsIgnoreCase("Custom FPGA")) {
-		setNumOutputs(32);
-		setNumInputs(0);
+		dataThread = new FPGAThread();
 	} else if (getName().equalsIgnoreCase("File Reader")) {
-		setNumOutputs(16);
-		setNumInputs(0);
+		dataThread = new FileReaderThread();
+	}
+
+	setNumInputs(0);
+
+	if (dataThread != 0) {
+		setNumOutputs(dataThread->getNumChannels());
+		inputBuffer = dataThread->getBufferAddress();
+	} else {
+		setNumOutputs(10);
 	}
 
 	setPlayConfigDetails(getNumInputs(), getNumOutputs(), 44100.0, 128);
 
-	//sendActionMessage("Intan Demo Board source created.");
-	//sendMessage("Intan Demo Board source created.");
+	if (dataThread != 0)
+	{
+		if (!dataThread->foundInputSource())
+		{
+			enabledState(false);
+		}
+	} else {
+		enabledState(false);
+	}
+
+	// check for input source every two seconds
+	startTimer(sourceCheckInterval); 
 
 }
 
 SourceNode::~SourceNode() 
 {
+	if (dataThread != 0)
+		deleteAndZero(dataThread);
+
 	config->removeDataSource(this);	
 }
 
 float SourceNode::getSampleRate()
 {
-	if (getName().equalsIgnoreCase("Intan Demo Board")) {
-		return 25000.0;
-	} else if (getName().equalsIgnoreCase("Custom FPGA")) {
-		return 25000.0;
-	} else if (getName().equalsIgnoreCase("File Reader")) {
-		return 40000.0;
-	} else {
-		return 44100.0;
-	}
 
+	if (dataThread != 0)
+		return dataThread->getSampleRate();
+	else
+		return 44100.0;
 }
 
-// void SourceNode::setName(const String name_)
-// {
-// 	name = name_;
+void SourceNode::enabledState(bool t)
+{
+	if (t && !dataThread->foundInputSource())
+	{
+		isEnabled = false;
+	} else {
+		isEnabled = t;
+	}
 
-// 	// Source node type determines configuration info
+
+}
 
 void SourceNode::setConfiguration(Configuration* cf)
 {
@@ -114,57 +135,78 @@ AudioProcessorEditor* SourceNode::createEditor()
 	//return 0;
 }
 
-// void SourceNode::setSourceNode(GenericProcessor* sn) 
-// {
-// 	sourceNode = 0;
-// }
-
-// void SourceNode::setDestNode(GenericProcessor* dn)
-// {
-// 	destNode = dn;
-// 	if (dn != 0)
-// 		dn->setSourceNode(this);
-// }
-
-//void SourceNode::createEditor() {
-	
-//}
+void SourceNode::timerCallback()
+{
+	if (dataThread->foundInputSource() && !isEnabled)
+	{
+		std::cout << "Input source found." << std::endl;
+		//stopTimer(); // check for input source every two seconds
+		enabledState(true);
+		GenericEditor* ed = (GenericEditor*) getEditor();
+		viewport->updateVisibleEditors(ed, 4);
+	} else if (!dataThread->foundInputSource() && isEnabled) {
+		std::cout << "No input source found." << std::endl;
+		enabledState(false);
+		GenericEditor* ed = (GenericEditor*) getEditor();
+		viewport->updateVisibleEditors(ed, 4);
+	}
+}
 
 bool SourceNode::enable() {
 	
 	std::cout << "Source node received enable signal" << std::endl;
 
-	bool return_code = true;
+	if (dataThread != 0)
+	{
+		if (dataThread->foundInputSource())
+		{
+			dataThread->startAcquisition();
+			return true;
+		} else {
+			return false;
+		}
+	} else {
+		return false;
+	}
 
-	if (getName().equalsIgnoreCase("Intan Demo Board")) {
+	stopTimer();
+
+	// bool return_code = true;
+
+	// if (getName().equalsIgnoreCase("Intan Demo Board")) {
 		
-		dataThread = new IntanThread();
-		inputBuffer = dataThread->getBufferAddress();
-		return_code = dataThread->threadStarted();
+	// 	dataThread = new IntanThread();
+	// 	inputBuffer = dataThread->getBufferAddress();
+	// 	return_code = dataThread->threadStarted();
 
-		if (!return_code)
-			deleteAndZero(dataThread);
+	// 	if (!return_code)
+	// 		deleteAndZero(dataThread);
 
-	} else if (getName().equalsIgnoreCase("Custom FPGA")) {
-		dataThread = new FPGAThread();
-		inputBuffer = dataThread->getBufferAddress();
-	} else if (getName().equalsIgnoreCase("File Reader")) {
-		dataThread = new FileReaderThread();
-		inputBuffer = dataThread->getBufferAddress();
-	}
+	// } else if (getName().equalsIgnoreCase("Custom FPGA")) {
+	// 	dataThread = new FPGAThread();
+	// 	inputBuffer = dataThread->getBufferAddress();
+	// } else if (getName().equalsIgnoreCase("File Reader")) {
+	// 	dataThread = new FileReaderThread();
+	// 	inputBuffer = dataThread->getBufferAddress();
+	// }
 
-	return return_code;
+	// return return_code;
 
 }
 
 bool SourceNode::disable() {
-	
+
 	std::cout << "Source node received disable signal" << std::endl;
 
-	if (dataThread != 0) {
-		delete dataThread;
-		dataThread = 0;
-	}
+	if (dataThread != 0)
+		dataThread->stopAcquisition();
+	
+	startTimer(2000);
+
+	// if (dataThread != 0) {
+	// 	delete dataThread;
+	// 	dataThread = 0;
+	// }
 
 	return true;
 }

Source/Processors/SourceNode.h

@@ -20,7 +20,8 @@
 #include "DataThreads/FileReaderThread.h"
 #include "GenericProcessor.h"
 
-class SourceNode : public GenericProcessor
+class SourceNode : public GenericProcessor,
+				   public Timer
 
 {
 public:
@@ -29,6 +30,8 @@ public:
 	SourceNode(const String& name);
 	~SourceNode();
 
+	void enabledState(bool t);
+
 	//void setName(const String name_);
 	
 	void prepareToPlay (double sampleRate, int estimatedSamplesPerBlock);
@@ -54,7 +57,10 @@ public:
 	
 private:
 
+	int sourceCheckInterval;
+
 	//const String name;
+	void timerCallback();
 
 	DataThread* dataThread;
 	DataBuffer* inputBuffer;