Add EGI NA400 AmpServer network acquisition

.github/workflows/run_unix.yml

@@ -261,6 +261,9 @@ jobs:
     # start testing
     - name: Run unit tests
       run: $GITHUB_WORKSPACE/build/tests/brainflow_tests
+    - name: EGI NA400 emulator
+      if: runner.os == 'Linux'
+      run: python3 $GITHUB_WORKSPACE/emulator/brainflow_emulator/egi_ampserver_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,
+        EGI_AMP_SERVER_BOARD = 72
     };
 
 

docs/SupportedBoards.rst

@@ -1581,3 +1581,56 @@ Common sensor bits:
 - :code:`0x000008`, EXG2 24-bit
 - :code:`0x000004`, GSR, exposed as EDA
 - :code:`0x002000`, battery
+
+EGI NA400 AmpServer (256 channels, 1000 Hz)
+-----------------------------------------
+
+``BoardIds.EGI_AMP_SERVER_BOARD`` (72) supports the NA400 AmpServer Packet
+Format 2 stream at 1000 Hz with a 256-channel net (net codes 2, 6 or 10), or
+the test connector (14). NA300, NA410, NA500, Packet Format 1, smaller nets,
+impedance mode and PIB/physio channels are not supported by this driver.
+
+Set ``ip_address`` to the AmpServer IPv4 address. ``ip_port`` is the command
+port (default 9877), and ``ip_port_aux`` is the data port (default 9879).
+``other_info`` optionally supplies a numeric amplifier ID (default ``"0"``).
+``timeout`` controls first-sample waiting in seconds (default 5); command replies
+use a five-second socket timeout. This NA400 path does not use the notification
+port or depend on NA500's power-on notifications.
+
+Preparation validates the amplifier's reported model, packet format and channel
+count. Starting stops/resets acquisition, configures a 1000 Hz decimated rate,
+powers on, sets the default acquisition state and starts listening. It therefore
+requires control of the amplifier; close other acquisition clients first.
+Stopping sends ``cmd_Stop``. Releasing closes the sockets. It does not manage
+AmpServer installation or silently reconnect after a lost connection.
+
+EEG rows contain microvolts, using the NA400 reference scaling factor
+0.00015522042 per signed 32-bit count. The 256 streamed EEG slots are returned
+without adding a synthetic Cz channel. ``other_channels`` contains the raw
+16-bit DIN word, its active-low-to-active-high inversion, TR, net code, and the
+raw device timestamp in microseconds. Device timestamps may be zero between
+clock reports and are not assumed to be Unix time. The BrainFlow timestamp is
+host receive time, with unique samples within a block spaced at 1000 Hz.
+Duplicate counters are discarded; counter discontinuities are logged.
+``config_board("get_amp_details")`` returns the preparation reply.
+
+.. code-block:: python
+
+    from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds
+
+    params = BrainFlowInputParams()
+    params.ip_address = "192.168.1.10"
+    board = BoardShim(BoardIds.EGI_AMP_SERVER_BOARD, params)
+    board.prepare_session()
+    try:
+        board.start_stream()
+        # Consume get_board_data() while recording.
+        board.stop_stream()
+    finally:
+        board.release_session()
+
+Protocol fixtures and a command/data loopback server test endian conversion,
+scaling, duplicate suppression, fragmented reads, model/net validation, timeout
+recovery and session restart. Physical NA400 validation remains outstanding.
+
+Reference: https://github.com/labstreaminglayer/App-EGIAmpServer

emulator/brainflow_emulator/egi_ampserver_test.py

