# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.
import os
import os.path as op
from collections import OrderedDict
from datetime import datetime, timezone
import numpy as np
from ..._fiff.constants import FIFF
from ..._fiff.meas_info import create_info
from ..._fiff.utils import _mult_cal_one
from ...annotations import Annotations
from ...utils import _check_fname, fill_doc, logger, verbose, warn
from ..base import BaseRaw
@fill_doc
def read_raw_persyst(fname, preload=False, verbose=None) -> "RawPersyst":
"""Reader for a Persyst (.lay/.dat) recording.
Parameters
----------
fname : path-like
Path to the Persyst header ``.lay`` file.
%(preload)s
%(verbose)s
Returns
-------
raw : instance of RawPersyst
A Raw object containing Persyst data.
See :class:`mne.io.Raw` for documentation of attributes and methods.
See Also
--------
mne.io.Raw : Documentation of attributes and methods of RawPersyst.
Notes
-----
It is assumed that the ``.lay`` and ``.dat`` file
are in the same directory. To get the correct file path to the
``.dat`` file, ``read_raw_persyst`` will get the corresponding dat
filename from the lay file, and look for that file inside the same
directory as the lay file.
"""
return RawPersyst(fname, preload, verbose)
@fill_doc
class RawPersyst(BaseRaw):
"""Raw object from a Persyst file.
Parameters
----------
fname : path-like
Path to the Persyst header (.lay) file.
%(preload)s
%(verbose)s
See Also
--------
mne.io.Raw : Documentation of attributes and methods.
"""
@verbose
def __init__(self, fname, preload=False, verbose=None):
fname = str(_check_fname(fname, "read", True, "fname"))
logger.info(f"Loading {fname}")
# make sure filename is the Lay file
if not fname.endswith(".lay"):
fname = fname + ".lay"
# get the current directory and Lay filename
curr_path, lay_fname = op.dirname(fname), op.basename(fname)
if not op.exists(fname):
raise FileNotFoundError(
f'The path you specified, "{lay_fname}",does not exist.'
)
# sections and subsections currently unused
keys, data, sections = _read_lay_contents(fname)
# these are the section headers in the Persyst file layout
# Note: We do not make use of "SampleTimes" yet
fileinfo_dict = OrderedDict()
channelmap_dict = OrderedDict()
patient_dict = OrderedDict()
comments_dict = OrderedDict()
# keep track of total number of comments
num_comments = 0
# loop through each line in the lay file
for key, val, section in zip(keys, data, sections):
if key == "":
continue
# Make sure key are lowercase for everything, but electrodes.
# We also do not want to lower-case comments because those
# are free-form text where casing may matter.
if key is not None and section not in ["channelmap", "comments"]:
key = key.lower()
# FileInfo
if section == "fileinfo":
# extract the .dat file name
if key == "file":
dat_fname = op.basename(val)
dat_fpath = op.join(curr_path, op.basename(dat_fname))
# determine if .dat file exists where it should
error_msg = (
f"The data path you specified "
f"does not exist for the lay path, "
f"{lay_fname}. Make sure the dat file "
f"is in the same directory as the lay "
f"file, and the specified dat filename "
f"matches."
)
if not op.exists(dat_fpath):
raise FileNotFoundError(error_msg)
fileinfo_dict[key] = val
# ChannelMap
elif section == "channelmap":
# channel map has <channel_name>=<number> for <key>=<val>
channelmap_dict[key] = val
# Patient (All optional)
elif section == "patient":
patient_dict[key] = val
# Comments (turned into mne.Annotations)
elif section == "comments":
comments_dict[key] = comments_dict.get(key, list()) + [val]
num_comments += 1
# get numerical metadata
# datatype is either 7 for 32 bit, or 0 for 16 bit
datatype = fileinfo_dict.get("datatype")
cal = float(fileinfo_dict.get("calibration"))
n_chs = int(fileinfo_dict.get("waveformcount"))
# Store subject information from lay file in mne format
# Note: Persyst also records "Physician", "Technician",
# "Medications", "History", and "Comments1" and "Comments2"
# and this information is currently discarded
subject_info = _get_subjectinfo(patient_dict)
# set measurement date
testdate = patient_dict.get("testdate")
if testdate is not None:
# TODO: Persyst may change its internal date schemas
# without notice
# These are the 3 "so far" possible datatime storage
# formats in Persyst .lay
if "/" in testdate:
testdate = datetime.strptime(testdate, "%m/%d/%Y")
elif "-" in testdate:
testdate = datetime.strptime(testdate, "%d-%m-%Y")
elif "." in testdate:
testdate = datetime.strptime(testdate, "%Y.%m.%d")
if not isinstance(testdate, datetime):
warn(
"Cannot read in the measurement date due "
"to incompatible format. Please set manually "
f"for {lay_fname} "
)
meas_date = None
else:
testtime = datetime.strptime(patient_dict.get("testtime"), "%H:%M:%S")
meas_date = datetime(
year=testdate.year,
month=testdate.month,
day=testdate.day,
hour=testtime.hour,
minute=testtime.minute,
second=testtime.second,
tzinfo=timezone.utc,
)
# Create mne structure
ch_names = list(channelmap_dict.keys())
if n_chs != len(ch_names):
raise RuntimeError(
"Channels in lay file do not "
"match the number of channels "
"in the .dat file."
) # noqa
# get rid of the "-Ref" in channel names
ch_names = [ch.upper().split("-REF")[0] for ch in ch_names]
# get the sampling rate and default channel types to EEG
sfreq = fileinfo_dict.get("samplingrate")
ch_types = "eeg"
info = create_info(ch_names, sfreq, ch_types=ch_types)
info.update(subject_info=subject_info)
with info._unlock():
for idx in range(n_chs):
# calibration brings to uV then 1e-6 brings to V
info["chs"][idx]["cal"] = cal * 1.0e-6
info["meas_date"] = meas_date
# determine number of samples in file
# Note: We do not use the lay file to do this
# because clips in time may be generated by Persyst that
# DO NOT modify the "SampleTimes" section
with open(dat_fpath, "rb") as f:
# determine the precision
if int(datatype) == 7:
# 32 bit
dtype = np.dtype("i4")
elif int(datatype) == 0:
# 16 bit
dtype = np.dtype("i2")
else:
raise RuntimeError(f"Unknown format: {datatype}")
# allow offset to occur
f.seek(0, os.SEEK_END)
n_samples = f.tell()
n_samples = n_samples // (dtype.itemsize * n_chs)
logger.debug(f"Loaded {n_samples} samples for {n_chs} channels.")
raw_extras = {"dtype": dtype, "n_chs": n_chs, "n_samples": n_samples}
# create Raw object
super().__init__(
info,
preload,
filenames=[dat_fpath],
last_samps=[n_samples - 1],
raw_extras=[raw_extras],
verbose=verbose,
)
# set annotations based on the comments read in
onset = np.zeros(num_comments, float)
duration = np.zeros(num_comments, float)
description = [""] * num_comments
# loop through comments dictionary, which may contain
# multiple events for the same "text" annotation
t_idx = 0
for _description, event_tuples in comments_dict.items():
for _onset, _duration in event_tuples:
# extract the onset, duration, description to
# create an Annotations object
onset[t_idx] = _onset
duration[t_idx] = _duration
description[t_idx] = _description
t_idx += 1
annot = Annotations(onset, duration, description)
self.set_annotations(annot)
def _read_segment_file(self, data, idx, fi, start, stop, cals, mult):
"""Read a segment of data from a file.
The Persyst software records raw data in either 16 or 32 bit
binary files. In addition, it stores the calibration to convert
data to uV in the lay file.
"""
dtype = self._raw_extras[fi]["dtype"]
n_chs = self._raw_extras[fi]["n_chs"]
dat_fname = self.filenames[fi]
# compute samples count based on start and stop
time_length_samps = stop - start
# read data from .dat file into array of correct size, then calibrate
# records = recnum rows x inf columns
count = time_length_samps * n_chs
# seek the dat file
with open(dat_fname, "rb") as dat_file_ID:
# allow offset to occur
dat_file_ID.seek(n_chs * dtype.itemsize * start, 1)
# read in the actual record starting at possibly offset
record = np.fromfile(dat_file_ID, dtype=dtype, count=count)
# chs * rows
# cast as float32; more than enough precision
record = np.reshape(record, (n_chs, -1), order="F").astype(np.float32)
# calibrate to convert to V and handle mult
_mult_cal_one(data, record, idx, cals, mult)
def _get_subjectinfo(patient_dict):
# attempt to parse out the birthdate, but if it doesn't
# meet spec, then it will set to None
birthdate = patient_dict.get("birthdate")
if "/" in birthdate:
try:
birthdate = datetime.strptime(birthdate, "%m/%d/%y")
except ValueError:
birthdate = None
print(f"Unable to process birthdate of {birthdate} ")
elif "-" in birthdate:
try:
birthdate = datetime.strptime(birthdate, "%d-%m-%y")
except ValueError:
birthdate = None
print(f"Unable to process birthdate of {birthdate} ")
subject_info = {
"first_name": patient_dict.get("first"),
"middle_name": patient_dict.get("middle"),
"last_name": patient_dict.get("last"),
"sex": patient_dict.get("sex"),
"hand": patient_dict.get("hand"),
"his_id": patient_dict.get("id"),
"birthday": birthdate,
}
subject_info = {key: val for key, val in subject_info.items() if val is not None}
# Recode sex values
sex_dict = dict(
m=FIFF.FIFFV_SUBJ_SEX_MALE,
male=FIFF.FIFFV_SUBJ_SEX_MALE,
f=FIFF.FIFFV_SUBJ_SEX_FEMALE,
female=FIFF.FIFFV_SUBJ_SEX_FEMALE,
)
subject_info["sex"] = sex_dict.get(subject_info["sex"], FIFF.FIFFV_SUBJ_SEX_UNKNOWN)
# Recode hand values
hand_dict = dict(
r=FIFF.FIFFV_SUBJ_HAND_RIGHT,
right=FIFF.FIFFV_SUBJ_HAND_RIGHT,
l=FIFF.FIFFV_SUBJ_HAND_LEFT,
left=FIFF.FIFFV_SUBJ_HAND_LEFT,
a=FIFF.FIFFV_SUBJ_HAND_AMBI,
ambidextrous=FIFF.FIFFV_SUBJ_HAND_AMBI,
ambi=FIFF.FIFFV_SUBJ_HAND_AMBI,
)
# no handedness is set when unknown
try:
subject_info["hand"] = hand_dict[subject_info["hand"]]
except KeyError:
subject_info.pop("hand")
return subject_info
def _read_lay_contents(fname):
"""Lay file are laid out like a INI file."""
# keep track of sections, keys and data
sections = []
keys, data = [], []
# initialize all section to empty str
section = ""
with open(fname) as fin:
for line in fin:
# break a line into a status, key and value
status, key, val = _process_lay_line(line, section)
# handle keys and values if they are
# Section, Subsections, or Line items
if status == 1: # Section was found
section = val.lower()
continue
# keep track of all sections, subsections,
# keys and the data of the file
sections.append(section)
data.append(val)
keys.append(key)
return keys, data, sections
def _process_lay_line(line, section):
"""Process a line read from the Lay (INI) file.
Each line in the .lay file will be processed
into a structured ``status``, ``key`` and ``value``.
Parameters
----------
line : str
The actual line in the Lay file.
section : str
The section in the Lay file.
Returns
-------
status : int
Returns the following integers based on status.
-1 => unknown string found
0 => empty line found
1 => section found
2 => key-value pair found
key : str
The string before the ``'='`` character. If section is "Comments",
then returns the text comment description.
value : str
The string from the line after the ``'='`` character. If section is
"Comments", then returns the onset and duration as a tuple.
Notes
-----
The lay file comprises of multiple "sections" that are documented with
bracket ``[]`` characters. For example, ``[FileInfo]`` and the lines
afterward indicate metadata about the data file itself. Within
each section, there are multiple lines in the format of
``<key>=<value>``.
For ``FileInfo``, ``Patient`` and ``ChannelMap``
each line will be denoted with a ``key`` and a ``value`` that
can be represented as a dictionary. The keys describe what sort
of data that line holds, while the values contain the corresponding
value. In some cases, the ``value``.
For ``SampleTimes``, the ``key`` and ``value`` pair indicate the
start and end time in seconds of the original data file.
For ``Comments`` section, this denotes an area where users through
Persyst actually annotate data in time. These are instead
represented as 5 data points that are ``,`` delimited. These
data points are ordered as:
1. time (in seconds) of the annotation
2. duration (in seconds) of the annotation
3. state (unused)
4. variable type (unused)
5. free-form text describing the annotation
"""
key = "" # default; only return value possibly not set
line = line.strip() # remove leading and trailing spaces
end_idx = len(line) - 1 # get the last index of the line
# empty sequence evaluates to false
if not line:
status = 0
key = ""
value = ""
return status, key, value
# section found
elif (line[0] == "[") and (line[end_idx] == "]") and (end_idx + 1 >= 3):
status = 1
value = line[1:end_idx].lower()
# key found
else:
# handle Comments section differently from all other sections
# TODO: utilize state and var_type in code.
# Currently not used
if section == "comments":
# Persyst Comments output 5 variables "," separated
time_sec, duration, state, var_type, text = line.split(",", 4)
del var_type, state
status = 2
key = text
value = (time_sec, duration)
# all other sections
else:
if "=" not in line:
raise RuntimeError(
f"The line {line} does not conform "
"to the standards. Please check the "
".lay file."
) # noqa
pos = line.index("=")
status = 2
# the line now is composed of a
# <key>=<value>
key = line[0:pos]
key.strip()
value = line[pos + 1 : end_idx + 1]
value.strip()
return status, key, value
Datasets
Models
