Finish Harp sync; enabled event translation across streams with hardware timestamps

Author: jsiegle

Source/Processors/EventTranslator/EventTranslator.cpp

@@ -266,9 +266,6 @@ void EventTranslator::handleTTLEvent (TTLEventPtr event)
 
                 int64 firstSampleNumberForBlock = getFirstSampleNumberForBlock (streamId);
 
-                if (newSampleNumber < firstSampleNumberForBlock)
-                    newSampleNumber = firstSampleNumberForBlock;
-
                 //std::cout << "translated sample number (" << streamId << "): " << newSampleNumber - firstSampleNumberForBlock << std::endl;
                 //std::cout << std::endl;
 
@@ -279,4 +276,34 @@ void EventTranslator::handleTTLEvent (TTLEventPtr event)
             }
         }
     }
+
+    if (synchronizer.getStatus (eventStreamKey) == SyncStatus::HARDWARE_SYNCED)
+    {
+        const bool state = event->getState();
+
+        double timestamp = synchronizer.convertSampleNumberToTimestamp (eventStreamKey, sampleNumber);
+
+        for (auto stream : getDataStreams())
+        {
+            const uint16 streamId = stream->getStreamId();
+            const String streamKey = stream->getKey();
+
+            if (streamKey == eventStreamKey || synchronizer.getStatus (streamKey) != SyncStatus::HARDWARE_SYNCED)
+                continue; // don't translate events to non-hardware-synced streams
+
+            int64 newSampleNumber = synchronizer.convertTimestampToSampleNumber (streamKey, timestamp);
+
+            //std::cout << "new sample number (" << streamKey << "): " << newSampleNumber << std::endl;
+
+            int64 firstSampleNumberForBlock = getFirstSampleNumberForBlock (streamId);
+
+            //std::cout << "translated sample number (" << streamKey << "): " << newSampleNumber - firstSampleNumberForBlock << std::endl;
+            //std::cout << std::endl;
+
+            TTLEventPtr translatedEvent =
+                settings[streamId]->createEvent (newSampleNumber, timestamp, ttlLine, state);
+
+            addEvent (translatedEvent, newSampleNumber - firstSampleNumberForBlock);
+        }
+    }
 }

Source/Processors/Parameter/ParameterEditor.cpp