@@ -0,0 +1,167 @@
+"""NA400 command/data loopback test; pass the native BoardController library path."""
+import ctypes
+import json
+import socket
+import struct
+import sys
+import threading
+import time
+
+
+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]
+    servers = []
+    for _ in range(2):
+        server = socket.socket()
+        server.bind(('127.0.0.1', 0))
+        server.listen()
+        server.settimeout(0.1)
+        servers.append(server)
+    stop = threading.Event()
+    streaming = threading.Event()
+    mode = ['valid']
+    workers = []
+    failures = []
+    commands = []
+    params = json.dumps(dict(serial_port='', mac_address='', ip_address='127.0.0.1',
+        ip_address_aux='', ip_address_anc='', ip_port=servers[0].getsockname()[1],
+        ip_port_aux=servers[1].getsockname()[1], ip_port_anc=0, ip_protocol=0,
+        other_info='', timeout=1, serial_number='', file='', file_aux='', file_anc='',
+        master_board=-100)).encode()
+
+    def command_peer(conn):
+        with conn:
+            pending = b''
+            conn.settimeout(0.1)
+            while not stop.is_set():
+                try:
+                    chunk = conn.recv(1024)
+                except socket.timeout:
+                    continue
+                if not chunk:
+                    return
+                pending += chunk
+                while b'\n' in pending:
+                    line, pending = pending.split(b'\n', 1)
+                    parts = line.decode().strip('()').split()
+                    assert len(parts) == 5 and parts[0] == 'sendCommand' and parts[2:4] == ['0', '0']
+                    name = parts[1]
+                    commands.append((name, parts[4]))
+                    response = '(sendCommand_return (status complete))\n'
+                    if name == 'cmd_GetAmpDetails':
+                        model = 'NA500' if mode[0] == 'model' else 'NA400'
+                        response = ('(sendCommand_return (status complete) (amp_details '
+                            '(amp_type %s) (packet_format 2) (number_of_channels 256)))\n') % model
+                    elif name == 'cmd_Stop':
+                        streaming.clear()
+                    elif name == 'cmd_Start':
+                        streaming.set()
+                    elif name == 'cmd_SetDecimatedRate':
+                        assert parts[4] == '1000'
+                    else:
+                        assert name in ('cmd_SetPower', 'cmd_DefaultAcquisitionState')
+                    for i in range(0, len(response), 5):
+                        conn.sendall(response[i:i + 5].encode())
+
+    def data_peer(conn):
+        with conn:
+            conn.settimeout(0.1)
+            pending = b''
+            while b'\n' not in pending and not stop.is_set():
+                try:
+                    chunk = conn.recv(1024)
+                except socket.timeout:
+                    continue
+                if not chunk:
+                    return
+                pending += chunk
+            assert pending.strip() == b'(sendCommand cmd_ListenToAmp 0 0 0)'
+            seq = 0
+            scenario = mode[0]
+            while not stop.wait(0.005):
+                if not streaming.is_set() or scenario == 'silent':
+                    continue
+                frames = []
+                for counter in [seq, seq, seq + 1]:  # duplicate must be discarded
+                    frame = bytearray(1264)
+                    struct.pack_into('<H', frame, 0, 0xfffe)
+                    struct.pack_into('<Q', frame, 25, counter)
+                    struct.pack_into('<Q', frame, 33, 123000 + counter * 1000)
+                    frame[41] = 5 if scenario == 'net' else 6
+                    struct.pack_into('<256i', frame, 80, *([-1000000] * 256))
+                    frames.append(frame)
+                body = b''.join(frames)
+                packet = struct.pack('>QQ', 2 if scenario == 'id' else 0, len(body)) + body
+                for i in range(0, len(packet), 113):
+                    conn.sendall(packet[i:i + 113])
+                seq += 2
+
+    def guarded(conn, handler):
+        try:
+            handler(conn)
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+        except Exception as exc:
+            failures.append(exc)
+
+    def accept(server, handler):
+        while not stop.is_set():
+            try:
+                conn, _ = server.accept()
+            except socket.timeout:
+                continue
+            worker = threading.Thread(target=guarded, args=(conn, handler))
+            workers.append(worker)
+            worker.start()
+
+    listeners = [threading.Thread(target=accept, args=(server, handler))
+        for server, handler in zip(servers, [command_peer, data_peer])]
+    for listener in listeners:
+        listener.start()
+    try:
+        mode[0] = 'model'
+        assert lib.prepare_session(72, params) == 6
+        mode[0] = 'valid'
+        assert lib.prepare_session(72, params) == 0
+        for _ in range(2):
+            assert lib.start_stream(1024, b'', 72, params) == 0
+            time.sleep(0.05)
+            data = (ctypes.c_double * (264 * 1024))()
+            count = ctypes.c_int()
+            assert lib.get_current_board_data(1024, 0, data, ctypes.byref(count), 72, params) == 0
+            assert count.value >= 2
+            assert all(data[count.value + i] == -1000000 * 0.00015522042 for i in range(count.value))
+            assert all(data[i + 1] - data[i] == 1 for i in range(count.value - 1))
+            assert all(data[258 * count.value + i] == 1 for i in range(count.value))
+            assert lib.stop_stream(72, params) == 0
+        for scenario, expected in [('id', 5), ('net', 5), ('silent', 18)]:
+            mode[0] = scenario
+            assert lib.start_stream(1024, b'', 72, params) == expected, scenario
+        mode[0] = 'valid'
+        assert lib.start_stream(1024, b'', 72, params) == 0
+        assert lib.release_session(72, params) == 0
+        assert lib.prepare_session(72, params) == 0
+        assert lib.start_stream(1024, b'', 72, params) == 0
+    finally:
+        lib.release_session(72, params)
+        stop.set()
+        for listener in listeners:
+            listener.join()
+        for worker in workers:
+            worker.join()
+        for server in servers:
+            server.close()
+    assert not failures, failures
+    assert ('cmd_SetDecimatedRate', '1000') in commands
+    assert commands[-1] == ('cmd_Stop', '0')
+    print('EGI NA400 command/data 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),
+    EGI_AMP_SERVER_BOARD(72);
 
     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
+    EGI_AMP_SERVER_BOARD = 72
 
 end
 

matlab_package/brainflow/BoardIds.m

@@ -68,5 +68,6 @@
         NEUROPAWN_KNIGHT_BOARD_IMU(66)
         MUSE_S_ATHENA_BOARD(67)
         SHIMMER3_BOARD(68)
+        EGI_AMP_SERVER_BOARD(72)
     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,
+    EGI_AMP_SERVER_BOARD = 72
 }
 
 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  #:
