add delete_streamer method (#560)

Signed-off-by: Andrey Parfenov <a1994ndrey@gmail.com>

cpp_package/examples/get_data/src/multiple_streamers.cpp

@@ -32,6 +32,14 @@ int main (int argc, char *argv[])
             "file://streamer_aux_2.csv:w", (int)BrainFlowPresets::AUXILIARY_PRESET);
         board->start_stream ();
 
+#ifdef _WIN32
+        Sleep (5000);
+#else
+        sleep (5);
+#endif
+        board->delete_streamer (
+            "file://streamer_aux_2.csv:w", (int)BrainFlowPresets::AUXILIARY_PRESET);
+
 #ifdef _WIN32
         Sleep (5000);
 #else

cpp_package/src/board_shim.cpp

@@ -133,6 +133,16 @@ void BoardShim::add_streamer (std::string streamer_params, int preset)
     }
 }
 
+void BoardShim::delete_streamer (std::string streamer_params, int preset)
+{
+    int res =
+        ::delete_streamer (streamer_params.c_str (), preset, board_id, serialized_params.c_str ());
+    if (res != (int)BrainFlowExitCodes::STATUS_OK)
+    {
+        throw BrainFlowException ("failed to delete streamer", res);
+    }
+}
+
 void BoardShim::start_stream (int buffer_size, std::string streamer_params)
 {
     int res = ::start_stream (

cpp_package/src/inc/board_shim.h

@@ -221,6 +221,15 @@ class BoardShim
      */
     void add_streamer (
         std::string streamer_params, int preset = (int)BrainFlowPresets::DEFAULT_PRESET);
+    /**
+     * delete streamer
+     * @param streamer_params use it to pass data packages further or store them directly during
+     streaming, supported values: "file://%file_name%:w", "file://%file_name%:a",
+     "streaming_board://%multicast_group_ip%:%port%"". Range for multicast addresses is from
+     "224.0.0.0" to "239.255.255.255"
+     */
+    void delete_streamer (
+        std::string streamer_params, int preset = (int)BrainFlowPresets::DEFAULT_PRESET);
     /// check if session is ready or not
     bool is_prepared ();
     /// stop streaming thread, doesnt release other resources

csharp_package/brainflow/brainflow/board_controller_library.cs

@@ -194,6 +194,8 @@ public static class BoardControllerLibrary64
         public static extern int get_board_presets (int board_id, int[] names, int[] len);
         [DllImport ("BoardController.dll", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
         public static extern int add_streamer (string streamer, int preset, int board_id, string input_json);
+        [DllImport ("BoardController.dll", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
+        public static extern int delete_streamer (string streamer, int preset, int board_id, string input_json);
     }
 
     public static class BoardControllerLibrary32
@@ -276,6 +278,8 @@ public static class BoardControllerLibrary32
         public static extern int get_board_presets (int board_id, int[] names, int[] len);
         [DllImport ("BoardController32.dll", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
         public static extern int add_streamer (string streamer, int preset, int board_id, string input_json);
+        [DllImport ("BoardController32.dll", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
+        public static extern int delete_streamer (string streamer, int preset, int board_id, string input_json);
     }
 
     public static class BoardControllerLibraryLinux
@@ -358,6 +362,8 @@ public static class BoardControllerLibraryLinux
         public static extern int get_board_presets (int board_id, int[] names, int[] len);
         [DllImport ("libBoardController.so", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
         public static extern int add_streamer (string streamer, int preset, int board_id, string input_json);
+        [DllImport ("libBoardController.so", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
+        public static extern int delete_streamer (string streamer, int preset, int board_id, string input_json);
     }
 
     public static class BoardControllerLibraryMac
@@ -440,6 +446,8 @@ public static class BoardControllerLibraryMac
         public static extern int get_board_presets (int board_id, int[] names, int[] len);
         [DllImport ("libBoardController.dylib", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
         public static extern int add_streamer (string streamer, int preset, int board_id, string input_json);
+        [DllImport ("libBoardController.dylib", SetLastError = true, CallingConvention = CallingConvention.Cdecl)]
+        public static extern int delete_streamer (string streamer, int preset, int board_id, string input_json);
     }
 
     public static class BoardControllerLibrary
@@ -666,6 +674,23 @@ public static int add_streamer (string streamer, int preset, int board_id, strin
             return (int)BrainFlowExitCodes.GENERAL_ERROR;
         }
 
+        public static int delete_streamer (string streamer, int preset, int board_id, string input_json)
+        {
+            switch (PlatformHelper.get_library_environment ())
+            {
+                case LibraryEnvironment.x64:
+                    return BoardControllerLibrary64.delete_streamer (streamer, preset, board_id, input_json);
+                case LibraryEnvironment.x86:
+                    return BoardControllerLibrary32.delete_streamer (streamer, preset, board_id, input_json);
+                case LibraryEnvironment.Linux:
+                    return BoardControllerLibraryLinux.delete_streamer (streamer, preset, board_id, input_json);
+                case LibraryEnvironment.MacOS:
+                    return BoardControllerLibraryMac.delete_streamer (streamer, preset, board_id, input_json);
+            }
+
+            return (int)BrainFlowExitCodes.GENERAL_ERROR;
+        }
+
         public static int set_log_file_board_controller (string log_file)
         {
             switch (PlatformHelper.get_library_environment ())

csharp_package/brainflow/brainflow/board_shim.cs

@@ -686,6 +686,19 @@ public void add_streamer (string streamer_params, int preset = (int)BrainFlowPre
             }
         }
 
+        /// <summary>
+        /// delete streamer
+        /// </summary>
+        /// <param name="streamer_params">supoprted formats file://filename:w file://filename:a streaming_board://ip:port</param>
+        public void delete_streamer (string streamer_params, int preset = (int)BrainFlowPresets.DEFAULT_PRESET)
+        {
+            int res = BoardControllerLibrary.delete_streamer (streamer_params, preset, board_id, input_json);
+            if (res != (int)BrainFlowExitCodes.STATUS_OK)
+            {
+                throw new BrainFlowError (res);
+            }
+        }
+
         /// <summary>
         /// insert marker to data array
         /// </summary>

java_package/brainflow/src/main/java/brainflow/BoardShim.java

@@ -30,6 +30,8 @@ private interface DllInterface extends Library
 
         int add_streamer (String streamer, int preset, int board_id, String params);
 
+        int delete_streamer (String streamer, int preset, int board_id, String params);
+
         int start_stream (int buffer_size, String streamer_params, int board_id, String params);
 
         int stop_stream (int board_id, String params);
@@ -1334,6 +1336,18 @@ public void add_streamer (String streamer, int preset) throws BrainFlowError
         }
     }
 
+    /**
+     * delete streamer
+     */
+    public void delete_streamer (String streamer, int preset) throws BrainFlowError
+    {
+        int ec = instance.delete_streamer (streamer, preset, board_id, input_json);
+        if (ec != BrainFlowExitCode.STATUS_OK.get_code ())
+        {
+            throw new BrainFlowError ("Error in delete_streamer", ec);
+        }
+    }
+
     public void add_streamer (String streamer, BrainFlowPresets preset) throws BrainFlowError
     {
         add_streamer (streamer, preset.get_code ());
@@ -1344,6 +1358,16 @@ public void add_streamer (String streamer) throws BrainFlowError
         add_streamer (streamer, BrainFlowPresets.DEFAULT_PRESET);
     }
 
+    public void delete_streamer (String streamer, BrainFlowPresets preset) throws BrainFlowError
+    {
+        delete_streamer (streamer, preset.get_code ());
+    }
+
+    public void delete_streamer (String streamer) throws BrainFlowError
+    {
+        delete_streamer (streamer, BrainFlowPresets.DEFAULT_PRESET);
+    }
+
     /**
      * send string to a board, use this method carefully and only if you understand
      * what you are doing

julia_package/brainflow/src/board_shim.jl

@@ -260,6 +260,11 @@ end
             streamer_params, Int32(preset), board_shim.board_id, board_shim.input_json)
 end
 
+@brainflow_rethrow function delete_streamer(streamer_params::String, board_shim::BoardShim, preset::PresetType=Integer(DEFAULT_PRESET))
+    ccall((:delete_streamer, BOARD_CONTROLLER_INTERFACE), Cint, (Ptr{UInt8}, Cint, Cint, Ptr{UInt8}),
+            streamer_params, Int32(preset), board_shim.board_id, board_shim.input_json)
+end
+
 @brainflow_rethrow function config_board(config::String, board_shim::BoardShim)
     resp_string = Vector{Cuchar}(undef, 4096)
     len = Vector{Cint}(undef, 1)

matlab_package/brainflow/BoardShim.m

@@ -373,6 +373,16 @@ function add_streamer(obj, streamer, preset)
             [exit_code, tmp] = calllib(lib_name, task_name, streamer, preset, obj.board_id, obj.input_params_json);
             BoardShim.check_ec(exit_code, task_name);
         end
+
+        function delete_streamer(obj, streamer, preset)
+            % delete streamer
+            task_name = 'delete_streamer';
+            lib_name = BoardShim.load_lib();
+            % no way to understand how it works in matlab used this link
+            % https://nl.mathworks.com/matlabcentral/answers/131446-what-data-type-do-i-need-to-calllib-with-pointer-argument-char%
+            [exit_code, tmp] = calllib(lib_name, task_name, streamer, preset, obj.board_id, obj.input_params_json);
+            BoardShim.check_ec(exit_code, task_name);
+        end
 
         function start_stream(obj, buffer_size, streamer_params)
             % start data acqusition

python_package/brainflow/board_shim.py

@@ -195,6 +195,15 @@ def __init__(self):
             ctypes.c_char_p
         ]
 
+        self.delete_streamer = self.lib.delete_streamer
+        self.delete_streamer.restype = ctypes.c_int
+        self.delete_streamer.argtypes = [
+            ctypes.c_char_p,
+            ctypes.c_int,
+            ctypes.c_int,
+            ctypes.c_char_p
+        ]
+
         self.stop_stream = self.lib.stop_stream
         self.stop_stream.restype = ctypes.c_int
         self.stop_stream.argtypes = [
@@ -1105,6 +1114,27 @@ def add_streamer(self, streamer_params: str, preset: int = BrainFlowPresets.DEFA
         if res != BrainFlowExitCodes.STATUS_OK.value:
             raise BrainFlowError('unable to add streamer', res)
 
+    def delete_streamer(self, streamer_params: str, preset: int = BrainFlowPresets.DEFAULT_PRESET) -> None:
+        """Delete streamer
+
+        :param preset: preset
+        :type preset: int
+        :param streamer_params parameter to stream data from brainflow, supported vals: "file://%file_name%:w", "file://%file_name%:a", "streaming_board://%multicast_group_ip%:%port%". Range for multicast addresses is from "224.0.0.0" to "239.255.255.255"
+        :type streamer_params: str
+        """
+
+        if streamer_params is None:
+            streamer = None
+        else:
+            try:
+                streamer = streamer_params.encode()
+            except BaseException:
+                streamer = streamer_params
+
+        res = BoardControllerDLL.get_instance().delete_streamer(streamer, preset, self.board_id, self.input_json)
+        if res != BrainFlowExitCodes.STATUS_OK.value:
+            raise BrainFlowError('unable to delete streamer', res)
+
     def start_stream(self, num_samples: int = 1800 * 250, streamer_params: str = None) -> None:
         """Start streaming data, this methods stores data in ringbuffer
 

python_package/examples/tests/multiple_streamers.py

@@ -0,0 +1,35 @@
+import time
+
+from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds, BrainFlowPresets
+from brainflow.data_filter import DataFilter
+
+
+def main():
+    BoardShim.enable_dev_board_logger()
+
+    # use synthetic board for demo
+    params = BrainFlowInputParams()
+    board_id = BoardIds.SYNTHETIC_BOARD.value
+
+    presets = BoardShim.get_board_presets(board_id)
+    print (presets)
+
+    board = BoardShim(board_id, params)
+    board.prepare_session()
+    board.add_streamer('file://streamer_default.csv:w')
+    board.add_streamer('file://streamer_default2.csv:w')
+    board.start_stream()
+    board.add_streamer('file://streamer_aux.csv:w', BrainFlowPresets.AUXILIARY_PRESET)
+    time.sleep(5)
+    board.delete_streamer('file://streamer_default2.csv:w')
+    time.sleep(5)
+    data_default = board.get_board_data(preset=BrainFlowPresets.DEFAULT_PRESET)
+    data_aux = board.get_board_data(preset=BrainFlowPresets.AUXILIARY_PRESET)
+    board.stop_stream()
+    board.release_session()
+    DataFilter.write_file(data_default, 'default.csv', 'w')
+    DataFilter.write_file(data_aux, 'aux.csv', 'w')
+
+
+if __name__ == "__main__":
+    main()

rust_package/brainflow/src/board_shim.rs

@@ -90,7 +90,7 @@ impl BoardShim {
         Ok(check_brainflow_exit_code(res)?)
     }
 
-    /// Start streaming data, this methods stores data in ringbuffer.
+    /// Add streamer from BrainFlow to file or streaming board.
     pub fn add_streamer<S: AsRef<str>>(
         &self,
         streamer_params: S,
@@ -108,6 +108,24 @@ impl BoardShim {
         Ok(check_brainflow_exit_code(res)?)
     }
 
+    /// Delete streamer registered streamer.
+    pub fn delete_streamer<S: AsRef<str>>(
+        &self,
+        streamer_params: S,
+        preset: BrainFlowPresets,
+    ) -> Result<()> {
+        let streamer_params = CString::new(streamer_params.as_ref())?;
+        let res = unsafe {
+            board_controller::delete_streamer(
+                streamer_params.as_ptr(),
+                preset as c_int,
+                self.board_id as c_int,
+                self.json_brainflow_input_params.as_ptr(),
+            )
+        };
+        Ok(check_brainflow_exit_code(res)?)
+    }
+
     /// Stop streaming data.
     pub fn stop_stream(&self) -> Result<()> {
         let res = unsafe {

rust_package/brainflow/src/ffi/board_controller.rs

@@ -265,6 +265,14 @@ extern "C" {
         json_brainflow_input_params: *const ::std::os::raw::c_char,
     ) -> ::std::os::raw::c_int;
 }
+extern "C" {
+    pub fn delete_streamer(
+        streamer: *const ::std::os::raw::c_char,
+        preset: ::std::os::raw::c_int,
+        board_id: ::std::os::raw::c_int,
+        json_brainflow_input_params: *const ::std::os::raw::c_char,
+    ) -> ::std::os::raw::c_int;
+}
 extern "C" {
     pub fn release_all_sessions() -> ::std::os::raw::c_int;
 }