diff --git a/README.md b/README.md index ff94ccd..0425fbf 100644 --- a/README.md +++ b/README.md @@ -73,12 +73,16 @@ records into versioned `TRK1` frames, isolating acquisition from brief transport stalls. CRC, packet and sample sequences, timestamps, and cumulative loss/overrun counters make permanent loss detectable by the receiver. -The current ESP-IDF USB VFS path reports physical disconnects, but a connected -host that stops draining can time out below stdio and still appear successful to -firmware. The host can detect resulting loss from packet/sample sequences and -CRC framing, but the device cannot count that case. A direct USB driver with -bounded drain waits, and ultimately receiver acknowledgements with replay, are -deferred to the common USB/BLE transport layer. +USB telemetry now uses ESP-IDF's interrupt-driven USB Serial/JTAG driver behind +a transport-neutral state machine. A complete frame is submitted atomically to +the driver ring and remains pending across bounded drain timeouts; firmware does +not resubmit it ambiguously or dequeue another frame. The 512-sample queue +therefore also protects a connected endpoint that temporarily stops draining. + +USB drain confirms that bytes left the device endpoint, not that the capture +application persisted them. Packet/sample sequences and CRC expose loss after +the fact. End-to-end receiver acknowledgements and replay remain part of the BLE +transport milestone. Measured end-to-end framing overhead is about 2.47 kB/s at 100 Hz, or 8.47 MiB/hour before BLE link overhead. @@ -132,10 +136,12 @@ Status bits: The binary capture tool auto-detects a single `/dev/cu.usbmodem*` device, stores only CRC-valid frames, renders CSV, and reports packet, sample, timing, status, -drop, overrun, timestamp-saturation, and trailing-partial-byte totals: +drop, overrun, timestamp-saturation, and trailing-partial-byte totals. An +optional `--wire` path preserves every received byte, including startup text and +damaged or partial frames, for forensic comparison: ```sh -python tools/capture_binary.py +python tools/capture_binary.py --wire captures/session.wire ``` An existing `.trk` stream can be decoded again without hardware: @@ -144,5 +150,8 @@ An existing `.trk` stream can be decoded again without hardware: python tools/decode_binary.py captures/session.trk captures/session.csv ``` +Live capture and offline decoding use the same integrity tracker, including +wrap-aware packet/sample gap classification. + `tools/capture_serial.py` remains available only for decoding captures from the older CSV-v3 firmware snapshots. diff --git a/docs/binary-record-v1.md b/docs/binary-record-v1.md index c3fa500..5da056d 100644 --- a/docs/binary-record-v1.md +++ b/docs/binary-record-v1.md @@ -88,15 +88,23 @@ the oldest retained data is sent first. If the queue fills, acquisition drops new samples rather than overwriting older ones; sequence gaps and the cumulative lost-sample counter expose that permanent loss. -That retry guarantee requires the transport's success result to mean that the -complete frame was accepted for eventual delivery. The current USB Serial/JTAG -VFS/stdio path does not fully satisfy that contract: if the host remains -connected but stops draining, its internal timeout can discard bytes while the -stdio write appears successful. CRC and sequence checks make that loss visible -to a receiver, but it does not increment the device's drop counter. A direct -driver path with bounded transmit-drain waits can report this condition; an -application acknowledgement and replay window is required for end-to-end -delivery confirmation. +The shared transport state machine distinguishes three nonfatal states. `RETRY` +means zero bytes were accepted and the complete frame may be submitted again. +`PENDING` means the backend owns an in-flight frame, so firmware may only poll +that transfer. `COMPLETE` permits the output task to reuse its packet buffer and +consume more samples. This prevents a timeout after partial progress from +causing an ambiguous whole-frame duplicate. + +The direct USB Serial/JTAG backend atomically copies a complete frame into its TX +ring, then polls a bounded transmit-drain wait. A timeout remains `PENDING`; it +does not trigger resubmission. This closes the VFS/stdio path's silent-discard +case for a connected host that stops draining. + +USB drain is not end-to-end application delivery confirmation. A host process +may attach after earlier frames have already left the endpoint, or fail after +the endpoint accepts them. CRC and sequence checks make resulting loss visible, +but an application acknowledgement and replay window are still required to +guarantee receipt. Receivers report bytes left in an incomplete trailing frame when capture ends. Those bytes cannot pass CRC validation and are not silently admitted as samples. diff --git a/docs/binary-transport-validation-2026-08-17.md b/docs/binary-transport-validation-2026-08-17.md index 302d0fa..e8600d7 100644 --- a/docs/binary-transport-validation-2026-08-17.md +++ b/docs/binary-transport-validation-2026-08-17.md @@ -39,11 +39,11 @@ bytes even though stdio reports a successful write. Firmware therefore cannot retain that particular frame or increment its drop counter. The receiver still detects the loss through CRC resynchronization and packet/sample sequence gaps. -ESP-IDF's direct USB Serial/JTAG driver provides bounded writes and an explicit -transmit-drain wait, allowing a connected stall to become observable to the -transport policy. That is a useful improvement for the common transport layer. -It is not proof of receiver delivery; application acknowledgements and replay -are needed for that stronger guarantee and are planned with BLE integration. +This limitation was subsequently closed at the device/endpoint boundary by the +shared transport state machine and direct USB Serial/JTAG driver described in +`transport-layer-validation-2026-08-17.md`. It is not proof of receiver delivery; +application acknowledgements and replay are still needed for that stronger +guarantee and are planned with BLE integration. ## Final hardware capture diff --git a/docs/transport-layer-validation-2026-08-17.md b/docs/transport-layer-validation-2026-08-17.md new file mode 100644 index 0000000..f0f273d --- /dev/null +++ b/docs/transport-layer-validation-2026-08-17.md @@ -0,0 +1,74 @@ +# Direct USB Transport Validation — 2026-08-17 + +This milestone moved framed telemetry off stdio and the USB Serial/JTAG VFS data +path. The output task now targets a transport-neutral state machine backed by +ESP-IDF's interrupt-driven USB Serial/JTAG driver. + +## Transaction contract + +- `RETRY`: the backend accepted zero bytes, so whole-frame resubmission is safe. +- `PENDING`: the backend owns an in-flight frame. Only completion polling is + allowed; the caller must retain and not modify the packet buffer. +- `COMPLETE`: the backend's completion criterion is satisfied and the caller may + reuse the packet buffer. +- `FATAL`: a programming or backend invariant failed. The output task stops + consuming the sample queue rather than silently discarding its in-flight data. + +For USB, submission uses a 512-byte TX ring and a bounded 50 ms write. ESP-IDF's +ring-buffer send is all-or-nothing for each `TRK1` frame. Once accepted, bounded +50 ms `usb_serial_jtag_wait_tx_done()` calls continue returning `PENDING` until +the host drains the endpoint. A timeout never resubmits the frame. + +The VFS is switched to driver mode after readable startup output so any +unexpected diagnostic output cannot race the driver's ISR by accessing the +hardware FIFO directly. Firmware logs are disabled before binary telemetry +tasks start, as before. + +## Verification + +The host C transport fixture compiles the production state machine with +`-Wall -Wextra -Werror` and verifies: + +- zero-accept submissions remain retryable; +- accepted transfers become pending; +- repeated pending polls never call submission again; +- completion returns the sender to idle; +- backend-confirmed safe retry returns to idle; +- invalid arguments and unknown backend states fail closed. + +The assembled ESP32-C3 prototype produced a normal 1,728-sample direct-driver +capture with no packet gaps, sample gaps, resets, CRC failures, reported drops, +loop overruns, trailing partial bytes, or timestamp-saturation frames. + +For the connected-stall case, the USB endpoint was left enumerated without a +serial reader long enough to overflow the 512-sample acquisition queue. When the +reader opened, delivery preserved the oldest block through sequence 511 and +resumed at sequence 1,706. The one observed gap and the cumulative device drop +count both equal 1,194 samples. The timestamp difference across that gap is +11,950,000 us, exactly 1,195 sample intervals. Packet gaps, CRC failures, loop +overruns, trailing partial bytes, and timestamp-saturation flags are zero. + +That exact validated stream is tracked as +`tests/fixtures/direct_usb_stall.trk`, SHA-256 +`40f874b7eaa7f705524ecdd75f832e8a724252366633116ac015fc75dfd16558`, and +its signature is asserted by the regression suite. + +## Remaining delivery boundary + +The stall capture begins at sample sequence 8. The flashing process still had +the serial endpoint open long enough to drain sequences 0 through 7 before the +capture application attached. This is not silent device-side loss; it precisely +demonstrates the boundary of USB drain confirmation. Only an application-level +ACK can prove that the intended receiver received and persisted a frame. + +The planned BLE backend will use the same `PENDING` ownership rule while waiting +for acknowledgements, retain unacknowledged frames for replay, and expose +per-cause transport/queue counters separately. + +## Host evidence handling + +Live capture and offline decode now share one integrity tracker for packet and +sample gaps/resets, timestamp anomalies, sensor status, saturation flags, drops, +and acquisition overruns. Sequence classification is wrap-aware. The capture +tool also accepts `--wire PATH` to preserve every serial byte before parsing, +including startup text, corrupt frames, and trailing fragments. diff --git a/main/CMakeLists.txt b/main/CMakeLists.txt index 5b72e01..b07d56c 100644 --- a/main/CMakeLists.txt +++ b/main/CMakeLists.txt @@ -1,5 +1,6 @@ idf_component_register( - SRCS "trikke_sensor_main.c" "trikke_protocol.c" + SRCS "trikke_sensor_main.c" "trikke_protocol.c" "trikke_transport.c" + "trikke_usb_transport.c" INCLUDE_DIRS "." REQUIRES adxl345 l3g4200d esp_timer esp_driver_gpio esp_driver_i2c esp_driver_usb_serial_jtag vfs diff --git a/main/trikke_sensor_main.c b/main/trikke_sensor_main.c index e0a0810..5bb0bf4 100644 --- a/main/trikke_sensor_main.c +++ b/main/trikke_sensor_main.c @@ -14,6 +14,8 @@ #include "freertos/task.h" #include "l3g4200d.h" #include "trikke_protocol.h" +#include "trikke_transport.h" +#include "trikke_usb_transport.h" // Seeed Studio XIAO ESP32-C3: D4/SDA = GPIO6, D5/SCL = GPIO7. #define TRIKKE_I2C_PORT I2C_NUM_0 @@ -54,6 +56,8 @@ typedef struct { QueueHandle_t sample_queue; atomic_uint_least32_t dropped_sample_count; atomic_uint_least32_t loop_overrun_count; + trikke_transport_t transport; + trikke_usb_transport_t usb_transport; } trikke_context_t; static trikke_context_t s_context; @@ -157,22 +161,26 @@ static void acquisition_task(void *argument) } } -static bool write_binary_packet(const uint8_t *packet, size_t packet_size) -{ - const size_t written = fwrite(packet, 1, packet_size, stdout); - const int flush_result = fflush(stdout); - const bool complete = written == packet_size && flush_result == 0; - if (!complete) { - clearerr(stdout); - } - return complete; -} - static void write_binary_packet_until_sent( + trikke_context_t *context, + trikke_transport_sender_t *sender, const uint8_t *packet, size_t packet_size) { - while (!write_binary_packet(packet, packet_size)) { + while (true) { + const trikke_transport_status_t status = + trikke_transport_sender_step( + sender, &context->transport, packet, packet_size); + if (status == TRIKKE_TRANSPORT_COMPLETE) { + return; + } + if (status == TRIKKE_TRANSPORT_FATAL) { + // Preserve the in-flight packet and stop consuming the queue. A + // fatal backend invariant is not safely recoverable or retryable. + while (true) { + vTaskDelay(portMAX_DELAY); + } + } vTaskDelay(pdMS_TO_TICKS(TRIKKE_TRANSPORT_RETRY_DELAY_MS)); } } @@ -185,12 +193,14 @@ static void output_task(void *argument) uint8_t packet[TRIKKE_WIRE_MAX_PACKET_SIZE] = {0}; uint32_t packet_sequence = 0; uint32_t sample_packet_count = 0; + trikke_transport_sender_t sender; + trikke_transport_sender_init(&sender); size_t packet_size = trikke_encode_metadata_packet( packet, sizeof(packet), packet_sequence++, esp_timer_get_time(), atomic_load(&context->dropped_sample_count), atomic_load(&context->loop_overrun_count), &TRIKKE_METADATA); - write_binary_packet_until_sent(packet, packet_size); + write_binary_packet_until_sent(context, &sender, packet, packet_size); while (true) { trikke_wire_sample_t samples[TRIKKE_WIRE_MAX_RECORDS] = {0}; @@ -227,14 +237,15 @@ static void output_task(void *argument) packet, sizeof(packet), packet_sequence++, esp_timer_get_time(), atomic_load(&context->dropped_sample_count), atomic_load(&context->loop_overrun_count), &TRIKKE_METADATA); - write_binary_packet_until_sent(packet, packet_size); + write_binary_packet_until_sent( + context, &sender, packet, packet_size); } packet_size = trikke_encode_sample_packet( packet, sizeof(packet), packet_sequence++, atomic_load(&context->dropped_sample_count), atomic_load(&context->loop_overrun_count), samples, sample_count); - write_binary_packet_until_sent(packet, packet_size); + write_binary_packet_until_sent(context, &sender, packet, packet_size); ++sample_packet_count; } } @@ -324,6 +335,17 @@ void app_main(void) // binary frames whenever a payload byte equals LF. usb_serial_jtag_vfs_set_tx_line_endings(ESP_LINE_ENDINGS_LF); + err = trikke_usb_transport_init( + &s_context.usb_transport, &s_context.transport); + if (err != ESP_OK) { + ESP_LOGE(TAG, "USB transport initialization failed: %s", + esp_err_to_name(err)); + l3g4200d_deinit(&s_context.gyroscope); + adxl345_deinit(&s_context.accelerometer); + i2c_del_master_bus(bus); + return; + } + s_context.sample_queue = xQueueCreate(TRIKKE_SAMPLE_QUEUE_DEPTH, sizeof(trikke_wire_sample_t)); if (s_context.sample_queue == NULL) { @@ -331,6 +353,7 @@ void app_main(void) l3g4200d_deinit(&s_context.gyroscope); adxl345_deinit(&s_context.accelerometer); i2c_del_master_bus(bus); + trikke_usb_transport_deinit(&s_context.usb_transport); return; } @@ -351,6 +374,7 @@ void app_main(void) l3g4200d_deinit(&s_context.gyroscope); adxl345_deinit(&s_context.accelerometer); i2c_del_master_bus(bus); + trikke_usb_transport_deinit(&s_context.usb_transport); return; } diff --git a/main/trikke_transport.c b/main/trikke_transport.c new file mode 100644 index 0000000..29eeac0 --- /dev/null +++ b/main/trikke_transport.c @@ -0,0 +1,43 @@ +#include "trikke_transport.h" + +static bool status_is_valid(trikke_transport_status_t status) +{ + return status >= TRIKKE_TRANSPORT_COMPLETE && + status <= TRIKKE_TRANSPORT_FATAL; +} + +void trikke_transport_sender_init(trikke_transport_sender_t *sender) +{ + if (sender != NULL) { + sender->pending = false; + } +} + +trikke_transport_status_t trikke_transport_sender_step( + trikke_transport_sender_t *sender, + const trikke_transport_t *transport, + const uint8_t *packet, + size_t packet_size) +{ + if (sender == NULL || transport == NULL || transport->begin == NULL || + transport->poll == NULL || packet == NULL || packet_size == 0) { + return TRIKKE_TRANSPORT_FATAL; + } + + const trikke_transport_status_t status = sender->pending + ? transport->poll(transport->context) + : transport->begin(transport->context, packet, packet_size); + if (!status_is_valid(status)) { + sender->pending = false; + return TRIKKE_TRANSPORT_FATAL; + } + + if (status == TRIKKE_TRANSPORT_PENDING) { + sender->pending = true; + } else { + // RETRY from poll is allowed only when the backend has discarded or + // otherwise resolved the old transfer and knows resubmission is safe. + sender->pending = false; + } + return status; +} diff --git a/main/trikke_transport.h b/main/trikke_transport.h new file mode 100644 index 0000000..b90fb28 --- /dev/null +++ b/main/trikke_transport.h @@ -0,0 +1,48 @@ +#pragma once + +#include +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef enum { + TRIKKE_TRANSPORT_COMPLETE = 0, + TRIKKE_TRANSPORT_RETRY, + TRIKKE_TRANSPORT_PENDING, + TRIKKE_TRANSPORT_FATAL, +} trikke_transport_status_t; + +typedef trikke_transport_status_t (*trikke_transport_begin_fn)( + void *context, + const uint8_t *packet, + size_t packet_size); + +typedef trikke_transport_status_t (*trikke_transport_poll_fn)(void *context); + +typedef struct { + void *context; + trikke_transport_begin_fn begin; + trikke_transport_poll_fn poll; +} trikke_transport_t; + +typedef struct { + bool pending; +} trikke_transport_sender_t; + +void trikke_transport_sender_init(trikke_transport_sender_t *sender); + +// Advances one bounded transport operation. The packet storage must remain valid +// and unchanged from the first PENDING result through COMPLETE. While pending, +// only poll is called: an ambiguous timeout can never duplicate a frame. +trikke_transport_status_t trikke_transport_sender_step( + trikke_transport_sender_t *sender, + const trikke_transport_t *transport, + const uint8_t *packet, + size_t packet_size); + +#ifdef __cplusplus +} +#endif diff --git a/main/trikke_usb_transport.c b/main/trikke_usb_transport.c new file mode 100644 index 0000000..9d17576 --- /dev/null +++ b/main/trikke_usb_transport.c @@ -0,0 +1,92 @@ +#include "trikke_usb_transport.h" + +#include "driver/usb_serial_jtag.h" +#include "driver/usb_serial_jtag_vfs.h" +#include "freertos/FreeRTOS.h" + +#define TRIKKE_USB_TX_BUFFER_SIZE 512 +#define TRIKKE_USB_RX_BUFFER_SIZE 128 +#define TRIKKE_USB_OPERATION_TIMEOUT_MS 50 + +static trikke_transport_status_t usb_begin_packet( + void *context, + const uint8_t *packet, + size_t packet_size) +{ + const trikke_usb_transport_t *usb = context; + if (usb == NULL || !usb->initialized || packet == NULL || packet_size == 0 || + packet_size > TRIKKE_USB_TX_BUFFER_SIZE) { + return TRIKKE_TRANSPORT_FATAL; + } + + const int written = usb_serial_jtag_write_bytes( + packet, packet_size, + pdMS_TO_TICKS(TRIKKE_USB_OPERATION_TIMEOUT_MS)); + if (written == (int)packet_size) { + return TRIKKE_TRANSPORT_PENDING; + } + if (written == 0) { + return TRIKKE_TRANSPORT_RETRY; + } + return TRIKKE_TRANSPORT_FATAL; +} + +static trikke_transport_status_t usb_poll_packet(void *context) +{ + const trikke_usb_transport_t *usb = context; + if (usb == NULL || !usb->initialized) { + return TRIKKE_TRANSPORT_FATAL; + } + + const esp_err_t err = usb_serial_jtag_wait_tx_done( + pdMS_TO_TICKS(TRIKKE_USB_OPERATION_TIMEOUT_MS)); + if (err == ESP_OK) { + return TRIKKE_TRANSPORT_COMPLETE; + } + if (err == ESP_ERR_TIMEOUT) { + return TRIKKE_TRANSPORT_PENDING; + } + return TRIKKE_TRANSPORT_FATAL; +} + +esp_err_t trikke_usb_transport_init( + trikke_usb_transport_t *usb, + trikke_transport_t *transport) +{ + if (usb == NULL || transport == NULL) { + return ESP_ERR_INVALID_ARG; + } + if (usb->initialized) { + return ESP_ERR_INVALID_STATE; + } + + usb_serial_jtag_driver_config_t config = { + .tx_buffer_size = TRIKKE_USB_TX_BUFFER_SIZE, + .rx_buffer_size = TRIKKE_USB_RX_BUFFER_SIZE, + }; + const esp_err_t err = usb_serial_jtag_driver_install(&config); + if (err != ESP_OK) { + return err; + } + + usb->initialized = true; + transport->context = usb; + transport->begin = usb_begin_packet; + transport->poll = usb_poll_packet; + + // Route any unexpected VFS output through the installed driver as well, so + // it cannot race the driver's ISR by touching the hardware FIFO directly. + usb_serial_jtag_vfs_use_driver(); + return ESP_OK; +} + +void trikke_usb_transport_deinit(trikke_usb_transport_t *usb) +{ + if (usb == NULL || !usb->initialized) { + return; + } + + usb_serial_jtag_vfs_use_nonblocking(); + usb_serial_jtag_driver_uninstall(); + usb->initialized = false; +} diff --git a/main/trikke_usb_transport.h b/main/trikke_usb_transport.h new file mode 100644 index 0000000..36e97c7 --- /dev/null +++ b/main/trikke_usb_transport.h @@ -0,0 +1,24 @@ +#pragma once + +#include + +#include "esp_err.h" +#include "trikke_transport.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct { + bool initialized; +} trikke_usb_transport_t; + +esp_err_t trikke_usb_transport_init( + trikke_usb_transport_t *usb, + trikke_transport_t *transport); + +void trikke_usb_transport_deinit(trikke_usb_transport_t *usb); + +#ifdef __cplusplus +} +#endif diff --git a/tests/fixtures/README.md b/tests/fixtures/README.md index 1448dd0..dd5b940 100644 --- a/tests/fixtures/README.md +++ b/tests/fixtures/README.md @@ -1,9 +1,11 @@ # Hardware outage fixtures -These captures came from the assembled XIAO ESP32-C3 prototype. A temporary -validation build made the packet writer report failure while acquisition kept -running; that failure injection was removed before the production firmware was -built and flashed. The files contain only complete, CRC-valid `TRK1` frames. +These captures came from the assembled XIAO ESP32-C3 prototype. The files +contain only complete, CRC-valid `TRK1` frames. + +The two forced-outage captures used a temporary validation build that made the +packet writer report failure while acquisition kept running. That injection was +removed before production firmware was built and flashed. - `forced_outage_3s.trk` — SHA-256 `01482816cdaa668e4681c33c8baa1df331d733b9bbcbc4f448ece25e88185ad6`. @@ -15,10 +17,24 @@ built and flashed. The files contain only complete, CRC-valid `TRK1` frames. 138 samples were dropped after the 512-entry queue filled, the cumulative drop count reached 138, and the corresponding timestamp delta is exactly 1,390,000 us. +- `direct_usb_stall.trk` — SHA-256 + `40f874b7eaa7f705524ecdd75f832e8a724252366633116ac015fc75dfd16558`. + This came from the direct USB driver build after leaving the enumerated USB + endpoint without a serial reader long enough to overflow the acquisition + queue. It contains 864 samples. The retained block ends at sequence 511, + delivery resumes at 1,706, and both the sole 1,194-sample gap and the device's + cumulative drop counter equal 1,194. There are no packet gaps, CRC failures, + loop overruns, trailing bytes, or timestamp-saturation flags. + + The first captured sample is sequence 8 because the flashing process still + had the endpoint open long enough to drain sequences 0 through 7 before the + capture application opened. That is deliberate evidence of the remaining + distinction: USB endpoint drain is observable, but application receipt + requires the planned acknowledgement/replay layer. `tests/test_trikke_protocol.py` verifies the hashes, parses the captures in fragmented chunks, and asserts these signatures so the hardware evidence remains -executable regression data. To inspect either file manually: +executable regression data. To inspect a file manually: ```sh python3 tools/decode_binary.py tests/fixtures/forced_outage_3s.trk /tmp/outage.csv diff --git a/tests/fixtures/direct_usb_stall.trk b/tests/fixtures/direct_usb_stall.trk new file mode 100644 index 0000000..3d0e5cf Binary files /dev/null and b/tests/fixtures/direct_usb_stall.trk differ diff --git a/tests/test_trikke_protocol.py b/tests/test_trikke_protocol.py index 208d63c..a79bbad 100644 --- a/tests/test_trikke_protocol.py +++ b/tests/test_trikke_protocol.py @@ -10,6 +10,7 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "tools")) from trikke_protocol import ( # noqa: E402 + IntegrityTracker, PACKET_FLAG_TIMESTAMP_DELTA_SATURATED, PACKET_TYPE_METADATA, PACKET_TYPE_SAMPLES, @@ -50,6 +51,34 @@ class ProtocolContractTest(unittest.TestCase): ) cls.encoded = fixture.stdout + transport_executable = Path(cls.tempdir.name) / "transport_fixture" + subprocess.run( + [ + compiler, + "-std=c11", + "-Wall", + "-Wextra", + "-Werror", + "-I", + str(ROOT / "main"), + str(ROOT / "main" / "trikke_transport.c"), + str(ROOT / "tests" / "transport_fixture.c"), + "-o", + str(transport_executable), + ], + check=True, + ) + transport_fixture = subprocess.run( + [str(transport_executable)], capture_output=True + ) + if transport_fixture.returncode != 0: + stderr = transport_fixture.stderr.decode(errors="replace").strip() + raise AssertionError( + "transport fixture exited " + f"{transport_fixture.returncode}: {stderr}" + ) + cls.transport_fixture_passed = True + @classmethod def tearDownClass(cls) -> None: cls.tempdir.cleanup() @@ -103,6 +132,18 @@ class ProtocolContractTest(unittest.TestCase): self.assertEqual((2, 1, 258), tuple(row[14:17])) self.assertEqual(3, row[-1]) + def test_transport_state_machine_contract(self) -> None: + self.assertTrue(self.transport_fixture_passed) + + def test_integrity_sequence_wrap_classification(self) -> None: + self.assertEqual( + (0, 0), IntegrityTracker._classify_sequence(0xFFFFFFFF, 0) + ) + self.assertEqual( + (2, 0), IntegrityTracker._classify_sequence(0xFFFFFFFE, 1) + ) + self.assertEqual((0, 1), IntegrityTracker._classify_sequence(1000, 0)) + def test_crc_failure_resynchronizes_to_next_frame(self) -> None: first_size = 36 + 48 damaged = bytearray(self.encoded[:first_size]) @@ -141,6 +182,7 @@ class ProtocolContractTest(unittest.TestCase): "forced_outage_3s.trk": { "sha256": "01482816cdaa668e4681c33c8baa1df331d733b9bbcbc4f448ece25e88185ad6", "sample_count": 2144, + "first_sequence": 0, "last_sequence": 2143, "max_dropped": 0, "gaps": [], @@ -148,10 +190,19 @@ class ProtocolContractTest(unittest.TestCase): "forced_outage_7s.trk": { "sha256": "2ea8a5742944bdebc13bec2ccdbceba75f0bb71e48c856b0f86285878e190cd3", "sample_count": 1840, + "first_sequence": 0, "last_sequence": 1977, "max_dropped": 138, "gaps": [(511, 650, 1_390_000)], }, + "direct_usb_stall.trk": { + "sha256": "40f874b7eaa7f705524ecdd75f832e8a724252366633116ac015fc75dfd16558", + "sample_count": 864, + "first_sequence": 8, + "last_sequence": 2065, + "max_dropped": 1194, + "gaps": [(511, 1706, 11_950_000)], + }, } for name, contract in expected.items(): @@ -175,7 +226,7 @@ class ProtocolContractTest(unittest.TestCase): samples = [sample for frame in frames for sample in frame.samples] self.assertEqual(contract["sample_count"], len(samples)) - self.assertEqual(0, samples[0].sequence) + self.assertEqual(contract["first_sequence"], samples[0].sequence) self.assertEqual(contract["last_sequence"], samples[-1].sequence) self.assertEqual( contract["max_dropped"], @@ -191,6 +242,21 @@ class ProtocolContractTest(unittest.TestCase): ) ) + integrity = IntegrityTracker() + for frame in frames: + integrity.observe(frame) + self.assertEqual(0, integrity.packet_gap_count) + self.assertEqual(0, integrity.packet_reset_count) + self.assertEqual( + contract["max_dropped"], integrity.sample_gap_count + ) + self.assertEqual(0, integrity.sample_reset_count) + self.assertEqual( + contract["max_dropped"], + integrity.final_dropped_sample_count, + ) + self.assertEqual(0, integrity.final_loop_overrun_count) + gaps = [ ( left.sequence, diff --git a/tests/transport_fixture.c b/tests/transport_fixture.c new file mode 100644 index 0000000..32f9c33 --- /dev/null +++ b/tests/transport_fixture.c @@ -0,0 +1,129 @@ +#include +#include +#include + +#include "trikke_transport.h" + +typedef struct { + trikke_transport_status_t begin_status; + trikke_transport_status_t poll_status; + unsigned int begin_calls; + unsigned int poll_calls; + const uint8_t *packet; + size_t packet_size; +} mock_transport_t; + +static int fail(int code, const char *message) +{ + fprintf(stderr, "transport fixture failure %d: %s\n", code, message); + return code; +} + +static trikke_transport_status_t mock_begin( + void *context, + const uint8_t *packet, + size_t packet_size) +{ + mock_transport_t *mock = context; + ++mock->begin_calls; + mock->packet = packet; + mock->packet_size = packet_size; + return mock->begin_status; +} + +static trikke_transport_status_t mock_poll(void *context) +{ + mock_transport_t *mock = context; + ++mock->poll_calls; + return mock->poll_status; +} + +int main(void) +{ + const uint8_t packet[] = {0x54, 0x52, 0x4B, 0x31}; + mock_transport_t mock = { + .begin_status = TRIKKE_TRANSPORT_RETRY, + .poll_status = TRIKKE_TRANSPORT_PENDING, + }; + const trikke_transport_t transport = { + .context = &mock, + .begin = mock_begin, + .poll = mock_poll, + }; + trikke_transport_sender_t sender; + trikke_transport_sender_init(&sender); + + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_RETRY || + sender.pending || mock.begin_calls != 1 || mock.poll_calls != 0 || + mock.packet != packet || mock.packet_size != sizeof(packet)) { + return fail(1, "zero-accept submission must remain retryable"); + } + + mock.begin_status = TRIKKE_TRANSPORT_PENDING; + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_PENDING || + !sender.pending || mock.begin_calls != 2 || mock.poll_calls != 0) { + return fail(2, "accepted submission must become pending"); + } + + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_PENDING || + !sender.pending || mock.begin_calls != 2 || mock.poll_calls != 1) { + return fail(3, "pending transfer must poll without resubmission"); + } + + mock.poll_status = TRIKKE_TRANSPORT_COMPLETE; + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_COMPLETE || + sender.pending || mock.begin_calls != 2 || mock.poll_calls != 2) { + return fail(4, "completed transfer must return to idle"); + } + + mock.begin_status = TRIKKE_TRANSPORT_COMPLETE; + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_COMPLETE || + sender.pending || mock.begin_calls != 3) { + return fail(5, "synchronous completion contract"); + } + + mock.begin_status = TRIKKE_TRANSPORT_PENDING; + mock.poll_status = TRIKKE_TRANSPORT_RETRY; + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_PENDING || + trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_RETRY || + sender.pending) { + return fail(6, "backend-confirmed safe retry must return to idle"); + } + + if (trikke_transport_sender_step( + NULL, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_FATAL || + trikke_transport_sender_step( + &sender, NULL, packet, sizeof(packet)) != TRIKKE_TRANSPORT_FATAL || + trikke_transport_sender_step( + &sender, &transport, NULL, sizeof(packet)) != + TRIKKE_TRANSPORT_FATAL || + trikke_transport_sender_step( + &sender, &transport, packet, 0) != TRIKKE_TRANSPORT_FATAL) { + return fail(7, "invalid arguments must fail closed"); + } + + mock.begin_status = (trikke_transport_status_t)99; + if (trikke_transport_sender_step( + &sender, &transport, packet, sizeof(packet)) != + TRIKKE_TRANSPORT_FATAL || + sender.pending) { + return fail(8, "unknown backend status must fail closed"); + } + + return 0; +} diff --git a/tools/capture_binary.py b/tools/capture_binary.py index 94394a5..5d7bbb2 100644 --- a/tools/capture_binary.py +++ b/tools/capture_binary.py @@ -6,6 +6,7 @@ import csv import glob import signal import sys +from contextlib import ExitStack from datetime import datetime from pathlib import Path @@ -13,9 +14,9 @@ import serial from trikke_protocol import ( CSV_COLUMNS, - PACKET_FLAG_TIMESTAMP_DELTA_SATURATED, PACKET_TYPE_METADATA, Frame, + IntegrityTracker, Metadata, StreamParser, sample_to_csv_row, @@ -28,6 +29,11 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--baud", type=int, default=115200) parser.add_argument("--output", type=Path, help="validated binary .trk output") parser.add_argument("--csv", type=Path, help="decoded CSV output") + parser.add_argument( + "--wire", + type=Path, + help="optional byte-for-byte serial capture, including startup text", + ) return parser.parse_args() @@ -57,6 +63,8 @@ def main() -> int: csv_output = args.csv or output.with_suffix(".csv") output.parent.mkdir(parents=True, exist_ok=True) csv_output.parent.mkdir(parents=True, exist_ok=True) + if args.wire is not None: + args.wire.parent.mkdir(parents=True, exist_ok=True) stop_requested = False @@ -72,83 +80,46 @@ def main() -> int: pending_frames: list[Frame] = [] sample_count = 0 metadata_count = 0 - packet_gap_count = 0 - packet_reset_count = 0 - sample_gap_count = 0 - sample_reset_count = 0 - timing_anomaly_count = 0 - timestamp_saturation_frame_count = 0 - accel_stale_count = 0 - accel_overrun_count = 0 - gyro_stale_count = 0 - gyro_overrun_count = 0 - previous_packet_sequence = None - previous_sample_sequence = None - previous_timestamp_us = None - final_dropped_count = 0 - final_loop_overrun_count = 0 + integrity = IntegrityTracker() serial_error: serial.SerialException | None = None def render_frame(frame: Frame, writer: csv.writer) -> None: - nonlocal sample_count, sample_gap_count, sample_reset_count - nonlocal timing_anomaly_count - nonlocal accel_stale_count, accel_overrun_count - nonlocal gyro_stale_count, gyro_overrun_count - nonlocal previous_sample_sequence, previous_timestamp_us + nonlocal sample_count if metadata is None: pending_frames.append(frame) return for sample in frame.samples: - if previous_sample_sequence is not None: - expected = (previous_sample_sequence + 1) & 0xFFFFFFFF - if sample.sequence != expected: - if sample.sequence > expected: - sample_gap_count += sample.sequence - expected - else: - sample_reset_count += 1 - if previous_timestamp_us is not None: - if sample.timestamp_us - previous_timestamp_us != 10_000: - timing_anomaly_count += 1 - if not sample.accel_status & 0x80: - accel_stale_count += 1 - if sample.accel_status & 0x01: - accel_overrun_count += 1 - if not sample.gyro_status & 0x08: - gyro_stale_count += 1 - if sample.gyro_status & 0x80: - gyro_overrun_count += 1 writer.writerow(sample_to_csv_row(sample, metadata, frame.loop_overrun_count)) - previous_sample_sequence = sample.sequence - previous_timestamp_us = sample.timestamp_us sample_count += 1 if sample_count % 500 == 0: print(f" {sample_count} samples captured", flush=True) print(f"Recording {port} to {output} and {csv_output}; press Ctrl-C to stop") try: - with serial.Serial(port, args.baud, timeout=0.25) as sensor, output.open( - "wb" - ) as raw_capture, csv_output.open("w", encoding="utf-8", newline="") as decoded: + with ExitStack() as stack: + sensor = stack.enter_context( + serial.Serial(port, args.baud, timeout=0.25) + ) + raw_capture = stack.enter_context(output.open("wb")) + decoded = stack.enter_context( + csv_output.open("w", encoding="utf-8", newline="") + ) + wire_capture = ( + stack.enter_context(args.wire.open("wb")) + if args.wire is not None + else None + ) writer = csv.writer(decoded) writer.writerow(CSV_COLUMNS) while not stop_requested: chunk = sensor.read(4096) if not chunk: continue + if wire_capture is not None: + wire_capture.write(chunk) for frame in parser.feed(chunk): raw_capture.write(frame.raw) - final_dropped_count = frame.dropped_sample_count - final_loop_overrun_count = frame.loop_overrun_count - if previous_packet_sequence is not None: - expected = (previous_packet_sequence + 1) & 0xFFFFFFFF - if frame.packet_sequence != expected: - if frame.packet_sequence > expected: - packet_gap_count += frame.packet_sequence - expected - else: - packet_reset_count += 1 - previous_packet_sequence = frame.packet_sequence - if frame.flags & PACKET_FLAG_TIMESTAMP_DELTA_SATURATED: - timestamp_saturation_frame_count += 1 + integrity.observe(frame) if frame.packet_type == PACKET_TYPE_METADATA: metadata = frame.metadata metadata_count += 1 @@ -159,16 +130,21 @@ def main() -> int: render_frame(frame, writer) raw_capture.flush() decoded.flush() + if wire_capture is not None: + wire_capture.flush() except serial.SerialException as exc: print(f"Serial error: {exc}", file=sys.stderr) serial_error = exc print( f"Stopped after {sample_count} samples and {metadata_count} metadata frames; " - f"packet_gaps={packet_gap_count}, packet_resets={packet_reset_count}, " - f"sample_gaps={sample_gap_count}, sample_resets={sample_reset_count}, " - f"timing_anomalies={timing_anomaly_count}, " - f"timestamp_saturation_frames={timestamp_saturation_frame_count}, " + f"packet_gaps={integrity.packet_gap_count}, " + f"packet_resets={integrity.packet_reset_count}, " + f"sample_gaps={integrity.sample_gap_count}, " + f"sample_resets={integrity.sample_reset_count}, " + f"timing_anomalies={integrity.timing_anomaly_count}, " + "timestamp_saturation_frames=" + f"{integrity.timestamp_saturation_frame_count}, " f"startup_crc_rejects={parser.startup_crc_errors}, " f"stream_crc_errors={parser.crc_errors}, " f"header_errors={parser.header_errors}, " @@ -176,12 +152,17 @@ def main() -> int: f"trailing_partial_bytes={parser.buffered_bytes}" ) print( - f"Status totals: accel_stale={accel_stale_count}, " - f"accel_overrun={accel_overrun_count}, gyro_stale={gyro_stale_count}, " - f"gyro_overrun={gyro_overrun_count}, dropped={final_dropped_count}, " - f"acquisition_loop_overruns={final_loop_overrun_count}" + f"Status totals: accel_stale={integrity.accel_stale_count}, " + f"accel_overrun={integrity.accel_overrun_count}, " + f"gyro_stale={integrity.gyro_stale_count}, " + f"gyro_overrun={integrity.gyro_overrun_count}, " + f"dropped={integrity.final_dropped_sample_count}, " + f"acquisition_loop_overruns={integrity.final_loop_overrun_count}" ) - print(f"Saved {output} and {csv_output}") + saved = f"Saved {output} and {csv_output}" + if args.wire is not None: + saved += f"; raw wire saved to {args.wire}" + print(saved) if serial_error is not None: return 1 return 0 if metadata is not None else 4 diff --git a/tools/decode_binary.py b/tools/decode_binary.py index 83f4a27..f86a204 100644 --- a/tools/decode_binary.py +++ b/tools/decode_binary.py @@ -7,9 +7,9 @@ from pathlib import Path from trikke_protocol import ( CSV_COLUMNS, - PACKET_FLAG_TIMESTAMP_DELTA_SATURATED, PACKET_TYPE_METADATA, Frame, + IntegrityTracker, StreamParser, sample_to_csv_row, ) @@ -40,13 +40,12 @@ def main() -> int: args.output.parent.mkdir(parents=True, exist_ok=True) sample_count = 0 - timestamp_saturation_frame_count = 0 + integrity = IntegrityTracker() with args.output.open("w", encoding="utf-8", newline="") as target: writer = csv.writer(target) writer.writerow(CSV_COLUMNS) for frame in frames: - if frame.flags & PACKET_FLAG_TIMESTAMP_DELTA_SATURATED: - timestamp_saturation_frame_count += 1 + integrity.observe(frame) if frame.packet_type == PACKET_TYPE_METADATA: if frame.metadata is not None: metadata = frame.metadata @@ -62,7 +61,19 @@ def main() -> int: f"startup_crc_rejects={stream.startup_crc_errors}, " f"stream_crc_errors={stream.crc_errors}, header_errors={stream.header_errors}, " f"skipped_nonframe_bytes={stream.skipped_bytes}, " - f"timestamp_saturation_frames={timestamp_saturation_frame_count}, " + f"packet_gaps={integrity.packet_gap_count}, " + f"packet_resets={integrity.packet_reset_count}, " + f"sample_gaps={integrity.sample_gap_count}, " + f"sample_resets={integrity.sample_reset_count}, " + f"timing_anomalies={integrity.timing_anomaly_count}, " + "timestamp_saturation_frames=" + f"{integrity.timestamp_saturation_frame_count}, " + f"accel_stale={integrity.accel_stale_count}, " + f"accel_overrun={integrity.accel_overrun_count}, " + f"gyro_stale={integrity.gyro_stale_count}, " + f"gyro_overrun={integrity.gyro_overrun_count}, " + f"dropped={integrity.final_dropped_sample_count}, " + f"acquisition_loop_overruns={integrity.final_loop_overrun_count}, " f"trailing_partial_bytes={stream.buffered_bytes}; saved {args.output}" ) return 0 diff --git a/tools/trikke_protocol.py b/tools/trikke_protocol.py index 5af8a02..a3dad5f 100644 --- a/tools/trikke_protocol.py +++ b/tools/trikke_protocol.py @@ -84,6 +84,75 @@ class Frame: raw: bytes +@dataclass +class IntegrityTracker: + packet_gap_count: int = 0 + packet_reset_count: int = 0 + sample_gap_count: int = 0 + sample_reset_count: int = 0 + timing_anomaly_count: int = 0 + timestamp_saturation_frame_count: int = 0 + accel_stale_count: int = 0 + accel_overrun_count: int = 0 + gyro_stale_count: int = 0 + gyro_overrun_count: int = 0 + final_dropped_sample_count: int = 0 + final_loop_overrun_count: int = 0 + _previous_packet_sequence: int | None = None + _previous_sample_sequence: int | None = None + _previous_timestamp_us: int | None = None + + @staticmethod + def _classify_sequence( + previous: int, + current: int, + ) -> tuple[int, int]: + expected = (previous + 1) & 0xFFFFFFFF + forward_distance = (current - expected) & 0xFFFFFFFF + if forward_distance == 0: + return (0, 0) + if forward_distance < 0x80000000: + return (forward_distance, 0) + return (0, 1) + + def observe(self, frame: Frame) -> None: + if self._previous_packet_sequence is not None: + gaps, resets = self._classify_sequence( + self._previous_packet_sequence, frame.packet_sequence + ) + self.packet_gap_count += gaps + self.packet_reset_count += resets + self._previous_packet_sequence = frame.packet_sequence + + if frame.flags & PACKET_FLAG_TIMESTAMP_DELTA_SATURATED: + self.timestamp_saturation_frame_count += 1 + self.final_dropped_sample_count = frame.dropped_sample_count + self.final_loop_overrun_count = frame.loop_overrun_count + + for sample in frame.samples: + if self._previous_sample_sequence is not None: + gaps, resets = self._classify_sequence( + self._previous_sample_sequence, sample.sequence + ) + self.sample_gap_count += gaps + self.sample_reset_count += resets + if ( + self._previous_timestamp_us is not None + and sample.timestamp_us - self._previous_timestamp_us != 10_000 + ): + self.timing_anomaly_count += 1 + if not sample.accel_status & 0x80: + self.accel_stale_count += 1 + if sample.accel_status & 0x01: + self.accel_overrun_count += 1 + if not sample.gyro_status & 0x08: + self.gyro_stale_count += 1 + if sample.gyro_status & 0x80: + self.gyro_overrun_count += 1 + self._previous_sample_sequence = sample.sequence + self._previous_timestamp_us = sample.timestamp_us + + def _lround(value: float) -> int: """Match C lroundf: nearest integer, halfway cases away from zero.""" return math.floor(value + 0.5) if value >= 0 else math.ceil(value - 0.5)