Restore events and spikes recording in Open Ephys format

Author: jsiegle

Source/Processors/RecordNode/OpenEphysFormat/OriginalRecording.cpp

@@ -69,7 +69,7 @@ String OriginalRecording::getEngineID() const
 
 void OriginalRecording::addSpikeElectrode(int index, const SpikeChannel* elec)
 {
-	spikeFileArray.add(nullptr);
+	//spikeFileArray.add(nullptr); // deprecated
 }
 
 void OriginalRecording::resetChannels()
@@ -91,7 +91,7 @@ void OriginalRecording::openFiles(File rootFolder, int experimentNumber, int rec
 	processorArray.clear();
 	lastProcId = 0;
 
-	//openFile(rootFolder, getEventChannel(0), 0);
+	openFile(rootFolder, getEventChannel(0), 0);
 	openMessageFile(rootFolder);
 
 	int nChannels = getNumRecordedChannels();
@@ -103,10 +103,15 @@ void OriginalRecording::openFiles(File rootFolder, int experimentNumber, int rec
 		blockIndex.add(0);
 		samplesSinceLastTimestamp.add(0);
 	}
-	for (int i = 0; i < spikeFileArray.size(); i++)
+
+	int nSpikes = getNumRecordedSpikes();
+
+	for (int i = 0; i < nSpikes; i++)
 	{
+		spikeFileArray.add(nullptr);
 		openSpikeFile(rootFolder, getSpikeChannel(i), i);
 	}
+
 }
 
 void OriginalRecording::openFile(File rootFolder, const InfoObjectCommon* ch, int channelIndex)
@@ -214,11 +219,14 @@ void OriginalRecording::openSpikeFile(File rootFolder, const SpikeChannel* elec,
 
 	if (!fileExists)
 	{
+
 		String header = generateSpikeHeader(elec);
 		fwrite(header.toUTF8(), 1, header.getNumBytesAsUTF8(), spFile);
+		std::cout << "Wrote header." << std::endl;
 	}
 	diskWriteLock.exit();
 	spikeFileArray.set(channelIndex, spFile);
+	std::cout << "Added file." << std::endl;
 
 }
 
@@ -364,7 +372,7 @@ String OriginalRecording::generateSpikeHeader(const SpikeChannel* elec)
 	header += "';\n";
 
 	header += "header.num_channels = ";
-	header += String(elec->getNumChannels());
+	header += String(c);
 	header += ";\n";
 
 	header += "header.sampleRate = ";
@@ -592,6 +600,7 @@ void OriginalRecording::closeFiles()
 		}
 	}
 	fileArray.clear();
+
 	blockIndex.clear();
 	samplesSinceLastTimestamp.clear();
 	for (int i = 0; i < spikeFileArray.size(); i++)
@@ -630,12 +639,18 @@ void OriginalRecording::closeFiles()
 
 void OriginalRecording::writeSpike(int electrodeIndex, const SpikeEvent* spike)
 {
+	//std::cout << "Electrode index: " << electrodeIndex << std::endl;
+
 	if (spikeFileArray[electrodeIndex] == nullptr)
 		return;
 
+	//std::cout << "Got spike" << std::endl;
+
 	HeapBlock<char> spikeBuffer;
 	const SpikeChannel* channel = getSpikeChannel(electrodeIndex);
 
+	//std::cout << "Got spike channel" << std::endl;
+
 	int totalSamples = channel->getTotalSamples() * channel->getNumChannels();
 	int numChannels = channel->getNumChannels();
 	int chanSamples = channel->getTotalSamples();
@@ -658,6 +673,8 @@ void OriginalRecording::writeSpike(int electrodeIndex, const SpikeEvent* spike)
 	zeromem(spikeBuffer.getData() + 32, 2 * sizeof(float));
 	*reinterpret_cast<uint16*>(spikeBuffer.getData() + 40) = channel->getSampleRate();
 
+	//std::cout << "Allocated memory" << std::endl;
+
 	int ptrIdx = 0;
 	uint16* dataIntPtr = reinterpret_cast<uint16*>(spikeBuffer.getData() + 42);
 	const float* spikeDataPtr = spike->getDataPointer();
@@ -683,6 +700,8 @@ void OriginalRecording::writeSpike(int electrodeIndex, const SpikeEvent* spike)
 		ptrIdx += sizeof(int16);
 	}
 
+	//std::cout << "Starting disk write" << std::endl;
+
 	diskWriteLock.enter();
 
 	fwrite(spikeBuffer, 1, totalBytes, spikeFileArray[electrodeIndex]);
