add interface

src/constantinople_lab_to_nwb/fiber_photometry/interfaces/doric_fiber_photometry_interface.py

@@ -0,0 +1,152 @@
+from pathlib import Path
+from typing import Union, Literal
+
+import h5py
+import numpy as np
+from ndx_fiber_photometry import FiberPhotometryResponseSeries
+from neuroconv import BaseTemporalAlignmentInterface
+from neuroconv.tools import get_module
+from neuroconv.utils import load_dict_from_file, dict_deep_update
+from pynwb import NWBFile
+
+from constantinople_lab_to_nwb.fiber_photometry.utils import add_fiber_photometry_table, add_fiber_photometry_devices
+
+
+class DoricFiberPhotometryInterface(BaseTemporalAlignmentInterface):
+    """Behavior interface for fiber photometry conversion"""
+
+    def __init__(
+        self,
+        file_path: Union[str, Path],
+        stream_name: str,
+        verbose: bool = True,
+    ):
+        self._timestamps = None
+        super().__init__(file_path=file_path, stream_name=stream_name, verbose=verbose)
+
+    def load(self, stream_name: str):
+        file_path = Path(self.source_data["file_path"])
+        # check if suffix is .doric
+        if file_path.suffix != ".doric":
+            raise ValueError(f"File '{file_path}' is not a .doric file.")
+
+        channel_group = h5py.File(file_path, mode="r")[stream_name]
+        if "Time" not in channel_group.keys():
+            raise ValueError(f"Time not found in '{stream_name}'.")
+        return channel_group
+
+    def get_original_timestamps(self) -> np.ndarray:
+        channel_group = self.load(stream_name=self.source_data["stream_name"])
+        return channel_group["Time"][:]
+
+    def get_timestamps(self, stub_test: bool = False) -> np.ndarray:
+        timestamps = self._timestamps if self._timestamps is not None else self.get_original_timestamps()
+        if stub_test:
+            return timestamps[:100]
+        return timestamps
+
+    def set_aligned_timestamps(self, aligned_timestamps: np.ndarray) -> None:
+        self._timestamps = np.array(aligned_timestamps)
+
+    def add_to_nwbfile(
+        self,
+        nwbfile: NWBFile,
+        metadata: dict,
+        fiber_photometry_series_name: str,
+        parent_container: Literal["acquisition", "processing/ophys"] = "acquisition",
+        stub_test: bool = False,
+    ) -> None:
+
+        add_fiber_photometry_devices(nwbfile=nwbfile, metadata=metadata)
+
+        fiber_photometry_metadata = metadata["Ophys"]["FiberPhotometry"]
+        traces_metadata = fiber_photometry_metadata["FiberPhotometryResponseSeries"]
+        trace_metadata = next(
+            (trace for trace in traces_metadata if trace["name"] == fiber_photometry_series_name),
+            None,
+        )
+        if trace_metadata is None:
+            raise ValueError(f"Trace metadata for '{fiber_photometry_series_name}' not found.")
+
+        add_fiber_photometry_table(nwbfile=nwbfile, metadata=metadata)
+        fiber_photometry_table = nwbfile.lab_meta_data["FiberPhotometry"].fiber_photometry_table
+
+        row_indices = trace_metadata["fiber_photometry_table_region"]
+        device_fields = [
+            "optical_fiber",
+            "excitation_source",
+            "photodetector",
+            "dichroic_mirror",
+            "indicator",
+            "excitation_filter",
+            "emission_filter",
+        ]
+        for row_index in row_indices:
+            row_metadata = fiber_photometry_metadata["FiberPhotometryTable"]["rows"][row_index]
+            row_data = {field: nwbfile.devices[row_metadata[field]] for field in device_fields if field in row_metadata}
+            row_data["location"] = row_metadata["location"]
+            if "coordinates" in row_metadata:
+                row_data["coordinates"] = row_metadata["coordinates"]
+            if "commanded_voltage_series" in row_metadata:
+                row_data["commanded_voltage_series"] = nwbfile.acquisition[row_metadata["commanded_voltage_series"]]
+            fiber_photometry_table.add_row(**row_data)
+
+        stream_name = trace_metadata["stream_name"]
+        stream_indices = trace_metadata["stream_indices"]
+
+        traces_to_add = []
+        data = self.load(stream_name=stream_name)
+        channel_names = list(data.keys())
+        for stream_index in stream_indices:
+            trace = data[channel_names[stream_index]]
+            trace = trace[:100] if stub_test else trace[:]
+            traces_to_add.append(trace)
+
+        traces = np.vstack(traces_to_add).T
+
+        fiber_photometry_table_region = fiber_photometry_table.create_fiber_photometry_table_region(
+            description=trace_metadata["fiber_photometry_table_region_description"],
+            region=trace_metadata["fiber_photometry_table_region"],
+        )
+
+        # Get the timing information
+        timestamps = self.get_timestamps(stub_test=stub_test)
+
+        fiber_photometry_response_series = FiberPhotometryResponseSeries(
+            name=trace_metadata["name"],
+            description=trace_metadata["description"],
+            data=traces,
+            unit=trace_metadata["unit"],
+            fiber_photometry_table_region=fiber_photometry_table_region,
+            timestamps=timestamps,
+        )
+
+        if parent_container == "acquisition":
+            nwbfile.add_acquisition(fiber_photometry_response_series)
+        elif parent_container == "processing/ophys":
+            ophys_module = get_module(
+                nwbfile,
+                name="ophys",
+                description="Contains the processed fiber photometry data.",
+            )
+            ophys_module.add(fiber_photometry_response_series)
+
+
+interface = DoricFiberPhotometryInterface(
+    file_path="/Volumes/T9/Constantinople/Preprocessed_data/J100/Raw/J100_rDAgAChDLSDMS_20240819_HJJ_0001.doric",
+    stream_name="/DataAcquisition/FPConsole/Signals/Series0001/AnalogIn/",
+)
+
+metadata = interface.get_metadata()
+metadata["NWBFile"]["session_start_time"] = "2024-08-19T00:00:00"
+
+# Update default metadata with the editable in the corresponding yaml file
+editable_metadata_path = Path(__file__).parent.parent / "metadata" / "fiber_photometry_metadata.yaml"
+editable_metadata = load_dict_from_file(editable_metadata_path)
+metadata = dict_deep_update(metadata, editable_metadata)
+
+interface.run_conversion(
+    nwbfile_path="/Volumes/T9/Constantinople/nwbfiles/test.nwb",
+    metadata=metadata,
+    fiber_photometry_series_name="fiber_photometry_response_series_green",
+)

