Restores spikes and event writing in Open Ephys format

Plugins/IntanRecordingController/RHD2000Editor.cpp

@@ -737,6 +737,7 @@ void RHD2000Editor::handleAsyncUpdate()
 	canvas->updateImpedance(impedanceData->streams, impedanceData->channels, impedanceData->magnitudes, impedanceData->phases);
     if (saveImpedances)
     {
+        // this may not work with new Record Node architecture
 		CoreServices::RecordNode::createNewrecordingDir();
 
 		String path(CoreServices::RecordNode::getRecordingPath().getFullPathName()

Source/Processors/MessageCenter/MessageCenter.cpp

@@ -36,22 +36,36 @@ GenericProcessor("Message Center"), newEventAvailable(false), isRecording(false)
                          0, // number of outputs
                          44100.0, // sampleRate
                          128);    // blockSize
+
+    eventChannel = nullptr;
+
 }
 
 MessageCenter::~MessageCenter()
 {
 
 }
 
-void MessageCenter::addSpecialProcessorChannels(Array<EventChannel*>& channels) 
+void MessageCenter::addSpecialProcessorChannels() 
 {
-	clearSettings();
-	EventChannel* chan = new EventChannel(EventChannel::TEXT, 1, MAX_MSG_LENGTH, CoreServices::getGlobalSampleRate(), this, 0);
-	chan->setName("GUI Messages");
-	chan->setDescription("Messages from the GUI Message Center");
-	channels.add(chan);
-	eventChannelArray.add(new EventChannel(*chan));
-	updateChannelIndexes();
+
+    if (eventChannel == nullptr)
+    {
+
+        clearSettings();
+
+        eventChannel = new EventChannel(EventChannel::TEXT, 
+                                            1, 
+                                            MAX_MSG_LENGTH, 
+                                            CoreServices::getGlobalSampleRate(), 
+                                            this, 0);
+
+        eventChannel->setName("GUI Messages");
+        eventChannel->setDescription("Messages from the GUI Message Center");
+        eventChannelArray.add(new EventChannel(*eventChannel));
+
+        updateChannelIndexes();
+    }
 }
 
 AudioProcessorEditor* MessageCenter::createEditor()
@@ -63,6 +77,11 @@ AudioProcessorEditor* MessageCenter::createEditor()
 
 }
 
+const EventChannel* MessageCenter::getMessageChannel()
+{
+    return getEventChannel(0);
+}
+
 void MessageCenter::setParameter(int parameterIndex, float newValue)
 {
     if (isRecording)
@@ -114,8 +133,10 @@ void MessageCenter::process(AudioSampleBuffer& buffer)
 		TextEventPtr event = TextEvent::createTextEvent(getEventChannel(0), CoreServices::getGlobalTimestamp(), eventString);
 		addEvent(getEventChannel(0), event, 0);
 
+        std::cout << "Message Center added event." << std::endl;
+
         newEventAvailable = false;
     }
 
 
-}
+}

Source/Processors/MessageCenter/MessageCenter.h

@@ -61,6 +61,8 @@ class MessageCenter : public GenericProcessor
     /** A pointer to the Message Center editor. */
     ScopedPointer<MessageCenterEditor> messageCenterEditor;
 
+    const EventChannel* getMessageChannel();
+
     bool enable() override;
     bool disable() override;
 
@@ -75,13 +77,15 @@ class MessageCenter : public GenericProcessor
         needsToSendTimestampMessage = false;
     }
 
-	void addSpecialProcessorChannels(Array<EventChannel*>& channel);
+	void addSpecialProcessorChannels();
 private:
 
     bool newEventAvailable;
     bool isRecording;
     bool needsToSendTimestampMessage;
 
+    ScopedPointer<EventChannel> eventChannel;
+
     JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR(MessageCenter);
 
 };

Source/Processors/MessageCenter/MessageCenterEditor.h

@@ -64,6 +64,8 @@ class MessageCenterEditor : public AudioProcessorEditor,
 
     String getLabelString();
 
+    MessageCenter* messageCenter;
+
 private:
 
     void buttonClicked(Button* button);
@@ -85,8 +87,6 @@ class MessageCenterEditor : public AudioProcessorEditor,
     /** A JUCE button used to send messages. */
     ScopedPointer<Button> sendMessageButton;
 
-    MessageCenter* messageCenter;
-
     Colour incomingBackground;
     Colour outgoingBackground;
 

Source/Processors/ProcessorGraph/ProcessorGraph.cpp

