Skip to content

Commit

Permalink
Merge pull request #1404 from PeterNSteinmetz/factorHeaderHandling
Browse files Browse the repository at this point in the history
NeuralynxIO: Factor out header handling.
  • Loading branch information
samuelgarcia authored Feb 27, 2024
2 parents c33de92 + 1ae5841 commit 0fcc37f
Show file tree
Hide file tree
Showing 3 changed files with 58 additions and 27 deletions.
18 changes: 6 additions & 12 deletions neo/rawio/neuralynxrawio/ncssections.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,19 +51,13 @@ class NcsSection:

_RECORD_SIZE = 512 # nb sample per signal record

def __init__(self):
self.startRec = -1 # index of starting record
self.startTime = -1 # starttime of first record
self.endRec = -1 # index of last record (inclusive)
self.endTime = -1 # end time of last record, that is, the end time of the last
# sampling period contained in the last record of the section

def __init__(self, sb, st, eb, et, ns):
self.startRec = sb
self.startTime = st
self.endRec = eb
self.endTime = et
self.n_samples = ns
self.startRec = sb # index of starting record
self.startTime = st # starttime of first record
self.endRec = eb # index of last record (inclusive)
self.endTime = et # end time of last record, that is, the end time of the last
# sampling period contained in the last record of the section
self.n_samples = ns # number of samples in record which are valid

def __eq__(self, other):
return (
Expand Down
51 changes: 37 additions & 14 deletions neo/rawio/neuralynxrawio/nlxheader.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,18 +124,46 @@ def _to_bool(txt):
),
}

def __init__(self, filename):
def __init__(self, filename, props_only=False):
"""
Factory function to build NlxHeader for a given file.
:param filename: name of Neuralynx file
:param props_only: if true, will not try and read time and date or check start
"""
super(OrderedDict, self).__init__()
with open(filename, "rb") as f:
txt_header = f.read(NlxHeader.HEADER_SIZE)
txt_header = txt_header.strip(b"\x00").decode("latin-1")

# must start with 8 # characters
assert txt_header.startswith("########"), "Neuralynx files must start with 8 # characters."
if not props_only and not txt_header.startswith("########"):
ValueError("Neuralynx files must start with 8 # characters.")

self.read_properties(filename, txt_header)

if not props_only:
self.readTimeDate(txt_header)

@staticmethod
def build_with_properties_only(filename):
"""
Builds a version of the header without time and date or other validity checking.
Intended mostly for utilities but may also be useful for some recalcitrant header formats.
:param filename: name of Neuralynx file.
:return: NlxHeader with properties from header text
"""
res = OrderedDict()


def read_properties(self, filename, txt_header):
"""
Read properties from header and place in OrderedDictionary which this object is.
:param filename: name of ncs file, used for extracting channel number
:param txt_header: header text
"""
# find keys
for k1, k2, type_ in NlxHeader.txt_header_keys:
pattern = r"-(?P<name>" + k1 + r")\s+(?P<value>[\S ]*)"
Expand All @@ -149,17 +177,14 @@ def __init__(self, filename):
if type_ is not None:
value = type_(value)
self[name] = value

# if channel_ids or s not in self then the filename is used
name = os.path.splitext(os.path.basename(filename))[0]

# convert channel ids
if "channel_ids" in self:
chid_entries = re.findall(r"\S+", self["channel_ids"])
self["channel_ids"] = [int(c) for c in chid_entries]
else:
self["channel_ids"] = ["unknown"]

# convert channel names
if "channel_names" in self:
name_entries = re.findall(r"\S+", self["channel_names"])
Expand All @@ -170,7 +195,6 @@ def __init__(self, filename):
), "Number of channel ids does not match channel names."
else:
self["channel_names"] = ["unknown"] * len(self["channel_ids"])

# version and application name
# older Cheetah versions with CheetahRev property
if "CheetahRev" in self:
Expand All @@ -192,11 +216,9 @@ def __init__(self, filename):
else:
self["ApplicationName"] = "Neuraview"
app_version = "2"

if " Development" in app_version:
app_version = app_version.replace(" Development", ".dev0")
self["ApplicationVersion"] = Version(app_version)

# convert bit_to_microvolt
if "bit_to_microVolt" in self:
btm_entries = re.findall(r"\S+", self["bit_to_microVolt"])
Expand All @@ -206,7 +228,6 @@ def __init__(self, filename):
assert len(self["bit_to_microVolt"]) == len(
self["channel_ids"]
), "Number of channel ids does not match bit_to_microVolt conversion factors."

if "InputRange" in self:
ir_entries = re.findall(r"\w+", self["InputRange"])
if len(ir_entries) == 1:
Expand All @@ -217,9 +238,13 @@ def __init__(self, filename):
chid_entries
), "Number of channel ids does not match input range values."

# Format of datetime depends on app name, app version
# :TODO: this works for current examples but is not likely actually related
# to app version in this manner.
def readTimeDate(self, txt_header):
"""
Read time and date from text of header appropriate for app name and version
:TODO: this works for current examples but is not likely actually related
to app version in this manner.
"""
an = self["ApplicationName"]
if an == "Cheetah":
av = self["ApplicationVersion"]
Expand All @@ -245,7 +270,6 @@ def __init__(self, filename):
an = "Unknown"
av = "NA"
hpd = NlxHeader.header_pattern_dicts["def"]

# opening time
sr = re.search(hpd["datetime1_regex"], txt_header)
if not sr:
Expand All @@ -257,7 +281,6 @@ def __init__(self, filename):
self["recording_opened"] = datetime.datetime.strptime(
dt1["date"] + " " + dt1["time"], hpd["datetimeformat"]
)

# close time, if available
if "datetime2_regex" in hpd:
sr = re.search(hpd["datetime2_regex"], txt_header)
Expand Down
16 changes: 15 additions & 1 deletion neo/test/rawiotest/test_neuralynxrawio.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ class TestNeuralynxRawIO(
"neuralynx/Cheetah_v5.5.1/original_data",
"neuralynx/Cheetah_v5.6.3/original_data",
"neuralynx/Cheetah_v5.7.4/original_data",
"neuralynx/Cheetah_v6.3.2/incomplete_blocks",
"neuralynx/Cheetah_v6.3.2/incomplete_blocks"
]

def test_scan_ncs_files(self):
Expand Down Expand Up @@ -356,5 +356,19 @@ def test_equality(self):
self.assertNotEqual(ns0, ns1)


# I comment this now and will put it back when files will be in gin.g-node
# class TestNlxHeader(TestNeuralynxRawIO, unittest.TestCase):
# def test_no_date_time(self):
# filename = self.get_local_path("neuralynx/NoDateHeader/NoDateHeader.nev")

# with self.assertRaises(IOError):
# hdr = NlxHeader(filename)

# hdr = NlxHeader(filename, props_only=True)

# self.assertEqual(len(hdr), 11)
# self.assertEqual(hdr['ApplicationName'], 'Pegasus')
# self.assertEqual(hdr['FileType'], 'Event')

if __name__ == "__main__":
unittest.main()

0 comments on commit 0fcc37f

Please sign in to comment.