Repository: catalystneuro/nwb-conversion-tools (language: python)

Issue:
[Feature]: neuralynx session_start_time
### What would you like to see added to nwb-conversion-tools?

pull session_start_time from neuralynx file header

### Is your feature request related to a problem?

_No response_

### What solution would you like?

pull session_start_time from neuralynx file header

### Do you have any interest in helping implement the feature?

No.

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/catalystneuro/nwb-conversion-tools/blob/main/.github/CODE_OF_CONDUCT.rst)
- [X] Have you ensured this bug was not already [reported](https://github.com/catalystneuro/nwb-conversion-tools/issues)?


Discussion / hints:
Neuralynx has a lot of versions: https://gin.g-node.org/NeuralEnsemble/ephy_testing_data/src/master/neuralynx

Do all of them support this, and is access to the header easy via neo or SI helpers?

Fix patch:
diff --git a/.github/workflows/update-caches.yml b/.github/workflows/update-caches.yml
index f9bee0c6e..8e907d898 100644
--- a/.github/workflows/update-caches.yml
+++ b/.github/workflows/update-caches.yml
@@ -48,13 +48,15 @@ jobs:
         id: cache-ecephys-datasets
         with:
           path: ./ephy_testing_data
-          key: ecephys-datasets-040122-${{ matrix.os }}-${{ steps.ecephys.outputs.HASH_EPHY_DATASET }}
+          key: ecephys-datasets-040422-${{ matrix.os }}-${{ steps.ecephys.outputs.HASH_EPHY_DATASET }}
       - name: "Force GIN: ecephys download"
         if: steps.cache-ecephys-datasets.outputs.cache-hit == false
         run: |
           datalad install https://gin.g-node.org/NeuralEnsemble/ephy_testing_data
           cd ephy_testing_data
           datalad get -r ./neuralynx/Cheetah_v5.7.4/original_data/
+          datalad get -r ./neuralynx/Cheetah_v5.6.3/original_data/
+          datalad get -r ./neuralynx/Cheetah_v5.4.0/original_data/
           datalad get -r ./neuroscope/test1/
           datalad get -r ./openephysbinary/v0.4.4.1_with_video_tracking/
           datalad get -r ./blackrock/
diff --git a/nwb_conversion_tools/datainterfaces/ecephys/neuralynx/neuralynxdatainterface.py b/nwb_conversion_tools/datainterfaces/ecephys/neuralynx/neuralynxdatainterface.py
index c5a92e923..27391223c 100644
--- a/nwb_conversion_tools/datainterfaces/ecephys/neuralynx/neuralynxdatainterface.py
+++ b/nwb_conversion_tools/datainterfaces/ecephys/neuralynx/neuralynxdatainterface.py
@@ -1,11 +1,80 @@
-"""Authors: Cody Baker."""
+"""Authors: Cody Baker and Ben Dichter."""
+import warnings
 from pathlib import Path
 from natsort import natsorted
+from dateutil import parser
+import json
 
 from spikeextractors import MultiRecordingChannelExtractor, NeuralynxRecordingExtractor
 
 from ..baserecordingextractorinterface import BaseRecordingExtractorInterface
 from ....utils import FolderPathType
+from ....utils.json_schema import dict_deep_update
+
+
+def parse_header(header):
+    header_dict = dict()
+    for line in header.split("\n")[1:]:
+        if line:
+            if line[0] == "-":
+                key, val = line[1:].split(" ", 1)
+                header_dict[key] = val
+    return header_dict
+
+
+def get_metadata(folder_path: FolderPathType) -> dict:
+    """
+    Parse the header of one of the .ncs files to get the session start time (without
+    timezone) and the session_id.
+
+    Parameters
+    ----------
+    folder_path: str or Path
+
+    Returns
+    -------
+    dict
+
+    """
+    folder_path = Path(folder_path)
+    csc_files = sorted(folder_path.glob("*.[nN]cs"))
+    file_path = csc_files[0]
+    with file_path.open(encoding="latin1") as file:
+        raw_header = file.read(1024)
+    header = parse_header(raw_header)
+    if header.get("FileVersion") == "3.4":
+        return dict(
+            session_start_time=parser.parse(header["TimeCreated"]),
+            session_id=header["SessionUUID"],
+        )
+    if header.get("FileVersion", "").startswith("3.3") or header["CheetahRev"].startswith("5.4"):
+        open_line = raw_header.split("\n")[2]
+        spliced_line = open_line[24:35] + open_line[-13:]
+        return dict(session_start_time=parser.parse(spliced_line, dayfirst=False))
+
+
+def get_filtering(channel_path: FolderPathType) -> str:
+    """Get the filtering metadata from an .nsc file.
+
+    Parameters
+    ----------
+    channel_path: str or Path
+        Filepath for an .nsc file
+
+    Returns
+    -------
+    str:
+        json dump of filter parameters. Uses the mu character, which may cause problems
+        for downstream things that expect ASCII.
+    """
+    channel_path = Path(channel_path)
+    with open(channel_path, "r", encoding="latin1") as file:
+        raw_header = file.read(1024)
+    header = parse_header(raw_header)
+
+    return json.dumps(
+        {key: val.strip(" ") for key, val in header.items() if key.lower().startswith("dsp")}, ensure_ascii=False
+    )
 
 
 class NeuralynxRecordingInterface(BaseRecordingExtractorInterface):
@@ -16,10 +85,23 @@ class NeuralynxRecordingInterface(BaseRecordingExtractorInterface):
     def __init__(self, folder_path: FolderPathType):
         self.subset_channels = None
         self.source_data = dict(folder_path=folder_path)
-        neuralynx_files = natsorted([str(x) for x in Path(folder_path).iterdir() if ".ncs" in x.suffixes])
-        extractors = [NeuralynxRecordingExtractor(filename=filename, seg_index=0) for filename in neuralynx_files]
+        nsc_files = natsorted([str(x) for x in Path(folder_path).iterdir() if ".ncs" in x.suffixes])
+        extractors = [NeuralynxRecordingExtractor(filename=filename, seg_index=0) for filename in nsc_files]
         gains = [extractor.get_channel_gains()[0] for extractor in extractors]
         for extractor in extractors:
             extractor.clear_channel_gains()
         self.recording_extractor = self.RX(extractors)
         self.recording_extractor.set_channel_gains(gains=gains)
+        try:
+            for i, filename in enumerate(nsc_files):
+                self.recording_extractor.set_channel_property(
+                    i,
+                    "filtering",
+                    get_filtering(filename),
+                )
+        except:
+            warnings.warn("filtering could not be extracted.")
+
+    def get_metadata(self):
+        new_metadata = dict(NWBFile=get_metadata(self.source_data["folder_path"]))
+        return dict_deep_update(super().get_metadata(), new_metadata)
diff --git a/setup.py b/setup.py
index 7fb54a132..e79cefe6b 100644
--- a/setup.py
+++ b/setup.py
@@ -1,7 +1,6 @@
 from setuptools import setup, find_packages
 from codecs import open
 import os
-from pathlib import Path
 from shutil import copy
 
 path = os.path.abspath(os.path.dirname(__file__))


Test patch:
diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml
index 66712641a..0b20cbeaf 100644
--- a/.github/workflows/testing.yml
+++ b/.github/workflows/testing.yml
@@ -54,13 +54,15 @@ jobs:
         id: cache-ecephys-datasets
         with:
           path: ./ephy_testing_data
-          key: ecephys-datasets-040122-${{ matrix.os }}-${{ steps.ecephys.outputs.HASH_EPHY_DATASET }}
+          key: ecephys-datasets-040422-${{ matrix.os }}-${{ steps.ecephys.outputs.HASH_EPHY_DATASET }}
       - name: "Force GIN: ecephys download"
         if: steps.cache-ecephys-datasets.outputs.cache-hit == false
         run: |
           datalad install https://gin.g-node.org/NeuralEnsemble/ephy_testing_data
           cd ephy_testing_data
           datalad get -r ./neuralynx/Cheetah_v5.7.4/original_data/
+          datalad get -r ./neuralynx/Cheetah_v5.6.3/original_data/
+          datalad get -r ./neuralynx/Cheetah_v5.4.0/original_data/
           datalad get -r ./neuroscope/test1/
           datalad get -r ./openephysbinary/v0.4.4.1_with_video_tracking/
           datalad get -r ./blackrock/
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_internals/__init__.py b/tests/test_internals/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_on_data/__init__.py b/tests/test_on_data/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_on_data/setup_paths.py b/tests/test_on_data/setup_paths.py
new file mode 100644
index 000000000..2830dbbd8
--- /dev/null
+++ b/tests/test_on_data/setup_paths.py
@@ -0,0 +1,39 @@
+from pathlib import Path
+import tempfile
+import os
+
+import pytest
+
+from nwb_conversion_tools.utils import load_dict_from_file
+
+
+# Load the configuration for the data tests
+file_path = Path(__file__).parent.parent.parent / "tests" / "test_on_data" / "gin_test_config.json"
+test_config_dict = load_dict_from_file(file_path)
+
+#  GIN dataset: https://gin.g-node.org/CatalystNeuro/behavior_testing_data
+if os.getenv("CI"):
+    LOCAL_PATH = Path(".")  # Must be set to "." for CI
+    print("Running GIN tests on Github CI!")
+else:
+    # Override LOCAL_PATH in the `gin_test_config.json` file to a point on your system that contains the dataset folder
+    # Use DANDIHub at hub.dandiarchive.org for open, free use of data found in the /shared/catalystneuro/ directory
+    LOCAL_PATH = Path(test_config_dict["LOCAL_PATH"])
+    print("Running GIN tests locally!")
+BEHAVIOR_DATA_PATH = LOCAL_PATH / "behavior_testing_data"
+if not BEHAVIOR_DATA_PATH.exists():
+    pytest.fail(f"No folder found in location: {BEHAVIOR_DATA_PATH}!")
+
+ECEPHY_DATA_PATH = LOCAL_PATH / "ephy_testing_data"
+if not ECEPHY_DATA_PATH.exists():
+    pytest.fail(f"No folder found in location: {ECEPHY_DATA_PATH}!")
+
+OPHYS_DATA_PATH = LOCAL_PATH / "ophys_testing_data"
+if not OPHYS_DATA_PATH.exists():
+    pytest.fail(f"No folder found in location: {OPHYS_DATA_PATH}!")
+
+if test_config_dict["SAVE_OUTPUTS"]:
+    OUTPUT_PATH = LOCAL_PATH / "example_nwb_output"
+    OUTPUT_PATH.mkdir(exist_ok=True)
+else:
+    OUTPUT_PATH = Path(tempfile.mkdtemp())
diff --git a/tests/test_on_data/test_YAML_conversion_specification.py b/tests/test_on_data/test_YAML_conversion_specification.py
index f4f283e94..c315bd137 100644
--- a/tests/test_on_data/test_YAML_conversion_specification.py
+++ b/tests/test_on_data/test_YAML_conversion_specification.py
@@ -1,42 +1,16 @@
-import os
 import sys
 from pathlib import Path
-from tempfile import mkdtemp
 from jsonschema import validate, RefResolver
 from datetime import datetime
 
 from hdmf.testing import TestCase
-import pytest
 from pynwb import NWBHDF5IO
 
 from nwb_conversion_tools import run_conversion_from_yaml
 from nwb_conversion_tools.utils import load_dict_from_file
 
-# Load the configuration for the data tests
-test_config_dict = load_dict_from_file(Path(__file__).parent / "gin_test_config.json")
-
-# GIN dataset: https://gin.g-node.org/NeuralEnsemble/ephy_testing_data
-if os.getenv("CI"):
-    LOCAL_PATH = Path(".")  # Must be set to "." for CI
-    print("Running GIN tests on Github CI!")
-else:
-    # Override LOCAL_PATH in the `gin_test_config.json` file to a point on your system that contains the dataset folder
-    # Use DANDIHub at hub.dandiarchive.org for open, free use of data found in the /shared/catalystneuro/ directory
-    LOCAL_PATH = Path(test_config_dict["LOCAL_PATH"])
-    print("Running GIN tests locally!")
-DATA_PATH = LOCAL_PATH / "ephy_testing_data"
-HAVE_DATA = DATA_PATH.exists()
-
-if test_config_dict["SAVE_OUTPUTS"]:
-    OUTPUT_PATH = LOCAL_PATH / "example_yaml_output"
-    OUTPUT_PATH.mkdir(exist_ok=True)
-else:
-    OUTPUT_PATH = Path(mkdtemp())
-DATA_PATH = LOCAL_PATH / "ephy_testing_data"
-HAVE_DATA = DATA_PATH.exists()
-
-if not HAVE_DATA:
-    pytest.fail(f"No ephy_testing_data folder found in location: {DATA_PATH}!")
+from .setup_paths import ECEPHY_DATA_PATH as DATA_PATH
+from .setup_paths import OUTPUT_PATH
 
 
 class TestYAMLConversionSpecification(TestCase):
diff --git a/tests/test_on_data/test_gin_behavior.py b/tests/test_on_data/test_gin_behavior.py
index ad55219c0..c5bb9a2e0 100644
--- a/tests/test_on_data/test_gin_behavior.py
+++ b/tests/test_on_data/test_gin_behavior.py
@@ -1,39 +1,12 @@
-import tempfile
 import unittest
 import os
-from pathlib import Path
 from datetime import datetime
 
-import pytest
 import numpy as np
 from pynwb import NWBHDF5IO
 from nwb_conversion_tools import NWBConverter, MovieInterface
-from nwb_conversion_tools.utils import load_dict_from_file
-
-
-# Load the configuration for the data tests
-test_config_dict = load_dict_from_file(Path(__file__).parent / "gin_test_config.json")
-
-#  GIN dataset: https://gin.g-node.org/CatalystNeuro/behavior_testing_data
-if os.getenv("CI"):
-    LOCAL_PATH = Path(".")  # Must be set to "." for CI
-    print("Running GIN tests on Github CI!")
-else:
-    # Override LOCAL_PATH in the `gin_test_config.json` file to a point on your system that contains the dataset folder
-    # Use DANDIHub at hub.dandiarchive.org for open, free use of data found in the /shared/catalystneuro/ directory
-    LOCAL_PATH = Path(test_config_dict["LOCAL_PATH"])
-    print("Running GIN tests locally!")
-BEHAVIOR_DATA_PATH = LOCAL_PATH / "behavior_testing_data"
-HAVE_BEHAVIOR_DATA = BEHAVIOR_DATA_PATH.exists()
-
-if test_config_dict["SAVE_OUTPUTS"]:
-    OUTPUT_PATH = LOCAL_PATH / "example_nwb_output"
-    OUTPUT_PATH.mkdir(exist_ok=True)
-else:
-    OUTPUT_PATH = Path(tempfile.mkdtemp())
-
-if not HAVE_BEHAVIOR_DATA:
-    pytest.fail(f"No oephys_testing_data folder found in location: {BEHAVIOR_DATA_PATH}!")
+
+from .setup_paths import OUTPUT_PATH, BEHAVIOR_DATA_PATH
 
 
 class TestMovieDataNwbConversions(unittest.TestCase):
diff --git a/tests/test_on_data/test_gin_ecephys.py b/tests/test_on_data/test_gin_ecephys.py
index 6096bdd2a..e59e940e9 100644
--- a/tests/test_on_data/test_gin_ecephys.py
+++ b/tests/test_on_data/test_gin_ecephys.py
@@ -1,13 +1,10 @@
-import tempfile
 import unittest
-import os
 from pathlib import Path
 from datetime import datetime
 
 import numpy as np
 import numpy.testing as npt
 
-import pytest
 from parameterized import parameterized, param
 
 from spikeextractors import NwbRecordingExtractor, NwbSortingExtractor, RecordingExtractor
@@ -15,7 +12,6 @@
 from spikeextractors.testing import check_recordings_equal, check_sortings_equal
 from spikeinterface.core.testing import check_recordings_equal as check_recordings_equal_si
 
-
 from pynwb import NWBHDF5IO
 
 from nwb_conversion_tools import (
@@ -34,32 +30,9 @@
     AxonaRecordingExtractorInterface,
     AxonaLFPDataInterface,
 )
-from nwb_conversion_tools.utils import load_dict_from_file
-
-
-# Load the configuration for the data tests
-test_config_dict = load_dict_from_file(Path(__file__).parent / "gin_test_config.json")
-
-# GIN dataset: https://gin.g-node.org/NeuralEnsemble/ephy_testing_data
-if os.getenv("CI"):
-    LOCAL_PATH = Path(".")  # Must be set to "." for CI
-    print("Running GIN tests on Github CI!")
-else:
-    # Override LOCAL_PATH in the `gin_test_config.json` file to a point on your system that contains the dataset folder
-    # Use DANDIHub at hub.dandiarchive.org for open, free use of data found in the /shared/catalystneuro/ directory
-    LOCAL_PATH = Path(test_config_dict["LOCAL_PATH"])
-    print("Running GIN tests locally!")
-DATA_PATH = LOCAL_PATH / "ephy_testing_data"
-HAVE_DATA = DATA_PATH.exists()
-
-if test_config_dict["SAVE_OUTPUTS"]:
-    OUTPUT_PATH = LOCAL_PATH / "example_nwb_output"
-    OUTPUT_PATH.mkdir(exist_ok=True)
-else:
-    OUTPUT_PATH = Path(tempfile.mkdtemp())
-
-if not HAVE_DATA:
-    pytest.fail(f"No ephy_testing_data folder found in location: {DATA_PATH}!")
+
+from .setup_paths import ECEPHY_DATA_PATH as DATA_PATH
+from .setup_paths import OUTPUT_PATH
 
 
 def custom_name_func(testcase_func, param_num, param):
diff --git a/tests/test_on_data/test_gin_ophys.py b/tests/test_on_data/test_gin_ophys.py
index 635dce75c..76c5df058 100644
--- a/tests/test_on_data/test_gin_ophys.py
+++ b/tests/test_on_data/test_gin_ophys.py
@@ -1,10 +1,6 @@
-import tempfile
 import unittest
-import os
-from pathlib import Path
 from datetime import datetime
 
-import pytest
 from parameterized import parameterized, param
 from roiextractors import NwbImagingExtractor, NwbSegmentationExtractor
 from roiextractors.testing import check_imaging_equal, check_segmentations_equal
@@ -19,32 +15,8 @@
     ExtractSegmentationInterface,
     Suite2pSegmentationInterface,
 )
