Skip to content

Commit 4db8803

Browse files
cboulaycboulay
committed
Improved timing and reduced overall latency.
1 parent 61006e4 commit 4db8803

6 files changed

Lines changed: 69 additions & 59 deletions

File tree

‎.gitignore‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,5 @@ ui_*.h
66
/out/
77
.DS_Store
88
.idea
9-
.cmake-build-*
9+
.cmake-build-*/
10+
.cache/

‎src/core/include/mccoutlet/Device.hpp‎

Lines changed: 14 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -61,19 +61,22 @@ class IDevice {
6161
virtual DeviceCapabilities getCapabilities() const = 0;
6262

6363
/**
64-
* @brief Retrieve data from the device
65-
* @param buffer Output buffer (channel-interleaved float samples)
64+
* @brief Retrieve all available data from the device
65+
* @param buffer Resized to fit all available channel-interleaved samples
66+
* @param timestamp Output: LSL timestamp of the most recent sample
6667
* @return true if data was retrieved, false on error or shutdown
6768
*
68-
* Blocks until enough data is available to fill the buffer.
69+
* Blocks until at least one scan is available, then drains everything
70+
* the device has buffered. The timestamp is captured at the moment
71+
* data availability is detected.
6972
*/
70-
virtual bool getData(std::vector<float>& buffer) = 0;
73+
virtual bool getData(std::vector<float>& buffer, double& timestamp) = 0;
7174

72-
/** @brief Retrieve raw data as int32 (for >16-bit ADC in raw mode) */
73-
virtual bool getDataInt32(std::vector<int32_t>& buffer) = 0;
75+
/** @brief Retrieve all available raw data as int32 (for >16-bit ADC in raw mode) */
76+
virtual bool getDataInt32(std::vector<int32_t>& buffer, double& timestamp) = 0;
7477

75-
/** @brief Retrieve raw data as int16 (for <=16-bit ADC in raw mode) */
76-
virtual bool getDataInt16(std::vector<int16_t>& buffer) = 0;
78+
/** @brief Retrieve all available raw data as int16 (for <=16-bit ADC in raw mode) */
79+
virtual bool getDataInt16(std::vector<int16_t>& buffer, double& timestamp) = 0;
7780
};
7881

