Nm database

.gitignore

@@ -48,6 +48,7 @@ coverage.xml
 *.py,cover
 .hypothesis/
 .pytest_cache/
+/test_data
 
 # Translations
 *.mo

examples/plot_0_first_demo.py

@@ -146,7 +146,7 @@ def generate_random_walk(NUM_CHANNELS, TIME_DATA_SAMPLES):
     line_noise=50,
 )
 
-features = stream.run(data)
+features = stream.run(data, save_csv=True)
 
 # %%
 # Feature Analysis

examples/plot_1_example_BIDS.py

@@ -118,6 +118,7 @@
     data=data,
     out_path_root=PATH_OUT,
     folder_name=RUN_NAME,
+    save_csv = True,
 )
 
 # %%

examples/plot_3_example_sharpwave_analysis.py

@@ -310,7 +310,7 @@
     verbose=True,
 )
 
-df_features = stream.run(data=data[:, :30000])
+df_features = stream.run(data=data[:, :30000], save_csv=True)
 
 # %%
 # We can then plot two exemplary features, prominence and interval, and see that the movement amplitude can be clustered with those two features alone:

examples/plot_4_example_gridPointProjection.py

@@ -87,6 +87,7 @@
     data=data[:, :int(sfreq*5)],
     out_path_root=PATH_OUT,
     folder_name=RUN_NAME,
+    save_csv=True, 
 )
 
 # %%

py_neuromodulation/nm_database.py

@@ -0,0 +1,107 @@
+import os
+import sqlite3
+from pathlib import Path
+import numpy as np
+import pandas as pd
+from py_neuromodulation.nm_types import _PathLike
+
+class NMDatabase:
+    """
+    Class to create a database and insert data into it.
+    Parameters
+    ----------
+    out_path_root : _PathLike
+        The root path to save the database.
+    folder_name : str
+        The folder name to save the database.
+    csv_path : str, optional   
+        The path to save the csv file. If not provided, it will be saved in the same folder as the database.
+    """
+    def __init__(
+            self, 
+            out_path_root: _PathLike,
+            folder_name: str, 
+            csv_path: _PathLike | None = None):
+        self.out_path_root = out_path_root
+        self.folder_name = folder_name
+        self.db_path = Path(out_path_root, folder_name, "stream.db") 
+        if csv_path is None:
+            self.csv_path = Path(out_path_root, folder_name, f"stream.csv")
+        else:
+            self.csv_path = Path(csv_path)
+
+        if os.path.exists(self.db_path):
+            os.remove(self.db_path)
+
+        self.conn = sqlite3.connect(self.db_path, isolation_level=None)
+        self.cursor = self.conn.cursor()
+
+    def infer_type(self, value):
+        """Infer the type of the value to create the table schema.
+        Parameters
+        ----------
+        value : int, float, str
+            The value to infer the type."""
+
+        if isinstance(value, (int, float)):
+            return "REAL"
+        elif isinstance(value, str):
+            return "TEXT"
+        else:
+            return "BLOB"
+
+    def cast_values(self, feature_dict: dict):
+        """Cast the int values of the dictionary to float.
+        Parameters
+        ----------
+        feature_dict : dict
+            The dictionary to cast the values."""
+        for key, value in feature_dict.items():
+            if isinstance(value, (int, float, np.int64)):
+                feature_dict[key] = float(value)
+        return feature_dict
+
+    def create_table(self, feature_dict: dict):
+        """
+        Create a table in the database.
+        Parameters
+        ----------
+        feature_dict : dict
+            The dictionary with the feature names and values.
+        """
+        columns_schema = ", ".join([f'"{column}" {self.infer_type(value)}' for column, value in feature_dict.items()])
+        self.cursor.execute(f"CREATE TABLE IF NOT EXISTS stream_table ({columns_schema})")
+
+    def insert_data(self, feature_dict: dict):
+        """
+        Insert data into the database.  
+        Parameters
+        ----------
+        feature_dict : dict
+            The dictionary with the feature names and values.
+        """
+        columns = ", ".join([f'"{column}"' for column in feature_dict.keys()])
+        placeholders = ", ".join(["?" for _ in feature_dict])
+        insert_sql = f"INSERT INTO stream_table ({columns}) VALUES ({placeholders})"
+        values = tuple(feature_dict.values())
+        self.cursor.execute(insert_sql, values)
+
+    def commit(self):
+        self.conn.commit()
+
+    def fetch_all(self):
+        """"
+        Fetch all the data from the database.
+        Returns
+        -------
+        pd.DataFrame
+            The data in a pandas DataFrame.
+        """
+        return pd.read_sql_query("SELECT * FROM stream_table", self.conn)
+
+    def save_as_csv(self):
+        df = self.fetch_all()
+        df.to_csv(self.csv_path, index=False)
+
+    def close(self):
+        self.conn.close()

