Generalize code to retrieve conversion parameters for all example data

src/cai_lab_to_nwb/zaki_2024/utils/__init__.py

@@ -1 +1,13 @@
 from .edf_slicing import get_session_slicing_time_range, get_session_run_time
+from .source_data_path_resolver import (
+    get_session_times_df,
+    get_experiment_dir_path,
+    get_edf_file_path,
+    get_video_file_path,
+    get_date_str_from_experiment_dir_path,
+    get_sleep_classification_file_path,
+    get_imaging_folder_path,
+    get_miniscope_folder_path,
+    get_freezing_output_file_path,
+)
+from .define_conversion_parameters import update_conversion_parameters_yaml

src/cai_lab_to_nwb/zaki_2024/utils/conversion_parameters.yaml

@@ -0,0 +1,21 @@
+Ca_EEG2-1:
+  Ca_EEG2-1_FC:
+    date_str: null
+    edf_file_path: null
+    experiment_dir_path: D:\Ca_EEG_Experiment\Ca_EEG2-1\Ca_EEG2-1_Sessions\Ca_EEG2-1_FC
+    freezing_output_file_path: D:\Ca_EEG_Experiment\Ca_EEG2-1\Ca_EEG2-1_Sessions\Ca_EEG2-1_FC\Ca_EEG2-1_FC_FreezingOutput.csv
+    imaging_folder_path: D:\Ca_EEG_Experiment\Ca_EEG2-1\Ca_EEG2-1_Sessions\Ca_EEG2-1_FC\10_11_24
+    minian_folder_path: D:\Ca_EEG_Calcium\Ca_EEG2-1\Ca_EEG2-1_FC\minian
+    output_dir_path: D:\cai_lab_conversion_nwb
+    session_id: Ca_EEG2-1_FC
+    shock_stimulus:
+      shock_amplitude: 1.5
+      shock_duration: 2.0
+      shock_times:
+      - 120.0
+      - 180.0
+      - 240.0
+    sleep_classification_file_path: null
+    subject_id: Ca_EEG2-1
+    time_str: '10_11_24'
+    video_file_path: D:\Ca_EEG_Experiment\Ca_EEG2-1\Ca_EEG2-1_Sessions\Ca_EEG2-1_FC\Ca_EEG2-1_FC.wmv

src/cai_lab_to_nwb/zaki_2024/utils/define_conversion_parameters.py

