# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.
import os
from datetime import datetime, timezone
import numpy as np
from ..._fiff.constants import FIFF
from ..._fiff.meas_info import _empty_info
from ..._fiff.utils import _file_size, _read_segments_file
from ...annotations import Annotations
from ...utils import _check_fname, fill_doc, logger, warn
from ..base import BaseRaw, _get_scaling
CH_TYPE_MAPPING = {
"CC": "SEEG",
}
# See https://blackrockneurotech.com/wp-content/uploads/LB-0023-7.00_NEV_File_Format.pdf
DATA_BYTE_SIZE = 2
ORIG_FORMAT = "short"
nsx_header_dict = {
"basic": [
("file_id", "S8"), # achFileType
# file specification split into major and minor version number
("ver_major", "uint8"),
("ver_minor", "uint8"),
# bytes of basic & extended header
("bytes_in_headers", "uint32"),
# label of the sampling group (e.g., "1 kS/s" or "LFP low")
("label", "S16"),
("comment", "S256"),
("period", "uint32"),
("timestamp_resolution", "uint32"),
# time origin: 2byte uint16 values for ...
("year", "uint16"),
("month", "uint16"),
("weekday", "uint16"),
("day", "uint16"),
("hour", "uint16"),
("minute", "uint16"),
("second", "uint16"),
("millisecond", "uint16"),
# number of channel_count match number of extended headers
("channel_count", "uint32"),
],
"extended": [
("type", "S2"),
("electrode_id", "uint16"),
("electrode_label", "S16"),
# used front-end amplifier bank (e.g., A, B, C, D)
("physical_connector", "uint8"),
# used connector pin (e.g., 1-37 on bank A, B, C or D)
("connector_pin", "uint8"),
# digital and analog value ranges of the signal
("min_digital_val", "int16"),
("max_digital_val", "int16"),
("min_analog_val", "int16"),
("max_analog_val", "int16"),
# units of the analog range values ("mV" or "uV")
("units", "S16"),
# filter settings used to create nsx from source signal
("hi_freq_corner", "uint32"),
("hi_freq_order", "uint32"),
("hi_freq_type", "uint16"), # 0=None, 1=Butterworth
("lo_freq_corner", "uint32"),
("lo_freq_order", "uint32"),
("lo_freq_type", "uint16"),
], # 0=None, 1=Butterworth,
"data>2.1<3": [
("header", "uint8"),
("timestamp", "uint32"),
("nb_data_points", "uint32"),
],
"data>=3": [
("header", "uint8"),
("timestamp", "uint64"),
("nb_data_points", "uint32"),
],
}
@fill_doc
def read_raw_nsx(
input_fname, stim_channel=True, eog=None, misc=None, preload=False, *, verbose=None
) -> "RawNSX":
"""Reader function for NSx (Blackrock Microsystems) files.
Parameters
----------
input_fname : str
Path to the NSx file.
stim_channel : ``'auto'`` | str | list of str | int | list of int
Defaults to ``'auto'``, which means that channels named ``'status'`` or
``'trigger'`` (case insensitive) are set to STIM. If str (or list of
str), all channels matching the name(s) are set to STIM. If int (or
list of ints), channels corresponding to the indices are set to STIM.
eog : list or tuple
Names of channels or list of indices that should be designated EOG
channels. Values should correspond to the electrodes in the file.
Default is None.
misc : list or tuple
Names of channels or list of indices that should be designated MISC
channels. Values should correspond to the electrodes in the file.
Default is None.
%(preload)s
%(verbose)s
Returns
-------
raw : instance of RawEDF
The raw instance.
See :class:`mne.io.Raw` for documentation of attributes and methods.
Notes
-----
NSx files with id (= NEURALSG), i.e., version 2.1 is currently not
supported.
If channels named 'status' or 'trigger' are present, they are considered as
STIM channels by default. Use func:`mne.find_events` to parse events
encoded in such analog stim channels.
"""
input_fname = _check_fname(
input_fname, overwrite="read", must_exist=True, name="input_fname"
)
if not input_fname.suffix.lower().startswith(".ns"):
raise NotImplementedError(
f"Only NSx files are supported, got {input_fname.suffix}."
)
return RawNSX(
input_fname, stim_channel, eog, misc, preload=preload, verbose=verbose
)
@fill_doc
class RawNSX(BaseRaw):
"""Raw object from NSx file from Blackrock Microsystems.
Parameters
----------
input_fname : str
Path to the NSx file.
stim_channel : ``'auto'`` | str | list of str | int | list of int
Defaults to ``'auto'``, which means that channels named ``'status'`` or
``'trigger'`` (case insensitive) are set to STIM. If str (or list of
str), all channels matching the name(s) are set to STIM. If int (or
list of ints), channels corresponding to the indices are set to STIM.
eog : list or tuple
Names of channels or list of indices that should be designated EOG
channels. Values should correspond to the electrodes in the file.
Default is None.
misc : list or tuple
Names of channels or list of indices that should be designated MISC
channels. Values should correspond to the electrodes in the file.
Default is None.
%(preload)s
%(verbose)s
Notes
-----
NSx files with id (= NEURALSG), i.e., version 2.1 is currently not
supported.
If channels named 'status' or 'trigger' are present, they are considered as
STIM channels by default. Use func:`mne.find_events` to parse events
encoded in such analog stim channels.
"""
def __init__(
self,
input_fname,
stim_channel="auto",
eog=None,
misc=None,
preload=False,
verbose=None,
):
logger.info(f"Extracting NSX parameters from {input_fname}...")
input_fname = os.path.abspath(input_fname)
(
info,
data_fname,
fmt,
n_samples,
orig_format,
raw_extras,
orig_units,
) = _get_hdr_info(input_fname, stim_channel=stim_channel, eog=eog, misc=misc)
raw_extras["orig_format"] = orig_format
first_samps = (raw_extras["timestamp"][0],)
super().__init__(
info,
first_samps=first_samps,
last_samps=[first_samps[0] + n_samples - 1],
filenames=[data_fname],
orig_format=orig_format,
preload=preload,
verbose=verbose,
raw_extras=[raw_extras],
orig_units=orig_units,
)
# Add annotations for in-data skips
if len(self._raw_extras[0]["timestamp"]) > 1:
starts = (
self._raw_extras[0]["timestamp"] + self._raw_extras[0]["nb_data_points"]
)[:-1] + 1
stops = self._raw_extras[0]["timestamp"][1:] - 1
durations = (stops - starts + 1) / self.info["sfreq"]
annot = Annotations(
onset=(starts / self.info["sfreq"]),
duration=durations,
description="BAD_ACQ_SKIP",
orig_time=self.info["meas_date"],
)
self.set_annotations(annot)
def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
"""Read a chunk of raw data."""
dtype = self._raw_extras[fi]["orig_format"]
first_samps = self._raw_extras[fi]["timestamp"]
recording_extents = self._raw_extras[fi]["nb_data_points"]
offsets = self._raw_extras[fi]["offset_to_data_block"]
for first_samp, recording_extent, offset in zip(
first_samps, recording_extents, offsets
):
if start > first_samp + recording_extent or stop < first_samp:
# There is nothing to read in this chunk
continue
i_start = max(start, first_samp)
i_stop = min(stop, first_samp + recording_extent)
_read_segments_file(
self,
data[:, i_start - start : i_stop - start],
idx,
fi,
i_start - first_samp,
i_stop - first_samp,
cals,
mult,
dtype,
n_channels=None,
offset=offset,
trigger_ch=None,
)
def _read_header(fname):
nsx_file_id = np.fromfile(fname, count=1, dtype=[("file_id", "S8")])[0][
"file_id"
].decode()
if nsx_file_id in ["NEURALCD", "BRSMPGRP"]:
basic_header = _read_header_22_and_above(fname)
elif nsx_file_id == "NEURALSG":
raise NotImplementedError(
"NSx file id (= NEURALSG), i.e., file"
" version 2.1 is currently not supported."
)
else:
raise ValueError(
f"NSx file id (={nsx_file_id}) does not match"
" with supported file ids:"
" ('NEURALCD', 'BRSMPGRP')"
)
time_origin = datetime(
*[
basic_header.pop(xx)
for xx in (
"year",
"month",
"day",
"hour",
"minute",
"second",
"millisecond",
)
],
tzinfo=timezone.utc,
)
basic_header["meas_date"] = time_origin
return basic_header
def _read_header_22_and_above(fname):
basic_header = {}
dtype0 = nsx_header_dict["basic"]
dtype1 = nsx_header_dict["extended"]
nsx_file_header = np.fromfile(fname, count=1, dtype=dtype0)[0]
basic_header.update(
{name: nsx_file_header[name] for name in nsx_file_header.dtype.names}
)
offset_dtype0 = np.dtype(dtype0).itemsize
shape = nsx_file_header["channel_count"]
basic_header["extended"] = np.memmap(
fname, shape=shape, offset=offset_dtype0, dtype=dtype1, mode="r"
)
# The following values are stored in mHz
# See:
# https://blackrockneurotech.com/wp-content/uploads/LB-0023-7.00_NEV_File_Format.pdf
basic_header["highpass"] = basic_header["extended"]["hi_freq_corner"]
basic_header["lowpass"] = basic_header["extended"]["lo_freq_corner"]
for x in ["highpass", "lowpass"]:
basic_header[x] = basic_header[x] * 1e-3
ver_major, ver_minor = basic_header.pop("ver_major"), basic_header.pop("ver_minor")
basic_header["spec"] = f"{ver_major}.{ver_minor}"
data_header = list()
index = 0
offset = basic_header["bytes_in_headers"]
filesize = _file_size(fname)
if float(basic_header["spec"]) < 3.0:
dtype2 = nsx_header_dict["data>2.1<3"]
else:
dtype2 = nsx_header_dict["data>=3"]
while offset < filesize:
dh = np.memmap(fname, dtype=dtype2, shape=1, offset=offset, mode="r")[0]
data_header.append(
{
"header": dh["header"],
"timestamp": dh["timestamp"],
"nb_data_points": dh["nb_data_points"],
"offset_to_data_block": offset + dh.dtype.itemsize,
}
)
# data size = number of data points * (data_bytes * number of channels)
# use of `int` avoids overflow problem
data_size = (
int(dh["nb_data_points"])
* int(basic_header["channel_count"])
* DATA_BYTE_SIZE
)
# define new offset (to possible next data block)
offset = data_header[index]["offset_to_data_block"] + data_size
index += 1
basic_header["data_header"] = data_header
return basic_header
def _get_hdr_info(fname, stim_channel=True, eog=None, misc=None):
"""Read header information NSx file."""
eog = eog if eog is not None else []
misc = misc if misc is not None else []
nsx_info = _read_header(fname)
ch_names = list(nsx_info["extended"]["electrode_label"])
ch_types = list(nsx_info["extended"]["type"])
ch_units = list(nsx_info["extended"]["units"])
ch_names, ch_types, ch_units = (
list(map(bytes.decode, xx)) for xx in (ch_names, ch_types, ch_units)
)
max_analog_val = nsx_info["extended"]["max_analog_val"].astype("double")
min_analog_val = nsx_info["extended"]["min_analog_val"].astype("double")
max_digital_val = nsx_info["extended"]["max_digital_val"].astype("double")
min_digital_val = nsx_info["extended"]["min_digital_val"].astype("double")
cals = (max_analog_val - min_analog_val) / (max_digital_val - min_digital_val)
stim_channel_idxs, _ = _check_stim_channel(stim_channel, ch_names)
nchan = int(nsx_info["channel_count"])
logger.info("Setting channel info structure...")
chs = list()
pick_mask = np.ones(len(ch_names))
orig_units = {}
for idx, ch_name in enumerate(ch_names):
chan_info = {}
chan_info["logno"] = int(nsx_info["extended"]["electrode_id"][idx])
chan_info["scanno"] = int(nsx_info["extended"]["electrode_id"][idx])
chan_info["ch_name"] = ch_name
chan_info["unit_mul"] = FIFF.FIFF_UNITM_NONE
ch_unit = ch_units[idx]
chan_info["unit"] = FIFF.FIFF_UNIT_V
# chan_info["range"] = _unit_range_dict[ch_units[idx]]
chan_info["range"] = 1 / _get_scaling("eeg", ch_units[idx])
chan_info["cal"] = cals[idx]
chan_info["coord_frame"] = FIFF.FIFFV_COORD_HEAD
chan_info["coil_type"] = FIFF.FIFFV_COIL_EEG
chan_info["kind"] = FIFF.FIFFV_SEEG_CH
# montage can't be stored in NSx so channel locs are unknown:
chan_info["loc"] = np.full(12, np.nan)
orig_units[ch_name] = ch_unit
# if the NSx info contained channel type information
# set it now. They are always set to 'CC'.
# If not inferable, set it to 'SEEG' with a warning.
ch_type = ch_types[idx]
ch_const = getattr(FIFF, f"FIFFV_{CH_TYPE_MAPPING.get(ch_type, 'SEEG')}_CH")
chan_info["kind"] = ch_const
# if user passes in explicit mapping for eog, misc and stim
# channels set them here.
if ch_name in eog or idx in eog or idx - nchan in eog:
chan_info["coil_type"] = FIFF.FIFFV_COIL_NONE
chan_info["kind"] = FIFF.FIFFV_EOG_CH
pick_mask[idx] = False
elif ch_name in misc or idx in misc or idx - nchan in misc:
chan_info["coil_type"] = FIFF.FIFFV_COIL_NONE
chan_info["kind"] = FIFF.FIFFV_MISC_CH
pick_mask[idx] = False
elif idx in stim_channel_idxs:
chan_info["coil_type"] = FIFF.FIFFV_COIL_NONE
chan_info["unit"] = FIFF.FIFF_UNIT_NONE
chan_info["kind"] = FIFF.FIFFV_STIM_CH
pick_mask[idx] = False
chan_info["ch_name"] = ch_name
ch_names[idx] = chan_info["ch_name"]
chs.append(chan_info)
sfreq = nsx_info["timestamp_resolution"] / nsx_info["period"]
info = _empty_info(sfreq)
info["meas_date"] = nsx_info["meas_date"]
info["chs"] = chs
info["ch_names"] = ch_names
highpass = nsx_info["highpass"][:128]
lowpass = nsx_info["lowpass"][:128]
_decode_online_filters(info, highpass, lowpass)
# Some keys to be consistent with FIF measurement info
info["description"] = None
info._unlocked = False
info._update_redundant()
orig_format = ORIG_FORMAT
raw_extras = {
key: [r[key] for r in nsx_info["data_header"]]
for key in nsx_info["data_header"][0]
}
for key in raw_extras:
raw_extras[key] = np.array(raw_extras[key], int)
good_data_packets = raw_extras.pop("header") == 1
if not good_data_packets.any():
raise RuntimeError("NSx file appears to be broken")
raw_extras = {key: raw_extras[key][good_data_packets] for key in raw_extras.keys()}
raw_extras["timestamp"] = raw_extras["timestamp"] // nsx_info["period"]
first_samp = raw_extras["timestamp"][0]
last_samp = raw_extras["timestamp"][-1] + raw_extras["nb_data_points"][-1]
n_samples = last_samp - first_samp
return (
info,
fname,
nsx_info["spec"],
n_samples,
orig_format,
raw_extras,
orig_units,
)
def _decode_online_filters(info, highpass, lowpass):
"""Decode low/high-pass filters that are applied online."""
if np.all(highpass == highpass[0]):
if highpass[0] == "NaN":
# Placeholder for future use. Highpass set in _empty_info.
pass
else:
hp = float(highpass[0])
info["highpass"] = hp
else:
info["highpass"] = float(np.max(highpass))
warn(
"Channels contain different highpass filters. Highest filter "
"setting will be stored."
)
if np.all(lowpass == lowpass[0]):
if lowpass[0] in ("NaN", "0", "0.0"):
# Placeholder for future use. Lowpass set in _empty_info.
pass
else:
info["lowpass"] = float(lowpass[0])
else:
info["lowpass"] = float(np.min(lowpass))
warn(
"Channels contain different lowpass filters. Lowest filter "
"setting will be stored."
)
def _check_stim_channel(stim_channel, ch_names):
"""Check that the stimulus channel exists in the current datafile."""
DEFAULT_STIM_CH_NAMES = ["status", "trigger"]
if stim_channel is None or stim_channel is False:
return [], []
if stim_channel is True: # convenient aliases
stim_channel = "auto"
if isinstance(stim_channel, str):
if stim_channel == "auto":
if "auto" in ch_names:
warn(
RuntimeWarning,
"Using `stim_channel='auto'` when auto"
" also corresponds to a channel name is ambiguous."
" Please use `stim_channel=['auto']`.",
)
else:
valid_stim_ch_names = DEFAULT_STIM_CH_NAMES
else:
valid_stim_ch_names = [stim_channel.lower()]
elif isinstance(stim_channel, int):
valid_stim_ch_names = [ch_names[stim_channel].lower()]
elif isinstance(stim_channel, list):
if all([isinstance(s, str) for s in stim_channel]):
valid_stim_ch_names = [s.lower() for s in stim_channel]
elif all([isinstance(s, int) for s in stim_channel]):
valid_stim_ch_names = [ch_names[s].lower() for s in stim_channel]
else:
raise ValueError("Invalid stim_channel")
else:
raise ValueError("Invalid stim_channel")
ch_names_low = [ch.lower() for ch in ch_names]
found = list(set(valid_stim_ch_names) & set(ch_names_low))
stim_channel_idxs = [ch_names_low.index(f) for f in found]
names = [ch_names[idx] for idx in stim_channel_idxs]
return stim_channel_idxs, names
Datasets
Models