src/constantinople_lab_to_nwb/fiber_photometry/metadata/fiber_photometry_metadata.yaml

@@ -0,0 +1,144 @@
+# The metadata of the fiber photometry setup
+# papers with similar setup: https://www.biorxiv.org/content/10.1101/2023.12.09.570945v1.full.pdf
+#                            https://www.biorxiv.org/content/10.1101/2024.05.03.592444v1.full.pdf
+Ophys:
+  FiberPhotometry:
+    OpticalFibers:
+      - name: optical_fiber
+        description: The optical fibers (Thor labs) with 400 μm core, 0.5 NA fiber optics were implanted over the injection site.
+        manufacturer: Thor labs
+        # model: unknown
+        numerical_aperture: 0.5
+        core_diameter_in_um: 400.0
+    ExcitationSources: # Can't find any information in the referred papers
+      - name: excitation_source
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+        illumination_type: LED
+        excitation_wavelength_in_nm: 0.0 # TBD
+      - name: excitation_source_isosbestic_control
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+        illumination_type: LED
+        excitation_wavelength_in_nm: 0.0 # TBD
+    Photodetectors: # Can't find any information in the referred papers
+      - name: photodetector
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+        detector_type: photodiode
+        detected_wavelength_in_nm: 0.0 # TBD
+        # gain: # TBD
+    BandOpticalFilters: # Can't find any information in the referred papers
+      - name: emission_filter
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+        center_wavelength_in_nm: 0.0 # TBD
+        bandwidth_in_nm: 0.0 # TBD
+        filter_type: Bandpass
+      - name: excitation_filter
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+        center_wavelength_in_nm: 0.0 # TBD
+        bandwidth_in_nm: 0.0 # TBD
+        filter_type: Bandpass
+      - name: isosbestic_excitation_filter
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+        center_wavelength_in_nm: 0.0 # TBD
+        bandwidth_in_nm: 0.0 # TBD
+        filter_type: Bandpass
+    DichroicMirrors: # Can't find any information in the referred papers
+      - name: dichroic_mirror
+        description: TBD
+        manufacturer: Doric Lenses
+        model: TBD
+    Indicators:
+      - name: grab_da_dms
+        description: "To measure dopamine activity, AAV9-hsyn-GRAB DA2h (AddGene #140554) was injected into the DMS (AP: -0.4 mm, ML: ±3.5 mm, DV: -4.6 mm from the skull surface at bregma at a 10 degrees angle)."
+        manufacturer: Addgene
+        label: AAV9-hSyn-GRAB
+        injection_location: DMS
+        injection_coordinates_in_mm: [-0.4, -3.5, -4.6]
+      - name: grab_ach_dms
+        description: "To measure acetylcholine activity, AAV9-hSyn-ACh4.3 (WZ Biosciences #YL10002) was injected into the DMS (AP: 1.7 mm, ML: ±2.8 mm, DV: -4.3 mm from the skull surface at bregma at a 10 degrees angle)."
+        manufacturer: WZ Biosciences
+        label: AAV9-hSyn-ACh4.3
+        injection_location: DMS
+        injection_coordinates_in_mm: [1.7, 2.8, -4.3]
+      - name: grab_da_nacc
+        description: "To measure dopamine activity, AAV9-hsyn-GRAB DA2h (AddGene #140554) was injected into the NAcc (AP: 1.3 mm, ML: 1.65 mm, DV:-6.9 mm with an 8-10 degrees angle from the midline for bilateral implants)."
+        manufacturer: Addgene
+        label: AAV9-hSyn-GRAB
+        injection_location: NAcc
+        injection_coordinates_in_mm: [1.3, 1.65, -6.9]
+      - name: mcherry
+        description: "The control fluorophore was mCherry (AAV9-CB7-CI-mCherry-WPRE-RBG, AddGene #105544)."
+        label: AAV1-CB7-CI-mCherry
+        manufacturer: Addgene
+    FiberPhotometryTable:
+      name: fiber_photometry_table
+      description: TBD
+      rows:
+        - name: 0
+          location: DMS
+          # coordinates: [0.8, 1.5, 2.8]
+          # commanded_voltage_series: commanded_voltage_series_dms_calcium_signal
+          indicator: grab_da_dms
+          optical_fiber: optical_fiber
+          excitation_source: excitation_source
+          photodetector: photodetector
+          excitation_filter: excitation_filter
+          emission_filter: emission_filter
+          dichroic_mirror: dichroic_mirror
+        - name: 1
+          location: DLS
+          # coordinates: [0.8, 1.5, 2.8]
+          indicator: grab_da_dms
+          optical_fiber: optical_fiber
+          excitation_source: excitation_source
+          photodetector: photodetector
+          excitation_filter: excitation_filter
+          emission_filter: emission_filter
+          dichroic_mirror: dichroic_mirror
+        - name: 2
+          location: DMS
+          # coordinates: [0.8, 1.5, 2.8]
+          # commanded_voltage_series: commanded_voltage_series_dms_isosbestic_control
+          indicator: mcherry
+          optical_fiber: optical_fiber
+          excitation_source: excitation_source_isosbestic_control
+          photodetector: photodetector
+          excitation_filter: isosbestic_excitation_filter
+          emission_filter: emission_filter
+          dichroic_mirror: dichroic_mirror
+        - name: 3
+          location: DLS
+            # coordinates: [0.8, 1.5, 2.8]
+          indicator: mcherry
+          optical_fiber: optical_fiber
+          excitation_source: excitation_source_isosbestic_control
+          photodetector: photodetector
+          excitation_filter: isosbestic_excitation_filter
+          emission_filter: emission_filter
+          dichroic_mirror: dichroic_mirror
+    FiberPhotometryResponseSeries:
+      - name: fiber_photometry_response_series_green
+        description: The raw fluorescence signal # TBD
+        stream_name: "/DataAcquisition/FPConsole/Signals/Series0001/AnalogIn/"
+        stream_indices: [1, 3]
+        unit: a.u.
+        fiber_photometry_table_region: [0, 1]
+        fiber_photometry_table_region_description: The region of the FiberPhotometryTable corresponding to the raw fluorescence signal.
+      - name: fiber_photometry_response_series_red
+        description: The raw fluorescence signal # TBD
+        stream_name: "/DataAcquisition/FPConsole/Signals/Series0001/AnalogIn/"
+        stream_indices: [0, 2]
+        unit: a.u.
+        fiber_photometry_table_region: [2, 3]
+        fiber_photometry_table_region_description: The region of the FiberPhotometryTable corresponding to the raw fluorescence signal.