@@ -0,0 +1,110 @@
+import re
+import yaml
+
+from source_data_path_resolver import *
+
+
+def update_conversion_parameters_yaml(
+    subject_id: str,
+    data_dir_path: Union[str, Path],
+    output_dir_path: Union[str, Path],
+    experiment_design_file_path: Union[str, Path],
+    session_types: list = (),
+):
+    """
+    Update a YAML file with parameters required for session-to-NWB conversion.
+
+    Parameters:
+    -----------
+    subject_id : str
+        The ID of the subject.
+    data_dir_path : Union[str, Path]
+        Path to the base data directory.
+    output_dir_path : Union[str, Path]
+        Path to the output directory for NWB files.
+    experiment_design_file_path : Union[str, Path]
+        Path to the experiment design file.
+    session_types : list, optional
+        List of session types to process. Defaults to an empty list.
+
+    Returns:
+    --------
+    None
+    """
+    yaml_file_path = Path(__file__).parent / "conversion_parameters.yaml"
+    subjects_df = pd.read_excel(experiment_design_file_path)
+    session_times_df = get_session_times_df(
+        subject_id=subject_id, data_dir_path=data_dir_path, session_types=session_types
+    )
+    for session_type in session_times_df["Session"]:
+        session_id = subject_id + "_" + session_type
+        session_row = session_times_df[session_times_df["Session"] == session_type].iloc[0]
+        date_str = session_row["Date"]
+        time_str = session_row["Time"]
+        experiment_dir_path = get_experiment_dir_path(subject_id, session_id, data_dir_path)
+        if "Offline" in session_id:
+            if date_str is None:
+                date_str = get_date_str_from_experiment_dir_path(experiment_dir_path=experiment_dir_path)
+            edf_file_path = get_edf_file_path(subject_id, date_str, data_dir_path)
+            sleep_classification_file_path = get_sleep_classification_file_path(subject_id, session_id, data_dir_path)
+            video_file_path = None
+            freezing_output_file_path = None
+            shock_stimulus = None
+        else:
+            edf_file_path = None
+            sleep_classification_file_path = None
+            video_file_path = get_video_file_path(subject_id, session_id, data_dir_path)
+            freezing_output_file_path = get_freezing_output_file_path(subject_id, session_id, data_dir_path)
+            if session_type == "FC":
+                shock_amplitude = subjects_df["Amplitude"][subjects_df["Mouse"] == subject_id].to_numpy()[0]
+                shock_amplitude = float(re.findall(r"[-+]?\d*\.\d+|\d+", shock_amplitude)[0])
+                shock_stimulus = dict(
+                    shock_times=[120.0, 180.0, 240.0], shock_amplitude=shock_amplitude, shock_duration=2.0
+                )
+            else:
+                shock_stimulus = None
+        imaging_folder_path = get_imaging_folder_path(subject_id, session_id, data_dir_path, time_str, date_str)
+        minian_folder_path = get_miniscope_folder_path(subject_id, session_id, data_dir_path)
+        session_to_nwb_kwargs_per_session = {
+            session_id: {
+                "output_dir_path": str(output_dir_path),
+                "subject_id": subject_id,
+                "session_id": session_id,
+                "date_str": date_str,
+                "time_str": time_str,
+                "experiment_dir_path": str(experiment_dir_path),
+                "imaging_folder_path": str(imaging_folder_path) if imaging_folder_path else None,
+                "minian_folder_path": str(minian_folder_path) if minian_folder_path else None,
+                "video_file_path": str(video_file_path) if video_file_path else None,
+                "freezing_output_file_path": str(freezing_output_file_path) if freezing_output_file_path else None,
+                "edf_file_path": str(edf_file_path) if edf_file_path else None,
+                "sleep_classification_file_path": (
+                    str(sleep_classification_file_path) if sleep_classification_file_path else None
+                ),
+                "shock_stimulus": shock_stimulus,
+            }
+        }
+
+        try:
+            with open(yaml_file_path, "r") as file:
+                yaml_content = yaml.safe_load(file) or {}
+        except FileNotFoundError:
+            yaml_content = {}
+
+        if subject_id not in yaml_content:
+            yaml_content[subject_id] = {}
+
+        yaml_content[subject_id].update(session_to_nwb_kwargs_per_session)
+
+        with open(yaml_file_path, "w") as file:
+            yaml.dump(yaml_content, file, default_flow_style=False)
+    return
+
+
+if __name__ == "__main__":
+    update_conversion_parameters_yaml(
+        subject_id="Ca_EEG2-1",
+        data_dir_path=Path("D:/"),
+        output_dir_path=Path("D:/cai_lab_conversion_nwb/"),
+        experiment_design_file_path=Path("D:/Ca_EEG_Design.xlsx"),
+    )

src/cai_lab_to_nwb/zaki_2024/utils/edf_slicing.py

@@ -42,6 +42,36 @@ def get_session_slicing_time_range(miniscope_metadata_json: Union[str, Path], ti
 
 
 def get_session_run_time(txt_file_path: Union[str, Path]):
+    """
+    Reads a text file, extracts the "Run Time" information, and converts it to seconds.
+
+    Parameters:
+    ----------
+    txt_file_path : Union[str, Path]
+        The file path to the text file containing the "Run Time" information.
+
+    Returns:
+    -------
+    float
+        The run time in seconds.
+
+    Raises:
+    ------
+    FileNotFoundError
+        If the file specified by `txt_file_path` does not exist.
+    ValueError
+        If the "Run Time" information is not found or improperly formatted in the file.
+
+    Notes:
+    -----
+    - The "Run Time" line in the file should follow the format: `Run Time : HH:MM:SS`.
+    - If the "Run Time" information is missing, an error message is printed, and no value is returned.
+
+    Example:
+    -------
+    If the text file contains the line `Run Time : 01:30:45`, the function will return `5445.0` (seconds).
+
+    """
     import re
 
     try: