import warnings
from datetime import date
from pydantic import FilePath
from ..baserecordingextractorinterface import BaseRecordingExtractorInterface
from ....tools import get_package
from ....utils import DeepDict
# EDF+ writes the month as an English abbreviation, so it is mapped rather than read with ``%b``, which
# goes through ``LC_TIME`` and would fail on a machine not running an English locale.
# TODO: nothing tests this. Asserting it needs a non-English locale, which no runner image ships, so the
# test has to come with a `locale-gen` step in `testing.yml` and a fixture that restores `LC_TIME`.
_MONTH_NUMBERS = {
name: number
for number, name in enumerate(
["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"], start=1
)
}
def _parse_birthdate(birthdate: str) -> str | None:
"""Return an EDF+ birthdate as an ISO 8601 date, or ``None`` where the header does not state one.
Parameters
----------
birthdate : str
The header's birthdate, which the readers hand back as ``"02 may 1951"``. EDF+ reserves ``"X"``
for a field the recording does not state, and a file written outside the spec can hold anything,
since it is a free-text patient field.
Returns
-------
str or None
The date as ``"1951-05-02"``, or ``None`` where the value is absent or not a date. It is returned
as a string because that is what the metadata schema declares; the conversion to a ``datetime``
happens once, where the subject is written.
"""
parts = birthdate.strip().split()
if len(parts) != 3:
return None
day, month_name, year = parts
month = _MONTH_NUMBERS.get(month_name.lower())
if month is None or not (day.isdigit() and year.isdigit()):
return None
try:
return date(year=int(year), month=month, day=int(day)).isoformat()
except ValueError: # A day the month does not have.
return None
[docs]
class EDFRecordingInterface(BaseRecordingExtractorInterface):
"""
Data interface class for converting European Data Format (EDF) data.
Uses the :py:func:`~spikeinterface.extractors.read_edf` reader from SpikeInterface.
Not supported on M1 macs.
"""
display_name = "EDF Recording"
keywords = BaseRecordingExtractorInterface.keywords + ("European Data Format",)
associated_suffixes = (".edf",)
info = "Interface for European Data Format (EDF) recording data."
[docs]
@classmethod
def get_source_schema(cls) -> dict:
source_schema = super().get_source_schema()
source_schema["properties"]["file_path"]["description"] = "Path to the .edf file."
return source_schema
[docs]
@classmethod
def get_stream_names(cls, file_path: FilePath) -> list[str]:
"""
Get the names of the streams available in an EDF file.
A stream is a set of channels that share a sampling rate, so a file that sampled some of
its signals at a different rate than the rest carries more than one.
Parameters
----------
file_path : FilePath
Path to the EDF file
Returns
-------
list of str
List of the stream names in the EDF file
"""
from spikeinterface.extractors.extractor_classes import EDFRecordingExtractor
stream_names, _ = EDFRecordingExtractor.get_streams(file_path=file_path)
return stream_names
[docs]
@staticmethod
def get_available_channel_ids(file_path: FilePath) -> list:
"""
Get all available channel names from an EDF file.
The names span the whole file. A file that sampled some of its signals at a different rate
than the rest holds them in separate streams, and an interface reads one stream at a time, so
the channels of the stream it holds are a subset of these. They are read from the file's
header, so this works on a file with more than one stream.
Parameters
----------
file_path : FilePath
Path to the EDF file
Returns
-------
list
List of all channel names in the EDF file
"""
from pyedflib import EdfReader
edf_reader = EdfReader(str(file_path))
try:
channel_names = edf_reader.getSignalLabels()
finally:
# EDFlib refuses to open a file it already has open, so the handle is released here
# rather than left to garbage collection.
edf_reader.close()
return channel_names
def _initialize_extractor(self, interface_kwargs: dict):
"""Override to add use_names_as_ids and pop channels_to_skip."""
self.extractor_kwargs = interface_kwargs.copy()
self.extractor_kwargs.pop("verbose", None)
self.extractor_kwargs.pop("es_key", None)
self.extractor_kwargs.pop("channels_to_skip")
self.extractor_kwargs["all_annotations"] = True
self.extractor_kwargs["use_names_as_ids"] = True
extractor_class = self.get_extractor_class()
extractor_instance = extractor_class(**self.extractor_kwargs)
return extractor_instance
def __init__(
self,
file_path: FilePath,
*args, # TODO: change to * (keyword only) on or after August 2026
verbose: bool = False,
es_key: str | None = None,
metadata_key: str | None = None,
channels_to_skip: list | None = None,
stream_name: str | None = None,
):
"""
Load and prepare data for EDF.
Currently, only continuous EDF+ files (EDF+C) and original EDF files (EDF) are supported
Parameters
----------
file_path : str or Path
Path to the edf file
verbose : bool, default: False
Allows verbose.
es_key : str, default: "ElectricalSeries"
Key for the ElectricalSeries metadata
metadata_key : str, optional
Key that indexes this interface's entries in the dict-based metadata. Defaults to
``"edf_recording"``.
channels_to_skip : list, default: None
Channels to skip when adding the data to the nwbfile. These parameter can be used to skip non-neural
channels that are present in the EDF file.
stream_name : str, optional
Name of the stream to read, as returned by ``get_stream_names``. A file that sampled some
of its signals at a different rate than the rest carries more than one stream and cannot
be read without naming one, since a single recording holds a single sampling rate.
"""
# Handle deprecated positional arguments
if args:
parameter_names = [
"verbose",
"es_key",
"channels_to_skip",
]
num_positional_args_before_args = 1 # file_path
if len(args) > len(parameter_names):
raise TypeError(
f"__init__() takes at most {len(parameter_names) + num_positional_args_before_args + 1} positional arguments but "
f"{len(args) + num_positional_args_before_args + 1} were given. "
"Note: Positional arguments are deprecated and will be removed on or after August 2026. "
"Please use keyword arguments."
)
positional_values = dict(zip(parameter_names, args))
passed_as_positional = list(positional_values.keys())
warnings.warn(
f"Passing arguments positionally to EDFRecordingInterface.__init__() is deprecated "
f"and will be removed on or after August 2026. "
f"The following arguments were passed positionally: {passed_as_positional}. "
"Please use keyword arguments instead.",
FutureWarning,
stacklevel=2,
)
verbose = positional_values.get("verbose", verbose)
es_key = positional_values.get("es_key", es_key)
channels_to_skip = positional_values.get("channels_to_skip", channels_to_skip)
get_package(
package_name="pyedflib",
excluded_platforms_and_python_versions=dict(darwin=dict(arm=["3.9"])),
)
super().__init__(
file_path=file_path,
verbose=verbose,
es_key=es_key,
metadata_key=metadata_key,
channels_to_skip=channels_to_skip,
stream_name=stream_name,
)
if metadata_key is None:
self.metadata_key = "edf_recording"
self.edf_header = self.recording_extractor.neo_reader.edf_header
# We remove the channels that are not neural
if channels_to_skip:
self.recording_extractor = self.recording_extractor.remove_channels(remove_channel_ids=channels_to_skip)