7982
/**
@@ -123,9 +126,9 @@ class MCCDevice : public IDevice {
123126
bool isConnected() const override;
124127
DeviceInfo getInfo() const override;
125128
DeviceCapabilities getCapabilities() const override;
126-
bool getData(std::vector<float>& buffer) override;
127-
bool getDataInt32(std::vector<int32_t>& buffer) override;
128-
bool getDataInt16(std::vector<int16_t>& buffer) override;
129+
bool getData(std::vector<float>& buffer, double& timestamp) override;
130+
bool getDataInt32(std::vector<int32_t>& buffer, double& timestamp) override;
131+
bool getDataInt16(std::vector<int16_t>& buffer, double& timestamp) override;
129132

130133
private:
131134
bool restartScan();

‎src/core/include/mccoutlet/LSLOutlet.hpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,9 @@ class LSLOutlet {
2222
LSLOutlet(LSLOutlet&&) noexcept = default;
2323
LSLOutlet& operator=(LSLOutlet&&) noexcept = default;
2424

25-
void pushChunk(const std::vector<float>& data);
26-
void pushChunk(const std::vector<int32_t>& data);
27-
void pushChunk(const std::vector<int16_t>& data);
25+
void pushChunk(const std::vector<float>& data, double timestamp = 0.0);
26+
void pushChunk(const std::vector<int32_t>& data, double timestamp = 0.0);
27+
void pushChunk(const std::vector<int16_t>& data, double timestamp = 0.0);
2828
void pushSample(const std::vector<float>& sample);
2929
std::string getStreamName() const;
3030
bool hasConsumers() const;

‎src/core/src/Device.cpp‎

Lines changed: 28 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#include "mccoutlet/Device.hpp"
22

3+
#include <lsl_cpp.h>
34
#include <uldaq.h>
45

56
#include <chrono>
@@ -463,11 +464,10 @@ DeviceCapabilities MCCDevice::getCapabilities() const {
463464
return capabilities_;
464465
}
465466

466-
bool MCCDevice::getData(std::vector<float>& buffer) {
467+
bool MCCDevice::getData(std::vector<float>& buffer, double& timestamp) {
467468
if (!connected_ || handle_ == 0) return false;
468469

469470
const int channelCount = config_.high_channel - config_.low_channel + 1;
470-
const int samples_needed = static_cast<int>(buffer.size()) / channelCount;
471471
const size_t total_buffer_elements = scan_buffer_.size();
472472

473473
while (!disconnecting_) {
@@ -477,42 +477,41 @@ bool MCCDevice::getData(std::vector<float>& buffer) {
477477
UlError err = ulAInScanStatus(handle_, &status, &xfer);
478478
if (err != ERR_NO_ERROR || status != SS_RUNNING) {
479479
if (disconnecting_) return false;
480-
// Overrun or other scan failure — attempt restart
481480
if (!restartScan()) return false;
482481
continue;
483482
}
484483

485-
// currentScanCount = per-channel samples transferred since scan start
486484
long long available = static_cast<long long>(xfer.currentScanCount) -
487485
static_cast<long long>(scans_read_);
488486

489-
if (available >= samples_needed) {
490-
// Calculate read position in the circular buffer
487+
if (available > 0) {
488+
timestamp = lsl::local_clock();
489+
490+
size_t num_elements = static_cast<size_t>(available) * channelCount;
491+
buffer.resize(num_elements);
492+
491493
size_t read_offset =
492494
(static_cast<size_t>(scans_read_) * channelCount) % total_buffer_elements;
493495

494-
// Copy data, converting double -> float, handling wrap-around
495-
for (size_t i = 0; i < buffer.size(); ++i) {
496+
for (size_t i = 0; i < num_elements; ++i) {
496497
buffer[i] = static_cast<float>(
497498
scan_buffer_[(read_offset + i) % total_buffer_elements]);
498499
}
499500

500-
scans_read_ += samples_needed;
501+
scans_read_ += available;
501502
return true;
502503
}
503504

504-
// Sleep briefly before polling again
505505
std::this_thread::sleep_for(std::chrono::milliseconds(1));
506506
}
507507

508508
return false;
509509
}
510510

511-
bool MCCDevice::getDataInt32(std::vector<int32_t>& buffer) {
511+
bool MCCDevice::getDataInt32(std::vector<int32_t>& buffer, double& timestamp) {
512512
if (!connected_ || handle_ == 0) return false;
513513

514514
const int channelCount = config_.high_channel - config_.low_channel + 1;
515-
const int samples_needed = static_cast<int>(buffer.size()) / channelCount;
516515
const size_t total_buffer_elements = scan_buffer_.size();
517516

518517
while (!disconnecting_) {
@@ -529,16 +528,21 @@ bool MCCDevice::getDataInt32(std::vector<int32_t>& buffer) {
529528
long long available = static_cast<long long>(xfer.currentScanCount) -
530529
static_cast<long long>(scans_read_);
531530

532-
if (available >= samples_needed) {
531+
if (available > 0) {
532+
timestamp = lsl::local_clock();
533+
534+
size_t num_elements = static_cast<size_t>(available) * channelCount;
535+
buffer.resize(num_elements);
536+
533537
size_t read_offset =
534538
(static_cast<size_t>(scans_read_) * channelCount) % total_buffer_elements;
535539

536-
for (size_t i = 0; i < buffer.size(); ++i) {
540+
for (size_t i = 0; i < num_elements; ++i) {
537541
buffer[i] = static_cast<int32_t>(
538542
scan_buffer_[(read_offset + i) % total_buffer_elements]);
539543
}
540544

541-
scans_read_ += samples_needed;
545+
scans_read_ += available;
542546
return true;
543547
}
544548

@@ -548,11 +552,10 @@ bool MCCDevice::getDataInt32(std::vector<int32_t>& buffer) {
548552
return false;
549553
}
550554

551-
bool MCCDevice::getDataInt16(std::vector<int16_t>& buffer) {
555+
bool MCCDevice::getDataInt16(std::vector<int16_t>& buffer, double& timestamp) {
552556
if (!connected_ || handle_ == 0) return false;
553557

554558
const int channelCount = config_.high_channel - config_.low_channel + 1;
555-
const int samples_needed = static_cast<int>(buffer.size()) / channelCount;
556559
const size_t total_buffer_elements = scan_buffer_.size();
557560

558561
while (!disconnecting_) {
@@ -569,16 +572,21 @@ bool MCCDevice::getDataInt16(std::vector<int16_t>& buffer) {
569572
long long available = static_cast<long long>(xfer.currentScanCount) -
570573
static_cast<long long>(scans_read_);
571574

572-
if (available >= samples_needed) {
575+
if (available > 0) {
576+
timestamp = lsl::local_clock();
577+
578+
size_t num_elements = static_cast<size_t>(available) * channelCount;
579+
buffer.resize(num_elements);
580+
573581
size_t read_offset =
574582
(static_cast<size_t>(scans_read_) * channelCount) % total_buffer_elements;
575583

576-
for (size_t i = 0; i < buffer.size(); ++i) {
584+
for (size_t i = 0; i < num_elements; ++i) {
577585
buffer[i] = static_cast<int16_t>(
578586
scan_buffer_[(read_offset + i) % total_buffer_elements]);
579587
}
580588

581-
scans_read_ += samples_needed;
589+
scans_read_ += available;
582590
return true;
583591
}
584592

‎src/core/src/LSLOutlet.cpp‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -60,21 +60,21 @@ LSLOutlet::LSLOutlet(const DeviceInfo& info)
6060

6161
LSLOutlet::~LSLOutlet() = default;
6262

63-
void LSLOutlet::pushChunk(const std::vector<float>& data) {
63+
void LSLOutlet::pushChunk(const std::vector<float>& data, double timestamp) {
6464
if (outlet_ && !data.empty()) {
65-
outlet_->push_chunk_multiplexed(data);
65+
outlet_->push_chunk_multiplexed(data, timestamp);
6666
}
6767
}
6868

69-
void LSLOutlet::pushChunk(const std::vector<int32_t>& data) {
69+
void LSLOutlet::pushChunk(const std::vector<int32_t>& data, double timestamp) {
7070
if (outlet_ && !data.empty()) {
71-
outlet_->push_chunk_multiplexed(data);
71+
outlet_->push_chunk_multiplexed(data, timestamp);
7272
}
7373
}
7474

75-
void LSLOutlet::pushChunk(const std::vector<int16_t>& data) {
75+
void LSLOutlet::pushChunk(const std::vector<int16_t>& data, double timestamp) {
7676
if (outlet_ && !data.empty()) {
77-
outlet_->push_chunk_multiplexed(data);
77+
outlet_->push_chunk_multiplexed(data, timestamp);
7878
}
7979
}
8080

‎src/core/src/StreamThread.cpp‎

Lines changed: 16 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -97,41 +97,39 @@ void StreamThread::threadFunction() {
9797
statusCallback_("LSL outlet created: " + info.name, false);
9898
}
9999

100-
// Buffer ~100ms of data per chunk, minimum 1 sample
101-
size_t samples_per_chunk = std::max(
102-
1,
103-
static_cast<int>(info.sample_rate * 0.1)
104-
);
105-
size_t chunk_elements = samples_per_chunk * info.channel_count;
100+
// Reserve ~1 second of capacity; getData resizes to actual available data
101+
size_t reserve_elements =
102+
static_cast<size_t>(info.sample_rate) * info.channel_count;
103+
double timestamp = 0.0;
106104

107105
if (info.scaled) {
108-
// Scaled mode: calibrated voltage as float
109-
std::vector<float> buffer(chunk_elements);
106+
std::vector<float> buffer;
107+
buffer.reserve(reserve_elements);
110108
while (!shutdown_) {
111-
if (device_->getData(buffer)) {
112-
outlet.pushChunk(buffer);
109+
if (device_->getData(buffer, timestamp)) {
110+
outlet.pushChunk(buffer, timestamp);
113111
} else if (!shutdown_) {
114112
if (statusCallback_) statusCallback_("Device acquisition error", true);
115113
break;
116114
}
117115
}
118116
} else if (info.resolution_bits <= 16) {
119-
// Raw mode, <=16-bit ADC: integer counts as int16
120-
std::vector<int16_t> buffer(chunk_elements);
117+
std::vector<int16_t> buffer;
118+
buffer.reserve(reserve_elements);
121119
while (!shutdown_) {
122-
if (device_->getDataInt16(buffer)) {
123-
outlet.pushChunk(buffer);
120+
if (device_->getDataInt16(buffer, timestamp)) {
121+
outlet.pushChunk(buffer, timestamp);
124122
} else if (!shutdown_) {
125123
if (statusCallback_) statusCallback_("Device acquisition error", true);
126124
break;
127125
}
128126
}
129127
} else {
130-
// Raw mode, >16-bit ADC: integer counts as int32
131-
std::vector<int32_t> buffer(chunk_elements);
128+
std::vector<int32_t> buffer;
129+
buffer.reserve(reserve_elements);
132130
while (!shutdown_) {
133-
if (device_->getDataInt32(buffer)) {
134-
outlet.pushChunk(buffer);
131+
if (device_->getDataInt32(buffer, timestamp)) {
132+
outlet.pushChunk(buffer, timestamp);
135133
} else if (!shutdown_) {
136134
if (statusCallback_) statusCallback_("Device acquisition error", true);
137135
break;

0 commit comments

Comments
 (0)