"""Interface for discrete events (digital IO) from Doric Neuroscience Studio CSV exports."""
from pydantic import FilePath, validate_call
from neuroconv.utils import DeepDict
from ..baseeventsinterface import BaseEventsInterface, _EventsData
from ....tools.events import (
_get_event_type_source_ids,
_resolve_detection_plan,
_validate_detection_configuration,
)
from ....tools.signal_processing import (
_condition_signal,
_detect_events,
_frames_to_seconds,
)
[docs]
class DoricCSVEventsInterface(BaseEventsInterface):
"""Convert discrete events from a Doric Neuroscience Studio CSV export to NWB.
A DoricStudio CSV export stores its channels under a grouped two-row header: the first row names
each channel's group (e.g. ``Analog In. | Ch.1``, ``Digital I/O | Ch.1``) and the second row names
each column (e.g. ``Time(s)``, ``DI/O-1``). The digital IO lines (the columns whose group is
``Digital I/O``) are sampled ``0``/``1`` traces on the shared ``Time(s)`` clock. Each such column is
a *signal*, and the events derived from it are set by ``detection_configuration``: one entry per
signal holding a list of detection specs, since a signal can yield more than one event type. Each
event type is written as its own ``pynwb.event.EventsTable`` into ``nwbfile.events``. By default
every line is read as a ``high_period`` (each rising edge is an event onset, its duration the span to
the next falling edge). A line that never toggles still yields its event type, written as a zero-row
table, since the type existed in the recording and nothing fired.
This reads the DoricStudio CSV export only; the ``.doric`` HDF5 layouts are handled by
:class:`.DoricEventsInterface`. The CSV export carries no session start time, so the user must
supply ``NWBFile/session_start_time`` via editable metadata.
"""
keywords = ("events", "Doric")
display_name = "DoricCSVEvents"
info = "Data Interface for converting discrete events (digital IO) from Doric Neuroscience Studio CSV exports."
associated_suffixes = ("csv",)
@validate_call
def __init__(
self,
file_path: FilePath,
*,
detection_configuration: dict | None = None,
metadata_key: str | None = None,
verbose: bool = False,
):
"""Initialize the DoricCSVEventsInterface.
Parameters
----------
file_path : FilePath
Path to the DoricStudio CSV export.
detection_configuration : dict, optional
Which digital lines to read and how, keyed by the line's ``signal_source_id`` (its column
name, e.g. ``{"DI/O-1": [{"signal_conditioning": {"binarize": "midpoint"}, "detection":
"high_period"}]}``). Each value is a **list** of
detection specs, one per event type derived from that line, since a line can yield more than
one. A spec's ``detection`` is one of ``"rising"`` / ``"falling"`` (a point event at each
edge) or ``"high_period"`` / ``"low_period"`` (a durative event, onset at one edge and
duration to the next opposite edge), and it is required. ``signal_conditioning`` is required
too and says how the signal becomes a line: a DoricStudio digital column is already
``0``/``1``, so it takes ``{"binarize": "midpoint"}``, whose cut falls strictly between the
two levels whatever they are. An optional
``event_name`` replaces the derived identifier and pins it against later edits. If None
(default), every digital line in the file is read as a ``high_period``, lossless for an
active-high line; use ``"low_period"`` for an active-low one. When given, only the named
lines are read.
metadata_key : str, optional
The key under ``metadata["Events"]`` that namespaces this interface's events metadata.
If None (default), ``"doric_events"`` is used.
verbose : bool, optional
Whether to print status messages, default = False.
"""
super().__init__(
file_path=file_path,
detection_configuration=detection_configuration,
verbose=verbose,
)
self.metadata_key = metadata_key or "doric_events"
self._time_column, digital_columns = self._discover_columns(self.source_data["file_path"])
# available_signals: signal_source_id (the column name, e.g. "DI/O-1") -> its {kind, column}
# descriptor. The header group "Digital I/O" makes every discovered signal a digital line, settled
# structurally with no data read, which is what lets the validator reject a bit carve on one.
# Whether a line fired is a property of this recording, not of intent, so a line that never
# toggles is still a (possibly empty) event type.
self._available_signals = {str(column[1]): {"kind": "line", "column": column} for column in digital_columns}
if detection_configuration is None:
# The default, used only when the caller passes none: read every discovered line as a
# "high_period", the lossless durative reading (onset at the rising edge, duration to the
# falling edge, for an active-high line). The "midpoint" cut is what a line takes: it falls
# strictly between the two levels whatever they are, so it needs no knowledge of the file.
detection_configuration = {
signal_source_id: [{"signal_conditioning": {"binarize": "midpoint"}, "detection": "high_period"}]
for signal_source_id in self._available_signals
}
# One construction-time check, on the default as well as on a caller-supplied configuration: the
# default is machine-built but its inputs are not, so it too can resolve two event types to the
# same identifier. Validation covers structure and identifier resolution (rules 4 and 5) alike.
_validate_detection_configuration(detection_configuration, self._available_signals)
self._detection_configuration = detection_configuration
[docs]
def get_event_type_source_ids(self) -> list[str]:
"""The event types the configuration resolves to, read from nothing."""
return _get_event_type_source_ids(self._detection_configuration)
@staticmethod
def _read_doric_csv(file_path):
"""Read the DoricStudio two-row-header CSV into a DataFrame with ``(group, name)`` MultiIndex columns.
DoricStudio writes a trailing comma on the header rows, so the header has one more field than the
data. Reading the two header rows and the data separately (each a consistent width) sidesteps
pandas' header/data length-mismatch warning; the MultiIndex is then trimmed to the data's column
count, dropping the phantom trailing column that comma leaves behind.
"""
import pandas as pd
header_rows = pd.read_csv(file_path, header=None, nrows=2, dtype=str) # the (group, name) header rows
data = pd.read_csv(file_path, header=None, skiprows=2) # data only -> no header/data length mismatch
width = data.shape[1]
data.columns = pd.MultiIndex.from_arrays([header_rows.iloc[0, :width], header_rows.iloc[1, :width]])
return data
@staticmethod
def _discover_columns(file_path):
"""Return ``(time_column, [digital_columns])`` from the CSV's grouped two-row header.
The columns are ``(group, name)`` pairs. The time column is the one whose name holds ``Time``
(the shared ``Time(s)`` clock); the digital lines are the columns whose group is ``Digital I/O``.
Analog columns (``Analog In.``/``Analog Out.``) and the trailing empty column are ignored. Each
digital column's name (e.g. ``DI/O-1``) is its ``event_type_source_id`` (identity-in-header).
"""
import pandas as pd
header_rows = pd.read_csv(file_path, header=None, nrows=2, dtype=str) # header only -> no length warning
columns = list(zip(header_rows.iloc[0], header_rows.iloc[1]))
time_columns = [column for column in columns if "time" in str(column[1]).lower()]
digital_columns = [column for column in columns if "digital" in str(column[0]).lower()]
time_column = time_columns[0] if time_columns else None
return time_column, digital_columns
def _get_events_data_dict(self) -> dict[str, _EventsData]:
"""Build the internal event representation by edge-detecting each digital line, cached.
Each selected digital line becomes one :class:`_EventsData` keyed by its ``event_type_source_id``
(the column name): its trace is read per the line's ``detection`` into onset frames and, for a
durative reading, offset frames. Both are then indexed into the shared ``Time(s)`` column, so a
duration is the elapsed clock time between the two edges rather than a frame count times an
assumed sampling period. A line that never toggles keeps its entry with empty timestamps, which
the writer renders as a zero-row table.
"""
if self._events_data_dict is not None:
return self._events_data_dict
dataframe = self._read_doric_csv(self.source_data["file_path"])
time = dataframe[self._time_column].to_numpy(dtype="float64")
# Built here rather than held on the interface: the configuration is the source of truth, and the
# plan is pure and cheap to rebuild. Grouped by signal, so a column is extracted once however
# many event types it yields.
detection_plan = _resolve_detection_plan(self._detection_configuration)
events_data_dict = {}
for signal_source_id, detection_specs in detection_plan.items():
column = self._available_signals[signal_source_id]["column"]
data = dataframe[column].to_numpy(dtype="float64")
for event_type_source_id, spec in detection_specs:
# A DoricStudio digital column is already 0/1, so its cut is the derived midpoint,
# which lands between the two levels and hands detection back the same line.
conditioned = _condition_signal(data, spec["signal_conditioning"])
onset_frames, offset_frames = _detect_events(conditioned, spec["detection"])
onsets, durations = _frames_to_seconds(onset_frames, offset_frames, time)
events_data_dict[event_type_source_id] = _EventsData(
event_type_source_id=event_type_source_id,
timestamps=onsets,
durations=durations,
)
self._events_data_dict = events_data_dict
return self._events_data_dict