@@ -693,6 +712,8 @@ void OriginalRecording::writeSpike(int electrodeIndex, const SpikeEvent* spike)
 		spikeFileArray[electrodeIndex]); // ptr to FILE object
 
 	diskWriteLock.exit();
+
+	//std::cout << "Wrote to file" << std::endl;
 }
 
 void OriginalRecording::writeXml()

Source/Processors/RecordNode/RecordNode.cpp

@@ -71,6 +71,7 @@ RecordNode::~RecordNode()
 
 void RecordNode::addInputChannel(const GenericProcessor* sourceNode, int chan)
 {
+	// not getting called
 
 	if (chan != AccessClass::getProcessorGraph()->midiChannelIndex)
 	{
@@ -85,13 +86,12 @@ void RecordNode::addInputChannel(const GenericProcessor* sourceNode, int chan)
 	}
 	else
 	{
-
 		for (int n = 0; n < sourceNode->getTotalEventChannels(); n++)
 		{
 			const EventChannel* orig = sourceNode->getEventChannel(n);
 			//only add to the record node the events originating from this processor, to avoid duplicates
 			if (orig->getSourceNodeID() == sourceNode->getNodeId())
-				eventChannelArray.add(new EventChannel(*orig));
+				nonOwnedEventChannelArray.add(new EventChannel(*orig));
 
 		}
 
@@ -357,6 +357,24 @@ void RecordNode::updateSubprocessorMap()
             synchronizer->setSyncChannel(chan->getSourceNodeID(), chan->getSubProcessorIdx(), ch);
         }
 
+    } 
+
+    for (int ch = 0; ch < nonOwnedEventChannelArray.size(); ch++)
+    {
+
+        EventChannel* chan = nonOwnedEventChannelArray[ch];
+        int sourceID = chan->getSourceNodeID();
+        int subProcID = chan->getSubProcessorIdx();
+
+        eventMap[sourceID][subProcID] = chan->getNumChannels();
+
+        if (dataChannelStates[sourceID][subProcID].size() && !syncChannelMap[sourceID][subProcID])
+        {
+            syncOrderMap[sourceID][subProcID] = ch;
+            syncChannelMap[sourceID][subProcID] = 0;
+            synchronizer->setSyncChannel(chan->getSourceNodeID(), chan->getSubProcessorIdx(), ch);
+        }
+
     }
 
 }

Source/Processors/RecordNode/RecordNode.h

@@ -122,6 +122,8 @@ class RecordNode : public GenericProcessor, public FilenameComponentListener
 	std::vector<std::vector<int>> subProcessorMap;
 	std::vector<int> startRecChannels;
 
+    Array<EventChannel*> nonOwnedEventChannelArray;
+
     bool isSyncReady;
 
     //TODO: Need to validate these new methods

Source/Processors/RecordNode/RecordThread.cpp

@@ -208,14 +208,24 @@ void RecordThread::writeSynchronizedData(const AudioSampleBuffer& dataBuffer, co
 	std::vector<SpikeMessagePtr> spikes;
 	int nSpikes = m_spikeQueue->getEvents(spikes, maxSpikes);
 
+	int total_null = 0;
+	int total_real = 0;
+
 	for (int sp = 0; sp < nSpikes; ++sp)
 	{
 		if (spikes[sp] == NULL)
-			std::cout << "Got NULL" << std::endl;
-		else
+		{
+			total_null++;
+		}
+		else{
+
 			m_engine->writeSpike(spikes[sp]->getExtra(), &spikes[sp]->getData());
+			total_real++;
+		}
 		//m_engine->writeSpike(0, &spikes[sp]->getData());
 	}
+	if (total_null > 0)
+		std::cout << "Total null: " << total_null << ", total real: " << total_real << std::endl;
 }
 
 void RecordThread::writeData(const AudioSampleBuffer& dataBuffer, int maxSamples, int maxEvents, int maxSpikes, bool lastBlock)
@@ -277,10 +287,19 @@ void RecordThread::writeData(const AudioSampleBuffer& dataBuffer, int maxSamples
 	std::vector<SpikeMessagePtr> spikes;
 	int nSpikes = m_spikeQueue->getEvents(spikes, maxSpikes);
 
+	//int total_null = 0;
+	//int total_real = 0;
+
 	for (int sp = 0; sp < nSpikes; ++sp)
 	{
-		m_engine->writeSpike(spikes[sp]->getExtra(), &spikes[sp]->getData());
+		if (spikes[sp] != NULL)
+		{
+			m_engine->writeSpike(spikes[sp]->getExtra(), &spikes[sp]->getData());
+		}
 	}
+
+	//if (total_null > 0)
+	//	std::cout << "Total null: " << total_null << ", total real: " << total_real << std::endl;
 
 }