-from nwb_conversion_tools.utils import load_dict_from_file
-
-
-# Load the configuration for the data tests
-test_config_dict = load_dict_from_file(Path(__file__).parent / "gin_test_config.json")
-
-#  GIN dataset: https://gin.g-node.org/CatalystNeuro/ophys_testing_data
-if os.getenv("CI"):
-    LOCAL_PATH = Path(".")  # Must be set to "." for CI
-    print("Running GIN tests on Github CI!")
-else:
-    # Override LOCAL_PATH in the `gin_test_config.json` file to a point on your system that contains the dataset folder
-    # Use DANDIHub at hub.dandiarchive.org for open, free use of data found in the /shared/catalystneuro/ directory
-    LOCAL_PATH = Path(test_config_dict["LOCAL_PATH"])
-    print("Running GIN tests locally!")
-OPHYS_DATA_PATH = LOCAL_PATH / "ophys_testing_data"
-HAVE_OPHYS_DATA = OPHYS_DATA_PATH.exists()
-
-if test_config_dict["SAVE_OUTPUTS"]:
-    OUTPUT_PATH = LOCAL_PATH / "example_nwb_output"
-    OUTPUT_PATH.mkdir(exist_ok=True)
-else:
-    OUTPUT_PATH = Path(tempfile.mkdtemp())
-
-if not OPHYS_DATA_PATH:
-    pytest.fail(f"No oephys_testing_data folder found in location: {OPHYS_DATA_PATH}!")
+
+from .setup_paths import OPHYS_DATA_PATH, OUTPUT_PATH
 
 
 def custom_name_func(testcase_func, param_num, param):
