diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/constantinople_lab_to_nwb/schierek_embargo_2024/schierek_embargo_2024_notes.md b/src/constantinople_lab_to_nwb/schierek_embargo_2024/schierek_embargo_2024_notes.md index bcfacf2..d0f1c1c 100644 --- a/src/constantinople_lab_to_nwb/schierek_embargo_2024/schierek_embargo_2024_notes.md +++ b/src/constantinople_lab_to_nwb/schierek_embargo_2024/schierek_embargo_2024_notes.md @@ -157,3 +157,35 @@ We are using this computed time shift to shift the ephys timestamps. The following UML diagram shows the mapping of source data to NWB. ![nwb mapping](schierek_embargo_2024_uml.png) + +## OpenEphys XML Channel Fixer + +This utility helps fix OpenEphys `settings.xml` files when channels are accidentally dropped by the OpenEphys GUI. +This issue can prevent proper data extraction using our extractors. The script identifies missing AP channels and adds +them back to the XML configuration with the correct electrode positions. + + +### Usage + +```python +from constantinople_to_nwb.utils import fix_settings_xml_missing_channels +import probeinterface +from pathlib import Path + +# Path to the settings.xml file +settings_path = Path("/path/to/your/Ephys Data/E003_2022-08-22_14-56-16/Record Node 103/settings.xml") + +# Fix the XML file +modified_path = fix_settings_xml_missing_channels( + settings_xml_file_path=settings_path, + # Optional: specify AP stream name to verify the file can be read with probeinterface + stream_name="Record Node 103#Neuropix-PXI-100.0" +) + +# Verify the channel configuration has been updated successfully +ap_stream_name = "Record Node 103#Neuropix-PXI-100.0" # The name of the AP recording stream +probe = probeinterface.read_openephys( + settings_file=settings_path, stream_name=ap_stream_name +) +assert len(probe.contact_ids) == 384 +``` diff --git a/src/constantinople_lab_to_nwb/utils/__init__.py b/src/constantinople_lab_to_nwb/utils/__init__.py index c4a1cd8..e2f0d0a 100644 --- a/src/constantinople_lab_to_nwb/utils/__init__.py +++ b/src/constantinople_lab_to_nwb/utils/__init__.py @@ -1 +1,2 @@ +from .fix_xml_openephys import fix_settings_xml_missing_channels from .get_subject_metadata import get_subject_metadata_from_rat_info_folder diff --git a/src/constantinople_lab_to_nwb/utils/fix_xml_openephys.py b/src/constantinople_lab_to_nwb/utils/fix_xml_openephys.py new file mode 100644 index 0000000..28b4bd1 --- /dev/null +++ b/src/constantinople_lab_to_nwb/utils/fix_xml_openephys.py @@ -0,0 +1,92 @@ +from pathlib import Path +from typing import Union + +import numpy as np +from lxml import etree + + +def fix_settings_xml_missing_channels( + settings_xml_file_path: Union[str, Path], + ap_stream_name: str = None, +): + """ + Modify OpenEphys settings file (settings.xml) to include missing AP channels and their electrode positions. + + This function: + 1. Loads an XML settings file + 2. Identifies missing AP channels + 3. Adds missing channels with appropriate X/Y positions based on detected patterns + 4. Overwrites the XML settings file with the updated configuration + 5. Optionally, verifies the result using probeinterface + + Args: + settings_xml_file_path (Union[str, Path]): Path to the input XML settings file + ap_stream_name (str): Name of the data stream for probeinterface verification (optional) + + Raises: + FileNotFoundError: If the input file doesn't exist + ValueError: If the XML structure is invalid or required tags are missing + AssertionError: If the final probe configuration is invalid + """ + settings_xml_file_path = Path(settings_xml_file_path) + if not settings_xml_file_path.exists(): + raise FileNotFoundError(f"Settings file not found: {settings_xml_file_path}") + + try: + tree = etree.parse(str(settings_xml_file_path)) + root = tree.getroot() + + # Locate the , , and tags + all_channels_from_channel_info = root.xpath(".//CHANNEL_INFO/CHANNEL") + # Extract AP channels + all_ap_channels = set( + sorted( + int(channel.attrib["number"]) + for channel in all_channels_from_channel_info + if "AP" in channel.attrib["name"] + ) + ) + + channels_tag = root.xpath(".//CHANNELS")[0] + electrode_xpos_tag = root.xpath(".//ELECTRODE_XPOS")[0] + electrode_ypos_tag = root.xpath(".//ELECTRODE_YPOS")[0] + + # Extract channel numbers from the attributes + channel_numbers = sorted(int(attr[2:]) for attr in channels_tag.attrib.keys()) + + # Identify missing channels + missing_channels = sorted(all_ap_channels - set(channel_numbers)) + + # Detect repeating pattern in values + xpos_values = [int(value) for value in electrode_xpos_tag.attrib.values()] + pattern_length = next( + (i for i in range(1, len(xpos_values) // 2) if xpos_values[:i] == xpos_values[i : 2 * i]), len(xpos_values) + ) + xpos_pattern = xpos_values[:pattern_length] + + # Detect repeating pattern in values + ypos_values = [int(value) for value in electrode_ypos_tag.attrib.values()] + ypos_step = np.unique(np.diff(sorted(set(ypos_values))))[0] + + # Insert missing channels + for missing_channel in missing_channels: + channels_tag.set(f"CH{missing_channel}", "0") + pattern_value = xpos_pattern[missing_channel % pattern_length] + electrode_xpos_tag.set(f"CH{missing_channel}", str(pattern_value)) + + pattern_value_ypos = (missing_channel // 2) * ypos_step # 20 + electrode_ypos_tag.set(f"CH{missing_channel}", str(pattern_value_ypos)) + + # Save the updated XML to a new file + tree.write(str(settings_xml_file_path), pretty_print=True) + + except Exception as e: + print(f"Failed to modify settings XML: {str(e)}") + raise + + if ap_stream_name is not None: + import probeinterface + + probe = probeinterface.read_openephys(settings_file=settings_xml_file_path, stream_name=ap_stream_name) + assert len(probe.contact_ids) == 384 + print("Probe configuration verified successfully.")