+    EGI_AMP_SERVER_BOARD = 72  #:
 
 
 class IpProtocolTypes(enum.IntEnum):

rust_package/brainflow/src/ffi/constants.rs

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

src/board_controller/board_controller.cpp

@@ -33,6 +33,7 @@
 #include "cyton_daisy.h"
 #include "cyton_daisy_wifi.h"
 #include "cyton_wifi.h"
+#include "egi_ampserver.h"
 #include "emotibit.h"
 #include "enophone.h"
 #include "explore.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::EGI_AMP_SERVER_BOARD:
+            board = std::shared_ptr<Board> (new EGIAmpServer (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"]["72"]["default"] =
+    {
+        {"name", "EGINA400"},
+        {"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, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127, 128, 129, 130, 131, 132, 133, 134, 135, 136, 137, 138, 139, 140, 141, 142, 143, 144, 145, 146, 147, 148, 149, 150, 151, 152, 153, 154, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172, 173, 174, 175, 176, 177, 178, 179, 180, 181, 182, 183, 184, 185, 186, 187, 188, 189, 190, 191, 192, 193, 194, 195, 196, 197, 198, 199, 200, 201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215, 216, 217, 218, 219, 220, 221, 222, 223, 224, 225, 226, 227, 228, 229, 230, 231, 232, 233, 234, 235, 236, 237, 238, 239, 240, 241, 242, 243, 244, 245, 246, 247, 248, 249, 250, 251, 252, 253, 254, 255, 256}},
+        {"other_channels", {257, 258, 259, 260, 261}},
+        {"timestamp_channel", 262},
+        {"marker_channel", 263},
+        {"num_rows", 264}
+    };
+
 }
 
 BrainFlowBoards boards_struct;

src/board_controller/build.cmake

@@ -88,6 +88,7 @@ SET (BOARD_CONTROLLER_SRC
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/neuropawn/knight_base.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/neuropawn/knight_imu.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/biolistener/biolistener.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/egi/egi_ampserver.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/shimmer/shimmer3.cpp
 )
 
@@ -157,6 +158,7 @@ target_include_directories (
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/synchroni/inc
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/neuropawn/inc
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/biolistener/inc
+    ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/egi/inc
     ${CMAKE_CURRENT_SOURCE_DIR}/src/board_controller/shimmer/inc
 )