MedpcInterface (#883)

Co-authored-by: Cody Baker <[email protected]>

CHANGELOG.md

@@ -12,6 +12,7 @@
 
 ### Features
 * Added docker image and tests for an automated Rclone configuration (with file stream passed via an environment variable). [PR #902](https://github.com/catalystneuro/neuroconv/pull/902)
+* Added MedPCInterface for operant behavioral output files. [PR #883](https://github.com/catalystneuro/neuroconv/pull/883)
 
 ### Bug fixes
 * Fixed the conversion option schema of a `SpikeGLXConverter` when used inside another `NWBConverter`. [PR #922](https://github.com/catalystneuro/neuroconv/pull/922)
@@ -48,7 +49,6 @@
 * Fixed bug causing overwrite of NWB GUIDE watermark. [PR #890](https://github.com/catalystneuro/neuroconv/pull/890)
 
 
-
 ## v0.4.9 (June 5, 2024)
 
 ### Deprecations

docs/api/interfaces.behavior.rst

@@ -28,3 +28,7 @@ SLEAP
 Video
 -----
 .. automodule:: neuroconv.datainterfaces.behavior.video.videodatainterface
+
+MedPC
+-----
+.. automodule:: neuroconv.datainterfaces.behavior.medpc.medpcdatainterface

docs/conversion_examples_gallery/behavior/medpc.rst

@@ -0,0 +1,86 @@
+MedPC data conversion
+---------------------
+
+MedPC output files contain information about operant behavior such as nose pokes and rewards.
+Install NeuroConv with the additional dependencies necessary for writing medpc behavioral data.
+
+.. code-block:: bash
+
+    pip install neuroconv[medpc]
+
+Convert MedPC output data to NWB using
+:py:class:`~.neuroconv.datainterfaces.behavior.medpc.medpcdatainterface.MedPCInterface`.
+
+.. code-block:: python
+
+    from datetime import datetime
+    from zoneinfo import ZoneInfo
+    from neuroconv.datainterfaces import MedPCInterface
+
+    # For this data interface we need to pass the output file from MedPC
+    file_path = f"{BEHAVIOR_DATA_PATH}/medpc/example_medpc_file_06_06_2024.txt"
+    # Change the folder_path to the appropriate location in your system
+    session_conditions = {"Start Date": "04/18/19", "Start Time": "10:41:42"}
+    start_variable = "Start Date",
+    metadata_medpc_name_to_info_dict = dict(
+        "Start Date": {"name": "start_date", "is_array": False},
+        "Start Time": {"name": "start_time", "is_array": False},
+        "Subject": {"name": "subject", "is_array": False},
+        "Box": {"name": "box", "is_array": False},
+        "MSN": {"name": "MSN", "is_array": False},
+    )
+    interface = MedPCInterface(
+        file_path=file_path,
+        session_conditions=session_conditions,
+        start_variable=start_variable,
+        metadata_medpc_name_to_info_dict=metadata_medpc_name_to_info_dict
+    )
+
+    # Extract what metadata we can from the source file
+    metadata = interface.get_metadata()
+    # We add the time zone information, which is required by NWB
+    session_start_time = metadata["NWBFile"]["session_start_time"].replace(tzinfo=ZoneInfo("US/Pacific"))
+    metadata["NWBFile"].update(session_start_time=session_start_time)
+    metadata["MedPC"]["medpc_name_to_info_dict"] = {
+            "A": {"name": "left_nose_poke_times", "is_array": True},
+            "B": {"name": "left_reward_times", "is_array": True},
+            "C": {"name": "right_nose_poke_times", "is_array": True},
+            "D": {"name": "right_reward_times", "is_array": True},
+            "E": {"name": "duration_of_port_entry", "is_array": True},
+            "G": {"name": "port_entry_times", "is_array": True},
+            "H": {"name": "footshock_times", "is_array": True},
+    }
+    metadata["MedPC"]["Events"] = [
+        {
+            "name": "left_nose_poke_times",
+            "description": "Left nose poke times.",
+        },
+        {
+            "name": "left_reward_times",
+            "description": "Left reward times.",
+        },
+        {
+            "name": "right_nose_poke_times",
+            "description": "Right nose poke times.",
+        },
+        {
+            "name": "right_reward_times",
+            "description": "Right reward times.",
+        },
+        {
+            "name": "footshock_times",
+            "description": "Footshock times.",
+        },
+    ]
+    metadata["MedPC"]["IntervalSeries"] = [
+        {
+            "name": "reward_port_intervals",
+            "description": "Interval of time spent in reward port (1 is entry, -1 is exit).",
+            "onset_name": "port_entry_times",
+            "duration_name": "duration_of_port_entry",
+        },
+    ]
+
+    # Choose a path for saving the nwb file and run the conversion
+    nwbfile_path = f"{path_to_save_nwbfile}"  # This should be something like: "./saved_file.nwb"
+    interface.run_conversion(nwbfile_path=nwbfile_path, metadata=metadata)

docs/conversion_examples_gallery/index.rst

@@ -96,6 +96,7 @@ Behavior
     Neuralynx NVT <behavior/neuralynx_nvt>
     SLEAP <behavior/sleap>
     Videos <behavior/video>
+    MedPC <behavior/medpc>
 
 
 Text

src/neuroconv/datainterfaces/__init__.py

@@ -5,6 +5,7 @@
 from .behavior.lightningpose.lightningposedatainterface import (
     LightningPoseDataInterface,
 )
+from .behavior.medpc.medpcdatainterface import MedPCInterface
 from .behavior.miniscope.miniscopedatainterface import MiniscopeBehaviorInterface
 from .behavior.neuralynx.neuralynx_nvt_interface import NeuralynxNvtInterface
 from .behavior.sleap.sleapdatainterface import SLEAPInterface
@@ -155,6 +156,7 @@
     FicTracDataInterface,
     NeuralynxNvtInterface,
     LightningPoseDataInterface,
+    MedPCInterface,
     # Text
     CsvTimeIntervalsInterface,
     ExcelTimeIntervalsInterface,
@@ -191,5 +193,6 @@
         # Text
         CsvTimeIntervals=CsvTimeIntervalsInterface,
         ExcelTimeIntervals=ExcelTimeIntervalsInterface,
+        MedPC=MedPCInterface,
     ),
 )

src/neuroconv/datainterfaces/behavior/medpc/medpc_helpers.py

@@ -0,0 +1,174 @@
+import numpy as np
+
+from neuroconv.utils import FilePathType
+
+
+def get_medpc_variables(file_path: FilePathType, variable_names: list) -> dict:
+    """
+    Get the values of the given single-line variables from a MedPC file for all sessions in that file.
+
+    Parameters
+    ----------
+    file_path : FilePathType
+        The path to the MedPC file.
+    variable_names : list
+        The names of the variables to get the values of.
+
+    Returns
+    -------
+    dict
+        A dictionary with the variable names as keys and a list of variable values as values.
+    """
+    with open(file_path, "r") as f:
+        lines = f.readlines()
+    medpc_variables = {name: [] for name in variable_names}
+    for line in lines:
+        for variable_name in variable_names:
+            if line.startswith(variable_name):
+                medpc_variables[variable_name].append(line.split(":", maxsplit=1)[1].strip())
+    return medpc_variables
+
+
+def _get_session_lines(lines: list, session_conditions: dict, start_variable: str) -> list:
+    """
+    Get the lines for a session from a MedPC file.
+
+    Parameters
+    ----------
+    lines : list
+        The lines of the MedPC file.
+    session_conditions : dict
+        The conditions that define the session. The keys are the names of the single-line variables (ex. 'Start Date')
+        and the values are the values of those variables for the desired session (ex. '11/09/18').
+    start_variable : str
+        The name of the variable that starts the session (ex. 'Start Date').
+
+    Returns
+    -------
+    list
+        The lines for the session.
+
+    Raises
+    ------
+    ValueError
+        If the session with the given conditions could not be found.
+    ValueError
+        If the start variable of the session with the given conditions could not be found.
+
+    Notes
+    -----
+    If multiple sessions satisfy the session_conditions, the first session that meets the conditions will be returned.
+    """
+    session_condition_has_been_met = {name: False for name in session_conditions}
+    start_line, end_line = None, len(lines)
+    for i, line in enumerate(lines):
+        line = line.strip()
+        if line.startswith(f"{start_variable}:"):
+            start_line = i
+        for condition_name, condition_value in session_conditions.items():
+            if line == f"{condition_name}: {condition_value}":
+                session_condition_has_been_met[condition_name] = True
+        if line == "" and all(session_condition_has_been_met.values()):
+            end_line = i
+            break
+        elif line == "":
+            session_condition_has_been_met = {name: False for name in session_conditions}
+            start_line = None
+    if not all(session_condition_has_been_met.values()):
+        raise ValueError(f"Could not find the session with conditions {session_conditions}")
+    if start_line is None:
+        raise ValueError(
+            f"Could not find the start variable ({start_variable}) of the session with conditions {session_conditions}"
+        )
+    session_lines = lines[start_line:end_line]
+    return session_lines
+
+
+def read_medpc_file(
+    file_path: FilePathType,
+    medpc_name_to_info_dict: dict,
+    session_conditions: dict,
+    start_variable: str,
+) -> dict:
+    """
+    Read a raw MedPC text file into a dictionary.
+
+    Parameters
+    ----------
+    file_path : FilePathType
+        The path to the MedPC file.
+    medpc_name_to_info_dict : dict
+        A dictionary where the keys are the MedPC variable names and the values are dictionaries with the keys 'name' and
+        'is_array'. 'name' is the name of the variable in the output dictionary and 'is_array' is a boolean indicating
+        whether the variable is an array.  Ex. {'Start Date': {'name': 'start_date', 'is_array': False}}
+    session_conditions : dict
+        The conditions that define the session. The keys are the names of the single-line variables (ex. 'Start Date')
+        and the values are the values of those variables for the desired session (ex. '11/09/18').
+    start_variable : str
+        The name of the variable that starts the session (ex. 'Start Date').
+
+    Returns
+    -------
+    dict
+        A dictionary with the variable names as keys and the data extracted from medpc output are the values.
+
+    Raises
+    ------
+    ValueError
+        If the session with the given conditions could not be found.
+    """
+    with open(file_path, "r") as f:
+        lines = f.readlines()
+    session_lines = _get_session_lines(lines, session_conditions=session_conditions, start_variable=start_variable)
+
+    # Parse the session lines into a dictionary
+    session_dict = {}
+    for i, line in enumerate(session_lines):
+        line = line.rstrip()
+        if line.startswith("\\"):  # \\ indicates a commented line in the MedPC file
+            continue
+        assert ":" in line, f"Could not find ':' in line {repr(line)}"
+        split_line = line.split(":", maxsplit=1)
+        medpc_name, data = split_line
+        data = data.strip()
+        if "\t" in data:  # some sessions have a bunch of garbage after the last datum in the line separated by tabs
+            data = data.split("\t")[0]
+        if line.find(":") == 6:  # multiline variable
+            if medpc_name == "     0":  # first line of multiline variable
+                multiline_variable_name = session_lines[i - 1].split(":")[0]
+                if multiline_variable_name in medpc_name_to_info_dict:
+                    output_name = medpc_name_to_info_dict[multiline_variable_name]["name"]
+                    session_dict[output_name] = []
+            if multiline_variable_name not in medpc_name_to_info_dict:
+                continue
+            data = data.split(" ")
+            for datum in data:
+                datum = datum.strip()
+                if datum == "":
+                    continue
+                output_name = medpc_name_to_info_dict[multiline_variable_name]["name"]
+                session_dict[output_name].append(datum)
+
+        # single line variable
+        elif medpc_name in medpc_name_to_info_dict:
+            output_name = medpc_name_to_info_dict[medpc_name]["name"]
+            session_dict[output_name] = data
+
+    # Convert the data types
+    for info in medpc_name_to_info_dict.values():
+        output_name = info["name"]
+        is_array = info["is_array"]
+        if output_name in session_dict:
+            if is_array:
+                if session_dict[output_name] == "":
+                    session_dict[output_name] = np.array([], dtype=float)
+                elif type(session_dict[output_name]) == str:  # not a multiline variable
+                    raise ValueError(
+                        f"Expected {output_name} to be a multiline variable, but found a single line variable."
+                    )
+                else:
+                    session_dict[output_name] = np.array(session_dict[output_name], dtype=float)
+                    session_dict[output_name] = np.trim_zeros(
+                        session_dict[output_name], trim="b"
+                    )  # MEDPC adds extra zeros to the end of the array
+    return session_dict