From 4c68343670af300757c5540890d4cbfabedc5e2d Mon Sep 17 00:00:00 2001 From: Andrey Parfenov Date: Tue, 10 Nov 2026 00:51:42 +0100 Subject: [PATCH 1/2] Add EGI NA400 AmpServer acquisition and protocol emulator tests --- .github/workflows/run_unix.yml | 3 + .../brainflow/board_controller_library.cs | 3 +- docs/SupportedBoards.rst | 53 ++++ .../brainflow_emulator/egi_ampserver_test.py | 167 +++++++++++ .../src/main/java/brainflow/BoardIds.java | 3 +- julia_package/brainflow/src/board_shim.jl | 1 + matlab_package/brainflow/BoardIds.m | 1 + nodejs_package/brainflow/brainflow.types.ts | 3 +- python_package/brainflow/board_shim.py | 1 + rust_package/brainflow/src/ffi/constants.rs | 1 + src/board_controller/board_controller.cpp | 4 + src/board_controller/brainflow_boards.cpp | 12 + src/board_controller/build.cmake | 2 + src/board_controller/egi/egi_ampserver.cpp | 259 ++++++++++++++++++ src/board_controller/egi/inc/egi_ampserver.h | 38 +++ src/board_controller/egi/inc/egi_protocol.h | 90 ++++++ .../egi_ampserver_unittest.cpp | 61 +++++ src/tests/build.cmake | 1 + src/utils/inc/brainflow_constants.h | 3 +- src/utils/inc/socket_client_tcp.h | 2 + src/utils/socket_client_tcp.cpp | 23 ++ .../Sources/BrainFlow/BrainFlowEnums.swift | 1 + 22 files changed, 728 insertions(+), 4 deletions(-) create mode 100644 emulator/brainflow_emulator/egi_ampserver_test.py create mode 100644 src/board_controller/egi/egi_ampserver.cpp create mode 100644 src/board_controller/egi/inc/egi_ampserver.h create mode 100644 src/board_controller/egi/inc/egi_protocol.h create mode 100644 src/tests/board_controller/egi_ampserver_unittest.cpp diff --git a/.github/workflows/run_unix.yml b/.github/workflows/run_unix.yml index db6accc74..bd8a057e6 100644 --- a/.github/workflows/run_unix.yml +++ b/.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: | diff --git a/csharp_package/brainflow/brainflow/board_controller_library.cs b/csharp_package/brainflow/brainflow/board_controller_library.cs index 83d134a4c..6aeccb6e3 100644 --- a/csharp_package/brainflow/brainflow/board_controller_library.cs +++ b/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 }; diff --git a/docs/SupportedBoards.rst b/docs/SupportedBoards.rst index 533d8fda3..c2db5e4f7 100644 --- a/docs/SupportedBoards.rst +++ b/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 diff --git a/emulator/brainflow_emulator/egi_ampserver_test.py b/emulator/brainflow_emulator/egi_ampserver_test.py new file mode 100644 index 000000000..cec9d981e --- /dev/null +++ b/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('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]) diff --git a/java_package/brainflow/src/main/java/brainflow/BoardIds.java b/java_package/brainflow/src/main/java/brainflow/BoardIds.java index e6b24ab11..93b7fac4c 100644 --- a/java_package/brainflow/src/main/java/brainflow/BoardIds.java +++ b/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 bi_map = new HashMap (); diff --git a/julia_package/brainflow/src/board_shim.jl b/julia_package/brainflow/src/board_shim.jl index b1e05f60c..c0d103239 100644 --- a/julia_package/brainflow/src/board_shim.jl +++ b/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 diff --git a/matlab_package/brainflow/BoardIds.m b/matlab_package/brainflow/BoardIds.m index a30390909..e7d78e1fc 100644 --- a/matlab_package/brainflow/BoardIds.m +++ b/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 diff --git a/nodejs_package/brainflow/brainflow.types.ts b/nodejs_package/brainflow/brainflow.types.ts index 5aabeaed5..a48f99dde 100644 --- a/nodejs_package/brainflow/brainflow.types.ts +++ b/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 { diff --git a/python_package/brainflow/board_shim.py b/python_package/brainflow/board_shim.py index e18211334..67924f1f2 100644 --- a/python_package/brainflow/board_shim.py +++ b/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): diff --git a/rust_package/brainflow/src/ffi/constants.rs b/rust_package/brainflow/src/ffi/constants.rs index 37d78ddfa..a2a3b057b 100644 --- a/rust_package/brainflow/src/ffi/constants.rs +++ b/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)] diff --git a/src/board_controller/board_controller.cpp b/src/board_controller/board_controller.cpp index 612965994..8b887c21c 100644 --- a/src/board_controller/board_controller.cpp +++ b/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 ( new KnightIMU ((int)BoardIds::NEUROPAWN_KNIGHT_BOARD_IMU, params)); break; + case BoardIds::EGI_AMP_SERVER_BOARD: + board = std::shared_ptr (new EGIAmpServer (params)); + break; case BoardIds::SHIMMER3_BOARD: board = std::shared_ptr (new Shimmer3 (params)); break; diff --git a/src/board_controller/brainflow_boards.cpp b/src/board_controller/brainflow_boards.cpp index 9493e3f9a..fa2925abe 100644 --- a/src/board_controller/brainflow_boards.cpp +++ b/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; diff --git a/src/board_controller/build.cmake b/src/board_controller/build.cmake index f3fb0d790..f0f41f05e 100644 --- a/src/board_controller/build.cmake +++ b/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 ) diff --git a/src/board_controller/egi/egi_ampserver.cpp b/src/board_controller/egi/egi_ampserver.cpp new file mode 100644 index 000000000..92533b53b --- /dev/null +++ b/src/board_controller/egi/egi_ampserver.cpp @@ -0,0 +1,259 @@ +#include "egi_ampserver.h" +#include "egi_protocol.h" +#include "timestamp.h" + +#include +#include + +EGIAmpServer::EGIAmpServer (struct BrainFlowInputParams params) + : Board ((int)BoardIds::EGI_AMP_SERVER_BOARD, params) + , keep_alive (false) + , stream_error ((int)BrainFlowExitCodes::SYNC_TIMEOUT_ERROR) + , ready (false) + , owns_stream (false) + , amplifier (0) +{ +} + +EGIAmpServer::~EGIAmpServer () +{ + skip_logs = true; + release_session (); +} + +bool EGIAmpServer::send_all (SocketClientTCP &socket, const std::string &message) +{ + size_t offset = 0; + while (offset < message.size ()) + { + int count = socket.send (message.data () + offset, int (message.size () - offset)); + if (count <= 0) + return false; + offset += size_t (count); + } + return true; +} + +bool EGIAmpServer::command ( + const std::string &name, const std::string &value, std::string &response) +{ + response.clear (); + std::string message = + "(sendCommand " + name + " " + std::to_string (amplifier) + " 0 " + value + ")\n"; + if (!send_all (*commands, message)) + return false; + int depth = 0; + auto deadline = std::chrono::steady_clock::now () + std::chrono::seconds (5); + while ((response.size () < 8192) && (std::chrono::steady_clock::now () < deadline)) + { + char byte; + if (commands->recv (&byte, 1) != 1) + return false; + if (response.empty () && std::isspace (static_cast (byte))) + continue; + if (response.empty () && (byte != '(')) + return false; + response += byte; + if (byte == '(') + ++depth; + else if (byte == ')') + --depth; + if (depth == 0) + return response.find ("(sendCommand_return") == 0 && + egi::field (response, "status") == "complete"; + } + return false; +} + +int EGIAmpServer::prepare_session () +{ + if (commands) + return (int)BrainFlowExitCodes::STATUS_OK; + struct in_addr address; + if ((params.ip_address.size () > 31) || + (inet_pton (AF_INET, params.ip_address.c_str (), &address) != 1) || (params.ip_port < 0) || + (params.ip_port > 65535) || (params.ip_port_aux < 0) || (params.ip_port_aux > 65535) || + (params.timeout < 0) || (params.timeout > 600)) + return (int)BrainFlowExitCodes::INVALID_ARGUMENTS_ERROR; + amplifier = 0; + if (!params.other_info.empty ()) + { + // Numeric amplifier ID, not a command string. + if ((params.other_info.size () > 4) || + (params.other_info.find_first_not_of ("0123456789") != std::string::npos)) + return (int)BrainFlowExitCodes::INVALID_ARGUMENTS_ERROR; + amplifier = std::stoi (params.other_info); + } + commands.reset ( + new SocketClientTCP (params.ip_address.c_str (), params.ip_port ? params.ip_port : 9877)); + if (commands->connect () != (int)SocketClientTCPReturnCodes::STATUS_OK) + { + commands.reset (); + return (int)BrainFlowExitCodes::BOARD_NOT_READY_ERROR; + } + if (!command ("cmd_GetAmpDetails", "0", details) || + (egi::field (details, "amp_type") != "NA400") || + (egi::field (details, "packet_format") != "2") || + (egi::field (details, "number_of_channels") != "256")) + { + safe_logger (spdlog::level::err, "EGI requires NA400, Packet Format 2, 256 channels"); + commands.reset (); + return (int)BrainFlowExitCodes::INITIAL_MSG_ERROR; + } + return (int)BrainFlowExitCodes::STATUS_OK; +} + +int EGIAmpServer::start_stream (int buffer_size, const char *streamer_params) +{ + if (!commands) + return (int)BrainFlowExitCodes::BOARD_NOT_READY_ERROR; + if (worker.joinable ()) + return (int)BrainFlowExitCodes::STREAM_ALREADY_RUN_ERROR; + int result = prepare_for_acquisition (buffer_size, streamer_params); + if (result != (int)BrainFlowExitCodes::STATUS_OK) + return result; + data_socket.reset (new SocketClientTCP ( + params.ip_address.c_str (), params.ip_port_aux ? params.ip_port_aux : 9879)); + if (data_socket->connect () != (int)SocketClientTCPReturnCodes::STATUS_OK) + { + data_socket.reset (); + return (int)BrainFlowExitCodes::BOARD_NOT_READY_ERROR; + } + std::string response; + if (!command ("cmd_Stop", "0", response) || !command ("cmd_SetPower", "0", response) || + !command ("cmd_SetDecimatedRate", "1000", response) || + !command ("cmd_SetPower", "1", response) || + !command ("cmd_DefaultAcquisitionState", "0", response) || + !command ("cmd_Start", "0", response)) + { + data_socket.reset (); + return (int)BrainFlowExitCodes::BOARD_WRITE_ERROR; + } + owns_stream = true; + if (!send_all ( + *data_socket, "(sendCommand cmd_ListenToAmp " + std::to_string (amplifier) + " 0 0)\n")) + { + command ("cmd_Stop", "0", response); + owns_stream = false; + data_socket.reset (); + return (int)BrainFlowExitCodes::BOARD_WRITE_ERROR; + } + ready = false; + stream_error = (int)BrainFlowExitCodes::SYNC_TIMEOUT_ERROR; + keep_alive = true; + worker = std::thread (&EGIAmpServer::read_thread, this); + std::unique_lock lock (mutex); + bool success = + received.wait_for (lock, std::chrono::seconds (params.timeout ? params.timeout : 5), + [this] { return ready || !keep_alive; }); + success = success && ready; + lock.unlock (); + if (!success) + { + int error = stream_error; + stop_stream (); + return error; + } + return (int)BrainFlowExitCodes::STATUS_OK; +} + +bool EGIAmpServer::read_exact (void *data, size_t size) +{ + size_t offset = 0; + while (keep_alive && (offset < size)) + { + int count = data_socket->recv (static_cast (data) + offset, int (size - offset)); + if (count <= 0) + return false; + offset += size_t (count); + } + return offset == size; +} + +void EGIAmpServer::read_thread () +{ + bool first = true; + uint64_t previous = 0; + while (keep_alive) + { + uint8_t header[16]; + size_t size; + if (!read_exact (header, sizeof (header)) || + !egi::header (header, uint64_t (amplifier), size)) + break; + std::vector body (size); + if (!read_exact (body.data (), body.size ())) + break; + std::vector samples; + bool valid = true; + for (size_t offset = 0; offset < size; offset += 1264) + { + egi::Sample sample; + if (!egi::decode (body.data () + offset, 1264, sample)) + { + valid = false; + break; + } + if (!first && sample.counter == previous) + continue; + if (!first && sample.counter != previous + 1) + safe_logger (spdlog::level::warn, "EGI sample discontinuity: {} -> {}", previous, + sample.counter); + previous = sample.counter; + first = false; + samples.push_back (sample); + } + if (!valid) + break; + double timestamp = get_timestamp (); + for (size_t i = 0; i < samples.size (); ++i) + { + samples[i].values[262] = timestamp - (samples.size () - 1 - i) / 1000.0; + push_package (samples[i].values.data ()); + } + if (!samples.empty ()) + { + std::lock_guard lock (mutex); + ready = true; + received.notify_all (); + } + } + stream_error = (int)BrainFlowExitCodes::INCOMMING_MSG_ERROR; + keep_alive = false; + received.notify_all (); +} + +int EGIAmpServer::stop_stream () +{ + if (!worker.joinable ()) + return (int)BrainFlowExitCodes::STREAM_THREAD_IS_NOT_RUNNING; + keep_alive = false; + data_socket->shutdown (); + worker.join (); + data_socket.reset (); + std::string response; + bool stopped = !owns_stream || command ("cmd_Stop", "0", response); + owns_stream = false; + return stopped ? (int)BrainFlowExitCodes::STATUS_OK : + (int)BrainFlowExitCodes::BOARD_WRITE_ERROR; +} + +int EGIAmpServer::release_session () +{ + if (worker.joinable ()) + stop_stream (); + data_socket.reset (); + commands.reset (); + free_packages (); + return (int)BrainFlowExitCodes::STATUS_OK; +} + +int EGIAmpServer::config_board (std::string config, std::string &response) +{ + if (!commands) + return (int)BrainFlowExitCodes::BOARD_NOT_READY_ERROR; + if (config != "get_amp_details") + return (int)BrainFlowExitCodes::INVALID_ARGUMENTS_ERROR; + response = details; + return (int)BrainFlowExitCodes::STATUS_OK; +} diff --git a/src/board_controller/egi/inc/egi_ampserver.h b/src/board_controller/egi/inc/egi_ampserver.h new file mode 100644 index 000000000..0e8409258 --- /dev/null +++ b/src/board_controller/egi/inc/egi_ampserver.h @@ -0,0 +1,38 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include "board.h" +#include "socket_client_tcp.h" + +class EGIAmpServer : public Board +{ + std::unique_ptr commands; + std::unique_ptr data_socket; + std::atomic keep_alive; + std::atomic stream_error; + std::thread worker; + std::mutex mutex; + std::condition_variable received; + bool ready; + bool owns_stream; + int amplifier; + std::string details; + bool send_all (SocketClientTCP &socket, const std::string &message); + bool command (const std::string &name, const std::string &value, std::string &response); + bool read_exact (void *data, size_t size); + void read_thread (); + +public: + explicit EGIAmpServer (struct BrainFlowInputParams params); + ~EGIAmpServer () override; + int prepare_session () override; + int start_stream (int buffer_size, const char *streamer_params) override; + int stop_stream () override; + int release_session () override; + int config_board (std::string config, std::string &response) override; +}; diff --git a/src/board_controller/egi/inc/egi_protocol.h b/src/board_controller/egi/inc/egi_protocol.h new file mode 100644 index 000000000..7d72d06c1 --- /dev/null +++ b/src/board_controller/egi/inc/egi_protocol.h @@ -0,0 +1,90 @@ +#pragma once + +#include +#include +#include +#include + +namespace egi +{ + inline uint64_t le64 (const uint8_t *data) + { + uint64_t result = 0; + for (size_t i = 0; i < 8; ++i) + result |= uint64_t (data[i]) << (8 * i); + return result; + } + + inline uint64_t be64 (const uint8_t *data) + { + uint64_t result = 0; + for (size_t i = 0; i < 8; ++i) + result = (result << 8) | data[i]; + return result; + } + + inline bool header (const uint8_t *data, uint64_t amplifier, size_t &length) + { + uint64_t size = be64 (data + 8); + if ((be64 (data) != amplifier) || (size == 0) || (size % 1264 != 0) || (size > 1264 * 4096)) + return false; + length = size_t (size); + return true; + } + + inline std::string field (const std::string &response, const std::string &key) + { + std::string prefix = "(" + key; + size_t offset = response.find (prefix); + if (offset == std::string::npos) + return ""; + offset += prefix.size (); + if ((offset == response.size ()) || + !std::isspace (static_cast (response[offset]))) + return ""; + while ((offset < response.size ()) && + std::isspace (static_cast (response[offset]))) + ++offset; + size_t start = offset; + while ((offset < response.size ()) && (response[offset] != ')') && + !std::isspace (static_cast (response[offset]))) + ++offset; + if ((offset == response.size ()) || (offset == start)) + return ""; + return response.substr (start, offset - start); + } + + struct Sample + { + uint64_t counter; + std::array values; + }; + + inline bool decode (const uint8_t *data, size_t size, Sample &sample) + { + if (size != 1264) + return false; + // Fixed 256-channel profile; test connector is useful for vendor signal tests. + if ((data[41] != 2) && (data[41] != 6) && (data[41] != 10) && (data[41] != 14)) + return false; + sample.values.fill (0); + sample.counter = le64 (data + 25); + sample.values[0] = double (sample.counter); + for (size_t channel = 0; channel < 256; ++channel) + { + const uint8_t *p = data + 80 + 4 * channel; + uint32_t word = uint32_t (p[0]) | (uint32_t (p[1]) << 8) | (uint32_t (p[2]) << 16) | + (uint32_t (p[3]) << 24); + int64_t value = (word & 0x80000000) ? int64_t (word) - 0x100000000LL : int64_t (word); + // NA400 factor from the EGI AmpServer LSL reference; NA500 uses a different gain. + sample.values[channel + 1] = double (value) * 0.00015522042; + } + uint16_t digital = uint16_t (data[0]) | (uint16_t (data[1]) << 8); + sample.values[257] = digital; + sample.values[258] = uint16_t (~digital); + sample.values[259] = data[2]; + sample.values[260] = data[41]; + sample.values[261] = double (le64 (data + 33)); + return true; + } +} diff --git a/src/tests/board_controller/egi_ampserver_unittest.cpp b/src/tests/board_controller/egi_ampserver_unittest.cpp new file mode 100644 index 000000000..a8ad51ce3 --- /dev/null +++ b/src/tests/board_controller/egi_ampserver_unittest.cpp @@ -0,0 +1,61 @@ +#include "egi_protocol.h" +#include +#include + +TEST (EGIAmpServer, ValidatesBigEndianBlockHeader) +{ + uint8_t data[16] = {0}; + data[7] = 2; + data[14] = 4; + data[15] = 0xf0; // 1264 + size_t length = 0; + ASSERT_TRUE (egi::header (data, 2, length)); + EXPECT_EQ (length, 1264u); + EXPECT_FALSE (egi::header (data, 0, length)); + data[15] = 0xf1; + EXPECT_FALSE (egi::header (data, 2, length)); + data[8] = 1; + EXPECT_FALSE (egi::header (data, 2, length)); +} + +TEST (EGIAmpServer, DecodesUnalignedLittleEndianNa400Frame) +{ + std::vector data (1264, 0); + data[0] = 0xfe; + data[1] = 0xff; + data[2] = 7; + data[25] = 0x78; + data[26] = 0x56; + data[27] = 0x34; + data[28] = 0x12; + data[29] = 1; + data[33] = 42; + data[41] = 6; + // Channel 1 = -1000000; channel 256 = 1000000. + const uint8_t negative[] = {0xc0, 0xbd, 0xf0, 0xff}; + const uint8_t positive[] = {0x40, 0x42, 0x0f, 0}; + std::copy (negative, negative + 4, data.begin () + 80); + std::copy (positive, positive + 4, data.begin () + 1100); + egi::Sample sample; + ASSERT_TRUE (egi::decode (data.data (), data.size (), sample)); + EXPECT_EQ (sample.counter, 0x112345678ULL); + EXPECT_DOUBLE_EQ (sample.values[1], -155.22042); + EXPECT_DOUBLE_EQ (sample.values[256], 155.22042); + EXPECT_EQ (sample.values[257], 65534); + EXPECT_EQ (sample.values[258], 1); + EXPECT_EQ (sample.values[259], 7); + EXPECT_EQ (sample.values[260], 6); + EXPECT_EQ (sample.values[261], 42); + EXPECT_FALSE (egi::decode (data.data (), 1263, sample)); + data[41] = 5; // 128-channel net is outside this fixed profile. + EXPECT_FALSE (egi::decode (data.data (), data.size (), sample)); +} + +TEST (EGIAmpServer, MatchesCompleteMetadataTokens) +{ + EXPECT_EQ (egi::field ("(status complete) (amp_type\t NA400)", "amp_type"), "NA400"); + EXPECT_EQ (egi::field ("(amp_type_other NA400)", "amp_type"), ""); + EXPECT_EQ (egi::field ("(amp_type NA400", "amp_type"), ""); + EXPECT_EQ (egi::field ("(amp_type )", "amp_type"), ""); + EXPECT_EQ (egi::field ("(status error)", "status"), "error"); +} diff --git a/src/tests/build.cmake b/src/tests/build.cmake index 37af0e0fe..59d8682f5 100644 --- a/src/tests/build.cmake +++ b/src/tests/build.cmake @@ -42,6 +42,7 @@ SET (TESTS_SRC ${CMAKE_CURRENT_SOURCE_DIR}/src/utils/multicast_server.cpp ${CMAKE_CURRENT_SOURCE_DIR}/src/utils/socket_client_udp.cpp ${CMAKE_CURRENT_SOURCE_DIR}/src/utils/timestamp.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/src/tests/board_controller/egi_ampserver_unittest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/src/tests/board_controller/brainalive_unittest.cpp ${CMAKE_CURRENT_SOURCE_DIR}/src/tests/utils/bluetooth/socket_bluetooth_test.cpp ${CMAKE_CURRENT_SOURCE_DIR}/src/tests/utils/bluetooth/bluetooth_functions_unittest.cpp diff --git a/src/utils/inc/brainflow_constants.h b/src/utils/inc/brainflow_constants.h index 8b31b0df2..b0570aaba 100644 --- a/src/utils/inc/brainflow_constants.h +++ b/src/utils/inc/brainflow_constants.h @@ -97,9 +97,10 @@ enum class BoardIds : int NEUROPAWN_KNIGHT_BOARD_IMU = 66, MUSE_S_ATHENA_BOARD = 67, SHIMMER3_BOARD = 68, + EGI_AMP_SERVER_BOARD = 72, // use it to iterate FIRST = PLAYBACK_FILE_BOARD, - LAST = SHIMMER3_BOARD + LAST = EGI_AMP_SERVER_BOARD }; enum class IpProtocolTypes : int diff --git a/src/utils/inc/socket_client_tcp.h b/src/utils/inc/socket_client_tcp.h index 282c9c93f..8da980535 100644 --- a/src/utils/inc/socket_client_tcp.h +++ b/src/utils/inc/socket_client_tcp.h @@ -35,6 +35,8 @@ class SocketClientTCP int connect (); int send (const char *data, int size); int recv (void *data, int size); + // Interrupt a blocked receive without closing/reusing its descriptor before the reader joins. + void shutdown (); void close (); int get_local_ip_addr (const char *local_ip); char *get_ip_addr () diff --git a/src/utils/socket_client_tcp.cpp b/src/utils/socket_client_tcp.cpp index e74258273..090673a0d 100644 --- a/src/utils/socket_client_tcp.cpp +++ b/src/utils/socket_client_tcp.cpp @@ -2,6 +2,15 @@ #include "socket_client_tcp.h" +void SocketClientTCP::shutdown () +{ +#ifdef _WIN32 + ::shutdown (connect_socket, SD_BOTH); +#else + ::shutdown (connect_socket, SHUT_RDWR); +#endif +} + /////////////////////////////// /////////// WINDOWS /////////// ////////////////////////////// @@ -35,6 +44,11 @@ int SocketClientTCP::connect () wsa_initialized = true; socket_addr.sin_family = AF_INET; socket_addr.sin_port = htons (port); +#ifdef SO_NOSIGPIPE + int suppress_sigpipe = 1; + setsockopt ( + connect_socket, SOL_SOCKET, SO_NOSIGPIPE, &suppress_sigpipe, sizeof (suppress_sigpipe)); +#endif if (inet_pton (AF_INET, ip_addr, &socket_addr.sin_addr) == 0) { return (int)SocketClientTCPReturnCodes::PTON_ERROR; @@ -118,6 +132,11 @@ int SocketClientTCP::connect () socket_addr.sin_family = AF_INET; socket_addr.sin_port = htons (port); +#ifdef SO_NOSIGPIPE + int suppress_sigpipe = 1; + setsockopt ( + connect_socket, SOL_SOCKET, SO_NOSIGPIPE, &suppress_sigpipe, sizeof (suppress_sigpipe)); +#endif if (inet_pton (AF_INET, ip_addr, &socket_addr.sin_addr) == 0) { return (int)SocketClientTCPReturnCodes::PTON_ERROR; @@ -140,7 +159,11 @@ int SocketClientTCP::connect () int SocketClientTCP::send (const char *data, int size) { +#ifdef MSG_NOSIGNAL + int res = ::send (connect_socket, data, size, MSG_NOSIGNAL); +#else int res = ::send (connect_socket, data, size, 0); +#endif return res; } diff --git a/swift_package/Sources/BrainFlow/BrainFlowEnums.swift b/swift_package/Sources/BrainFlow/BrainFlowEnums.swift index 9e98fa9a0..cf8f6f22e 100644 --- a/swift_package/Sources/BrainFlow/BrainFlowEnums.swift +++ b/swift_package/Sources/BrainFlow/BrainFlowEnums.swift @@ -66,6 +66,7 @@ public enum BoardIds: Int, CaseIterable, Sendable { case NEUROPAWN_KNIGHT_BOARD_IMU = 66 case MUSE_S_ATHENA_BOARD = 67 case SHIMMER3_BOARD = 68 + case EGI_AMP_SERVER_BOARD = 72 public var code: Int { rawValue } } From c59fddefd3927f2c253c601813ba514391e2e778 Mon Sep 17 00:00:00 2001 From: Andrey Parfenov Date: Tue, 10 Nov 2026 01:04:58 +0100 Subject: [PATCH 2/2] Make protocol parser header independently includable --- src/board_controller/egi/inc/egi_protocol.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/board_controller/egi/inc/egi_protocol.h b/src/board_controller/egi/inc/egi_protocol.h index 7d72d06c1..0ddfadb4a 100644 --- a/src/board_controller/egi/inc/egi_protocol.h +++ b/src/board_controller/egi/inc/egi_protocol.h @@ -1,5 +1,7 @@ #pragma once +#include + #include #include #include