diff --git a/tests/test_on_data/test_metadata/__init__.py b/tests/test_on_data/test_metadata/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/test_on_data/test_metadata/test_neuralynx_metadata.py b/tests/test_on_data/test_metadata/test_neuralynx_metadata.py
new file mode 100644
index 000000000..1ec2b5032
--- /dev/null
+++ b/tests/test_on_data/test_metadata/test_neuralynx_metadata.py
@@ -0,0 +1,67 @@
+from pathlib import Path
+import datetime
+
+from nwb_conversion_tools.datainterfaces.ecephys.neuralynx.neuralynxdatainterface import (
+    get_metadata,
+    get_filtering,
+    NeuralynxRecordingInterface,
+)
+
+from ..setup_paths import ECEPHY_DATA_PATH
+
+NLX_PATH = ECEPHY_DATA_PATH / "neuralynx"
+
+
+def test_neuralynx_cheetah_v574_metadata():
+    folder_path = NLX_PATH / "Cheetah_v5.7.4" / "original_data"
+    assert get_metadata(str(folder_path)) == dict(
+        session_start_time=datetime.datetime(2017, 2, 16, 17, 56, 4), session_id="d8ba8eef-8d11-4cdc-86dc-05f50d4ba13d"
+    )
+
+
+def test_neuralynx_cheetah_v563_metadata():
+    folder_path = NLX_PATH / "Cheetah_v5.6.3" / "original_data"
+    assert get_metadata(str(folder_path)) == dict(
+        session_start_time=datetime.datetime(2016, 11, 28, 21, 50, 33, 322000)
+    )
+
+
+def test_neuralynx_cheetah_v540_metadata():
+    folder_path = NLX_PATH / "Cheetah_v5.4.0" / "original_data"
+    assert get_metadata(str(folder_path)) == dict(
+        session_start_time=datetime.datetime(2001, 1, 1, 0, 0),
+    )
+
+
+def test_neuralynx_filtering():
+    file_path = NLX_PATH / "Cheetah_v5.7.4" / "original_data" / "CSC1.ncs"
+    assert (
+        get_filtering(str(file_path)) == '{"DSPLowCutFilterEnabled": "True", '
+        '"DspLowCutFrequency": "10", '
+        '"DspLowCutNumTaps": "0", '
+        '"DspLowCutFilterType": "DCO", '
+        '"DSPHighCutFilterEnabled": "True", '
+        '"DspHighCutFrequency": "9000", '
+        '"DspHighCutNumTaps": "64", '
+        '"DspHighCutFilterType": "FIR", '
+        '"DspDelayCompensation": "Enabled", '
+        '"DspFilterDelay_µs": "984"}'
+    )
+
+
+def test_neuralynx_filtering_recording_extractor():
+
+    ri = NeuralynxRecordingInterface(folder_path=(NLX_PATH / "Cheetah_v5.7.4" / "original_data"))
+
+    assert (
+        ri.recording_extractor.get_channel_property(0, "filtering") == '{"DSPLowCutFilterEnabled": "True", '
+        '"DspLowCutFrequency": "10", '
+        '"DspLowCutNumTaps": "0", '
+        '"DspLowCutFilterType": "DCO", '
+        '"DSPHighCutFilterEnabled": "True", '
+        '"DspHighCutFrequency": "9000", '
+        '"DspHighCutNumTaps": "64", '
+        '"DspHighCutFilterType": "FIR", '
+        '"DspDelayCompensation": "Enabled", '
+        '"DspFilterDelay_µs": "984"}'
+    )