py_neuromodulation/nm_stream.py

@@ -5,6 +5,7 @@
 import pandas as pd
 from pathlib import Path
 from py_neuromodulation.nm_stream_abc import NMStream
+from py_neuromodulation.nm_database import NMDatabase
 from py_neuromodulation.nm_types import _PathLike
 from py_neuromodulation import logger
 
@@ -14,7 +15,7 @@
 
 class _GenericStream(NMStream):
     """_GenericStream base class.
-    This class can be inherited for different types of offline streams
+    This class can be inhereted for different types of offline streams
 
     Parameters
     ----------
@@ -84,10 +85,13 @@ def _run(
         is_stream_lsl: bool = True,
         stream_lsl_name: str = None,
         plot_lsl: bool = False,
+        save_csv: bool = False,
+        save_interval: int = 10,
     ) -> pd.DataFrame:
-        from py_neuromodulation.nm_generator import raw_data_generator
+        # from py_neuromodulation.nm_database import NMDatabase
 
         if not is_stream_lsl:
+            from py_neuromodulation.nm_generator import raw_data_generator
             generator = raw_data_generator(
                 data=data,
                 settings=self.settings,
@@ -117,9 +121,11 @@ def _run(
 
             generator = self.lsl_stream.get_next_batch()
 
-        l_features: list[dict] = []
         last_time = None
 
+        buff_cnt: int = 0
+        db = NMDatabase(out_path_root, folder_name)
+        data_acquired = False
         while True:
             next_item = next(generator, None)
 
@@ -134,23 +140,38 @@ def _run(
                 data_batch.astype(np.float64)
             )
             if is_stream_lsl:
-                feature_dict["time"] = time_[-1]
                 if self.verbose:
                     if last_time is not None:
                         logger.debug("%.3f seconds of new data processed", time_[-1] - last_time)
+                        feature_dict["time"] = time_[-1] - last_time
+                    else:
+                        feature_dict["time"] = 0
                     last_time = time_[-1]
             else:
-                feature_dict["time"] = np.ceil(time_[-1] * 1000 +1 ).astype(int)
+                feature_dict["time"] = np.ceil(time_[-1] * 1000 +1 )
                 logger.info("Time: %.2f", feature_dict["time"]/1000)
 
 
             self._add_target(feature_dict, data_batch)
+            buff_cnt += 1
 
-            l_features.append(feature_dict)
+            feature_dict = db.cast_values(feature_dict)
+            if not data_acquired:
+                db.create_table(feature_dict)
+                data_acquired = True
+            db.insert_data(feature_dict)
 
-        feature_df = pd.DataFrame(l_features)
+            if buff_cnt >= save_interval:
+                db.commit()
+                buff_cnt = 0
 
-        self.save_after_stream(out_path_root, folder_name, feature_df)
+        db.commit()
+        feature_df = db.fetch_all()
+
+
+        self.save_after_stream(out_path_root, folder_name, feature_df, save_csv=save_csv)
+
+        db.close()
 
         return feature_df
 
@@ -295,7 +316,9 @@ def run(
         folder_name: str = "sub",
         stream_lsl: bool = False,
         stream_lsl_name: str = None,
+        save_csv: bool = False,
         plot_lsl: bool = False,
+        save_interval: float = 1.0,
     ) -> pd.DataFrame:
         """Call run function for offline stream.
 
