applications: gesture_recognition: track lost samples

Prepend ID to samples send of NUS to keep track of potentially lost
samples. Add scripts/check_nus_data.py to verify that received samples
are consecutive and trim the added IDs.

Jira: NCSDK-37426
Signed-off-by: Szymon Antkowiak <szymon.antkowiak@nordicsemi.no>

Author: Szynkaa

applications/gesture_recognition/README.rst

@@ -270,6 +270,11 @@ You can find raw datasets used for model training on the `training dataset`_ pag
 
    In this mode, you must have an additional development kit running the `Nordic central UART sample`_ to receive the NUS data.
 
+   .. note::
+      When using Bluetooth LE NUS mode, some samples may be lost due to RF noise or increased distance between the device and the central.
+      To help track lost samples, the application prepends a sequential ID to each sample transmitted over NUS.
+      Use the :file:`scripts/check_nus_data.py` script to check for dropped samples and strip the ID from the received data.
+
 Building and running
 ********************
 

applications/gesture_recognition/src/ble/nus/ble_nus.c

@@ -145,9 +145,11 @@ int ble_nus_send(const int16_t *input_data)
 		return -ENOTCONN;
 	}
 
-	len = snprintf(buffer, sizeof(buffer), "%d,%d,%d,%d,%d,%d\r\n",
-		       input_data[0], input_data[1], input_data[2],
-		       input_data[3], input_data[4], input_data[5]);
+	static uint32_t id;
+
+	id++;
+	len = snprintf(buffer, sizeof(buffer), "%u %d,%d,%d,%d,%d,%d\r\n", id, input_data[0],
+		       input_data[1], input_data[2], input_data[3], input_data[4], input_data[5]);
 	if ((len <= 0) || (len >= (int)sizeof(buffer))) {
 		return -EINVAL;
 	}

scripts/check_nus_data.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+#
+# Copyright (c) 2026 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+#
+
+import argparse
+from pathlib import Path
+
+
+def check_consecutive_ids(log_path: Path) -> tuple[list[str], int]:
+    """Return (issues, parsed_count) after validating consecutive IDs."""
+    issues: list[str] = []
+    parsed_count = 0
+    prev_id: int | None = None
+
+    with log_path.open("r", encoding="utf-8") as f:
+        for line_no, raw_line in enumerate(f, start=1):
+            line = raw_line.strip()
+
+            if not line:
+                continue
+
+            first_token = line.split(maxsplit=1)[0]
+            try:
+                curr_id = int(first_token)
+            except ValueError:
+                issues.append(
+                    f"Line {line_no}: could not parse entry ID from first token '{first_token}'"
+                )
+                continue
+
+            parsed_count += 1
+
+            if prev_id is not None:
+                expected = prev_id + 1
+                if curr_id != expected:
+                    issues.append(f"Line {line_no}: expected ID {expected}, got {curr_id}")
+
+            prev_id = curr_id
+
+    return issues, parsed_count
+
+
+def create_raw_log(log_path: Path) -> tuple[Path, int]:
+    """Create a copy with leading ID removed from each parsable line."""
+    raw_path = log_path.with_name(f"{log_path.stem}_data{log_path.suffix}")
+    converted_count = 0
+
+    with log_path.open("r", encoding="utf-8") as src, raw_path.open("w", encoding="utf-8") as dst:
+        for raw_line in src:
+            line = raw_line.rstrip("\n")
+            parts = line.split(maxsplit=1)
+
+            if len(parts) == 2:
+                try:
+                    int(parts[0])
+                except ValueError:
+                    dst.write(raw_line)
+                else:
+                    dst.write(parts[1] + "\n")
+                    converted_count += 1
+            else:
+                dst.write(raw_line)
+
+    return raw_path, converted_count
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="""
+            Check NUS serial output for consecutive entry IDs.
+            Expected format is "<id> <data>".
+            If no problems were detected, than creates a '_data' file with IDs removed.
+        """,
+        allow_abbrev=False,
+    )
+    parser.add_argument(
+        "log_file",
+        help="Path to serial log file",
+    )
+    args = parser.parse_args()
+
+    log_path = Path(args.log_file)
+    if not log_path.is_file():
+        print(f"Error: file not found: {log_path}")
+        return 2
+
+    issues, parsed_count = check_consecutive_ids(log_path)
+
+    if parsed_count == 0:
+        print("No valid log entries found.")
+        return 1
+
+    if issues:
+        print(f"Found {len(issues)} issue(s) in {parsed_count} parsed entries:")
+        for issue in issues:
+            print(f"- {issue}")
+        return 1
+
+    print(f"OK: {parsed_count} entries have consecutive IDs.")
+
+    raw_path, converted_count = create_raw_log(log_path)
+    print(f"Created raw log: {raw_path} ({converted_count} entries converted)")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())