Add BrainVision RDA DATA32 acquisition gateway

.github/workflows/run_unix.yml

@@ -261,6 +261,9 @@ jobs:
     # start testing
     - name: Run unit tests
       run: $GITHUB_WORKSPACE/build/tests/brainflow_tests
+    - name: BrainVision RDA emulator
+      if: runner.os == 'Linux'
+      run: python3 $GITHUB_WORKSPACE/emulator/brainflow_emulator/brainvision_rda_test.py $GITHUB_WORKSPACE/installed/lib/libBoardController.so
     - name: Synthetic C# Test MacOS
       if: runner.os == 'macOS'
       run: |

csharp_package/brainflow/brainflow/board_controller_library.cs

@@ -124,7 +124,8 @@ public enum BoardIds
         IRONBCI_32_BOARD = 65,
         NEUROPAWN_KNIGHT_BOARD_IMU = 66,
         MUSE_S_ATHENA_BOARD = 67,
-        SHIMMER3_BOARD = 68
+        SHIMMER3_BOARD = 68,
+        BRAINVISION_RDA_BOARD = 71
     };
 
 

docs/SupportedBoards.rst

@@ -1581,3 +1581,61 @@ Common sensor bits:
 - :code:`0x000008`, EXG2 24-bit
 - :code:`0x000004`, GSR, exposed as EDA
 - :code:`0x002000`, battery
+
+BrainVision RDA (32 channels, 1000 Hz)
+-------------------------------------
+
+``BoardIds.BRAINVISION_RDA_BOARD`` (71) receives the TCP DATA32 stream from
+BrainVision Recorder's Remote Data Access interface. This is a Recorder gateway,
+not a direct USB driver. Enable RDA, configure **exactly 32 channels at 1000 Hz**,
+and start Recorder acquisition before calling ``start_stream``. Other channel
+counts/rates and DATA16 are rejected. Set ``ip_address`` to the Recorder IPv4
+address; ``ip_port`` defaults to 51244. ``timeout`` is the first-sample timeout
+in seconds (default 5). Preparation checks connectivity; each start reconnects
+to obtain fresh metadata and discard samples queued before recording.
+
+Rows 1-32 follow Recorder channel order and use its per-channel resolution to
+convert to microvolts. Use ``config_board("get_channel_names")`` after starting
+to get the channel names as JSON. Configure EEG-only channels: this profile does
+not classify arbitrary auxiliary channels supplied by Recorder.
+
+``package_num_channel`` contains the RDA block number, repeated for each sample
+in that block. ``other_channels`` contains the RDA marker count at each sample
+and the zero-based sample offset within its block. ``config_board("get_marker")``
+pops one marker as JSON (block, position, points, channel, type, description),
+or returns ``{}`` when empty. Channel -1 means all channels. The queue retains
+256 events; overflow drops the oldest event with a warning. Coincident markers
+remain separate. Each name is limited to 512 bytes and the complete JSON name list to 4000 bytes.
+The combined marker type/description is limited to 512 bytes so each event fits
+standard binding buffers. Invalid UTF-8 is replaced when serializing JSON.
+The normal BrainFlow marker row remains available for ``insert_marker``.
+
+Timestamps anchor the last sample of each block to host receipt and backdate
+samples at 1000 Hz. They are not hardware timestamps or clock synchronization.
+A Recorder STOP, malformed message, unsupported profile or lost connection ends
+the reader. Call ``stop_stream`` and ``start_stream`` to reconnect after restoring
+Recorder acquisition; there is no silent automatic reconnect.
+
+.. code-block:: python
+
+    from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds
+
+    params = BrainFlowInputParams()
+    params.ip_address = "127.0.0.1"
+    board = BoardShim(BoardIds.BRAINVISION_RDA_BOARD, params)
+    board.prepare_session()
+    try:
+        board.start_stream()
+        print(board.config_board("get_channel_names"))
+        # Consume get_board_data() and drain get_marker while recording.
+        board.stop_stream()
+    finally:
+        board.release_session()
+
+Validation uses protocol fixtures and a loopback RDA server, including fragmented
+TCP reads, scaling, coincident markers, profile rejection, timeouts and restarting.
+Physical Recorder/amplifier validation remains outstanding.
+
+Protocol references:
+https://github.com/brain-products/LSL-BrainVisionRDA and
+https://github.com/fieldtrip/fieldtrip/blob/master/realtime/src/buffer/src/rdaserver.c

emulator/brainflow_emulator/brainvision_rda_test.py

