Repository navigation
feat(export): let a live stream be recorded straight to CSV - #657
Conversation
Core shipped both halves of a recording and nothing joining them: an IAsyncEnumerable<LiveSample> from ILiveSampleSource in one namespace, a CsvExporter asking for an ISampleSource in another, and no supported way to connect the two. Streaming from a device and recording it to a file — the most basic DAQ workflow there is — had no answer in the package. Adds LiveSampleSource, an ISampleSource over a live stream that yields rows as frames decode rather than buffering the session, and IStreamingDevice.RecordLiveSamplesToCsvAsync, which drives it through CsvExporter and reports what reached the file and what did not (dropped, unmapped, non-monotonic). Purely additive: no existing public signature changes. closes #639 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
/agentic_review |
PR Summary by QodoRecord live device streams directly to CSV (no session buffering)
AI Description
Diagram
High-Level Assessment
Files changed (5)
|
|
/agentic_review |
Code Review by Qodo
1.
|
Qodo round 2 found two real things on #657. `DroppedLiveSampleCount` is device-wide and cumulative across every enumeration. When no channel is enabled the exporter returns without ever enumerating, yet the recording still took a delta over that shared counter -- so a second live consumer's drops were reported as this recording's, which contradicts the method's own documented all-zeros result for that case. The delta is now taken only when there was something to record, and the `DroppedSampleCount` doc says plainly that with a concurrent consumer the number is a delta over a shared counter rather than a per-recording one. The real-decode-path test waited for the exporter to subscribe with a fixed `Task.Delay(400)`, which loses the leading frames on a loaded machine -- the same shape #633 removed elsewhere today. `LiveSampleStream` now publishes an internal `LiveSubscriptionCount`, incremented once every sample handler is attached, so the test waits for the subscription itself instead of guessing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
Qodo round 2 was right on both counts, so both are fixed in 3958ddb rather than rebutted. Drop count misattributed (High). Real, and it broke this method's own contract. Fixed-delay test flakiness (Medium). Also real, and the same shape #633 removed elsewhere today. Bench validation re-run fresh on hardware this round (fw 3.7.2, sn 9090539562006014104), not inherited:
Also re-verified independently this round: the change is still purely additive against current Full suite green on the new head: 3879 Core on net9.0 and net10.0, 217 MCP, 0 failures; solution builds |
|
/agentic_review |
|
Code review by qodo was updated up to the latest commit 3958ddb |
The subscription-wait loop introduced in 3958ddb gated its timeout on DateTime.UtcNow. Wall-clock time is not monotonic: an NTP step during the wait either cuts it short (losing the leading frames the loop exists to protect) or stretches it past the intended bound. Stopwatch is the clock every other wait loop in this suite uses, and PR #602 replaced a UtcNow deadline for exactly this reason. Test-only; no production code touched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
Qodo round 3 found one new bug and it is valid, so it is fixed in Non-monotonic test deadline (Reliability). The subscription-wait loop added in Test-only change; no production code touched, so the hardware bench result on Full suite on the new head: 3879 Core passed on net9.0 and net10.0, 217 MCP, 0 failures, solution Round 2's two findings remain fixed and are shown as resolved. |
|
/agentic_review |
Confirmed. finding 2 is validly fixed in |
|
Code review by qodo was updated up to the latest commit ec52604 |
|
Qodo-clean, CI green — ready for review. Round 4 on Test-merged current Bench validation was fresh against the real board, not inherited: 100 Hz / 5 s gave rows=395, samples=790, dropped=0, CSV timestamps exactly 10.000 ms apart; a deliberately starved 1000 Hz / 4 s at |
Core shipped both halves of a recording and nothing that joined them.
ILiveSampleSourcehanded you anIAsyncEnumerable<LiveSample>in one namespace;CsvExporterasked for anISampleSourcein another; there was no supported way to connect the two. So the most basic DAQ workflow there is — stream from the device and record it to a file — had no answer in the package. Anyone who wanted it had to write the adapter themselves, and the adapter is fiddly enough that DAQiFi's own MCP server, the example app, and the test project each ended up with their own copy.This adds the join.
LiveSampleSourceis a publicISampleSourceover a live stream that yields rows as frames decode rather than buffering the session, so a recording's memory does not grow with its length.IStreamingDevice.RecordLiveSamplesToCsvAsync(writer, options, duration, bufferCapacity, ct)drives it throughCsvExporterand returns aLiveCsvRecordingResultcounting what reached the file and what did not — dropped, unmapped, non-monotonic. A recording with silent gaps is worse than one that failed, so nothing is swallowed.Purely additive. Two new types, one extension method, one README recipe. No existing public signature moved — the breaking-change question from #612/#616 is left to you. ADR 0002's source-compatibility promise only constrains how
LiveCsvRecordingResultmay grow later (positional appends), which is what it permits.What a reviewer should push back on
The extension hangs off
IStreamingDevicebut needsILiveSampleSourcetoo, so it does a runtimeis not ILiveSampleSourcecheck and throwsArgumentException. The recorder needs channels and identity from one interface and samples from the other, and neither implies the other (DaqifiStreamingDeviceimplements both). The alternatives were passing the same object twice, or a generic constraint that fails inference for anyone holding anIStreamingDevice. The runtime check has repo precedent inDaqifiAgent.RequireLiveSamples. It is the least-bad of three imperfect options and worth a second opinion.durationis the clean stop;cancellationTokenis the abort. The window arrives as a cancellation (the read it interrupts is unbounded) and is absorbed inside the private iterator, not at the exporter — that is what makes a timed recording end with a complete file, because the exporter sees the enumeration finish normally and flushes its last accumulated line. Letting the OCE reach the exporter silently discards that row. There is deliberately no graceful "stop signal" for a user-driven Stop button yet; today that meansduration, or cancel and accept an abort.GetSampleCountAsyncreturns 0 — exactly the "count unavailable" case the interface documents, and what makesCsvExporterskip percentage progress instead of dividing by an unknown total. Consequently there is noIProgress<int>parameter: a percentage of an unbounded stream is meaningless, andRowCount/SampleCountare the progress.RowCountis timestamps, not samples. The exporter collapses consecutive rows sharing a timestamp into one line, so counting samples would over-report by the channel count. That correctness note now lives in Core next to the exporter instead of being rediscovered per consumer.Also: values are written as
ScaledValue(a user who configured a transducer asked for PSI, not the volts underneath); samples match channels by(Type, ChannelNumber)rather than object identity, because a device rebuilds its channel objects on status repopulation; blank device name/serial normalize to"unknown", the same disposition as #627.Verification
Solution builds with
-warnaserrorunder the rootDirectory.Build.propsfrom #647: 0 warnings. Full suite green — 3878 Core tests on net9.0 and net10.0, 217 MCP tests, 0 failures. 45 new tests acrossLiveSampleSourceTests(24) andLiveCsvRecordingTests(21), weighted to edge cases: null/blank/duplicate channels, type-vs-number collision, empty stream, unmapped samples, backwards timestamps, three flavours of cancellation, no-enabled-channels, writer flush/non-dispose, buffer-capacity forwarding.Every test was proved able to fail via a 19-mutation matrix, each mutation reverted after its run. Which mutation covered which test:
RowCount++made unconditionalScaledValue→ValuebufferCapacityvalidation droppeddurationvalidation droppedbufferCapacitynot forwardedM16 initially killed nothing: the
&& !cancellationToken.IsCancellationRequestedhalf of the filter only matters when both tokens are cancelled, which no test reached. Rather than delete a defensible guard or leave it unpinned, a test was added that gates the scripted stream on aTaskCompletionSourceso it decides exactly which tokens are cancelled when the enumeration throws.M17 is killed by nothing, and is kept deliberately:
CsvExporter.ExportAsynchappens to make the same check, so no test can distinguish it — but the recorder's contract should not depend on a collaborator's internal ordering. Flagging it rather than quietly leaving it.Bench-validated on real hardware (fw 3.7.2, sn 9090539562006014104), via a scratch console app project-referencing this branch, since the example CLI has no live-CSV command:
rows=395 samples=790 dropped=0 unmapped=0 nonMonotonic=0.samples == 2 * rowsexactly — the per-frame collapse holds on hardware. CSV timestamps exactly 10.000 ms apart across all 395 rows.bufferCapacity: 1(deliberately starved):rows=3006 samples=3025 dropped=3323. This is the "dropped-sample count surfaced to the caller" criterion demonstrated on hardware — the losses are reported, not silent.All runs non-destructive: connect, enable channels, stream, record, stop, disconnect.
Deferred
This is step 2 of #639. Step 1 — promoting
Daqifi.Mcp's internalSdCardSampleSourceinto Core — is filed as #655 and deliberately not here: it is a cross-project move (delete, re-point, keep MCP's tests green), and mixing that with new public API makes one PR a reviewer has to evaluate in two modes. #655 also records new evidence the issue did not have: the example app carries a third hand-rolled copy of the same adapter.closes #639
Not merging — for review.