@@ -346,6 +346,9 @@ void ProcessorGraph::updateConnections(Array<SignalChainTabButton*, CriticalSect
                 //TODO: This is will be removed when probe based audio node added. 
                 connectProcessorToAudioNode(source);
 
+                //if (source->isRecordNode())
+                 //   connectProcessorToMessageCenter(source);
+
                 // find the next dest that's not a merger or splitter
                 GenericProcessor* prev = source;
 
@@ -447,14 +450,16 @@ void ProcessorGraph::updateConnections(Array<SignalChainTabButton*, CriticalSect
             connectProcessors(conn.source, dest, conn.connectContinuous, conn.connectEvents);
         }
     }
+
+    //OwnedArray<EventChannel> extraChannels;
+    getMessageCenter()->addSpecialProcessorChannels();
 
 	getAudioNode()->updatePlaybackBuffer();
 
-	Array<EventChannel*> extraChannels;
-	getMessageCenter()->addSpecialProcessorChannels(extraChannels);
-
+    /*
     for (auto& recordNode : getRecordNodes())
         recordNode->addSpecialProcessorChannels(extraChannels);
+    */
 
 } // end method
 
@@ -553,6 +558,19 @@ void ProcessorGraph::connectProcessorToAudioNode(GenericProcessor* source)
 
 }
 
+
+void ProcessorGraph::connectProcessorToMessageCenter(GenericProcessor* source)
+{
+
+    // connect event channel
+    addConnection(getMessageCenter()->getNodeId(),    // sourceNodeID
+                  midiChannelIndex,       // sourceNodeChannelIndex
+                  source->getNodeId(),          // destNodeID
+                  midiChannelIndex);      // destNodeChannelIndex
+
+}
+
+
 GenericProcessor* ProcessorGraph::createProcessorFromDescription(Array<var>& description)
 {
 	GenericProcessor* processor = nullptr;

Source/Processors/ProcessorGraph/ProcessorGraph.h

@@ -119,6 +119,7 @@ class ProcessorGraph    : public AudioProcessorGraph
     void connectProcessors(GenericProcessor* source, GenericProcessor* dest,
         bool connectContinuous, bool connectEvents);
     void connectProcessorToAudioNode(GenericProcessor* source);
+    void connectProcessorToMessageCenter(GenericProcessor* source);
 
 	int64 m_startSoftTimestamp{ 0 };
 	const GenericProcessor* m_timestampSource{ nullptr };

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.cpp

@@ -695,4 +695,4 @@ RecordEngineManager* BinaryRecording::getEngineManager()
 void BinaryRecording::setParameter(EngineParameter& parameter)
 {
 	boolParameter(0, m_saveTTLWords);
-}
+}

Source/Processors/RecordNode/BinaryFormat/NpyFile.cpp

@@ -79,6 +79,19 @@ bool NpyFile::openFile(String path)
     // output stream buffer size defaults to 32768 bytes, but is irrelevant because
     // each updateHeader() call triggers a m_file->flush() to disk:
     m_file = file.createOutputStream();
+
+    if (m_file == nullptr)
+    {
+        LOGD("FAILED to open file @", path);
+    }
+    else
+    {
+        String pad = "";
+        for (int i = 0; i < 162 - path.length(); i++)
+            pad += " ";
+        LOGD("Successfully opened file @", path, pad, m_file);
+    }
+
     if (!m_file)
         return false;
 
@@ -156,7 +169,8 @@ void NpyFile::writeHeader(const Array<NpyType>& typeList)
 
 void NpyFile::updateHeader()
 {
-    if (m_file != NULL)
+
+    if (true)
     {
         // overwrite the shape part of the header - even without explicitly calling
         // m_file->flush(), overwriting seems to trigger a flush to disk,

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;
 
 }
 
@@ -257,8 +265,8 @@ String OriginalRecording::getFileName(int channelIndex)
 {
 	String filename;
 	const DataChannel* ch = getDataChannel(channelIndex);
-
-	filename += String(static_cast<int>(ch->getCurrentNodeID()));
+    
+	filename += String(static_cast<int>(ch->getSourceNodeID()));
 	filename += "_";
 	if (renameFiles)
 		filename += renamedPrefix + String(getDataChannel(channelIndex)->getCurrentNodeChannelIdx() + 1);
@@ -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/RecordEngine.h

@@ -65,7 +65,7 @@ class PLUGIN_API RecordEngine
 	/** All the public methods (except registerManager) are called by RecordNode or RecordingThread:
 	When acquisition starts (in the specified order):
 	1-resetChannels
-	2-registerProcessor, addChannel, registerSpikeSource, addspikeelectrode
+	2-registerProcessor, addChannel, registerSpikeSource, addSpikeElectrode
 	3-configureEngine (which calls setParameter)
 	3-startAcquisition
 	When recording starts (in the specified order):