@@ -0,0 +1,132 @@
+"""Hardware-free RDA session test. Pass the native BoardController library path."""
+import ctypes
+import json
+import socket
+import struct
+import sys
+import threading
+import time
+
+UUID = bytes.fromhex('8e45584396c9864caf4a98bbf6c91450')
+
+
+def message(kind, body):
+    return UUID + struct.pack('<II', len(body) + 24, kind) + body
+
+
+def main(library):
+    lib = ctypes.CDLL(library)
+    lib.prepare_session.argtypes = [ctypes.c_int, ctypes.c_char_p]
+    lib.start_stream.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p]
+    lib.stop_stream.argtypes = lib.prepare_session.argtypes
+    lib.release_session.argtypes = lib.prepare_session.argtypes
+    lib.get_current_board_data.argtypes = [ctypes.c_int, ctypes.c_int,
+        ctypes.POINTER(ctypes.c_double), ctypes.POINTER(ctypes.c_int), ctypes.c_int, ctypes.c_char_p]
+    lib.config_board.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.POINTER(ctypes.c_int),
+        ctypes.c_int, ctypes.c_int, ctypes.c_char_p]
+    server = socket.socket()
+    server.bind(('127.0.0.1', 0))
+    server.listen()
+    server.settimeout(0.1)
+    stop = threading.Event()
+    mode = ['valid']
+    workers = []
+    failures = []
+    params = json.dumps(dict(serial_port='', mac_address='', ip_address='127.0.0.1',
+        ip_address_aux='', ip_address_anc='', ip_port=server.getsockname()[1], ip_port_aux=0,
+        ip_port_anc=0, ip_protocol=0, other_info='', timeout=1, serial_number='',
+        file='', file_aux='', file_anc='', master_board=-100)).encode()
+
+    def peer(conn, scenario):
+        try:
+            with conn:
+                if scenario == 'silent':
+                    stop.wait(3)
+                    return
+                if scenario == 'oversized':
+                    conn.sendall(UUID + struct.pack('<II', 0xffffffff, 4))
+                    return
+                channels = 31 if scenario == 'profile' else 32
+                start = struct.pack('<Id', channels, 1000) + struct.pack('<32d', *([0.5] * 32))
+                start += b''.join(('C%d\0' % i).encode() for i in range(32))
+                packet = message(1, start)
+                for offset in range(0, len(packet), 7):
+                    conn.sendall(packet[offset:offset + 7])
+                if scenario == 'profile':
+                    return
+                block = 0
+                while not stop.is_set():
+                    label = b'S\xff12' if scenario == 'encoding' else b'S 12'
+                    marker = struct.pack('<IIII', 30, 1, 1, 0xffffffff) + b'Stimulus\0' + label + b'\0'
+                    body = struct.pack('<III', block, 2, 2) + struct.pack('<64f', *([-3.5] * 64))
+                    packet = message(4, body + marker * 2)
+                    for offset in range(0, len(packet), 17):
+                        conn.sendall(packet[offset:offset + 17])
+                    if scenario == 'stop':
+                        conn.sendall(message(3, b''))
+                        return
+                    block += 1
+                    stop.wait(0.004)
+        except (BrokenPipeError, ConnectionResetError):
+            pass  # prepare/start and stop deliberately close earlier connections
+        except Exception as exc:
+            failures.append(exc)
+
+    def accept():
+        while not stop.is_set():
+            try:
+                conn, _ = server.accept()
+            except socket.timeout:
+                continue
+            worker = threading.Thread(target=peer, args=(conn, mode[0]))
+            workers.append(worker)
+            worker.start()
+
+    listener = threading.Thread(target=accept)
+    listener.start()
+    try:
+        assert lib.prepare_session(71, params) == 0
+        for _ in range(2):
+            assert lib.start_stream(1024, b'', 71, params) == 0
+            time.sleep(0.05)
+            data = (ctypes.c_double * (37 * 1024))()
+            count = ctypes.c_int()
+            assert lib.get_current_board_data(1024, 0, data, ctypes.byref(count), 71, params) == 0
+            assert count.value >= 2
+            assert all(data[count.value + i] == -1.75 for i in range(count.value))
+            assert 2 in data[33 * count.value:34 * count.value]
+            response = ctypes.create_string_buffer(8192)
+            length = ctypes.c_int()
+            assert lib.config_board(b'get_marker', response, ctypes.byref(length), 8192, 71, params) == 0
+            event = json.loads(response.value)
+            assert event['description'] == 'S 12' and event['position'] == 1
+            assert lib.config_board(b'get_channel_names', response, ctypes.byref(length), 8192, 71, params) == 0
+            assert json.loads(response.value) == ['C%d' % i for i in range(32)]
+            assert lib.stop_stream(71, params) == 0
+        for scenario, expected in [('encoding', 0), ('profile', 5), ('oversized', 5), ('silent', 18), ('stop', 0)]:
+            mode[0] = scenario
+            before = time.monotonic()
+            assert lib.start_stream(1024, b'', 71, params) == expected, scenario
+            assert time.monotonic() - before < 2.5, 'stop did not interrupt recv'
+            if scenario == 'encoding':
+                assert lib.config_board(b'get_marker', response, ctypes.byref(length), 8192, 71, params) == 0
+                assert json.loads(response.value)['description'] == 'S\ufffd12'
+            if expected == 0:
+                assert lib.stop_stream(71, params) == 0
+        assert lib.release_session(71, params) == 0
+        mode[0] = 'valid'
+        assert lib.prepare_session(71, params) == 0
+        assert lib.start_stream(1024, b'', 71, params) == 0
+    finally:
+        lib.release_session(71, params)
+        stop.set()
+        listener.join()
+        for worker in workers:
+            worker.join()
+        server.close()
+    assert not failures, failures
+    print('RDA protocol and session lifecycle passed')
+
+
+if __name__ == '__main__':
+    main(sys.argv[1])