@@ -911,7 +911,22 @@ void SyncControlButton::paintButton (Graphics& g, bool isMouseOver, bool isButto
         }
         case SyncStatus::HARP_CLOCK:
         {
-            Colour harpClockColour = Colour (200, 150, 255); // Light purplse
+            Colour harpClockColour = Colour (200, 150, 255); // Light purple
+
+            if (isMouseOver)
+            {
+                g.setColour (harpClockColour);
+            }
+            else
+            {
+                g.setColour (harpClockColour.darker (0.5f));
+            }
+            break;
+        }
+
+        case SyncStatus::HARP_DETECTING:
+        {
+            Colour harpClockColour = Colour (105, 87, 107); // Purple-gray
 
             if (isMouseOver)
             {

Source/Processors/RecordNode/BinaryFormat/BinaryRecording.cpp

@@ -801,7 +801,7 @@ void BinaryRecording::writeContinuousDataBatch (const int* writeChannels,
         int newSamples = jmax (numSamples, m_batchBufferSamples);
         int newChannels = jmax (numChannels, m_batchBufferChannels);
 
-        LOGC ("BinaryRecording::writeContinuousDataBatch: Resizing batch buffer to ", 
+        LOGD ("BinaryRecording::writeContinuousDataBatch: Resizing batch buffer to ", 
               newChannels, " channels x ", newSamples, " samples");
 
         m_batchIntBuffer.malloc (newSamples * newChannels);

Source/Processors/RecordNode/RecordNode.cpp

@@ -1157,6 +1157,7 @@ void RecordNode::process (AudioBuffer<float>& buffer)
                 {
                     first = getFirstTimestampForBlock (streamId);
                     second = first + 1 / stream->getSampleRate();
+                    synchronizer.setHardwareTimestamp (sampleNumber, first, streamKey);
                 }
                 // Each per-stream queue has only 1 timestamp stream (index 0)
                 dataQueues[streamIndex]->writeSynchronizedTimestamps (

Source/Processors/Synchronizer/Synchronizer.cpp

@@ -187,7 +187,7 @@ void SyncStream::reset (String mainStreamKey)
     latestSyncSampleNumber = 0;
     latestGlobalSyncTime = 0.0;
     latestSyncMillis = -1;
-    
+
     // Reset Harp detection state
     harpState = HarpDetectionState::IDLE;
     completedBarcodes.clear();
@@ -211,10 +211,21 @@ void SyncStream::reset (String mainStreamKey)
     }
     else
     {
-        actualSampleRate = -1.0;
-        globalStartTime = 0.0;
         overrideHardwareTimestamps = syncLine > -1; // override hardware timestamps for other streams if sync line is set
-        isSynchronized = generatesTimestamps && !overrideHardwareTimestamps; // if the stream generates its own timestamps, it is synchronized unless it overrides hardware timestamps
+
+        if (generatesTimestamps && ! overrideHardwareTimestamps)
+        {
+            // if the stream generates its own timestamps, it is synchronized unless it overrides hardware timestamps
+            actualSampleRate = expectedSampleRate;
+            isSynchronized = true;
+        }
+        else
+        {
+            actualSampleRate = -1.0;
+            isSynchronized = false;
+        }
+
+        globalStartTime = 0.0;
     }
 }
 
@@ -260,6 +271,43 @@ void SyncStream::addEvent (int64 sampleNumber, bool state)
     }
 }
 
+void SyncStream::setHardwareTimestamp (int64 sampleNumber, double timestamp)
+{
+
+    if (generatesTimestamps && ! overrideHardwareTimestamps)
+    {
+        latestSyncSampleNumber = sampleNumber;
+        latestGlobalSyncTime = timestamp;
+
+        if (! baselineMatchingPulse.complete)
+        {
+            baselineMatchingPulse.localSampleNumber = sampleNumber;
+            baselineMatchingPulse.globalTimestamp = timestamp;
+            baselineMatchingPulse.complete = true;
+
+            actualSampleRate = expectedSampleRate; // safe initial
+            globalStartTime = timestamp - (double (sampleNumber) / expectedSampleRate);
+            latestSyncMillis = Time::currentTimeMillis();
+            return;
+        }
+
+        double dt = timestamp - baselineMatchingPulse.globalTimestamp;
+        int64 ds = sampleNumber - baselineMatchingPulse.localSampleNumber;
+
+        if (dt <= 1e-6 || ds <= 0)
+            return;
+
+        double estimated = double (ds) / dt;
+
+        if (std::abs (estimated - expectedSampleRate) / expectedSampleRate < 0.01)
+        {
+            actualSampleRate = estimated;
+            latestSyncMillis = Time::currentTimeMillis();
+        }
+    }
+
+}
+
 double SyncStream::getLatestSyncTime()
 {
     //LOGC ("Getting latest sync time for stream ", streamKey, "...");
@@ -281,16 +329,26 @@ double SyncStream::getLatestSyncTime()
 
 double SyncStream::getSyncAccuracy()
 {
+
+    if (generatesTimestamps && ! overrideHardwareTimestamps)
+        return 0.0; // or compute error vs. expectedSampleRate drift
+
     if (pulses.size() > 0)
     {
 
-        //LOGD ("Sync accuracy for stream ", streamKey);
+        //if (!isMainStream)
+        //{
+            //LOGC ("Sync accuracy for stream ", streamKey);
+
+            //LOGC ("latestSyncSampleNumber: ", latestSyncSampleNumber);
+            //LOGC ("latestGlobalSyncTime: ", latestGlobalSyncTime);
+            //LOGC ("globalStartTime: ", globalStartTime);
+            //LOGC ("actualSampleRate: ", actualSampleRate);
+            //LOGC ("baselineMatchingPulse.globalTimestamp: ", baselineMatchingPulse.globalTimestamp);
+            //LOGC ("baselineMatchingPulse.localSampleNumber: ", baselineMatchingPulse.localSampleNumber);
+            //LOGC (" ");
+        //}
 
-        //LOGD ("latestSyncSampleNumber: ", latestSyncSampleNumber);
-        //LOGD ("latestGlobalSyncTime: ", latestGlobalSyncTime);
-        //LOGD ("globalStartTime: ", globalStartTime);
-        //LOGD ("actualSampleRate: ", actualSampleRate);
-        //LOGD ("baselineMatchingPulse.globalTimestamp: ", baselineMatchingPulse.globalTimestamp);
 
         // NEW CALCULATION:
         double estimatedGlobalTime = double(latestSyncSampleNumber - baselineMatchingPulse.localSampleNumber)
@@ -309,8 +367,8 @@ double SyncStream::getSyncAccuracy()
 
 void SyncStream::syncWith (const SyncStream* mainStream)
 {
-    //LOGD ("Synchronizing ", streamKey, " with ", mainStream->streamKey, "...");
-    //LOGD ("Expected sample rate: ", expectedSampleRate);
+    //LOGC ("Synchronizing ", streamKey, " with ", mainStream->streamKey, "...");
+    //LOGC ("Expected sample rate: ", expectedSampleRate);
 
     if (mainStream->pulses.size() < 2 || pulses.size() < 2)
     {
@@ -333,20 +391,19 @@ void SyncStream::syncWith (const SyncStream* mainStream)
                 {
                     if (comparePulses (pulse, mainPulse)) // putative match
                     {
-                        if (pulses.size() > localIndex + 3 && mainStream->pulses.size() > index + 3)
+                        if (pulses.size() > localIndex + 2 && mainStream->pulses.size() > index + 2)
                         {
-                            // previous three pulses also match
+                            // previous two pulses also match
                             if (comparePulses (pulses[localIndex + 1], mainStream->pulses[index + 1])
-                                && comparePulses (pulses[localIndex + 2], mainStream->pulses[index + 2])
-                                && comparePulses (pulses[localIndex + 3], mainStream->pulses[index + 3]))
+                                && comparePulses (pulses[localIndex + 2], mainStream->pulses[index + 2]))
                             {
                                 pulse.matchingPulseIndex = index;
                                 pulse.globalTimestamp = mainPulse.localTimestamp;
                                 latestSyncSampleNumber = pulse.localSampleNumber;
                                 latestGlobalSyncTime = pulse.globalTimestamp;
                                 latestSyncMillis = pulse.computerTimeMillis;
-                                //LOGD ("Pulse at ", pulse.localTimestamp, " matches with 4 main pulses at ", index);
-                                //LOGD ("latestSyncSampleNumber: ", latestSyncSampleNumber, ", latestGlobalSyncTime: ", latestGlobalSyncTime);
+                                //LOGC ("Pulse at ", pulse.localTimestamp, " matches with 3 main pulses at ", index);
+                                //LOGC ("latestSyncSampleNumber: ", latestSyncSampleNumber, ", latestGlobalSyncTime: ", latestGlobalSyncTime);
 
 
                                 if (baselineMatchingPulse.complete == false)
@@ -456,7 +513,9 @@ void SyncStream::syncWith (const SyncStream* mainStream)
 
                         {
                             globalStartTime = estimatedGlobalStartTime;
+
                             isSynchronized = true;
+
                         }
                     }
                     else
@@ -603,11 +662,15 @@ void SyncStream::attemptBarcodeDecoding()
     // decode last barcode
     if (harpDecoder.decodeBarcode(completedBarcodes.back(), expectedSampleRate))
     {
-       // LOGD ("Successful decoding...setting isHarpStream to true.");
-        isHarpStream = true;
-        isSynchronized = true;
 
-        
+        if (!isHarpStream)
+        {
+            LOGD ("Successful decoding...setting isHarpStream to true.");
+            isHarpStream = true;
+            isSynchronized = false;
+
+        }
+
         // Validate timing
         if (!validateBarcodeTimestamp(completedBarcodes.back()))
         {
@@ -648,12 +711,12 @@ void SyncStream::syncWithHarp()
 
     if (std::abs (estimatedSampleRate - expectedSampleRate) / expectedSampleRate < 0.05)
     {
-        LOGC (streamKey, " total barcodes = ", completedBarcodes.size(), "; estimated sample rate: ", estimatedSampleRate);
+        //LOGC (streamKey, " total barcodes = ", completedBarcodes.size(), "; estimated sample rate: ", estimatedSampleRate);
         actualSampleRate = estimatedSampleRate;
 
         // Calculate global start time
         //LOGD ("Estimated global start time: ", double (firstBarcode.encodedTime) - firstBarcode.localStartTimestamp);
-        globalStartTime = (double (firstBarcode.encodedTime) - firstBarcode.localStartTimestamp) / 1000;
+        globalStartTime = (double (firstBarcode.encodedTime) - firstBarcode.localStartTimestamp) / 1000; // divide by 1000 so that time appears in seconds in the stream info view
         baselineMatchingPulse.globalTimestamp = double (firstBarcode.encodedTime);
         baselineMatchingPulse.localSampleNumber = firstBarcode.localStartSample;
         latestSyncSampleNumber = lastBarcode.barcodeEvents.front().first;
@@ -666,7 +729,7 @@ void SyncStream::syncWithHarp()
     }
     else
     {
-        LOGC (streamKey, " estimated sample rate out of range; clearing Harp barcodes.");
+        LOGD (streamKey, " estimated sample rate out of range; clearing Harp barcodes.");
         isSynchronized = false;
         completedBarcodes.clear();
     }
@@ -796,6 +859,14 @@ void Synchronizer::addEvent (String streamKey,
     }
 }
 
+
+void Synchronizer::setHardwareTimestamp (int64 sampleNumber, double timestamp, String streamKey)
+{
+    const ScopedLock sl (synchronizerLock);
+    streams[streamKey]->setHardwareTimestamp (sampleNumber, timestamp);
+}
+
+
 double Synchronizer::convertSampleNumberToTimestamp (String streamKey, int64 sampleNumber)
 {
     if (streams[streamKey]->isSynchronized)
@@ -889,7 +960,12 @@ SyncStatus Synchronizer::getStatus (String streamKey)
         return SyncStatus::HARDWARE_SYNCED;
 
     if (streams[streamKey]->isHarpStream)
-        return SyncStatus::HARP_CLOCK;
+    {
+        if (isStreamSynced (streamKey))
+            return SyncStatus::HARP_CLOCK;
+        else
+            return SyncStatus::HARP_DETECTING;
+    }
 
     if (isStreamSynced (streamKey))
         return SyncStatus::SYNCED;

Source/Processors/Synchronizer/Synchronizer.h

@@ -201,6 +201,9 @@ class SyncStream
     /** Synchronize this stream with Harp timestamps */
     void syncWithHarp ();
 
+    /** Sets hardware timestamps (hardware-synced streams only) */
+    void setHardwareTimestamp (int64 sampleNumber, double timestamp);
+
     /** Compares pulses; returns true if a match is found */
     bool comparePulses (const SyncPulse& pulse1, const SyncPulse& pulse2);
 
@@ -397,6 +400,9 @@ class PLUGIN_API Synchronizer : public HighResolutionTimer
     /** Adds an event for a stream ID / line combination */
     void addEvent (String streamKey, int ttlLine, int64 sampleNumber, bool state);
 
+    /** Updates the synchronizer for a hardware-synced stream */
+    void setHardwareTimestamp (int64 sampleNumber, double timestamp, String streamKey);
+
     /** Signals start of acquisition */
     void startAcquisition();