Repository: catalystneuro/neuroconv

Issue:
[New Format]: Lightning Pose
### What format would you like to see added to NeuroConv?

Support writing `PoseEstimationSeries` from [Lightning Pose](https://github.com/danbider/lightning-pose/tree/main) output.

- add helper method `add_pose_estimation_series()` in `neuroconv.tools` which uses [ndx-pose](https://github.com/rly/ndx-pose)
- add LightningPoseInterface in `neuroconv.datainterfaces.behavior`
- extract metadata from csv: scorer, key point names
- extract timestamps from video (should also be able to override)

Example data: https://github.com/danbider/lightning-pose/tree/main/data/mirror-mouse-example



### Does the format have any documentation?

https://github.com/danbider/lightning-pose/blob/main/docs/README.md

### Existing APIs for format

_No response_

### Do you have any example files you are willing to share?

Example data: https://github.com/danbider/lightning-pose/tree/main/data/mirror-mouse-example

### Do you have any interest in helping implement the feature?

Yes.

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/catalystneuro/neuroconv/blob/main/.github/CODE_OF_CONDUCT.rst)
- [X] Have you ensured this bug was not already [reported](https://github.com/catalystneuro/neuroconv/issues)?

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 88fae63d..b5aa91f2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@
 ### Features
 * Changed the `Suite2pSegmentationInterface` to support multiple plane segmentation outputs. The interface now has a `plane_name` and `channel_name` arguments to determine which plane output and channel trace add to the NWBFile. [PR #601](https://github.com/catalystneuro/neuroconv/pull/601)
 * Added tool function `configure_datasets` for configuring all datasets of an in-memory `NWBFile` to be backend specific. [PR #571](https://github.com/catalystneuro/neuroconv/pull/571)
+* Added `LightningPoseConverter` to add pose estimation data and the original and the optional labeled video added as ImageSeries to NWB. [PR #633](https://github.com/catalystneuro/neuroconv/pull/633)
 
 ### Improvements
 * `nwbinspector` has been removed as a minimal dependency. It becomes an extra (optional) dependency with `neuroconv[dandi]`. [PR #672](https://github.com/catalystneuro/neuroconv/pull/672)
@@ -14,6 +15,7 @@
 * Added compression to `FicTracDataInterface` [PR #678](https://github.com/catalystneuro/neuroconv/pull/678)
 
 
+
 # v0.4.6 (November 30, 2023)
 
 ### Features
diff --git a/docs/api/interfaces.behavior.rst b/docs/api/interfaces.behavior.rst
index 95cdfa9a..a2aefddc 100644
--- a/docs/api/interfaces.behavior.rst
+++ b/docs/api/interfaces.behavior.rst
@@ -13,6 +13,10 @@ FicTrac
 -------
 .. automodule:: neuroconv.datainterfaces.behavior.fictrac.fictracdatainterface
 
+LightningPose
+-------------
+.. automodule:: neuroconv.datainterfaces.behavior.lightningpose.lightningposeconverter
+
 SLEAP
 -----
 .. automodule:: neuroconv.datainterfaces.behavior.sleap.sleapdatainterface
diff --git a/docs/conversion_examples_gallery/behavior/lightningpose.rst b/docs/conversion_examples_gallery/behavior/lightningpose.rst
new file mode 100644
index 00000000..b8c1a66e
--- /dev/null
+++ b/docs/conversion_examples_gallery/behavior/lightningpose.rst
@@ -0,0 +1,34 @@
+LightningPose data conversion
+--------------------------
+
+Install NeuroConv with the additional dependencies necessary for reading LightningPose data.
+
+.. code-block:: bash
+
+    pip install neuroconv[lightningpose]
+
+Convert LightningPose pose estimation data to NWB using :py:class:`~neuroconv.datainterfaces.behavior.lightningpose.lightningposeconverter.LightningPoseConverter`.
+
+.. code-block:: python
+
+    >>> from datetime import datetime
+    >>> from dateutil import tz
+    >>> from pathlib import Path
+    >>> from neuroconv.converters import LightningPoseConverter
+
+    >>> folder_path = BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds"
+    >>> file_path = str(folder_path / "test_vid.csv")
+    >>> original_video_file_path = str(folder_path / "test_vid.mp4")
+    >>> # The labeled video file path is optional
+    >>> labeled_video_file_path = str(folder_path / "labeled_videos/test_vid_labeled.mp4")
+
+    >>> converter = LightningPoseConverter(file_path=file_path, original_video_file_path=original_video_file_path, labeled_video_file_path=labeled_video_file_path, verbose=False)
+
+    >>> metadata = converter.get_metadata()
+    >>> # For data provenance we add the time zone information to the conversion
+    >>> session_start_time = metadata["NWBFile"]["session_start_time"]
+    >>> tzinfo = tz.gettz("US/Pacific")
+    >>> metadata["NWBFile"].update(session_start_time=session_start_time.replace(tzinfo=tzinfo))
+    >>>
+    >>> # Choose a path for saving the nwb file and run the conversion
+    >>> converter.run_conversion(nwbfile_path=path_to_save_nwbfile, metadata=metadata)
diff --git a/docs/conversion_examples_gallery/conversion_example_gallery.rst b/docs/conversion_examples_gallery/conversion_example_gallery.rst
index f60e7672..b6dd7c65 100644
--- a/docs/conversion_examples_gallery/conversion_example_gallery.rst
+++ b/docs/conversion_examples_gallery/conversion_example_gallery.rst
@@ -92,6 +92,7 @@ Behavior
     Audio <behavior/audio>
     DeepLabCut <behavior/deeplabcut>
     FicTrac <behavior/fictrac>
+    LightningPose <behavior/lightningpose>
     SLEAP <behavior/sleap>
     Videos <behavior/video>
 
diff --git a/docs/index.rst b/docs/index.rst
index 2ddb29f1..55b5bd4b 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -13,7 +13,7 @@ of proprietary formats to the Neurodata Without Borders (NWB) standard.
 
 Features:
 
-* Reads data from 41 popular neurophysiology data formats and writes to NWB using best practices.
+* Reads data from 42 popular neurophysiology data formats and writes to NWB using best practices.
 * Extracts relevant metadata from each format.
 * Handles large data volume by reading datasets piece-wise.
 * Minimizes the size of the NWB files by automatically applying chunking and lossless compression.
diff --git a/src/neuroconv/converters/__init__.py b/src/neuroconv/converters/__init__.py
index 8a1ec0ad..a47b67b2 100644
--- a/src/neuroconv/converters/__init__.py
+++ b/src/neuroconv/converters/__init__.py
@@ -1,3 +1,6 @@
+from ..datainterfaces.behavior.lightningpose.lightningposeconverter import (
+    LightningPoseConverter,
+)
 from ..datainterfaces.ecephys.spikeglx.spikeglxconverter import SpikeGLXConverterPipe
 from ..datainterfaces.ophys.brukertiff.brukertiffconverter import (
     BrukerTiffMultiPlaneConverter,
diff --git a/src/neuroconv/datainterfaces/__init__.py b/src/neuroconv/datainterfaces/__init__.py
index 4228e927..c173b203 100644
--- a/src/neuroconv/datainterfaces/__init__.py
+++ b/src/neuroconv/datainterfaces/__init__.py
@@ -2,6 +2,9 @@
 from .behavior.audio.audiointerface import AudioInterface
 from .behavior.deeplabcut.deeplabcutdatainterface import DeepLabCutInterface
 from .behavior.fictrac.fictracdatainterface import FicTracDataInterface
+from .behavior.lightningpose.lightningposedatainterface import (
+    LightningPoseDataInterface,
+)
 from .behavior.miniscope.miniscopedatainterface import MiniscopeBehaviorInterface
 from .behavior.neuralynx.neuralynx_nvt_interface import NeuralynxNvtInterface
 from .behavior.sleap.sleapdatainterface import SLEAPInterface
@@ -151,6 +154,7 @@ interface_list = [
     MiniscopeBehaviorInterface,
     FicTracDataInterface,
     NeuralynxNvtInterface,
+    LightningPoseDataInterface,
     # Text
     CsvTimeIntervalsInterface,
     ExcelTimeIntervalsInterface,
@@ -183,6 +187,7 @@ interfaces_by_category = dict(
         DeepLabCut=DeepLabCutInterface,
         SLEAP=SLEAPInterface,
         FicTrac=FicTracDataInterface,
+        LightningPose=LightningPoseDataInterface,
         # Text
         CsvTimeIntervals=CsvTimeIntervalsInterface,
         ExcelTimeIntervals=ExcelTimeIntervalsInterface,
diff --git a/src/neuroconv/datainterfaces/behavior/lightningpose/__init__.py b/src/neuroconv/datainterfaces/behavior/lightningpose/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposeconverter.py b/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposeconverter.py
new file mode 100644
index 00000000..6c1814c5
--- /dev/null
+++ b/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposeconverter.py
@@ -0,0 +1,201 @@
+from copy import deepcopy
+from typing import Optional
+
+from pynwb import NWBFile
+
+from neuroconv import NWBConverter
+from neuroconv.datainterfaces import LightningPoseDataInterface, VideoInterface
+from neuroconv.tools.nwb_helpers import make_or_load_nwbfile
+from neuroconv.utils import (
+    DeepDict,
+    FilePathType,
+    dict_deep_update,
+    get_schema_from_method_signature,
+)
+
+
+class LightningPoseConverter(NWBConverter):
+    """Primary conversion class for handling Lightning Pose data streams."""
+
+    @classmethod
+    def get_source_schema(cls):
+        return get_schema_from_method_signature(cls)
+
+    def __init__(
+        self,
+        file_path: FilePathType,
+        original_video_file_path: FilePathType,
+        labeled_video_file_path: Optional[FilePathType] = None,
+        image_series_original_video_name: Optional[str] = None,
+        image_series_labeled_video_name: Optional[str] = None,
+        verbose: bool = True,
+    ):
+        """
+        The converter for Lightning Pose format to convert the pose estimation data
+        along with the original and the optional labeled video added as ImageSeries to NWB.
+
+        Parameters
+        ----------
+        file_path : a string or a path
+            Path to the .csv file that contains the predictions from Lightning Pose.
+        original_video_file_path : a string or a path
+            Path to the original video file (.mp4).
+        labeled_video_file_path : a string or a path, optional
+            Path to the labeled video file (.mp4).
+        image_series_original_video_name: string, optional
+            The name of the ImageSeries to add for the original video.
+        image_series_labeled_video_name: string, optional
+            The name of the ImageSeries to add for the labeled video.
+        verbose : bool, default: True
+            controls verbosity. ``True`` by default.
+        """
+        self.verbose = verbose
+        self.data_interface_objects = dict(
+            OriginalVideo=VideoInterface(file_paths=[original_video_file_path]),
+            PoseEstimation=LightningPoseDataInterface(
+                file_path=file_path,
+                original_video_file_path=original_video_file_path,
+                labeled_video_file_path=labeled_video_file_path,
+            ),
+        )
+        self.original_video_name = image_series_original_video_name or "ImageSeriesOriginalVideo"
+        self.labeled_video_name = None
+        if labeled_video_file_path:
+            self.labeled_video_name = image_series_labeled_video_name or "ImageSeriesLabeledVideo"
+            self.data_interface_objects.update(dict(LabeledVideo=VideoInterface(file_paths=[labeled_video_file_path])))
+
+    def get_conversion_options_schema(self) -> dict:
+        conversion_options = self.data_interface_objects["PoseEstimation"].get_conversion_options_schema()
+        conversion_options = dict_deep_update(
+            conversion_options, self.data_interface_objects["OriginalVideo"].get_conversion_options_schema()
+        )
+
+        starting_frames = conversion_options["properties"].pop("starting_frames")
+        conversion_options["properties"].update(
+            dict(
+                starting_frames_original_videos=starting_frames,
+                starting_frames_labeled_videos=starting_frames,
+            )
+        )
+
+        return conversion_options
+
+    def get_metadata(self) -> DeepDict:
+        metadata = self.data_interface_objects["PoseEstimation"].get_metadata()
+        original_video_interface = self.data_interface_objects["OriginalVideo"]
+        original_videos_metadata = original_video_interface.get_metadata()
+        metadata = dict_deep_update(metadata, original_videos_metadata)
+
+        original_videos_metadata["Behavior"]["Videos"][0].update(
+            name=self.original_video_name,
+            description="The original video used for pose estimation.",
+        )
+
+        if "LabeledVideo" in self.data_interface_objects:
+            labeled_video_interface = self.data_interface_objects["LabeledVideo"]
+            labeled_videos_metadata = labeled_video_interface.get_metadata()
+            labeled_videos_metadata["Behavior"]["Videos"][0].update(
+                name=self.labeled_video_name,
+                description="The video recorded by camera with the pose estimation labels.",
+            )
+
+            metadata = dict_deep_update(metadata, labeled_videos_metadata)
+
+        return metadata
+
+    def add_to_nwbfile(
+        self,
+        nwbfile: NWBFile,
+        metadata: dict,
+        reference_frame: Optional[str] = None,
+        confidence_definition: Optional[str] = None,
+        external_mode: bool = True,
+        starting_frames_original_videos: Optional[list] = None,
+        starting_frames_labeled_videos: Optional[list] = None,
+        stub_test: bool = False,
+    ):
+        original_video_interface = self.data_interface_objects["OriginalVideo"]
+
+        original_video_metadata = next(
+            video_metadata
+            for video_metadata in metadata["Behavior"]["Videos"]
+            if video_metadata["name"] == self.original_video_name
+        )
+        if original_video_metadata is None:
+            raise ValueError(f"Metadata for '{self.original_video_name}' not found in metadata['Behavior']['Videos'].")
+        metadata_copy = deepcopy(metadata)
+        metadata_copy["Behavior"]["Videos"] = [original_video_metadata]
+        original_video_interface.add_to_nwbfile(
+            nwbfile=nwbfile,
+            metadata=metadata_copy,
+            stub_test=stub_test,
+            external_mode=external_mode,
+            starting_frames=starting_frames_original_videos,
+        )
+
+        if "LabeledVideo" in self.data_interface_objects:
+            labeled_video_interface = self.data_interface_objects["LabeledVideo"]
+            labeled_video_metadata = next(
+                video_metadata
+                for video_metadata in metadata["Behavior"]["Videos"]
+                if video_metadata["name"] == self.labeled_video_name
+            )
+            if labeled_video_metadata is None:
+                raise ValueError(
+                    f"Metadata for '{self.labeled_video_name}' not found in metadata['Behavior']['Videos']."
+                )
+            metadata_copy["Behavior"]["Videos"] = [labeled_video_metadata]
+            labeled_video_interface.add_to_nwbfile(
+                nwbfile=nwbfile,
+                metadata=metadata_copy,
+                stub_test=stub_test,
+                external_mode=external_mode,
+                starting_frames=starting_frames_labeled_videos,
+                module_name="behavior",
+            )
+
+        self.data_interface_objects["PoseEstimation"].add_to_nwbfile(
+            nwbfile=nwbfile,
+            metadata=metadata,
+            reference_frame=reference_frame,
+            confidence_definition=confidence_definition,
+            stub_test=stub_test,
+        )
+
+    def run_conversion(
+        self,
+        nwbfile_path: Optional[str] = None,
+        nwbfile: Optional[NWBFile] = None,
+        metadata: Optional[dict] = None,
+        overwrite: bool = False,
+        reference_frame: Optional[str] = None,
+        confidence_definition: Optional[str] = None,
+        external_mode: bool = True,
+        starting_frames_original_videos: Optional[list] = None,
+        starting_frames_labeled_videos: Optional[list] = None,
+        stub_test: bool = False,
+    ) -> None:
+        if metadata is None:
+            metadata = self.get_metadata()
+
+        self.validate_metadata(metadata=metadata)
+
+        self.temporally_align_data_interfaces()
+
+        with make_or_load_nwbfile(
+            nwbfile_path=nwbfile_path,
+            nwbfile=nwbfile,
+            metadata=metadata,
+            overwrite=overwrite,
+            verbose=self.verbose,
+        ) as nwbfile_out:
+            self.add_to_nwbfile(
+                nwbfile=nwbfile_out,
+                metadata=metadata,
+                reference_frame=reference_frame,
+                confidence_definition=confidence_definition,
+                external_mode=external_mode,
+                starting_frames_original_videos=starting_frames_original_videos,
+                starting_frames_labeled_videos=starting_frames_labeled_videos,
+                stub_test=stub_test,
+            )
diff --git a/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py b/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py
new file mode 100644
index 00000000..8eb50973
--- /dev/null
+++ b/src/neuroconv/datainterfaces/behavior/lightningpose/lightningposedatainterface.py
@@ -0,0 +1,263 @@
+import re
+from copy import deepcopy
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, Tuple
+
+import numpy as np
+from hdmf.backends.hdf5 import H5DataIO
+from pynwb import NWBFile
+
+from ....basetemporalalignmentinterface import BaseTemporalAlignmentInterface
+from ....tools import get_module
+from ....utils import (
+    DeepDict,
+    FilePathType,
+    calculate_regular_series_rate,
+    get_base_schema,
+)
+
+
+class LightningPoseDataInterface(BaseTemporalAlignmentInterface):
+    """Data interface for Lightning Pose datasets."""
+
+    def get_metadata_schema(self) -> dict:
+        metadata_schema = super().get_metadata_schema()
+        metadata_schema["properties"]["Behavior"] = get_base_schema(tag="Behavior")
+
+        metadata_schema["properties"]["Behavior"].update(
+            required=["PoseEstimation"],
+            properties=dict(
+                PoseEstimation=dict(
+                    type="object",
+                    required=["name"],
+                    properties=dict(
+                        name=dict(type="string", default="PoseEstimation"),
+                        description=dict(type="string"),
+                        scorer=dict(type="string"),
+                        source_software=dict(type="string", default="LightningPose"),
+                    ),
+                    patternProperties={
+                        "^(?!(name|description|scorer|source_software)$)[a-zA-Z0-9_]+$": dict(
+                            title="PoseEstimationSeries",
+                            type="object",
+                            properties=dict(name=dict(type="string"), description=dict(type="string")),
+                            minProperties=1,
+                            additionalProperties=False,
+                        )
+                    },
+                    minProperties=2,
+                    additionalProperties=False,
+                )
+            ),
+        )
+
+        return metadata_schema
+
+    def __init__(
+        self,
+        file_path: FilePathType,
+        original_video_file_path: FilePathType,
+        labeled_video_file_path: Optional[FilePathType] = None,
+        verbose: bool = True,
+    ):
+        """
+        Interface for writing pose estimation data from the Lightning Pose algorithm.
+
+        Parameters
+        ----------
+        file_path : a string or a path
+            Path to the .csv file that contains the predictions from Lightning Pose.
+        original_video_file_path : a string or a path
+            Path to the original video file (.mp4).
+        labeled_video_file_path : a string or a path, optional
+            Path to the labeled video file (.mp4).
+        verbose : bool, default: True
+            controls verbosity. ``True`` by default.
+        """
+        from neuroconv.datainterfaces.behavior.video.video_utils import (
+            VideoCaptureContext,
+        )
+
+        self._vc = VideoCaptureContext
+
+        self.file_path = Path(file_path)
+        assert self.file_path.exists(), f"The file '{self.file_path}' does not exist."
+        self.original_video_file_path = Path(original_video_file_path)
+        assert (
+            self.original_video_file_path.exists()
+        ), f"The original video file '{self.original_video_file_path}' does not exist."
+
+        super().__init__(
+            verbose,
+            file_path=file_path,
+            original_video_file_path=original_video_file_path,
+            labeled_video_file_path=labeled_video_file_path,
+        )
+
+        # dimension is width by height
+        self.dimension = self._get_original_video_shape()
+
+        pose_estimation_data = self._load_source_data()
+        _, self.scorer_name = pose_estimation_data.columns.get_level_values(0).drop_duplicates()
+        self.pose_estimation_data = pose_estimation_data[self.scorer_name]
+        self.keypoint_names = self.pose_estimation_data.columns.get_level_values(0).drop_duplicates().tolist()
+
+        self._times = None
+
+    def _load_source_data(self):
+        import pandas as pd
+
+        # The order of the header is "scorer", "bodyparts", "coords"
+        pose_estimation_data = pd.read_csv(self.file_path, header=[0, 1, 2])
+        return pose_estimation_data
+
+    def _get_original_video_shape(self) -> Tuple[int, int]:
+        with self._vc(file_path=str(self.original_video_file_path)) as video:
+            video_shape = video.get_frame_shape()
+        # image size of the original video is in height x width
+        return video_shape[0], video_shape[1]
+
+    def get_original_timestamps(self, stub_test: bool = False) -> np.ndarray:
+        max_frames = 10 if stub_test else None
+        with self._vc(file_path=str(self.original_video_file_path)) as video:
+            timestamps = video.get_video_timestamps(max_frames=max_frames)
+        return timestamps
+
+    def get_timestamps(self, stub_test: bool = False) -> np.ndarray:
+        max_frames = 10 if stub_test else None
+        if self._times is None:
+            return self.get_original_timestamps(stub_test=stub_test)
+
+        timestamps = self._times if not stub_test else self._times[:max_frames]
+        return timestamps
+
+    def set_aligned_timestamps(self, aligned_timestamps: np.ndarray):
+        self._times = aligned_timestamps
+
+    def get_metadata(self) -> DeepDict:
+        metadata = super().get_metadata()
+
+        # Update the session start time if folder structure is saved in the format: YYYY-MM-DD/HH-MM-SS
+        pattern = r"(?P<date_time>\d{4}-\d{2}-\d{2}/\d{2}-\d{2}-\d{2})"
+        # Convert the file path parts to a string with forward slashes
+        file_path = "/".join(self.file_path.parts)
+        match = re.search(pattern, file_path)
+        if match and "session_start_time" not in metadata["NWBFile"]:
+            datetime_str = match.group("date_time")
+            session_start_time = datetime.strptime(datetime_str, "%Y-%m-%d/%H-%M-%S")
+            metadata["NWBFile"].update(session_start_time=session_start_time)
+
+        metadata["Behavior"]["PoseEstimation"].update(
+            name="PoseEstimation",
+            description="Contains the pose estimation series for each keypoint.",
+            scorer=self.scorer_name,
+            source_software="LightningPose",
+        )
+        for keypoint_name in self.keypoint_names:
+            keypoint_name_without_spaces = keypoint_name.replace(" ", "")
+            pose_estimation_series_metadata = {
+                keypoint_name: dict(
+                    name=f"PoseEstimationSeries{keypoint_name_without_spaces}",
+                    description=f"The estimated position (x, y) of {keypoint_name} over time.",
+                )
+            }
+            metadata["Behavior"]["PoseEstimation"].update(pose_estimation_series_metadata)
+
+        return metadata
+
+    def add_to_nwbfile(
+        self,
+        nwbfile: NWBFile,
+        metadata: Optional[dict] = None,
+        reference_frame: Optional[str] = None,
+        confidence_definition: Optional[str] = None,
+        stub_test: Optional[bool] = False,
+    ) -> None:
+        """
+        Add the pose estimation data to the nwbfile.
+
+        Parameters
+        ----------
+        nwbfile : NWBFile
+            The nwbfile to which the pose estimation data is added.
+        metadata : dict, optional
+            The metadata for the pose estimation data.
+        reference_frame : str, optional
+            The description defining what the (0, 0) coordinate corresponds to.
+        confidence_definition : str, optional
+            The description of how the confidence was computed, e.g., 'Softmax output of the deep neural network'.
+        """
+        from ndx_pose import PoseEstimation, PoseEstimationSeries
+
+        metadata_copy = deepcopy(metadata)
+
+        # The parameters for the pose estimation container
+        pose_estimation_metadata = metadata_copy["Behavior"]["PoseEstimation"]
+
+        behavior = get_module(nwbfile, "behavior")
+
+        pose_estimation_name = pose_estimation_metadata["name"]
+        if pose_estimation_name in behavior.data_interfaces:
+            raise ValueError(f"The nwbfile already contains a data interface with the name '{pose_estimation_name}'.")
+
+        if "Videos" not in metadata_copy["Behavior"]:
+            original_video_name = str(self.original_video_file_path)
+        else:
+            original_video_name = metadata_copy["Behavior"]["Videos"][0]["name"]
+
+        pose_estimation_kwargs = dict(
+            name=pose_estimation_metadata["name"],
+            description=pose_estimation_metadata["description"],
+            source_software=pose_estimation_metadata["source_software"],
+            scorer=pose_estimation_metadata["scorer"],
+            original_videos=[original_video_name],
+            dimensions=[self.dimension],
+        )
+
+        pose_estimation_data = self.pose_estimation_data if not stub_test else self.pose_estimation_data.head(n=10)
+        timestamps = self.get_timestamps(stub_test=stub_test)
+        rate = calculate_regular_series_rate(series=timestamps)
+        if rate:
+            pose_estimation_series_kwargs = dict(rate=rate, starting_time=timestamps[0])
+        else:
+            assert len(timestamps) == len(
+                pose_estimation_data
+            ), f"The length of timestamps ({len(timestamps)}) and pose estimation data ({len(pose_estimation_data)}) must be equal."
+            pose_estimation_series_kwargs = dict(timestamps=H5DataIO(data=timestamps, compression="gzip"))
+
+        pose_estimation_series = []
+        for keypoint_name in self.keypoint_names:
+            pose_estimation_series_data = pose_estimation_data[keypoint_name]
+            keypoint_name_without_spaces = keypoint_name.replace(" ", "")
+
+            pose_estimation_series_kwargs.update(
+                name=pose_estimation_metadata[keypoint_name_without_spaces]["name"],
+                description=pose_estimation_metadata[keypoint_name_without_spaces]["description"],
+                data=pose_estimation_series_data[["x", "y"]].values,
+                confidence=pose_estimation_series_data["likelihood"].values,
+                reference_frame=reference_frame or "(0,0) is unknown.",
+                unit="px",
+            )
+
+            if confidence_definition:
+                pose_estimation_series_kwargs.update(confidence_definition=confidence_definition)
+
+            pose_estimation_series.append(PoseEstimationSeries(**pose_estimation_series_kwargs))
+
+        pose_estimation_kwargs.update(
+            pose_estimation_series=pose_estimation_series,
+        )
+
+        if self.source_data["labeled_video_file_path"]:
+            if "Videos" not in metadata_copy["Behavior"]:
+                labeled_video_name = str(self.source_data["labeled_video_file_path"])
+            else:
+                labeled_video_name = metadata_copy["Behavior"]["Videos"][1]["name"]
+
+            pose_estimation_kwargs.update(labeled_videos=[labeled_video_name])
+
+        # Create the container for pose estimation
+        pose_estimation = PoseEstimation(**pose_estimation_kwargs)
+
+        behavior.add(pose_estimation)
diff --git a/src/neuroconv/datainterfaces/behavior/lightningpose/requirements.txt b/src/neuroconv/datainterfaces/behavior/lightningpose/requirements.txt
new file mode 100644
index 00000000..73aa3208
--- /dev/null
+++ b/src/neuroconv/datainterfaces/behavior/lightningpose/requirements.txt
@@ -0,0 +1,2 @@
+ndx-pose==0.1.1
+neuroconv[video]


Test patch:
diff --git a/tests/test_on_data/test_behavior_interfaces.py b/tests/test_on_data/test_behavior_interfaces.py
index 12f9e90b..7b046dc7 100644
--- a/tests/test_on_data/test_behavior_interfaces.py
+++ b/tests/test_on_data/test_behavior_interfaces.py
@@ -3,11 +3,13 @@ from datetime import datetime, timezone
 from pathlib import Path
 
 import numpy as np
+import pandas as pd
 import sleap_io
 from hdmf.testing import TestCase
 from natsort import natsorted
 from ndx_miniscope import Miniscope
 from ndx_miniscope.utils import get_timestamps
+from ndx_pose import PoseEstimation, PoseEstimationSeries
 from numpy.testing import assert_array_equal
 from parameterized import param, parameterized
 from pynwb import NWBHDF5IO
@@ -17,6 +19,7 @@ from neuroconv import NWBConverter
 from neuroconv.datainterfaces import (
     DeepLabCutInterface,
     FicTracDataInterface,
+    LightningPoseDataInterface,
     MiniscopeBehaviorInterface,
     NeuralynxNvtInterface,
     SLEAPInterface,
@@ -28,6 +31,7 @@ from neuroconv.tools.testing.data_interface_mixins import (
     TemporalAlignmentMixin,
     VideoInterfaceMixin,
 )
+from neuroconv.utils import DeepDict
 
 try:
     from .setup_paths import BEHAVIOR_DATA_PATH, OPHYS_DATA_PATH, OUTPUT_PATH
@@ -35,6 +39,122 @@ except ImportError:
     from setup_paths import BEHAVIOR_DATA_PATH, OUTPUT_PATH
 
 
+class TestLightningPoseDataInterface(DataInterfaceTestMixin, TemporalAlignmentMixin, unittest.TestCase):
+    data_interface_cls = LightningPoseDataInterface
+    interface_kwargs = dict(
+        file_path=str(BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.csv"),
+        original_video_file_path=str(
+            BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.mp4"
+        ),
+    )
+    conversion_options = dict(reference_frame="(0,0) corresponds to the top left corner of the video.")
+    save_directory = OUTPUT_PATH
+
+    @classmethod
+    def setUpClass(cls):
+        cls.pose_estimation_name = "PoseEstimation"
+        cls.original_video_height = 406
+        cls.original_video_width = 396
+        cls.expected_keypoint_names = [
+            "paw1LH_top",
+            "paw2LF_top",
+            "paw3RF_top",
+            "paw4RH_top",
+            "tailBase_top",
+            "tailMid_top",
+            "nose_top",
+            "obs_top",
+            "paw1LH_bot",
+            "paw2LF_bot",
+            "paw3RF_bot",
+            "paw4RH_bot",
+            "tailBase_bot",
+            "tailMid_bot",
+            "nose_bot",
+            "obsHigh_bot",
+            "obsLow_bot",
+        ]
+        cls.expected_metadata = DeepDict(
+            PoseEstimation=dict(
+                name=cls.pose_estimation_name,
+                description="Contains the pose estimation series for each keypoint.",
+                scorer="heatmap_tracker",
+                source_software="LightningPose",
+            )
+        )
+        cls.expected_metadata[cls.pose_estimation_name].update(
+            {
+                keypoint_name: dict(
+                    name=f"PoseEstimationSeries{keypoint_name}",
+                    description=f"The estimated position (x, y) of {keypoint_name} over time.",
+                )
+                for keypoint_name in cls.expected_keypoint_names
+            }
+        )
+
+        cls.test_data = pd.read_csv(cls.interface_kwargs["file_path"], header=[0, 1, 2])["heatmap_tracker"]
+
+    def check_extracted_metadata(self, metadata: dict):
+        self.assertEqual(
+            metadata["NWBFile"]["session_start_time"],
+            datetime(2023, 11, 9, 10, 14, 37, 0),
+        )
+        self.assertIn(self.pose_estimation_name, metadata["Behavior"])
+        self.assertEqual(
+            metadata["Behavior"][self.pose_estimation_name], self.expected_metadata[self.pose_estimation_name]
+        )
+
+    def check_read_nwb(self, nwbfile_path: str):
+        with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io:
+            nwbfile = io.read()
+            self.assertIn("behavior", nwbfile.processing)
+            self.assertIn(self.pose_estimation_name, nwbfile.processing["behavior"].data_interfaces)
+            pose_estimation_container = nwbfile.processing["behavior"].data_interfaces[self.pose_estimation_name]
+            self.assertIsInstance(pose_estimation_container, PoseEstimation)
+
+            pose_estimation_metadata = self.expected_metadata[self.pose_estimation_name]
+            self.assertEqual(pose_estimation_container.description, pose_estimation_metadata["description"])
+            self.assertEqual(pose_estimation_container.scorer, pose_estimation_metadata["scorer"])
+            self.assertEqual(pose_estimation_container.source_software, pose_estimation_metadata["source_software"])
+            assert_array_equal(
+                pose_estimation_container.dimensions[:], [[self.original_video_height, self.original_video_width]]
+            )
+
+            self.assertEqual(len(pose_estimation_container.pose_estimation_series), len(self.expected_keypoint_names))
+            for keypoint_name in self.expected_keypoint_names:
+                series_metadata = pose_estimation_metadata[keypoint_name]
+                self.assertIn(series_metadata["name"], pose_estimation_container.pose_estimation_series)
+                pose_estimation_series = pose_estimation_container.pose_estimation_series[series_metadata["name"]]
+                self.assertIsInstance(pose_estimation_series, PoseEstimationSeries)
+                self.assertEqual(pose_estimation_series.unit, "px")
+                self.assertEqual(pose_estimation_series.description, series_metadata["description"])
+                self.assertEqual(pose_estimation_series.reference_frame, self.conversion_options["reference_frame"])
+
+                test_data = self.test_data[keypoint_name]
+                assert_array_equal(pose_estimation_series.data[:], test_data[["x", "y"]].values)
+                assert_array_equal(pose_estimation_series.confidence[:], test_data["likelihood"].values)
+
+
+class TestLightningPoseDataInterfaceWithStubTest(DataInterfaceTestMixin, TemporalAlignmentMixin, unittest.TestCase):
+    data_interface_cls = LightningPoseDataInterface
+    interface_kwargs = dict(
+        file_path=str(BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.csv"),
+        original_video_file_path=str(
+            BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds/test_vid.mp4"
+        ),
+    )
+    conversion_options = dict(stub_test=True)
+    save_directory = OUTPUT_PATH
+
+    def check_read_nwb(self, nwbfile_path: str):
+        with NWBHDF5IO(path=nwbfile_path, mode="r", load_namespaces=True) as io:
+            nwbfile = io.read()
+            pose_estimation_container = nwbfile.processing["behavior"].data_interfaces["PoseEstimation"]
+            for pose_estimation_series in pose_estimation_container.pose_estimation_series.values():
+                self.assertEqual(pose_estimation_series.data.shape[0], 10)
+                self.assertEqual(pose_estimation_series.confidence.shape[0], 10)
+
+
 class TestFicTracDataInterface(DataInterfaceTestMixin, unittest.TestCase):
     data_interface_cls = FicTracDataInterface
     interface_kwargs = [
diff --git a/tests/test_on_data/test_format_converters/test_lightningpose_converter.py b/tests/test_on_data/test_format_converters/test_lightningpose_converter.py
new file mode 100644
index 00000000..4d0f8ab8
--- /dev/null
+++ b/tests/test_on_data/test_format_converters/test_lightningpose_converter.py
@@ -0,0 +1,209 @@
+import shutil
+import tempfile
+from datetime import datetime
+from pathlib import Path
+from warnings import warn
+
+from hdmf.testing import TestCase
+from ndx_pose import PoseEstimation
+from pynwb import NWBHDF5IO
+from pynwb.image import ImageSeries
+
+from neuroconv import ConverterPipe, NWBConverter
+from neuroconv.converters import LightningPoseConverter
+from neuroconv.tools import get_module
+from neuroconv.utils import DeepDict
+
+from ..setup_paths import BEHAVIOR_DATA_PATH
+
+
+class TestLightningPoseConverter(TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        lightning_pose_folder_path = BEHAVIOR_DATA_PATH / "lightningpose" / "outputs/2023-11-09/10-14-37/video_preds"
+        cls.file_path = str(lightning_pose_folder_path / "test_vid.csv")
+        cls.original_video_file_path = str(lightning_pose_folder_path / "test_vid.mp4")
+        cls.labeled_video_file_path = str(lightning_pose_folder_path / "labeled_videos/test_vid_labeled.mp4")
+
+        cls.pose_estimation_name = "PoseEstimation"
+        cls.original_video_name = "original_video_name"
+        cls.labeled_video_name = "labeled_video_name"
+
+        cls.original_video_height = 406
+        cls.original_video_width = 396
+
+        cls.converter = LightningPoseConverter(
+            file_path=cls.file_path,
+            original_video_file_path=cls.original_video_file_path,
+            labeled_video_file_path=cls.labeled_video_file_path,
+            image_series_original_video_name=cls.original_video_name,
+            image_series_labeled_video_name=cls.labeled_video_name,
+        )
+
+        cls.expected_keypoint_names = [
+            "paw1LH_top",
+            "paw2LF_top",
+            "paw3RF_top",
+            "paw4RH_top",
+            "tailBase_top",
+            "tailMid_top",
+            "nose_top",
+            "obs_top",
+            "paw1LH_bot",
+            "paw2LF_bot",
+            "paw3RF_bot",
+            "paw4RH_bot",
+            "tailBase_bot",
+            "tailMid_bot",
+            "nose_bot",
+            "obsHigh_bot",
+            "obsLow_bot",
+        ]
+
+        cls.pose_estimation_metadata = DeepDict(
+            name=cls.pose_estimation_name,
+            description="Contains the pose estimation series for each keypoint.",
+            scorer="heatmap_tracker",
+            source_software="LightningPose",
+        )
+
+        cls.pose_estimation_metadata.update(
+            {
+                keypoint_name: dict(
+                    name=f"PoseEstimationSeries{keypoint_name}",
+                    description=f"The estimated position (x, y) of {keypoint_name} over time.",
+                )
+                for keypoint_name in cls.expected_keypoint_names
+            }
+        )
+
+        cls.converter_metadata = dict(
+            Behavior=dict(
+                PoseEstimation=cls.pose_estimation_metadata,
+                Videos=[
+                    dict(
+                        description="The original video used for pose estimation.",
+                        name="original_video_name",
+                        unit="Frames",
+                    ),
+                    dict(
+                        description="The video recorded by camera with the pose estimation labels.",
+                        name="labeled_video_name",
+                        unit="Frames",
+                    ),
+                ],
+            )
+        )
+
+        cls.test_dir = Path(tempfile.mkdtemp())
+
+        cls.conversion_options = dict(stub_test=True)
+
+        cls.pose_estimation_name = "PoseEstimation"
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        try:
+            shutil.rmtree(cls.test_dir)
+        except PermissionError:
+            warn(f"Unable to cleanup testing data at {cls.test_dir}! Please remove it manually.")
+
+    def test_expected_metadata(self):
+        metadata = self.converter.get_metadata()
+        videos_metadata = metadata["Behavior"]["Videos"]
+        self.assertEqual(len(videos_metadata), 2)
+        self.assertDictEqual(metadata["Behavior"], self.converter_metadata["Behavior"])
+
+    def test_run_conversion(self):
+        nwbfile_path = str(self.test_dir / "test_lightningpose_converter.nwb")
+        self.converter.run_conversion(nwbfile_path=nwbfile_path)
+
+        self.assertNWBFileStructure(nwbfile_path=nwbfile_path)
+
+    def test_run_conversion_add_conversion_options(self):
+        nwbfile_path = str(self.test_dir / "test_lightningpose_converter_conversion_options.nwb")
+
+        conversion_options = dict(
+            **self.conversion_options, reference_frame="(0,0) corresponds to the top left corner of the video."
+        )
+        self.converter.run_conversion(
+            nwbfile_path=nwbfile_path,
+            **conversion_options,
+        )
+
+        self.assertNWBFileStructure(nwbfile_path=nwbfile_path, **self.conversion_options)
+
+    def assertNWBFileStructure(self, nwbfile_path: str, stub_test: bool = False):
+        with NWBHDF5IO(path=nwbfile_path) as io:
+            nwbfile = io.read()
+
+            self.assertEqual(nwbfile.session_start_time, datetime(2023, 11, 9, 10, 14, 37).astimezone())
+
+            # Check original video added to acquisition
+            self.assertIn(self.original_video_name, nwbfile.acquisition)
+            image_series = nwbfile.acquisition[self.original_video_name]
+            self.assertIsInstance(image_series, ImageSeries)
+            self.assertEqual(image_series.external_file[:], self.original_video_file_path)
+            self.assertEqual(image_series.description, "The original video used for pose estimation.")
+
+            # Check labeled video added to behavior processing module
+            behavior = get_module(nwbfile=nwbfile, name="behavior")
+            self.assertIn(self.labeled_video_name, behavior.data_interfaces)
+            image_series_labeled_video = behavior.data_interfaces[self.labeled_video_name]
+            self.assertIsInstance(image_series_labeled_video, ImageSeries)
+            self.assertEqual(
+                image_series_labeled_video.external_file[:],
+                self.labeled_video_file_path,
+            )
+            self.assertEqual(
+                image_series_labeled_video.description,
+                "The video recorded by camera with the pose estimation labels.",
+            )
+
+            # Check pose estimation added to behavior processing module
+            self.assertIn(self.pose_estimation_name, behavior.data_interfaces)
+            self.assertIsInstance(behavior[self.pose_estimation_name], PoseEstimation)
+            pose_estimation_container = nwbfile.processing["behavior"]["PoseEstimation"]
+
+            # The current link between the pose estimation container "original_videos" and "labeled_videos" and the
+            # ImageSeries is the name of the ImageSeries. TODO: update this when ndx-pose 0.2.0 is released.
+            self.assertEqual(pose_estimation_container.original_videos[:], [image_series.name])
+            self.assertEqual(pose_estimation_container.labeled_videos[:], [image_series_labeled_video.name])
+
+            num_frames = 994 if not stub_test else 10
+            for pose_estimation_series in pose_estimation_container.pose_estimation_series.values():
+                self.assertEqual(pose_estimation_series.data.shape[0], num_frames)
+                self.assertEqual(pose_estimation_series.confidence.shape[0], num_frames)
+
+    def test_converter_in_converter(self):
+        class TestConverter(NWBConverter):
+            data_interface_classes = dict(TestPoseConverter=LightningPoseConverter)
+
+        converter = TestConverter(
+            source_data=dict(
+                TestPoseConverter=dict(
+                    file_path=self.file_path,
+                    original_video_file_path=self.original_video_file_path,
+                    labeled_video_file_path=self.labeled_video_file_path,
+                    image_series_original_video_name=self.original_video_name,
+                    image_series_labeled_video_name=self.labeled_video_name,
+                )
+            )
+        )
+
+        conversion_options = dict(TestPoseConverter=self.conversion_options)
+
+        nwbfile_path = str(self.test_dir / "test_lightningpose_converter_in_nwbconverter.nwb")
+        converter.run_conversion(nwbfile_path=nwbfile_path, conversion_options=conversion_options)
+
+        self.assertNWBFileStructure(nwbfile_path, **self.conversion_options)
+
+    def test_converter_in_converter_pipe(self):
+        converter_pipe = ConverterPipe(data_interfaces=[self.converter])
+
+        nwbfile_path = self.test_dir / "test_lightningpose_converter_in_converter_pipe.nwb"
+
+        conversion_options = dict(LightningPoseConverter=self.conversion_options)
+        converter_pipe.run_conversion(nwbfile_path=nwbfile_path, conversion_options=conversion_options)
+
+        self.assertNWBFileStructure(nwbfile_path=nwbfile_path, **self.conversion_options)