@@ -314,6 +337,10 @@ def run(
             stream name, by default None
         plot_lsl : bool, optional
             plot data with mne_lsl stream_viewer
+        save_csv : bool, optional
+            save csv file, by default False
+        save_interval : int, optional
+            save interval in number of samples, by default 10
 
         Returns
         -------
@@ -343,5 +370,7 @@ def run(
             folder_name,
             is_stream_lsl=stream_lsl,
             stream_lsl_name=stream_lsl_name,
-            plot_lsl=plot_lsl,
+            save_csv = save_csv,
+            plot_lsl = plot_lsl,
+            save_interval = save_interval,
         )

py_neuromodulation/nm_stream_abc.py

@@ -142,6 +142,7 @@ def save_after_stream(
         out_path_root: _PathLike = "",
         folder_name: str = "sub",
         feature_arr: pd.DataFrame | None = None,
+        save_csv : bool = False,
     ) -> None:
         """Save features, settings, nm_channels and sidecar after run"""
 
@@ -155,6 +156,9 @@ def save_after_stream(
 
         self.save_sidecar(out_path_root, folder_name)
 
+        if not save_csv:
+            feature_arr = feature_arr.head()
+
         if feature_arr is not None:
             self.save_features(out_path_root, folder_name, feature_arr)
 

tests/test_all_features.py

@@ -27,7 +27,7 @@ def test_all_features_random_array():
     arr = np.random.random([2, 2000])
     stream = get_example_stream(arr)
 
-    df = stream.run(arr)
+    df = stream.run(arr, out_path_root="./test_data", folder_name="test_all_features_random_array")
 
     assert df.shape[0] != 0  # terrible test
 
@@ -38,7 +38,7 @@ def test_all_features_zero_array():
     stream = get_example_stream(arr)
     stream.settings.features.fooof = False # Can't use fooof with zero values (log(0) undefined)
 
-    df = stream.run(arr)
+    df = stream.run(arr, out_path_root="./test_data", folder_name="test_all_features_zero_array")
 
     assert df.shape[0] != 0  # terrible test
 
@@ -50,6 +50,6 @@ def test_all_features_NaN_array():
     stream = get_example_stream(arr)
     stream.settings.features.fooof = False # Can't use fooof nan values
 
-    df = stream.run(arr)
+    df = stream.run(arr, out_path_root="./test_data", folder_name="test_all_features_NaN_array")
 
     assert df.shape[0] != 0  # terrible test

tests/test_bispectra.py

@@ -55,6 +55,6 @@ def test_bispectrum():
         coord_names=coord_names,
     )
 
-    features = stream.run(np.expand_dims(data[3, :], axis=0))
+    features = stream.run(np.expand_dims(data[3, :], axis=0), out_path_root="./test_data", folder_name="test_bispectrum")
 
     assert features["ECOG_RIGHT_1_Bispectrum_phase_mean_whole_fband_range"].sum() != 0

tests/test_feature_sampling_rates.py

@@ -24,7 +24,7 @@ def test_different_sampling_rate_100Hz():
         sfreq=1000, nm_channels=nm_channels, settings=settings, verbose=True
     )
 
-    df = stream.run(arr_test)
+    df = stream.run(arr_test, out_path_root="./test_data", folder_name="test_different_sampling_rate_100Hz")
 
     # check the difference between time points
     #print(df["time"].iloc[:2])
@@ -44,7 +44,7 @@ def test_different_sampling_rate_10Hz():
         sfreq=1000, nm_channels=nm_channels, settings=settings, verbose=True
     )
 
-    df = stream.run(arr_test)
+    df = stream.run(arr_test, out_path_root="./test_data", folder_name="test_different_sampling_rate_10Hz")
 
     # check the difference between time points
 
@@ -62,7 +62,7 @@ def test_different_sampling_rate_1Hz():
         sfreq=1000, nm_channels=nm_channels, settings=settings, verbose=True
     )
 
-    df = stream.run(arr_test)
+    df = stream.run(arr_test, out_path_root="./test_data", folder_name="test_different_sampling_rate_1Hz")
 
     # check the difference between time points
 
@@ -80,7 +80,7 @@ def test_different_sampling_rate_0DOT1Hz():
         sfreq=1000, nm_channels=nm_channels, settings=settings, verbose=True
     )
 
-    df = stream.run(arr_test)
+    df = stream.run(arr_test, out_path_root="./test_data", folder_name="test_different_sampling_rate_0DOT1Hz")
 
     # check the difference between time points
 
@@ -114,7 +114,7 @@ def test_different_segment_lengths():
         sfreq=1000, nm_channels=nm_channels, settings=settings, verbose=True
     )
 
-    df_seglength_800 = stream.run(arr_test)
+    df_seglength_800 = stream.run(arr_test, out_path_root="./test_data", folder_name="test_different_segment_lengths_800")
 
     segment_length_features_ms = 1000
 
@@ -128,7 +128,7 @@ def test_different_segment_lengths():
         sfreq=1000, nm_channels=nm_channels, settings=settings, verbose=True
     )
 
-    df_seglength_1000 = stream.run(arr_test)
+    df_seglength_1000 = stream.run(arr_test, out_path_root="./test_data", folder_name="test_different_segment_lengths_1000")
     # check the difference between time points
 
     print(df_seglength_1000.columns)

tests/test_lsl_stream.py

@@ -25,7 +25,7 @@ def test_offline_lsl(setup_default_stream_fast_compute, setup_lsl_player, setup_
 
     data, stream = setup_default_stream_fast_compute
 
-    features = stream.run(stream_lsl = True, plot_lsl= False, stream_lsl_name='offline_test')
+    features = stream.run(stream_lsl = True, plot_lsl= False, stream_lsl_name='offline_test', out_path_root="./test_data", folder_name="test_offline_lsl")
     # check sfreq
     assert raw.info['sfreq'] == stream.sfreq, "Expected same sampling frequency in the stream and input file"
     assert player.player.info['sfreq'] == stream.sfreq, "Expected same sampling frequency in the stream and player"

tests/test_nan_values.py

@@ -12,7 +12,7 @@ def test_stream_with_none_data():
 
     stream = nm.Stream(fs, data)
 
-    features = stream.run(data)
+    features = stream.run(data, out_path_root="./test_data", folder_name="test_stream_with_none_data")
 
     # assert if all features if name ch0 are None
     assert len(