diff --git a/scripts/audio_latency_test/README.md b/scripts/audio_latency_test/README.md index 2819eab..ae00d2c 100644 --- a/scripts/audio_latency_test/README.md +++ b/scripts/audio_latency_test/README.md @@ -65,8 +65,32 @@ run_test.py --labrecorder-cli /path/to/LabRecorderCLI # Re-analyze an existing recording (no hardware): run_test.py --analyze-xdf recordings/audio_latency_xxx.xdf + +# Verify timestamps and exit non-zero if the MCC stream is bad: +run_test.py --analyze-xdf recordings/audio_latency_xxx.xdf --fail-on-bad-timestamps ``` +## Timestamp verification + +Every run (live or `--analyze-xdf`) prints a **timestamp health check** for the +MCC stream. Unlike the latency analysis — which loads with +`dejitter_timestamps=True` and so *cleans up* the timestamps — this check loads +the **raw** timestamps exactly as the outlet emitted them +(`synchronize_clocks=False, dejitter_timestamps=False`) and reports: + +- **negative absolute timestamps** — a genuinely bad clock value; +- **non-monotonic (backward) steps** — consecutive samples whose timestamp goes + *backwards*. These are the artifact of stamping each chunk's most-recent + sample with `local_clock()` at read time and letting liblsl back-date the + rest; when one chunk's back-dated start lands before the previous chunk's end, + time runs backward. In a consumer that does **not** dejitter (e.g. Orion), or + across machines, those backward steps are what surface as *negative + timestamps* / negative inter-sample intervals. + +Pass `--fail-on-bad-timestamps` to make the run exit non-zero when the MCC +stream has negative or non-monotonic raw timestamps (useful as a regression +gate once the outlet is fixed to emit monotonic, device-anchored timestamps). + Outputs land in `scripts/audio_latency_test/recordings/`: `audio_latency_.xdf`, `latency_.png`, `latency_.npz`, and the `mcc_.log` CLI log. diff --git a/scripts/audio_latency_test/analysis.py b/scripts/audio_latency_test/analysis.py index 0683669..da65cf8 100644 --- a/scripts/audio_latency_test/analysis.py +++ b/scripts/audio_latency_test/analysis.py @@ -45,6 +45,83 @@ def load_from_xdf(path, mcc_name, marker_type="Markers", mcc_channel=0): return marker_times, mcc_ts, sig +# --------------------------------------------------------------------------- # +# Timestamp health check +# --------------------------------------------------------------------------- # +def check_timestamps(path, mcc_name, marker_type="Markers"): + """Verify the *raw* (un-synced, un-dejittered) timestamps are sound. + + ``load_from_xdf`` deliberately loads with ``dejitter_timestamps=True``, + which refits each stream's timestamps onto a clean monotonic line and so + *hides* upstream defects. This routine instead loads the timestamps exactly + as the outlet emitted them and checks for the two problems that surface + downstream (e.g. in Orion) as "negative timestamps": + + * **negative absolute timestamps** -- a genuinely bad clock value; + * **non-monotonic (backward) steps** -- consecutive samples whose timestamp + goes *backwards*. These become negative inter-sample intervals (and, in a + consumer that derives sample times by accumulating dt or that aligns + without dejittering, negative timestamps). They are the expected artifact + of stamping each chunk's most-recent sample with ``local_clock()`` at read + time and letting liblsl back-date the rest: when one chunk's back-dated + start lands before the previous chunk's end, time runs backward. + + Returns ``(ok, results)`` where ``ok`` is False if any stream has negative + timestamps, and ``results`` maps stream name -> stats dict. Prints a + human-readable summary. + """ + import pyxdf + + streams, _ = pyxdf.load_xdf( + path, synchronize_clocks=False, dejitter_timestamps=False) + + results = {} + ok = True + lines = ["Timestamp health check (raw, as emitted -- no sync/dejitter):"] + for s in streams: + name = s["info"]["name"][0] + ts = np.asarray(s["time_stamps"], dtype=float) + if ts.size == 0: + continue + neg = int(np.count_nonzero(ts < 0)) + dt = np.diff(ts) + back = dt < 0 + n_back = int(np.count_nonzero(back)) + max_back_ms = float(-dt[back].min() * 1e3) if n_back else 0.0 + results[name] = { + "n": int(ts.size), + "t_min": float(ts.min()), "t_max": float(ts.max()), + "n_negative": neg, + "n_backward": n_back, "max_backward_ms": max_back_ms, + "pct_backward": 100.0 * n_back / max(dt.size, 1), + } + if neg: + status = "FAIL (negative timestamps)" + ok = False + elif n_back: + status = "WARN (non-monotonic)" + else: + status = "OK" + detail = f" backward_steps={n_back}" + if n_back: + detail += (f" ({100.0 * n_back / max(dt.size, 1):.2f}% of samples," + f" max {max_back_ms:.3f} ms back)") + lines.append( + f" {name:14} n={ts.size:>8d} t_min={ts.min():14.4f}" + f" neg={neg}{detail} -> {status}") + + mcc = results.get(mcc_name) + if mcc is not None and mcc["n_negative"] == 0 and mcc["n_backward"] > 0: + lines.append( + " NOTE: no negative timestamps on this machine, but the MCC stream " + "is non-monotonic. Across machines (or in a consumer that does not " + "dejitter, e.g. Orion) those backward steps are what appear as " + "negative timestamps. Fix upstream by making the outlet emit " + "monotonic, device-anchored timestamps.") + print("\n".join(lines)) + return ok, results + + # --------------------------------------------------------------------------- # # Core: per-marker onset detection # --------------------------------------------------------------------------- # diff --git a/scripts/audio_latency_test/run_test.py b/scripts/audio_latency_test/run_test.py index 923bae3..b2be9d0 100644 --- a/scripts/audio_latency_test/run_test.py +++ b/scripts/audio_latency_test/run_test.py @@ -111,6 +111,11 @@ def parse_args(argv=None): p.add_argument("--analyze-xdf", default=None, help="Analyze an existing XDF and exit (no hardware)") + # Timestamp verification + p.add_argument("--fail-on-bad-timestamps", action="store_true", + help="Exit non-zero if the MCC stream has negative or " + "non-monotonic (backward) raw timestamps") + return p.parse_args(argv) @@ -156,6 +161,22 @@ def report(args, latencies, marker_times, mcc_ts, mcc_sig, info, tag): return stats +def verify_timestamps(args, xdf_path): + """Run the raw-timestamp health check; return an exit-code contribution. + + Always prints the report. Returns 1 (so callers can OR it into their exit + code) when --fail-on-bad-timestamps is set and the MCC stream has negative + or non-monotonic timestamps; otherwise 0. + """ + print() + ok, results = analysis.check_timestamps(xdf_path, args.mcc_name) + if not args.fail_on_bad_timestamps: + return 0 + mcc = results.get(args.mcc_name, {}) + bad = (not ok) or mcc.get("n_backward", 0) > 0 + return 1 if bad else 0 + + def _terminate(proc, timeout=5.0): if proc is None or proc.poll() is not None: return @@ -185,7 +206,7 @@ def run_analyze_only(args): tag = os.path.splitext(os.path.basename(args.analyze_xdf))[0] os.makedirs(args.outdir, exist_ok=True) report(args, latencies, marker_times, mcc_ts, mcc_sig, info, tag) - return 0 + return verify_timestamps(args, args.analyze_xdf) # --------------------------------------------------------------------------- # @@ -286,7 +307,7 @@ def run_live(args): window=args.window, pre=args.pre, thresh_frac=args.thresh_frac, abs_threshold=args.abs_threshold) report(args, latencies, marker_times, mcc_ts, mcc_sig, info_recs, tag) - return 0 + return verify_timestamps(args, xdf_path) finally: for closer in ( diff --git a/src/core/CMakeLists.txt b/src/core/CMakeLists.txt index 2596707..57123aa 100644 --- a/src/core/CMakeLists.txt +++ b/src/core/CMakeLists.txt @@ -6,6 +6,18 @@ add_library(mccoutlet_core STATIC src/StreamThread.cpp ) +# Power management: keep the host awake during acquisition. macOS needs the +# Objective-C++ implementation (NSProcessInfo activity, built with ARC) linked +# against Foundation; other platforms use the no-op stub. +if(APPLE) + target_sources(mccoutlet_core PRIVATE src/PowerAssertion.mm) + set_source_files_properties(src/PowerAssertion.mm + PROPERTIES COMPILE_OPTIONS "-fobjc-arc") + target_link_libraries(mccoutlet_core PRIVATE "-framework Foundation") +else() + target_sources(mccoutlet_core PRIVATE src/PowerAssertion.cpp) +endif() + target_include_directories(mccoutlet_core PUBLIC $ diff --git a/src/core/include/mccoutlet/Device.hpp b/src/core/include/mccoutlet/Device.hpp index c73b024..0ed74ab 100644 --- a/src/core/include/mccoutlet/Device.hpp +++ b/src/core/include/mccoutlet/Device.hpp @@ -8,6 +8,7 @@ */ #include +#include #include #include #include @@ -57,6 +58,17 @@ class IDevice { virtual bool startAcquisition() = 0; virtual void disconnect() = 0; virtual bool isConnected() const = 0; + + /** + * @brief Request that any in-progress getData*() call return promptly. + * + * Safe to call from a thread other than the one running getData*(). + * Causes the blocking acquisition loops to break out so the streaming + * thread can observe its shutdown flag and exit before disconnect() + * and the subsequent join(). Without this, a wedged device leaves + * getData*() spinning and stop() deadlocks on the join. + */ + virtual void requestStop() = 0; virtual DeviceInfo getInfo() const = 0; virtual DeviceCapabilities getCapabilities() const = 0; @@ -123,6 +135,7 @@ class MCCDevice : public IDevice { bool connect() override; bool startAcquisition() override; void disconnect() override; + void requestStop() override; bool isConnected() const override; DeviceInfo getInfo() const override; DeviceCapabilities getCapabilities() const override; @@ -141,6 +154,10 @@ class MCCDevice : public IDevice { std::atomic disconnecting_{false}; int overrun_count_ = 0; + // Coalesce overrun status reports: time of the last one emitted to the + // callback. Default-constructed (epoch) means none emitted yet. + std::chrono::steady_clock::time_point last_overrun_report_{}; + // ulAInScan circular buffer (double precision, as required by uldaq) std::vector scan_buffer_; int scan_buffer_samples_per_chan_ = 0; diff --git a/src/core/include/mccoutlet/PowerAssertion.hpp b/src/core/include/mccoutlet/PowerAssertion.hpp new file mode 100644 index 0000000..5d5d958 --- /dev/null +++ b/src/core/include/mccoutlet/PowerAssertion.hpp @@ -0,0 +1,37 @@ +#pragma once +/** + * @file PowerAssertion.hpp + * @brief RAII guard that keeps the host awake during acquisition. + */ + +#include + +namespace mccoutlet { + +/** + * @brief Keeps the system awake for as long as the object is alive. + * + * On macOS, idle system sleep suspends the USB bus, which wedges the DAQ's + * in-flight transfers (uldaq then logs "##### error still xfer pending" and + * acquisition cannot be recovered). App Nap can likewise throttle the libusb + * event thread enough to overrun the device's hardware FIFO. This guard opts + * the process out of both via @c -[NSProcessInfo beginActivityWithOptions:reason:] + * for the duration of streaming, releasing the activity on destruction. + * + * On non-Apple platforms it is a no-op (idle sleep does not suspend USB the + * same way; add a platform implementation here if that changes). + */ +class PowerAssertion { +public: + /// @param reason Human-readable reason shown in power diagnostics. + explicit PowerAssertion(const std::string& reason); + ~PowerAssertion(); + + PowerAssertion(const PowerAssertion&) = delete; + PowerAssertion& operator=(const PowerAssertion&) = delete; + +private: + void* token_ = nullptr; ///< retained activity token (id on macOS) +}; + +} // namespace mccoutlet diff --git a/src/core/src/Device.cpp b/src/core/src/Device.cpp index e017781..6b23eac 100644 --- a/src/core/src/Device.cpp +++ b/src/core/src/Device.cpp @@ -18,6 +18,11 @@ constexpr int MAX_DEV_COUNT = 100; // Larger values tolerate more jitter but use more memory. constexpr int SCAN_BUFFER_SECONDS = 10; +// Give up after this many consecutive scan restarts within a single getData*() +// call. A genuine overrun recovers in one restart; an unrecoverable USB state +// (e.g. stuck transfers) otherwise spins here forever, hammering ulAInScanStop. +constexpr int MAX_RESTART_ATTEMPTS = 3; + std::string ulErrorString(UlError err) { char msg[ERR_MSG_LEN]{}; ulGetErrMsg(err, msg); @@ -413,7 +418,13 @@ bool MCCDevice::restartScan() { scans_read_ = 0; overrun_count_++; - if (statusCallback_) { + // Coalesce reports to at most one per second so a recovery storm cannot + // flood the consumer (e.g. the GUI's queued-event status updates). + auto now = std::chrono::steady_clock::now(); + if (statusCallback_ && + (last_overrun_report_.time_since_epoch().count() == 0 || + now - last_overrun_report_ >= std::chrono::seconds(1))) { + last_overrun_report_ = now; statusCallback_( "Device FIFO overrun detected, scan restarted (overrun #" + std::to_string(overrun_count_) + ")", true); @@ -422,6 +433,12 @@ bool MCCDevice::restartScan() { return true; } +void MCCDevice::requestStop() { + // Break the getData*() polling loops so the streaming thread can exit + // before disconnect()/join(). Set from the caller's thread. + disconnecting_ = true; +} + void MCCDevice::disconnect() { disconnecting_ = true; @@ -470,6 +487,7 @@ bool MCCDevice::getData(std::vector& buffer, double& timestamp) { const int channelCount = config_.high_channel - config_.low_channel + 1; const size_t total_buffer_elements = scan_buffer_.size(); + int restart_attempts = 0; while (!disconnecting_) { ScanStatus status{}; TransferStatus xfer{}; @@ -477,6 +495,15 @@ bool MCCDevice::getData(std::vector& buffer, double& timestamp) { UlError err = ulAInScanStatus(handle_, &status, &xfer); if (err != ERR_NO_ERROR || status != SS_RUNNING) { if (disconnecting_) return false; + if (++restart_attempts > MAX_RESTART_ATTEMPTS) { + if (statusCallback_) { + statusCallback_( + "Acquisition could not be recovered after " + + std::to_string(MAX_RESTART_ATTEMPTS) + + " restart attempts; stopping stream", true); + } + return false; + } if (!restartScan()) return false; continue; } @@ -514,6 +541,7 @@ bool MCCDevice::getDataInt32(std::vector& buffer, double& timestamp) { const int channelCount = config_.high_channel - config_.low_channel + 1; const size_t total_buffer_elements = scan_buffer_.size(); + int restart_attempts = 0; while (!disconnecting_) { ScanStatus status{}; TransferStatus xfer{}; @@ -521,6 +549,15 @@ bool MCCDevice::getDataInt32(std::vector& buffer, double& timestamp) { UlError err = ulAInScanStatus(handle_, &status, &xfer); if (err != ERR_NO_ERROR || status != SS_RUNNING) { if (disconnecting_) return false; + if (++restart_attempts > MAX_RESTART_ATTEMPTS) { + if (statusCallback_) { + statusCallback_( + "Acquisition could not be recovered after " + + std::to_string(MAX_RESTART_ATTEMPTS) + + " restart attempts; stopping stream", true); + } + return false; + } if (!restartScan()) return false; continue; } @@ -560,6 +597,7 @@ bool MCCDevice::getDataInt16(std::vector& buffer, double& timestamp) { const int channelCount = config_.high_channel - config_.low_channel + 1; const size_t total_buffer_elements = scan_buffer_.size(); + int restart_attempts = 0; while (!disconnecting_) { ScanStatus status{}; TransferStatus xfer{}; @@ -567,6 +605,15 @@ bool MCCDevice::getDataInt16(std::vector& buffer, double& timestamp) { UlError err = ulAInScanStatus(handle_, &status, &xfer); if (err != ERR_NO_ERROR || status != SS_RUNNING) { if (disconnecting_) return false; + if (++restart_attempts > MAX_RESTART_ATTEMPTS) { + if (statusCallback_) { + statusCallback_( + "Acquisition could not be recovered after " + + std::to_string(MAX_RESTART_ATTEMPTS) + + " restart attempts; stopping stream", true); + } + return false; + } if (!restartScan()) return false; continue; } diff --git a/src/core/src/PowerAssertion.cpp b/src/core/src/PowerAssertion.cpp new file mode 100644 index 0000000..89475b7 --- /dev/null +++ b/src/core/src/PowerAssertion.cpp @@ -0,0 +1,10 @@ +// Non-Apple no-op implementation of PowerAssertion. +// macOS uses PowerAssertion.mm instead (selected in src/core/CMakeLists.txt). +#include "mccoutlet/PowerAssertion.hpp" + +namespace mccoutlet { + +PowerAssertion::PowerAssertion(const std::string& /*reason*/) {} +PowerAssertion::~PowerAssertion() = default; + +} // namespace mccoutlet diff --git a/src/core/src/PowerAssertion.mm b/src/core/src/PowerAssertion.mm new file mode 100644 index 0000000..a320c0e --- /dev/null +++ b/src/core/src/PowerAssertion.mm @@ -0,0 +1,33 @@ +// macOS implementation of PowerAssertion. Compiled with ARC (-fobjc-arc); +// see src/core/CMakeLists.txt. +#include "mccoutlet/PowerAssertion.hpp" + +#import + +namespace mccoutlet { + +PowerAssertion::PowerAssertion(const std::string& reason) { + // NSActivityUserInitiated -> disables App Nap and idle system sleep. + // NSActivityLatencyCritical -> also disables timer throttling so the + // libusb event thread keeps servicing high-rate transfers promptly. + NSActivityOptions options = + NSActivityUserInitiated | NSActivityLatencyCritical; + + NSString* why = [NSString stringWithUTF8String:reason.c_str()]; + if (why == nil) why = @"DAQ acquisition"; + + id token = [[NSProcessInfo processInfo] beginActivityWithOptions:options + reason:why]; + // Hand the token to the C++ object; balanced by __bridge_transfer below. + token_ = (__bridge_retained void*)token; +} + +PowerAssertion::~PowerAssertion() { + if (token_) { + id token = (__bridge_transfer id)token_; + [[NSProcessInfo processInfo] endActivity:token]; + token_ = nullptr; + } +} + +} // namespace mccoutlet diff --git a/src/core/src/StreamThread.cpp b/src/core/src/StreamThread.cpp index f7e27bb..5815926 100644 --- a/src/core/src/StreamThread.cpp +++ b/src/core/src/StreamThread.cpp @@ -1,4 +1,5 @@ #include "mccoutlet/StreamThread.hpp" +#include "mccoutlet/PowerAssertion.hpp" #include namespace mccoutlet { @@ -61,6 +62,14 @@ void StreamThread::stop() { shutdown_ = true; + // Break any in-progress getData*() call before joining. Otherwise a + // wedged device leaves the worker spinning inside getData*() (which only + // checks the device's own abort flag, not shutdown_) and join() blocks + // forever, freezing the caller (the GUI thread). + if (device_) { + device_->requestStop(); + } + if (thread_ && thread_->joinable()) { thread_->join(); } @@ -89,6 +98,14 @@ DeviceInfo StreamThread::getDeviceInfo() const { } void StreamThread::threadFunction() { + // Keep the host awake for the whole streaming session. Idle system sleep + // suspends the USB bus and wedges the DAQ's in-flight transfers (uldaq then + // reports "still xfer pending" and the scan cannot be recovered), and App + // Nap can throttle the libusb event thread enough to overrun the device + // FIFO. Released automatically when this function returns (stop, error, or + // shutdown). No-op on non-Apple platforms. + PowerAssertion keep_awake("Streaming MCC DAQ to LSL"); + try { auto info = device_->getInfo(); LSLOutlet outlet(info);