java_package/brainflow/src/main/java/brainflow/BoardIds.java

@@ -74,7 +74,8 @@ public enum BoardIds
     IRONBCI_32_BOARD(65),
     NEUROPAWN_KNIGHT_BOARD_IMU(66),
     MUSE_S_ATHENA_BOARD(67),
-    SHIMMER3_BOARD(68);
+    SHIMMER3_BOARD(68),
+    BRAINVISION_RDA_BOARD(71);
 
     private final int board_id;
     private static final Map<Integer, BoardIds> bi_map = new HashMap<Integer, BoardIds> ();

julia_package/brainflow/src/board_shim.jl

@@ -70,6 +70,7 @@ export BrainFlowInputParams
     NEUROPAWN_KNIGHT_BOARD_IMU = 66
     MUSE_S_ATHENA_BOARD = 67
     SHIMMER3_BOARD = 68
+    BRAINVISION_RDA_BOARD = 71
 
 end
 

matlab_package/brainflow/BoardIds.m

@@ -68,5 +68,6 @@
         NEUROPAWN_KNIGHT_BOARD_IMU(66)
         MUSE_S_ATHENA_BOARD(67)
         SHIMMER3_BOARD(68)
+        BRAINVISION_RDA_BOARD(71)
     end
 end

nodejs_package/brainflow/brainflow.types.ts

@@ -77,7 +77,8 @@ export enum BoardIds {
     IRONBCI_32_BOARD = 65,
     NEUROPAWN_KNIGHT_BOARD_IMU = 66,
     MUSE_S_ATHENA_BOARD = 67,
-    SHIMMER3_BOARD = 68
+    SHIMMER3_BOARD = 68,
+    BRAINVISION_RDA_BOARD = 71
 }
 
 export enum IpProtocolTypes {

python_package/brainflow/board_shim.py

@@ -83,6 +83,7 @@ class BoardIds(enum.IntEnum):
     NEUROPAWN_KNIGHT_BOARD_IMU = 66  #:
     MUSE_S_ATHENA_BOARD = 67  #:
     SHIMMER3_BOARD = 68  #:
+    BRAINVISION_RDA_BOARD = 71  #:
 
 
 class IpProtocolTypes(enum.IntEnum):

rust_package/brainflow/src/ffi/constants.rs

@@ -104,6 +104,7 @@ pub enum BoardIds {
     NeuropawnKnightBoardImu = 66,
     MuseSAthenaBoard = 67,
     Shimmer3Board = 68,
+    BrainvisionRdaBoard = 71,
 }
 #[repr(i32)]
 #[derive(FromPrimitive, ToPrimitive, Debug, Copy, Clone, Hash, PartialEq, Eq)]

src/board_controller/board_controller.cpp

@@ -26,6 +26,7 @@
 #include "brainflow_constants.h"
 #include "brainflow_input_params.h"
 #include "brainflow_version.h"
+#include "brainvision_rda.h"
 #include "callibri_ecg.h"
 #include "callibri_eeg.h"
 #include "callibri_emg.h"
@@ -315,6 +316,9 @@ int prepare_session (int board_id, const char *json_brainflow_input_params)
             board = std::shared_ptr<Board> (
                 new KnightIMU ((int)BoardIds::NEUROPAWN_KNIGHT_BOARD_IMU, params));
             break;
+        case BoardIds::BRAINVISION_RDA_BOARD:
+            board = std::shared_ptr<Board> (new BrainVisionRDA (params));
+            break;
         case BoardIds::SHIMMER3_BOARD:
             board = std::shared_ptr<Board> (new Shimmer3 (params));
             break;

src/board_controller/brainflow_boards.cpp

@@ -1217,6 +1217,18 @@ BrainFlowBoards::BrainFlowBoards()
         {"battery_channel", 16},
         {"other_channels", {17}}
     };
+    brainflow_boards_json["boards"]["71"]["default"] =
+    {
+        {"name", "BrainVisionRDA32"},
+        {"sampling_rate", 1000},
+        {"package_num_channel", 0},
+        {"eeg_channels", {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32}},
+        {"other_channels", {33, 34}},
+        {"timestamp_channel", 35},
+        {"marker_channel", 36},
+        {"num_rows", 37}
+    };
+
 }
 
 BrainFlowBoards boards_struct;