src/constantinople_lab_to_nwb/fiber_photometry/utils/__init__.py

@@ -0,0 +1 @@
+from .add_fiber_photometry import add_fiber_photometry_devices, add_fiber_photometry_table

src/constantinople_lab_to_nwb/fiber_photometry/utils/add_fiber_photometry.py

@@ -0,0 +1,48 @@
+from ndx_fiber_photometry import FiberPhotometryTable, FiberPhotometry
+from neuroconv.tools.fiber_photometry import add_fiber_photometry_device
+from pynwb import NWBFile
+
+
+def add_fiber_photometry_devices(nwbfile: NWBFile, metadata: dict):
+    fiber_photometry_metadata = metadata["Ophys"]["FiberPhotometry"]
+    # Add Devices
+    device_types = [
+        "OpticalFiber",
+        "ExcitationSource",
+        "Photodetector",
+        "BandOpticalFilter",
+        "EdgeOpticalFilter",
+        "DichroicMirror",
+        "Indicator",
+    ]
+    for device_type in device_types:
+        devices_metadata = fiber_photometry_metadata.get(device_type + "s", [])
+        for device_metadata in devices_metadata:
+            add_fiber_photometry_device(
+                nwbfile=nwbfile,
+                device_metadata=device_metadata,
+                device_type=device_type,
+            )
+
+
+def add_fiber_photometry_table(nwbfile: NWBFile, metadata: dict):
+    fiber_photometry_metadata = metadata["Ophys"]["FiberPhotometry"]
+    fiber_photometry_table_metadata = fiber_photometry_metadata["FiberPhotometryTable"]
+
+    if "FiberPhotometry" in nwbfile.lab_meta_data:
+        return
+
+    fiber_photometry_table = FiberPhotometryTable(
+        name=fiber_photometry_table_metadata["name"],
+        description=fiber_photometry_table_metadata["description"],
+    )
+    # fiber_photometry_table.add_column(
+    #     name="additional_column_name",
+    #     description="additional_column_description",
+    # )
+
+    fiber_photometry_lab_meta_data = FiberPhotometry(
+        name="FiberPhotometry",
+        fiber_photometry_table=fiber_photometry_table,
+    )
+    nwbfile.add_lab_meta_data(fiber_photometry_lab_meta_data)