Add support for writing explicit per-sample timestamps in Record Node

Author: anjaldoshi

Source/Processors/RecordNode/DataQueue.cpp

@@ -185,6 +185,30 @@ float DataQueue::writeSynchronizedTimestamps (double start, double step, int des
     return 1.0f - (float) m_FTSFifos[destChannel]->getFreeSpace() / (float) m_FTSFifos[destChannel]->getTotalSize();
 }
 
+float DataQueue::writeSynchronizedTimestamps (const double* timestamps, int destChannel, int nSamples)
+{
+    int index1, size1, index2, size2;
+
+    m_FTSFifos[destChannel]->prepareToWrite (nSamples, index1, size1, index2, size2);
+
+    if ((size1 + size2) < nSamples)
+    {
+        LOGE (__FUNCTION__, " Recording Data Queue Overflow: sz1: ", size1, " sz2: ", size2, " nSamples: ", nSamples);
+    }
+
+    double* writePtr = m_FTSBuffer.getWritePointer (destChannel);
+
+    if (size1 > 0)
+        memcpy (writePtr + index1, timestamps, (size_t) size1 * sizeof (double));
+
+    if (size2 > 0)
+        memcpy (writePtr + index2, timestamps + size1, (size_t) size2 * sizeof (double));
+
+    m_FTSFifos[destChannel]->finishedWrite (size1 + size2);
+
+    return 1.0f - (float) m_FTSFifos[destChannel]->getFreeSpace() / (float) m_FTSFifos[destChannel]->getTotalSize();
+}
+
 float DataQueue::writeChannel (const AudioBuffer<float>& buffer,
                                int srcChannel,
                                int destChannel,

Source/Processors/RecordNode/DataQueue.h

@@ -88,6 +88,9 @@ class DataQueue
     /** Writes an array of timestamps for one stream */
     float writeSynchronizedTimestamps (double start, double step, int destChannel, int nSamples);
 
+    /** Writes explicit per-sample timestamps for one stream */
+    float writeSynchronizedTimestamps (const double* timestamps, int destChannel, int nSamples);
+
     /** Returns the number of samples available to read (minimum across all channels) */
     int getNumSamplesReady() const;
 

Source/Processors/RecordNode/RecordNode.cpp

@@ -990,15 +990,7 @@ void RecordNode::handleTTLEvent (TTLEventPtr event)
         size_t size = event->getChannelInfo()->getDataSize() + event->getChannelInfo()->getTotalEventMetadataSize() + EVENT_BASE_SIZE;
         uint16 streamId = event->getStreamId();
 
-        double ts = -1.0;
-        if (synchronizer.streamGeneratesTimestamps (streamKey))
-        {
-            ts = getFirstTimestampForBlock (streamId) + (sampleNumber - getFirstSampleNumberForBlock (streamId)) / getDataStream (streamId)->getSampleRate();
-        }
-        else
-        {
-            ts = synchronizer.convertSampleNumberToTimestamp (streamKey, sampleNumber);
-        }
+        double ts = resolveRecordedTimestamp (streamKey, streamId, sampleNumber);
 
         HeapBlock<char> buffer (size);
         event->setTimestampInSeconds (ts);
@@ -1020,7 +1012,7 @@ void RecordNode::handleEvent (const EventChannel* eventInfo, const EventPacket&
 
         String streamKey = getDataStream (eventInfo->getStreamId())->getKey();
 
-        Event::setTimestampInSeconds (packet, synchronizer.convertSampleNumberToTimestamp (streamKey, sampleNumber));
+        Event::setTimestampInSeconds (packet, resolveRecordedTimestamp (streamKey, eventInfo->getStreamId(), sampleNumber));
 
         eventQueue->addEvent (packet, sampleNumber, eventIndex);
     }
@@ -1037,15 +1029,7 @@ void RecordNode::handleSpike (SpikePtr spike)
         uint16 streamId = spike->getStreamId();
         int64 sampleNumber = spike->getSampleNumber();
 
-        double ts = -1.0;
-        if (synchronizer.streamGeneratesTimestamps (streamKey))
-        {
-            ts = getFirstTimestampForBlock (streamId) + (sampleNumber - getFirstSampleNumberForBlock (streamId)) / getDataStream (streamId)->getSampleRate();
-        }
-        else
-        {
-            ts = synchronizer.convertSampleNumberToTimestamp (streamKey, sampleNumber);
-        }
+        double ts = resolveRecordedTimestamp (streamKey, streamId, sampleNumber);
 
         spike->setTimestampInSeconds (ts);
         writeSpike (spike, spike->getChannelInfo());
@@ -1143,19 +1127,36 @@ void RecordNode::process (AudioBuffer<float>& buffer)
                 {
                     first = synchronizer.convertSampleNumberToTimestamp (streamKey, sampleNumber);
                     second = synchronizer.convertSampleNumberToTimestamp (streamKey, sampleNumber + 1);
+
+                    dataQueues[streamIndex]->writeSynchronizedTimestamps (
+                        first,
+                        second - first,
+                        0,
+                        numSamples);
                 }
                 else
                 {
-                    first = getFirstTimestampForBlock (streamId);
-                    second = first + 1 / stream->getSampleRate();
+                    const double* blockTimestamps = getTimestampsForBlock (streamId);
+
+                    if (blockTimestamps != nullptr)
+                    {
+                        first = blockTimestamps[0];
+                        dataQueues[streamIndex]->writeSynchronizedTimestamps (blockTimestamps, 0, (int) numSamples);
+                    }
+                    else
+                    {
+                        first = getFirstTimestampForBlock (streamId);
+                        second = first + 1 / stream->getSampleRate();
+
+                        dataQueues[streamIndex]->writeSynchronizedTimestamps (
+                            first,
+                            second - first,
+                            0,
+                            numSamples);
+                    }
+
                     synchronizer.setHardwareTimestamp (sampleNumber, first, streamKey);
                 }
-                // Each per-stream queue has only 1 timestamp stream (index 0)
-                dataQueues[streamIndex]->writeSynchronizedTimestamps (
-                    first,
-                    second - first,
-                    0, // timestamp stream index within this queue
-                    numSamples);
             }
 
             if (numSamples > 0 && recordChanCount > 0 && streamSourceChannels[streamIndex] != nullptr)
@@ -1210,6 +1211,22 @@ void RecordNode::process (AudioBuffer<float>& buffer)
     }
 }
 
+double RecordNode::resolveRecordedTimestamp (const String& streamKey, uint16 streamId, int64 sampleNumber)
+{
+    if (synchronizer.streamGeneratesTimestamps (streamKey))
+    {
+        double timestamp = -1.0;
+
+        if (getTimestampForSample (streamId, sampleNumber, timestamp))
+            return timestamp;
+
+        return getFirstTimestampForBlock (streamId)
+               + (sampleNumber - getFirstSampleNumberForBlock (streamId)) / getDataStream (streamId)->getSampleRate();
+    }
+
+    return synchronizer.convertSampleNumberToTimestamp (streamKey, sampleNumber);
+}
+
 // called in RecordNode::handleSpike
 void RecordNode::writeSpike (const Spike* spike, const SpikeChannel* spikeElectrode)
 {

Source/Processors/RecordNode/RecordNode.h

@@ -256,6 +256,8 @@ class TESTABLE RecordNode : public GenericProcessor,
     static bool overrideTimestampWarningShown;
 
 private:
+    double resolveRecordedTimestamp (const String& streamKey, uint16 streamId, int64 sampleNumber);
+
     /** Handles other types of events (text, sync texts, etc.) */
     void handleEvent (const EventChannel* channel, const EventPacket& eventPacket);